diff --git a/packages/Core/Core/Http/RetryExecutor.cs b/packages/Core/Core/Http/RetryExecutor.cs index 5910442b..93598185 100644 --- a/packages/Core/Core/Http/RetryExecutor.cs +++ b/packages/Core/Core/Http/RetryExecutor.cs @@ -12,13 +12,18 @@ public static class RetryExecutor private static readonly ThreadLocal RandomProvider = new(() => new Random(Guid.NewGuid().GetHashCode())); /// Sends via , called fresh per attempt since a sent can't be reused. + public static Task SendAsync(HttpClient client, Func requestFactory, + RetryOptions options, CancellationToken cancellationToken = default) => + SendAsync(client, requestFactory, options, HttpCompletionOption.ResponseContentRead, cancellationToken); + + /// Sends a request with the specified . public static async Task SendAsync(HttpClient client, Func requestFactory, - RetryOptions options, CancellationToken cancellationToken = default) + RetryOptions options, HttpCompletionOption completionOption, CancellationToken cancellationToken) { var attempt = 0; while (true) { - var (response, exception) = await Attempt(client, requestFactory, cancellationToken).ConfigureAwait(false); + var (response, exception) = await Attempt(client, requestFactory, completionOption, cancellationToken).ConfigureAwait(false); if (!ShouldRetry(response, exception, attempt, options)) return response ?? throw exception!; @@ -29,12 +34,12 @@ public static async Task SendAsync(HttpClient client, Func< } private static async Task<(HttpResponseMessage? Response, HttpRequestException? Exception)> Attempt( - HttpClient client, Func requestFactory, CancellationToken cancellationToken) + HttpClient client, Func requestFactory, HttpCompletionOption completionOption, CancellationToken cancellationToken) { using var request = requestFactory(); try { - return (await client.SendAsync(request, cancellationToken).ConfigureAwait(false), null); + return (await client.SendAsync(request, completionOption, cancellationToken).ConfigureAwait(false), null); } catch (HttpRequestException e) { diff --git a/packages/Core/Core/PublicAPI.Unshipped.txt b/packages/Core/Core/PublicAPI.Unshipped.txt index 7dc5c581..cd381361 100644 --- a/packages/Core/Core/PublicAPI.Unshipped.txt +++ b/packages/Core/Core/PublicAPI.Unshipped.txt @@ -1 +1,2 @@ #nullable enable +static Supabase.Core.Http.RetryExecutor.SendAsync(System.Net.Http.HttpClient! client, System.Func! requestFactory, Supabase.Core.Http.RetryOptions! options, System.Net.Http.HttpCompletionOption completionOption, System.Threading.CancellationToken cancellationToken) -> System.Threading.Tasks.Task! diff --git a/packages/Functions/Functions.Tests/ClientContractTests.cs b/packages/Functions/Functions.Tests/ClientContractTests.cs index 42ff2f57..6ae0fc36 100644 --- a/packages/Functions/Functions.Tests/ClientContractTests.cs +++ b/packages/Functions/Functions.Tests/ClientContractTests.cs @@ -10,6 +10,7 @@ using Supabase.Functions; using Supabase.Functions.Exceptions; using WireMock; +using WireMock.Models; using WireMock.RequestBuilders; using WireMock.ResponseBuilders; using WireMock.Server; @@ -28,6 +29,7 @@ namespace Functions.Tests; public class ClientContractTests { private const string FunctionName = "hello"; + private static readonly TimeSpan Deadline = TimeSpan.FromSeconds(5); private WireMockServer server = null!; private Client client = null!; @@ -251,10 +253,62 @@ public async Task Invoke_ShouldThrowOperationCanceledException_GivenCallerCancel "caller-triggered cancellation must not be mis-reported as an HttpTimeout"); } + [TestMethod] + public async Task InvokeStream_ShouldReturnBeforeBodyCompletes() + { + var holdBodyOpen = new TaskCompletionSource(); + this.RespondWithEvents(async queue => + { + queue.Write("data: first\n\n"); + await holdBodyOpen.Task; + }); + + using var response = await this.client.InvokeStream(FunctionName).WaitAsync(Deadline); + holdBodyOpen.SetResult(); + + (await response.Content.ReadAsStringAsync()).Should().Be("data: first\n\n"); + } + + [TestMethod] + public async Task InvokeStream_ShouldBufferTheErrorBody_GivenServerError() + { + this.RespondWith(500, "internal boom"); + var act = () => this.client.InvokeStream(FunctionName); + var exception = (await act.Should().ThrowAsync()).Which; + (await exception.Response!.Content.ReadAsStringAsync()).Should().Be("internal boom"); + } + + [TestMethod] + public async Task InvokeStream_ShouldThrowTimeoutException_GivenAStalledErrorBody() + { + var holdBodyOpen = new TaskCompletionSource(); + // WireMock forces 200 on an SSE body, so a relay error is what makes this one fail. + this.RespondWithEvents(async queue => + { + queue.Write("partial"); + await holdBodyOpen.Task; + }, Response.Create().WithHeader("x-relay-error", "true")); + + var act = () => this.client.InvokeStream(FunctionName, options: new InvokeFunctionOptions { HttpTimeout = TimeSpan.FromMilliseconds(100) }); + var exception = (await act.Should().ThrowAsync().WaitAsync(Deadline)).Which; + holdBodyOpen.SetResult(); + + exception.InnerException.Should().BeOfType(); + } + private void RespondWith(int statusCode, string body) => this.server.Given(Request.Create().WithPath($"/functions/v1/{FunctionName}").UsingAnyMethod()) .RespondWith(Response.Create().WithStatusCode(statusCode).WithHeader("Content-Type", "application/json").WithBody(body)); + private void RespondWithEvents(Func, Task> body, IResponseBuilder? response = null) => + this.server.Given(Request.Create().WithPath($"/functions/v1/{FunctionName}").UsingAnyMethod()) + .RespondWith((response ?? Response.Create()).WithHeader("Content-Type", "text/event-stream") + .WithSseBody(async (_, queue) => + { + await body(queue); + queue.Close(); + })); + private IRequestMessage SingleRequest() => this.server.LogEntries.Should().ContainSingle().Which.RequestMessage!; private string HeaderOf(string name) => this.SingleRequest().Headers![name][0]; diff --git a/packages/Functions/Functions/Client.cs b/packages/Functions/Functions/Client.cs index a1bffc08..70f2c865 100644 --- a/packages/Functions/Functions/Client.cs +++ b/packages/Functions/Functions/Client.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Diagnostics; +using System.IO; using System.Net.Http; using System.Runtime.CompilerServices; using System.Text; @@ -82,6 +83,22 @@ public async Task RawInvoke( return (await this.HandleRequest(functionName, url, token, options, cancellationToken)).Content; } + /// + /// Invokes a function and returns a successful response after its headers arrive. + /// covers the request and any error body. + /// + /// Function name, appended to the base URL. + /// Bearer token. + /// Invocation options. + /// Cancels the request and error-body reads. Cancel successful body reads separately. + /// The response, which the caller must dispose. + public Task InvokeStream( + string functionName, + string? token = null, + InvokeFunctionOptions? options = null, + CancellationToken cancellationToken = default + ) => this.HandleRequest(functionName, $"{this.baseUrl}/{functionName}", token, options, cancellationToken, HttpCompletionOption.ResponseHeadersRead); + /// /// Invokes a function and returns the Text content of the response. /// @@ -136,6 +153,7 @@ public async Task Invoke( /// /// /// + /// /// /// private async Task HandleRequest( @@ -143,7 +161,8 @@ private async Task HandleRequest( string url, string? token = null, InvokeFunctionOptions? options = null, - CancellationToken cancellationToken = default + CancellationToken cancellationToken = default, + HttpCompletionOption completionOption = HttpCompletionOption.ResponseContentRead ) { options ??= new InvokeFunctionOptions(); @@ -178,7 +197,7 @@ private async Task HandleRequest( try { - var response = await RetryExecutor.SendAsync(this.httpClient, () => BuildRequestMessage(options, uri), this.Options.Retry, linkedCts.Token); + var response = await RetryExecutor.SendAsync(this.httpClient, () => BuildRequestMessage(options, uri), this.Options.Retry, completionOption, linkedCts.Token); statusCode = (int) response.StatusCode; var isRelayError = response.Headers.Contains("x-relay-error"); activity.SetHttpResponseTags(statusCode.Value); @@ -199,7 +218,9 @@ private async Task HandleRequest( errorType = statusCode.Value.ToString(); } - var content = await response.Content.ReadAsStringAsync(); + var content = completionOption == HttpCompletionOption.ResponseContentRead + ? await response.Content.ReadAsStringAsync() + : await BufferErrorBody(response, linkedCts.Token); var exception = new FunctionsException(content) { Content = content, @@ -229,6 +250,31 @@ private async Task HandleRequest( } } + // Keep the error body readable through FunctionsException.Response. + private static async Task BufferErrorBody(HttpResponseMessage response, CancellationToken cancellationToken) + { + try + { + using var buffer = new MemoryStream(); + var stream = await response.Content.ReadAsStreamAsync(); + await stream.CopyToAsync(buffer, cancellationToken); + + var buffered = new ByteArrayContent(buffer.ToArray()); + foreach (var header in response.Content.Headers) + buffered.Headers.TryAddWithoutValidation(header.Key, header.Value); + + response.Content.Dispose(); + response.Content = buffered; + + return await buffered.ReadAsStringAsync(); + } + catch + { + response.Dispose(); + throw; + } + } + private static Uri BuildUri(string url) { var builder = new UriBuilder(url); diff --git a/packages/Functions/Functions/Interfaces/IFunctionsClient.cs b/packages/Functions/Functions/Interfaces/IFunctionsClient.cs index 5e86ec6a..8d2d9b70 100644 --- a/packages/Functions/Functions/Interfaces/IFunctionsClient.cs +++ b/packages/Functions/Functions/Interfaces/IFunctionsClient.cs @@ -40,4 +40,14 @@ public interface IFunctionsClient : IGettableHeaders /// /// Task RawInvoke(string url, string? token = null, Client.InvokeFunctionOptions? options = null, CancellationToken cancellationToken = default); + + /// + /// Invokes a function and returns a successful response after its headers arrive. + /// + /// Function name, appended to the base URL. + /// Bearer token. + /// Invocation options. + /// Cancels the request and error-body reads. Cancel successful body reads separately. + /// The response, which the caller must dispose. + Task InvokeStream(string url, string? token = null, Client.InvokeFunctionOptions? options = null, CancellationToken cancellationToken = default); } diff --git a/packages/Functions/Functions/PublicAPI.Unshipped.txt b/packages/Functions/Functions/PublicAPI.Unshipped.txt index 7dc5c581..0150b6c3 100644 --- a/packages/Functions/Functions/PublicAPI.Unshipped.txt +++ b/packages/Functions/Functions/PublicAPI.Unshipped.txt @@ -1 +1,3 @@ #nullable enable +Supabase.Functions.Client.InvokeStream(string! functionName, string? token = null, Supabase.Functions.Client.InvokeFunctionOptions? options = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! +Supabase.Functions.Interfaces.IFunctionsClient.InvokeStream(string! url, string? token = null, Supabase.Functions.Client.InvokeFunctionOptions? options = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task! diff --git a/packages/Functions/README.md b/packages/Functions/README.md index c6e1c538..28ee35d6 100644 --- a/packages/Functions/README.md +++ b/packages/Functions/README.md @@ -58,6 +58,19 @@ var options = new Client.InvokeFunctionOptions var result = await client.Invoke("hello-world", token: SUPABASE_ANON_KEY, options); ``` +### Streaming a response + +`InvokeStream` returns successful responses after the headers arrive. `HttpTimeout` and request cancellation +cover the request and error-body reads. Cancel successful body reads separately and dispose the response when finished. + +```csharp +using var response = await client.InvokeStream("chat", token: SUPABASE_ANON_KEY); +using var reader = new StreamReader(await response.Content.ReadAsStreamAsync()); +string? line; +while ((line = await reader.ReadLineAsync()) != null) + Console.WriteLine(line); +``` + ## Observability (OpenTelemetry) The client emits traces and metrics through `System.Diagnostics`, so you can wire them into diff --git a/sdk-compliance.yaml b/sdk-compliance.yaml index 1436b73a..72c01d1e 100644 --- a/sdk-compliance.yaml +++ b/sdk-compliance.yaml @@ -393,10 +393,10 @@ features: symbols: - InvokeFunctionOptions.HttpMethod functions.invocation.streaming_response: - status: partially_implemented - note: "RawInvoke returns raw HttpContent, but the request uses the default buffered completion (no incremental text/event-stream streaming)." + status: implemented + note: "InvokeStream returns response bodies for incremental reading." symbols: - - Client.RawInvoke + - Client.InvokeStream functions.invocation.region_selection: status: implemented note: "Full AWS region list; per-call override or client default; emits x-region." @@ -628,4 +628,4 @@ features: symbols: - FileOptions.Metadata - FileOptions.CacheControl - - FileOptions.ContentType \ No newline at end of file + - FileOptions.ContentType