Add IAsyncReplicator interface and async support in HttpReplicator (client-side async API, related to #XXXX)

Author: NehanPathan

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

@@ -200,6 +200,29 @@ protected virtual HttpResponseMessage ExecuteGet(string request, params string[]
             return Execute(req);
         }
 
+        /// <summary>
+        /// Execute a GET request asynchronously with an array of parameters.
+        /// </summary>
+        protected Task<HttpResponseMessage> ExecuteGetAsync(string action, string[] parameters, CancellationToken cancellationToken)
+        {
+            var url = BuildUrl(action, parameters);
+            return Client.GetAsync(url, cancellationToken);
+        }
+
+        /// <summary>
+        /// Execute a GET request asynchronously with up to 3 name/value parameters.
+        /// </summary>
+        protected Task<HttpResponseMessage> ExecuteGetAsync(
+            string action,
+            string param1, string value1,
+            string param2 = null, string value2 = null,
+            string param3 = null, string value3 = null,
+            CancellationToken cancellationToken = default)
+        {
+            var url = BuildUrl(action, param1, value1, param2, value2, param3, value3);
+            return Client.GetAsync(url, cancellationToken);
+        }
+
         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
@@ -217,6 +240,31 @@ private string QueryString(string request, params string[] parameters)
                 .Join("&", parameters.Select(WebUtility.UrlEncode).InPairs((key, val) => string.Format("{0}={1}", key, val))));
         }
 
+        // Add this property so subclasses can access the HttpClient instance
+        protected HttpClient Client => httpc;
+
+        // BuildUrl helpers (mirror the QueryString overloads)
+        protected virtual string BuildUrl(string action, string[] parameters)
+        {
+            // QueryString has signature: QueryString(string request, params string[] parameters)
+            return QueryString(action, parameters);
+        }
+
+        protected virtual string BuildUrl(
+            string action,
+            string param1, string value1,
+            string param2 = null, string value2 = null,
+            string param3 = null, string value3 = null)
+        {
+            // Forward to QueryString which accepts params string[]
+            if (param2 == null && param3 == null)
+            {
+                return QueryString(action, param1, value1);
+            }
+            return QueryString(action, param1, value1, param2, value2, param3, value3);
+        }
+
+
         /// <summary>
         /// Internal utility: input stream of the provided response.
         /// </summary>
@@ -262,6 +310,37 @@ public virtual Stream GetResponseStream(HttpResponseMessage response, bool consu
             return result;
         }
 
+        /// <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 NET8_0_OR_GREATER
+            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>
+        public virtual async Task<Stream> GetResponseStreamAsync(HttpResponseMessage response, bool consume, CancellationToken cancellationToken = default)
+        {
+        #if NET8_0_OR_GREATER
+            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>
         /// 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
@@ -319,6 +398,59 @@ protected virtual T DoAction<T>(HttpResponseMessage response, bool consume, Func
             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
+            {
+                VerifyStatus(response);
+                return await call().ConfigureAwait(false);
+            }
+            catch (Exception t) when (t.IsThrowable())
+            {
+                th = t;
+            }
+            finally
+            {
+                try
+                {
+                    VerifyStatus(response);
+                }
+                finally
+                {
+                    if (consume)
+                    {
+                        try
+                        {
+                            ConsumeQuietly(response);
+                        }
+                        catch
+                        {
+                            // ignore on purpose
+                        }
+                    }
+                }
+            }
+
+            if (Debugging.AssertsEnabled) Debugging.Assert(th != null);
+            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"/>.

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

@@ -2,6 +2,9 @@
 using System;
 using System.IO;
 using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+
 
 namespace Lucene.Net.Replicator.Http
 {
@@ -28,7 +31,7 @@ namespace Lucene.Net.Replicator.Http
     /// <remarks>
     /// @lucene.experimental
     /// </remarks>
-    public class HttpReplicator : HttpClientBase, IReplicator
+    public class HttpReplicator : HttpClientBase, IReplicator, IAsyncReplicator
     {
         /// <summary>
         /// Creates a new <see cref="HttpReplicator"/> with the given host, port and path.
@@ -106,5 +109,95 @@ public virtual void Release(string sessionId)
             // do not remove this call: as it is still validating for us!
             DoAction<object>(response, () => null);
         }
+
+        #region Async methods (IAsyncReplicator)
+
+        /// <summary>
+        /// Checks for updates at the remote host asynchronously.
+        /// </summary>
+        /// <param name="currentVersion">The current index version.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        /// <returns>
+        /// A <see cref="SessionToken"/> if updates are available; otherwise, <c>null</c>.
+        /// </returns>
+        public async Task<SessionToken?> CheckForUpdateAsync(string currentVersion, CancellationToken cancellationToken = default)
+        {
+            string[] parameters = currentVersion != null
+                ? new[] { ReplicationService.REPLICATE_VERSION_PARAM, currentVersion }
+                : null;
+
+            using var response = await ExecuteGetAsync(
+                ReplicationService.ReplicationAction.UPDATE.ToString(),
+                parameters,
+                cancellationToken: cancellationToken).ConfigureAwait(false);
+
+            return await DoActionAsync(response, async () =>
+            {
+                using var inputStream = new DataInputStream(
+                    await GetResponseStreamAsync(response, cancellationToken).ConfigureAwait(false));
+
+                return inputStream.ReadByte() == 0 ? null : new SessionToken(inputStream);
+            }).ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Obtains the given file from the remote host asynchronously.
+        /// </summary>
+        /// <param name="sessionId">The session ID.</param>
+        /// <param name="source">The source of the file.</param>
+        /// <param name="fileName">The file name.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        /// <returns>A <see cref="Stream"/> of the requested file.</returns>
+        public async Task<Stream> ObtainFileAsync(string sessionId, string source, string fileName, CancellationToken cancellationToken = default)
+        {
+            using var response = await ExecuteGetAsync(
+                ReplicationService.ReplicationAction.OBTAIN.ToString(),
+                ReplicationService.REPLICATE_SESSION_ID_PARAM, sessionId,
+                ReplicationService.REPLICATE_SOURCE_PARAM, source,
+                ReplicationService.REPLICATE_FILENAME_PARAM, fileName,
+                cancellationToken: cancellationToken).ConfigureAwait(false);
+
+            return await DoActionAsync(response, async () =>
+            {
+                return await GetResponseStreamAsync(response, cancellationToken).ConfigureAwait(false);
+            }).ConfigureAwait(false);
+        }
+
+        /// <summary>
+        /// Publishes a new <see cref="IRevision"/> asynchronously.
+        /// Not supported in this implementation.
+        /// </summary>
+        /// <param name="revision">The revision to publish.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        /// <returns>A <see cref="Task"/> representing the operation.</returns>
+        /// <exception cref="NotSupportedException">Always thrown.</exception>
+        public Task PublishAsync(IRevision revision, CancellationToken cancellationToken = default)
+        {
+            throw UnsupportedOperationException.Create(
+                "this replicator implementation does not support remote publishing of revisions");
+        }
+
+        /// <summary>
+        /// Releases the session at the remote host asynchronously.
+        /// </summary>
+        /// <param name="sessionId">The session ID to release.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        /// <returns>A <see cref="Task"/> representing the operation.</returns>
+        public async Task ReleaseAsync(string sessionId, CancellationToken cancellationToken = default)
+        {
+            using var response = await ExecuteGetAsync(
+                ReplicationService.ReplicationAction.RELEASE.ToString(),
+                ReplicationService.REPLICATE_SESSION_ID_PARAM, sessionId,
+                cancellationToken: cancellationToken).ConfigureAwait(false);
+
+            await DoActionAsync(response, () =>
+            {
+                // No actual response content needed — just verification
+                return Task.FromResult<object?>(null);
+            }).ConfigureAwait(false);
+        }
+
+        #endregion
+
     }
 }

src/Lucene.Net.Replicator/IAsyncReplicator.cs

@@ -0,0 +1,35 @@
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Lucene.Net.Replicator
+{
+    /// <summary>
+    /// Async version of <see cref="IReplicator"/> for non-blocking replication operations.
+    /// </summary>
+    public interface IAsyncReplicator
+    {
+        /// <summary>
+        /// Check whether the given version is up-to-date and returns a
+        /// <see cref="SessionToken"/> which can be used for fetching the revision files.
+        /// </summary>
+        /// <param name="currentVersion">Current version of the index.</param>
+        /// <param name="cancellationToken">Cancellation token.</param>
+        Task<SessionToken?> CheckForUpdateAsync(string currentVersion, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Returns a stream for the requested file and source.
+        /// </summary>
+        Task<Stream> ObtainFileAsync(string sessionId, string source, string fileName, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Notify that the specified session is no longer needed.
+        /// </summary>
+        Task ReleaseAsync(string sessionId, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Publishing revisions is not supported in HttpReplicator; throw if called.
+        /// </summary>
+        Task PublishAsync(IRevision revision, CancellationToken cancellationToken = default);
+    }
+}