2222namespace Amazon . Lambda . RuntimeSupport
2323{
2424 /// <summary>
25- /// Internal implementation of IResponseStream with true streaming.
26- /// Writes data directly to the HTTP output stream as chunked transfer encoding .
25+ /// A write-only, non-seekable <see cref="Stream"/> subclass that streams response data
26+ /// to the Lambda Runtime API. Returned by <see cref="LambdaResponseStreamFactory.CreateStream"/> .
2727 /// </summary>
28- internal class ResponseStream : IResponseStream
28+ public partial class LambdaResponseStream : Stream , ILambdaResponseStream
2929 {
3030 private static readonly byte [ ] CrlfBytes = Encoding . ASCII . GetBytes ( "\r \n " ) ;
3131
32- private readonly long _maxResponseSize ;
3332 private long _bytesWritten ;
3433 private bool _isCompleted ;
3534 private bool _hasError ;
@@ -41,14 +40,26 @@ internal class ResponseStream : IResponseStream
4140 private readonly SemaphoreSlim _httpStreamReady = new SemaphoreSlim ( 0 , 1 ) ;
4241 private readonly SemaphoreSlim _completionSignal = new SemaphoreSlim ( 0 , 1 ) ;
4342
43+ /// <summary>
44+ /// The number of bytes written to the Lambda response stream so far.
45+ /// </summary>
4446 public long BytesWritten => _bytesWritten ;
47+
48+ /// <summary>
49+ /// Gets a value indicating whether the operation has completed.
50+ /// </summary>
4551 public bool IsCompleted => _isCompleted ;
52+
53+ /// <summary>
54+ /// Gets a value indicating whether an error has occurred.
55+ /// </summary>
4656 public bool HasError => _hasError ;
57+
58+
4759 internal Exception ReportedError => _reportedError ;
4860
49- public ResponseStream ( long maxResponseSize )
61+ internal LambdaResponseStream ( )
5062 {
51- _maxResponseSize = maxResponseSize ;
5263 }
5364
5465 /// <summary>
@@ -69,6 +80,13 @@ internal async Task WaitForCompletionAsync()
6980 await _completionSignal . WaitAsync ( ) ;
7081 }
7182
83+ /// <summary>
84+ /// Asynchronously writes a byte array to the response stream.
85+ /// </summary>
86+ /// <param name="buffer">The byte array to write.</param>
87+ /// <param name="cancellationToken">Optional cancellation token.</param>
88+ /// <returns>A task representing the asynchronous operation.</returns>
89+ /// <exception cref="InvalidOperationException">Thrown if the stream is already completed or an error has been reported.</exception>
7290 public async Task WriteAsync ( byte [ ] buffer , CancellationToken cancellationToken = default )
7391 {
7492 if ( buffer == null )
@@ -77,7 +95,16 @@ public async Task WriteAsync(byte[] buffer, CancellationToken cancellationToken
7795 await WriteAsync ( buffer , 0 , buffer . Length , cancellationToken ) ;
7896 }
7997
80- public async Task WriteAsync ( byte [ ] buffer , int offset , int count , CancellationToken cancellationToken = default )
98+ /// <summary>
99+ /// Asynchronously writes a portion of a byte array to the response stream.
100+ /// </summary>
101+ /// <param name="buffer">The byte array containing data to write.</param>
102+ /// <param name="offset">The zero-based byte offset in buffer at which to begin copying bytes.</param>
103+ /// <param name="count">The number of bytes to write.</param>
104+ /// <param name="cancellationToken">Optional cancellation token.</param>
105+ /// <returns>A task representing the asynchronous operation.</returns>
106+ /// <exception cref="InvalidOperationException">Thrown if the stream is already completed or an error has been reported.</exception>
107+ public override async Task WriteAsync ( byte [ ] buffer , int offset , int count , CancellationToken cancellationToken = default )
81108 {
82109 if ( buffer == null )
83110 throw new ArgumentNullException ( nameof ( buffer ) ) ;
@@ -93,14 +120,6 @@ public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationT
93120 lock ( _lock )
94121 {
95122 ThrowIfCompletedOrError ( ) ;
96-
97- if ( _bytesWritten + count > _maxResponseSize )
98- {
99- throw new InvalidOperationException (
100- $ "Writing { count } bytes would exceed the maximum response size of { _maxResponseSize } bytes (20 MiB). "+
101- $ "Current size: { _bytesWritten } bytes.") ;
102- }
103-
104123 _bytesWritten += count ;
105124 }
106125
@@ -120,13 +139,14 @@ public async Task WriteAsync(byte[] buffer, int offset, int count, CancellationT
120139 }
121140 }
122141
123- public async Task WriteAsync ( ReadOnlyMemory < byte > buffer , CancellationToken cancellationToken = default )
124- {
125- // Convert to array and delegate — small overhead but keeps the API simple
126- var array = buffer . ToArray ( ) ;
127- await WriteAsync ( array , 0 , array . Length , cancellationToken ) ;
128- }
129-
142+ /// <summary>
143+ /// Reports an error that occurred during streaming.
144+ /// This will send error information via HTTP trailing headers.
145+ /// </summary>
146+ /// <param name="exception">The exception to report.</param>
147+ /// <param name="cancellationToken">Optional cancellation token.</param>
148+ /// <returns>A task representing the asynchronous operation.</returns>
149+ /// <exception cref="InvalidOperationException">Thrown if the stream is already completed or an error has already been reported.</exception>
130150 public Task ReportErrorAsync ( Exception exception , CancellationToken cancellationToken = default )
131151 {
132152 if ( exception == null )
@@ -145,6 +165,7 @@ public Task ReportErrorAsync(Exception exception, CancellationToken cancellation
145165
146166 // Signal completion so StreamingHttpContent can write error trailers and finish
147167 _completionSignal . Release ( ) ;
168+
148169 return Task . CompletedTask ;
149170 }
150171
@@ -166,10 +187,17 @@ private void ThrowIfCompletedOrError()
166187 throw new InvalidOperationException ( "Cannot write to a stream after an error has been reported." ) ;
167188 }
168189
169- public void Dispose ( )
190+ // ── Dispose ──────────────────────────────────────────────────────────
191+
192+ /// <inheritdoc/>
193+ protected override void Dispose ( bool disposing )
170194 {
171- // Ensure completion is signaled if not already
172- try { _completionSignal . Release ( ) ; } catch ( SemaphoreFullException ) { }
195+ if ( disposing )
196+ {
197+ try { _completionSignal . Release ( ) ; } catch ( SemaphoreFullException ) { }
198+ }
199+
200+ base . Dispose ( disposing ) ;
173201 }
174202 }
175203}
0 commit comments