Add cancellation token support to CompleteTaskOnCloseStream

This update enhances CompleteTaskOnCloseStream by incorporating an optional CancellationTokenSource. The token ensures proper resource management and cleanup by disposing of the source upon stream disposal. This improves the handling of async operations with better cancellation support.

Author: Mpdreamz

src/Elastic.Transport/Components/TransportClient/Content/RequestDataContent.cs

@@ -123,7 +123,7 @@ async Task SerializeToStreamAsync(Stream stream, TransportContext context, Cance
 	{
 		var source = CancellationTokenSource.CreateLinkedTokenSource(_token, cancellationToken);
 		var serializeToStreamTask = new TaskCompletionSource<bool>();
-		var wrappedStream = new CompleteTaskOnCloseStream(stream, serializeToStreamTask);
+		var wrappedStream = new CompleteTaskOnCloseStream(stream, serializeToStreamTask, source);
 		await _onStreamAvailableAsync(_boundConfiguration, _postData, wrappedStream, this, context, source.Token).ConfigureAwait(false);
 		await serializeToStreamTask.Task.ConfigureAwait(false);
 	}
@@ -153,18 +153,21 @@ protected override bool TryComputeLength(out long length)
 	internal class CompleteTaskOnCloseStream : DelegatingStream
 	{
 		private readonly TaskCompletionSource<bool> _serializeToStreamTask;
+		private readonly CancellationTokenSource? _source;
 
-		public CompleteTaskOnCloseStream(Stream innerStream, TaskCompletionSource<bool> serializeToStreamTask)
+		public CompleteTaskOnCloseStream(Stream innerStream, TaskCompletionSource<bool> serializeToStreamTask, CancellationTokenSource? source = null)
 			: base(innerStream)
 		{
 			Contract.Assert(serializeToStreamTask != null);
 			_serializeToStreamTask = serializeToStreamTask;
+			_source = source;
 		}
 
 		protected override void Dispose(bool disposing)
 		{
 			_serializeToStreamTask.TrySetResult(true);
 			base.Dispose();
+			_source?.Dispose();
 		}
 
 		public override void Close() => _serializeToStreamTask.TrySetResult(true);