Merge branch 'normj/response-streaming' into normj/aspnetcore-responsestreaming

Author: normj

.autover/changes/c27a62e6-91ca-4a59-9406-394866cdfa62.json

@@ -4,8 +4,15 @@
       "Name": "Amazon.Lambda.RuntimeSupport",
       "Type": "Minor",
       "ChangelogMessages": [
-        "Add response streaming support"
+        "(Preview) Add response streaming support"
       ]
-    }
+    },
+    {
+      "Name": "Amazon.Lambda.Core",
+      "Type": "Minor",
+      "ChangelogMessages": [
+        "(Preview) Add response streaming support"
+      ]
+    }	
   ]
 }

Libraries/src/Amazon.Lambda.Core/ResponseStreaming/HttpResponseStreamPrelude.cs

@@ -11,7 +11,7 @@ namespace Amazon.Lambda.Core.ResponseStreaming
     /// <summary>
     /// The HTTP response prelude to be sent as the first chunk of a streaming response when using <see cref="LambdaResponseStreamFactory.CreateHttpStream"/>.
     /// </summary>
-    [RequiresPreviewFeatures(LambdaResponseStreamFactory.ParameterizedPreviewMessage)]
+    [RequiresPreviewFeatures(LambdaResponseStreamFactory.PreviewMessage)]
     public class HttpResponseStreamPrelude
     {
         /// <summary>

Libraries/src/Amazon.Lambda.Core/ResponseStreaming/LambdaResponseStream.cs

@@ -15,7 +15,7 @@ namespace Amazon.Lambda.Core.ResponseStreaming
     /// to the Lambda Runtime API. Returned by <see cref="LambdaResponseStreamFactory.CreateStream"/>.
     /// Integrates with standard .NET stream consumers such as <see cref="System.IO.StreamWriter"/>.
     /// </summary>
-    [RequiresPreviewFeatures(LambdaResponseStreamFactory.ParameterizedPreviewMessage)]
+    [RequiresPreviewFeatures(LambdaResponseStreamFactory.PreviewMessage)]
     public class LambdaResponseStream : Stream
     {
         private readonly ILambdaResponseStream _responseStream;
@@ -82,24 +82,24 @@ public override async Task WriteAsync(byte[] buffer, int offset, int count, Canc
         /// <exception cref="NotSupportedException">Always thrown.</exception>
         public override long Position
         {
-            get => throw new NotSupportedException("LambdaResponseStream does not support seeking.");
-            set => throw new NotSupportedException("LambdaResponseStream does not support seeking.");
+            get => throw new NotSupportedException($"{nameof(LambdaResponseStream)} does not support seeking.");
+            set => throw new NotSupportedException($"{nameof(LambdaResponseStream)} does not support seeking.");
         }
 
         /// <summary>Not supported.</summary>
         /// <exception cref="NotImplementedException">Always thrown.</exception>
         public override long Seek(long offset, SeekOrigin origin)
-            => throw new NotImplementedException("LambdaResponseStream does not support seeking.");
+            => throw new NotImplementedException($"{nameof(LambdaResponseStream)} does not support seeking.");
 
         /// <summary>Not supported.</summary>
         /// <exception cref="NotImplementedException">Always thrown.</exception>
         public override int Read(byte[] buffer, int offset, int count)
-            => throw new NotImplementedException("LambdaResponseStream does not support reading.");
+            => throw new NotImplementedException($"{nameof(LambdaResponseStream)} does not support reading.");
 
         /// <summary>Not supported.</summary>
         /// <exception cref="NotImplementedException">Always thrown.</exception>
         public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
-            => throw new NotImplementedException("LambdaResponseStream does not support reading.");
+            => throw new NotImplementedException($"{nameof(LambdaResponseStream)} does not support reading.");
 
         /// <summary>
         /// Writes a sequence of bytes to the stream. Delegates to the async path synchronously.
@@ -116,7 +116,7 @@ public override void Flush() { }
         /// <summary>Not supported.</summary>
         /// <exception cref="NotSupportedException">Always thrown.</exception>
         public override void SetLength(long value)
-            => throw new NotSupportedException("LambdaResponseStream does not support SetLength.");
+            => throw new NotSupportedException($"{nameof(LambdaResponseStream)} does not support SetLength.");
         #endregion
     }
 }

Libraries/src/Amazon.Lambda.Core/ResponseStreaming/LambdaResponseStreamFactory.cs

@@ -10,15 +10,19 @@ namespace Amazon.Lambda.Core.ResponseStreaming
     /// <summary>
     /// Factory to create Lambda response streams for writing streaming responses in AWS Lambda functions. The created streams are write-only and non-seekable.
     /// </summary>
-    [RequiresPreviewFeatures(LambdaResponseStreamFactory.ParameterizedPreviewMessage)]
+    [RequiresPreviewFeatures(LambdaResponseStreamFactory.PreviewMessage)]
     public class LambdaResponseStreamFactory
     {
-        internal const string ParameterizedPreviewMessage =
+        internal const string PreviewMessage =
             "Response streaming is in preview till a new version of .NET Lambda runtime client that supports response streaming " +
             "has been deployed to the .NET Lambda managed runtime. Till deployment has been made the feature can be used by deploying as an " +
             "executable including the latest version of Amazon.Lambda.RuntimeSupport and setting the \"EnablePreviewFeatures\" in the Lambda " +
             "project file to \"true\"";
 
+        internal const string UninitializedFactoryMessage =
+            "LambdaResponseStreamFactory is not initialized. This is caused by mismatch versions of Amazon.Lambda.Core and Amazon.Lambda.RuntimeSupport. " +
+            "Update both packages to the current version to address the issue.";
+
         private static Func<byte[], ILambdaResponseStream> _streamFactory;
 
         internal static void SetLambdaResponseStream(Func<byte[], ILambdaResponseStream> streamFactory)
@@ -27,19 +31,22 @@ internal static void SetLambdaResponseStream(Func<byte[], ILambdaResponseStream>
         }
 
         /// <summary>
-        /// Creates a <see cref="Stream"/> that can be used to write streaming responses back to callers of the Lambda function. Once
+        /// Creates a <see cref="LambdaResponseStream"/> a subclass of <see cref="Stream"/> that can be used to write streaming responses back to callers of the Lambda function. Once
         /// a Lambda function creates a response stream all output must be returned by writing to the stream; the Lambda function's handler
         /// return value will be ignored. The stream is write-only and non-seekable.
         /// </summary>
         /// <returns></returns>
-        public static Stream CreateStream()
+        public static LambdaResponseStream CreateStream()
         {
+            if (_streamFactory == null)
+                throw new InvalidOperationException(UninitializedFactoryMessage);
+
             var runtimeResponseStream = _streamFactory(Array.Empty<byte>());
             return new LambdaResponseStream(runtimeResponseStream);
         }
 
         /// <summary>
-        /// Create a <see cref="Stream"/> for writing streaming responses, with an HTTP response prelude containing status code and headers. This should be used for
+        /// Creates a <see cref="LambdaResponseStream"/> a subclass of <see cref="Stream"/> for writing streaming responses, with an HTTP response prelude containing status code and headers. This should be used for
         /// Lambda functions using response streaming that are invoked via the Lambda Function URLs or API Gateway HTTP APIs, where the response format is expected to be an HTTP response.
         /// The prelude will be serialized and sent as the first chunk of the response stream, and should contain any necessary HTTP status code and headers.
         /// <para>
@@ -49,8 +56,14 @@ public static Stream CreateStream()
         /// </summary>
         /// <param name="prelude">The HTTP response prelude including status code and headers.</param>
         /// <returns></returns>
-        public static Stream CreateHttpStream(HttpResponseStreamPrelude prelude)
+        public static LambdaResponseStream CreateHttpStream(HttpResponseStreamPrelude prelude)
         {
+            if (_streamFactory == null)
+                throw new InvalidOperationException(UninitializedFactoryMessage);
+
+            if (prelude is null)
+                throw new ArgumentNullException(nameof(prelude));
+
             var runtimeResponseStream = _streamFactory(prelude.ToByteArray());
             return new LambdaResponseStream(runtimeResponseStream);
         }

Libraries/src/Amazon.Lambda.RuntimeSupport/Bootstrap/LambdaBootstrap.cs

@@ -425,6 +425,7 @@ internal async Task InvokeOnceAsync(CancellationToken cancellationToken = defaul
                         {
                             // Wait for the streaming response to finish sending before allowing the next invocation to be processed. This ensures that responses are sent in the order the invocations were received.
                             await sendTask;
+                            sendTask.Result.Dispose();
                         }
 
                         streamIfCreated.Dispose();
@@ -466,10 +467,7 @@ internal async Task InvokeOnceAsync(CancellationToken cancellationToken = defaul
                 }
                 finally
                 {
-                    if (runtimeApiClient != null)
-                    {
-                        ResponseStreamFactory.CleanupInvocation(isMultiConcurrency);
-                    }
+                    ResponseStreamFactory.CleanupInvocation(isMultiConcurrency);
                     invocation.Dispose();
                 }
             };