Add client-side async API for replication (IAsyncReplicator) (#1182)

Co-authored-by: Paul Irwin <paulirwin@gmail.com>

Author: NehanPathan

Directory.Build.targets

@@ -52,6 +52,9 @@
   <PropertyGroup Condition=" $(TargetFramework.StartsWith('net8.')) Or $(TargetFramework.StartsWith('net9.')) Or $(TargetFramework.StartsWith('net10.')) ">
 
     <DefineConstants>$(DefineConstants);FEATURE_ASPNETCORE_TESTHOST</DefineConstants>
+    <DefineConstants>$(DefineConstants);FEATURE_CANCELLATIONTOKENSOURCE_CANCELASYNC</DefineConstants>
+    <DefineConstants>$(DefineConstants);FEATURE_HTTPCONTENT_READASSTREAM</DefineConstants>
+    <DefineConstants>$(DefineConstants);FEATURE_HTTPCONTENT_READASSTREAM_CANCELLATIONTOKEN</DefineConstants>
 
   </PropertyGroup>
 

src/Lucene.Net.Replicator/Http/HttpClientBase.cs

@@ -1,4 +1,3 @@
-using Lucene.Net.Diagnostics;
 using Lucene.Net.Support;
 using System;
 using System.IO;
@@ -7,6 +6,9 @@
 using System.Net.Http;
 using System.Threading;
 using System.Threading.Tasks;
+// ReSharper disable VirtualMemberNeverOverridden.Global
+// ReSharper disable MemberCanBePrivate.Global
+
 #nullable enable
 
 namespace Lucene.Net.Replicator.Http
@@ -173,20 +175,38 @@ protected virtual void ThrowKnownError(HttpResponseMessage response)
         }
 
         /// <summary>
-        /// <b>Internal:</b> Execute a request and return its result.
+        /// <b>Internal:</b> Execute a POST request with custom HttpContent and return its result.
         /// The <paramref name="parameters"/> argument is treated as: name1,value1,name2,value2,...
         /// </summary>
         protected virtual HttpResponseMessage ExecutePost(string request, HttpContent content, params string[]? parameters)
         {
             EnsureOpen();
 
-            var req = new HttpRequestMessage(HttpMethod.Post, QueryString(request, parameters));
-
-            req.Content = content;
+            var req = new HttpRequestMessage(HttpMethod.Post, QueryString(request, parameters))
+            {
+                Content = content
+            };
 
             return Execute(req);
         }
 
+        /// <summary>
+        /// <b>Internal:</b> Execute a POST request asynchronously with custom HttpContent.
+        /// The <paramref name="parameters"/> argument is treated as: name1,value1,name2,value2,...
+        /// </summary>
+        protected virtual async Task<HttpResponseMessage> ExecutePostAsync(string request, HttpContent content, CancellationToken cancellationToken, params string[]? parameters)
+        {
+            EnsureOpen();
+
+            var req = new HttpRequestMessage(HttpMethod.Post, QueryString(request, parameters))
+            {
+                Content = content
+            };
+
+            return await ExecuteAsync(req, cancellationToken).ConfigureAwait(false);
+        }
+
+
         /// <summary>
         /// <b>Internal:</b> Execute a request and return its result.
         /// The <paramref name="parameters"/> argument is treated as: name1,value1,name2,value2,...
@@ -200,6 +220,16 @@ protected virtual HttpResponseMessage ExecuteGet(string request, params string[]
             return Execute(req);
         }
 
+        /// <summary>
+        /// Execute a GET request asynchronously with an array of parameters.
+        /// </summary>
+        protected virtual async Task<HttpResponseMessage> ExecuteGetAsync(string action, string[]? parameters, CancellationToken cancellationToken)
+        {
+            EnsureOpen();
+            var req = new HttpRequestMessage(HttpMethod.Get, QueryString(action, parameters));
+            return await ExecuteAsync(req, cancellationToken).ConfigureAwait(false);
+        }
+
         private HttpResponseMessage Execute(HttpRequestMessage request)
         {
             //.NET Note: Bridging from Async to Sync, this is not ideal and we could consider changing the interface to be Async or provide Async overloads
@@ -209,6 +239,16 @@ private HttpResponseMessage Execute(HttpRequestMessage request)
             return response;
         }
 
+        // LUCENENET specific - async counterpart to Execute used by ExecuteGetAsync/ExecutePostAsync.
+        // Uses ResponseHeadersRead so callers can stream large response bodies (e.g. replication files)
+        // rather than buffering them into memory, and verifies status (matching sync Execute).
+        private async Task<HttpResponseMessage> ExecuteAsync(HttpRequestMessage request, CancellationToken cancellationToken)
+        {
+            var response = await httpc.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cancellationToken).ConfigureAwait(false);
+            VerifyStatus(response);
+            return response;
+        }
+
         private string QueryString(string request, params string[]? parameters)
         {
             return parameters is null || parameters.Length == 0
@@ -255,7 +295,11 @@ public virtual Stream GetResponseStream(HttpResponseMessage response) // LUCENEN
         /// <exception cref="IOException"></exception>
         public virtual Stream GetResponseStream(HttpResponseMessage response, bool consume) // LUCENENET: This was ResponseInputStream in Lucene
         {
+#if FEATURE_HTTPCONTENT_READASSTREAM
+            Stream result = response.Content.ReadAsStream();
+#else
             Stream result = response.Content.ReadAsStreamAsync().ConfigureAwait(false).GetAwaiter().GetResult();
+#endif
 
             if (consume)
             {
@@ -265,6 +309,70 @@ public virtual Stream GetResponseStream(HttpResponseMessage response, bool consu
             return result;
         }
 
+        /// <summary>
+        /// Internal utility: input stream of the provided response.
+        /// The returned stream takes ownership of the response and will dispose it
+        /// when the stream is disposed.
+        /// </summary>
+        /// <exception cref="IOException"></exception>
+        protected virtual Stream GetResponseStreamWithOwnership(HttpResponseMessage response)
+        {
+#if FEATURE_HTTPCONTENT_READASSTREAM
+            Stream result = response.Content.ReadAsStream();
+#else
+            Stream result = response.Content.ReadAsStreamAsync().ConfigureAwait(false).GetAwaiter().GetResult();
+#endif
+            return new ResponseOwningStream(result, response);
+        }
+
+        /// <summary>
+        /// Internal utility: input stream of the provided response asynchronously.
+        /// </summary>
+        /// <exception cref="IOException"></exception>
+        public virtual async Task<Stream> GetResponseStreamAsync(HttpResponseMessage response, CancellationToken cancellationToken = default)
+        {
+#if FEATURE_HTTPCONTENT_READASSTREAM_CANCELLATIONTOKEN
+            Stream result = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
+#else
+            Stream result = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
+#endif
+            return result;
+        }
+
+        /// <summary>
+        /// Internal utility: input stream of the provided response asynchronously, which optionally
+        /// consumes the response's resources when the input stream is exhausted.
+        /// </summary>
+        /// <exception cref="IOException"></exception>
+        // ReSharper disable once UnusedMember.Global - public API
+        public virtual async Task<Stream> GetResponseStreamAsync(HttpResponseMessage response, bool consume, CancellationToken cancellationToken = default)
+        {
+#if FEATURE_HTTPCONTENT_READASSTREAM_CANCELLATIONTOKEN
+            Stream result = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
+#else
+            Stream result = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
+#endif
+            if (consume)
+                result = new ConsumingStream(result);
+            return result;
+        }
+
+        /// <summary>
+        /// Internal utility: input stream of the provided response asynchronously.
+        /// The returned stream takes ownership of the response and will dispose it
+        /// when the stream is disposed.
+        /// </summary>
+        /// <exception cref="IOException"></exception>
+        protected virtual async Task<Stream> GetResponseStreamWithOwnershipAsync(HttpResponseMessage response, CancellationToken cancellationToken = default)
+        {
+#if FEATURE_HTTPCONTENT_READASSTREAM_CANCELLATIONTOKEN
+            Stream result = await response.Content.ReadAsStreamAsync(cancellationToken).ConfigureAwait(false);
+#else
+            Stream result = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
+#endif
+            return new ResponseOwningStream(result, response);
+        }
+
         /// <summary>
         /// Returns <c>true</c> if this instance was <see cref="Dispose(bool)"/>ed, otherwise
         /// returns <c>false</c>. Note that if you override <see cref="Dispose(bool)"/>, you must call
@@ -310,11 +418,56 @@ protected virtual T DoAction<T>(HttpResponseMessage response, bool consume, Func
                     }
                 }
             }
-            if (Debugging.AssertsEnabled) Debugging.Assert(th != null); // extra safety - if we get here, it means the Func<T> failed
+
+            // if (Debugging.AssertsEnabled) Debugging.Assert(th != null); // LUCENENET: Removed assertion because it'll never be null here, ensured by NRT
             Util.IOUtils.ReThrow(th);
             return default!; // silly, if we're here, IOUtils.reThrow always throws an exception
         }
 
+        /// <summary>
+        /// Do a specific async action and validate after the action that the status is still OK,
+        /// and if not, attempt to extract the actual server side exception. Optionally
+        /// release the response at exit, depending on <paramref name="consume"/> parameter.
+        /// </summary>
+        protected virtual async Task<T> DoActionAsync<T>(HttpResponseMessage response, bool consume, Func<Task<T>> call)
+        {
+            Exception? th /* = null */;
+            try
+            {
+                return await call().ConfigureAwait(false);
+            }
+            catch (Exception t) when (t.IsThrowable())
+            {
+                th = t;
+            }
+            finally
+            {
+                try
+                {
+                    VerifyStatus(response);
+                }
+                finally
+                {
+                    if (consume)
+                    {
+                        ConsumeQuietly(response);
+                    }
+                }
+            }
+
+            // if (Debugging.AssertsEnabled) Debugging.Assert(th != null); // LUCENENET: Removed assertion because it'll never be null here, ensured by NRT
+            Util.IOUtils.ReThrow(th);
+            return default!; // never reached, rethrow above always throws
+        }
+
+        /// <summary>
+        /// Calls the overload <see cref="DoActionAsync{T}(HttpResponseMessage, bool, Func{Task{T}})"/> passing <c>true</c> to consume.
+        /// </summary>
+        protected virtual Task<T> DoActionAsync<T>(HttpResponseMessage response, Func<Task<T>> call)
+        {
+            return DoActionAsync(response, true, call);
+        }
+
         /// <summary>
         /// Disposes this <see cref="HttpClientBase"/>.
         /// When called with <code>true</code>, this disposes the underlying <see cref="HttpClient"/>.
@@ -350,14 +503,70 @@ private static void ConsumeQuietly(HttpResponseMessage response)
             }
         }
 
+        /// <summary>
+        /// Wraps a stream and disposes the associated <see cref="HttpResponseMessage"/>
+        /// when the stream is disposed.
+        /// </summary>
+        private sealed class ResponseOwningStream : Stream
+        {
+            private readonly Stream input;
+            private readonly HttpResponseMessage response;
+            private bool disposed;
+
+            public ResponseOwningStream(Stream input, HttpResponseMessage response)
+            {
+                this.input = input ?? throw new ArgumentNullException(nameof(input));
+                this.response = response ?? throw new ArgumentNullException(nameof(response));
+            }
+
+            public override bool CanRead => input.CanRead;
+            public override bool CanSeek => input.CanSeek;
+            public override bool CanWrite => input.CanWrite;
+            public override long Length => input.Length;
+            public override long Position
+            {
+                get => input.Position;
+                set => input.Position = value;
+            }
+
+            public override void Flush() => input.Flush();
+            public override int Read(byte[] buffer, int offset, int count) => input.Read(buffer, offset, count);
+            public override long Seek(long offset, SeekOrigin origin) => input.Seek(offset, origin);
+            public override void SetLength(long value) => input.SetLength(value);
+            public override void Write(byte[] buffer, int offset, int count) => input.Write(buffer, offset, count);
+
+            public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+                => input.ReadAsync(buffer, offset, count, cancellationToken);
+
+#if FEATURE_STREAM_READ_SPAN
+            public override int Read(Span<byte> buffer) => input.Read(buffer);
+            public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+                => input.ReadAsync(buffer, cancellationToken);
+#endif
+
+            protected override void Dispose(bool disposing)
+            {
+                if (!disposed)
+                {
+                    if (disposing)
+                    {
+                        input.Dispose();
+                        response.Dispose();
+                    }
+                    disposed = true;
+                }
+                base.Dispose(disposing);
+            }
+        }
+
         /// <summary>
         /// Wraps a stream and consumes (flushes) and disposes automatically
         /// when the last call to a Read overload occurs.
         /// </summary>
         private class ConsumingStream : Stream
         {
             private readonly Stream input;
-            private bool consumed = false;
+            private bool consumed /* = false */;
 
             public ConsumingStream(Stream input)
             {