Add async PerformAsync and FlushAsync methods with CancellationToken support (#1170)

* Add async PerformAsync and FlushAsync methods with CancellationToken support

* Refactor replication service and stream extensions for fully async I/O, updating PerformAsync, SessionToken, and ASP.NET Core response handling.

* refactor: rename BigEndian methods and use ArrayPool for buffer allocations in StreamExtensions

* feat: Add async read/write support in ReplicationServlet and corresponding J2N-based async tests in TestStreamExtensions

* Add XML Summary For PerformAsync and remove dupilcate comment code

* Lucene.Net.Tests.Replicator: Refactored to test both synchronous and asynchronous APIs as well as both Startup class and Middleware configurations (when both are supported).

* Lucene.Net.Tests.Replicator: Added an HttpListener-based server so we don't have to rely on Microsoft.AspNetcore.TestHost prior to .NET Core. Bumped Microsoft.AspNetcore.TestHost package  to 9.0.8. Refactored tests to separate ASP.NET Core functionality from HttpListener functionality.

* Lucene.Net.Tests.Replicator: Only reference Lucene.Net.Replicator.AspNetCore on .NET Core 8 or higher

* Lucene.Net.Replicator.AspNetCore: Dropped support for all target framworks but net8.0.

* Lucene.Net.Replicator.AspNetCore: Removed PackageReference to transitive dependency System.Text.Encodings.Web, which only existed to bump the package to one without known security vulnerabilities. Bumped Microsoft.AspNetCore.Http.Abstractions to 2.3.0.

* Add IAsyncReplicationServer/IAsyncReplicationResponse, update ReplicationService, Middleware, and TestServer to support async replication with conditional sync/async DI.

* Revert "Add IAsyncReplicationServer/IAsyncReplicationResponse, update ReplicationService, Middleware, and TestServer to support async replication with conditional sync/async DI."

This reverts commit 349621f.

* Revert async interfaces, unify IReplicationService with Perform/PerformAsync, update ReplicationServlet, Middleware, and TestServer to use single interface

---------

Co-authored-by: Shad Storhaug <shad@shadstorhaug.com>

Author: NehanPathan

.build/dependencies.props

@@ -1,4 +1,4 @@
-<!--
+<!--
 
  Licensed to the Apache Software Foundation (ASF) under one
  or more contributor license agreements.  See the NOTICE file
@@ -39,11 +39,8 @@
     <!-- J2N will break binary compatibility in 3.0.0 to fix the APIs of collection types -->
     <J2NPackageVersion>[2.1.0, 3.0.0)</J2NPackageVersion>
     <LiquidTestReportsMarkdownPackageVersion>1.0.9</LiquidTestReportsMarkdownPackageVersion>
-    <MicrosoftAspNetCoreHttpAbstractionsPackageVersion>2.1.1</MicrosoftAspNetCoreHttpAbstractionsPackageVersion>
-    <MicrosoftAspNetCoreHttpPackageVersion>2.1.34</MicrosoftAspNetCoreHttpPackageVersion>
-    <MicrosoftAspNetCoreTestHostPackageVersion>6.0.0</MicrosoftAspNetCoreTestHostPackageVersion>
-    <MicrosoftAspNetCoreTestHostPackageVersion Condition=" $(TargetFramework.StartsWith('net4')) ">2.1.1</MicrosoftAspNetCoreTestHostPackageVersion>
-    <MicrosoftAspNetCoreTestHostPackageVersion Condition=" '$(TargetFramework)' == 'net5.0' ">3.1.32</MicrosoftAspNetCoreTestHostPackageVersion>
+    <MicrosoftAspNetCoreHttpAbstractionsPackageVersion>2.3.0</MicrosoftAspNetCoreHttpAbstractionsPackageVersion>
+    <MicrosoftAspNetCoreTestHostPackageVersion>8.0.19</MicrosoftAspNetCoreTestHostPackageVersion>
     <MicrosoftCodeAnalysisAnalyzersPackageVersion>2.9.8</MicrosoftCodeAnalysisAnalyzersPackageVersion>
     <MicrosoftCodeAnalysisCSharpPackageVersion>2.6.1</MicrosoftCodeAnalysisCSharpPackageVersion>
     <MicrosoftCodeAnalysisCSharpWorkspacesPackageVersion>$(MicrosoftCodeAnalysisCSharpPackageVersion)</MicrosoftCodeAnalysisCSharpWorkspacesPackageVersion>
@@ -71,7 +68,6 @@
     <RandomizedTestingGeneratorsPackageVersion>2.7.8</RandomizedTestingGeneratorsPackageVersion>
     <SharpZipLibPackageVersion>1.4.2</SharpZipLibPackageVersion>
     <Spatial4nPackageVersion>0.4.1.1</Spatial4nPackageVersion>
-    <SystemIOPipelinesPackageVersion>8.0.0</SystemIOPipelinesPackageVersion>
     <SystemMemoryPackageVersion>4.5.5</SystemMemoryPackageVersion>
     <SystemNetHttpPackageVersion>4.3.4</SystemNetHttpPackageVersion>
     <SystemReflectionEmitPackageVersion>4.3.0</SystemReflectionEmitPackageVersion>
@@ -80,7 +76,6 @@
     <SystemRuntimeInteropServicesRuntimeInformationPackageVersion>4.3.0</SystemRuntimeInteropServicesRuntimeInformationPackageVersion>
     <SystemTextEncodingCodePagesPackageVersion>4.3.0</SystemTextEncodingCodePagesPackageVersion>
     <SystemTextEncodingCodePagesPackageVersion Condition=" '$(TargetFramework)' == 'net472' ">5.0.0</SystemTextEncodingCodePagesPackageVersion>
-    <SystemTextEncodingsWebPackageVersion>8.0.0</SystemTextEncodingsWebPackageVersion>
     <SystemTextJsonPackageVersion>6.0.10</SystemTextJsonPackageVersion>
     <SystemTextRegularExpressionsPackageVersion>4.3.1</SystemTextRegularExpressionsPackageVersion>
     <TimeZoneConverterPackageVersion>6.1.0</TimeZoneConverterPackageVersion>

Directory.Build.targets

@@ -40,6 +40,7 @@
   <!-- Features in .NET 8.x and .NET 9.x only -->
   <PropertyGroup Condition=" $(TargetFramework.StartsWith('net8.')) Or $(TargetFramework.StartsWith('net9.')) ">
 
+    <DefineConstants>$(DefineConstants);FEATURE_ASPNETCORE_TESTHOST</DefineConstants>
     <DefineConstants>$(DefineConstants);FEATURE_UTF8_TOUTF16</DefineConstants>
 
   </PropertyGroup>
@@ -56,7 +57,6 @@
   <!-- Features in .NET 5.x, .NET 6.x, .NET 7.x, .NET 8.x, and .NET 9.x only -->
   <PropertyGroup Condition=" $(TargetFramework.StartsWith('net5.')) Or $(TargetFramework.StartsWith('net6.')) Or $(TargetFramework.StartsWith('net7.')) Or $(TargetFramework.StartsWith('net8.')) Or $(TargetFramework.StartsWith('net9.')) ">
 
-    <DefineConstants>$(DefineConstants);FEATURE_ASPNETCORE_ENDPOINT_CONFIG</DefineConstants>
     <DefineConstants>$(DefineConstants);FEATURE_READONLYSET</DefineConstants>
 
   </PropertyGroup>

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

@@ -7,6 +7,9 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Net;
+using System.Threading;
+using System.Threading.Tasks;
+
 
 namespace Lucene.Net.Replicator.Http
 {
@@ -118,62 +121,54 @@ private static string ExtractRequestParam(IReplicationRequest request, string pa
             return param;
         }
 
-        // LUCENENET specific - copy method not used
-
-        /// <summary>
-        /// Executes the replication task.
-        /// </summary>
-        /// <exception cref="InvalidOperationException">required parameters are missing</exception>
-        public virtual void Perform(IReplicationRequest request, IReplicationResponse response)
+        // method to avoid code duplication in sync and async Perform methods
+        private async Task ExecuteReplicationAsync(
+            IReplicationRequest request,
+            IReplicationResponse response,
+            Func<Stream, Task> copyStreamFunc,
+            Func<SessionToken, Task> writeTokenFunc,
+            Func<Task> flushFunc)
         {
             string[] pathElements = GetPathElements(request);
             if (pathElements.Length != 2)
-            {
                 throw ServletException.Create("invalid path, must contain shard ID and action, e.g. */s1/update");
-            }
 
             if (!Enum.TryParse(pathElements[ACTION_IDX], true, out ReplicationAction action))
-            {
                 throw ServletException.Create("Unsupported action provided: " + pathElements[ACTION_IDX]);
-            }
 
             if (!replicators.TryGetValue(pathElements[SHARD_IDX], out IReplicator replicator))
-            {
                 throw ServletException.Create("unrecognized shard ID " + pathElements[SHARD_IDX]);
-            }
 
             // SOLR-8933 Don't close this stream.
             try
             {
                 switch (action)
                 {
                     case ReplicationAction.OBTAIN:
-                        string sessionId = ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM);
-                        string fileName = ExtractRequestParam(request, REPLICATE_FILENAME_PARAM);
-                        string source = ExtractRequestParam(request, REPLICATE_SOURCE_PARAM);
-                        using (Stream stream = replicator.ObtainFile(sessionId, source, fileName))
-                            stream.CopyTo(response.Body);
-                        break;
+                        {
+                            string sessionId = ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM);
+                            string fileName = ExtractRequestParam(request, REPLICATE_FILENAME_PARAM);
+                            string source = ExtractRequestParam(request, REPLICATE_SOURCE_PARAM);
 
-                    case ReplicationAction.RELEASE:
-                        replicator.Release(ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM));
-                        break;
+                            using (Stream stream = replicator.ObtainFile(sessionId, source, fileName))
+                                await copyStreamFunc(stream);
+                            break;
+                        }
 
-                    case ReplicationAction.UPDATE:
-                        string currentVersion = request.QueryParam(REPLICATE_VERSION_PARAM);
-                        SessionToken token = replicator.CheckForUpdate(currentVersion);
-                        if (token is null)
+                    case ReplicationAction.RELEASE:
                         {
-                            response.Body.Write(new byte[] { 0 }, 0, 1); // marker for null token
+                            replicator.Release(ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM));
+                            break;
                         }
-                        else
+
+                    case ReplicationAction.UPDATE:
                         {
-                            response.Body.Write(new byte[] { 1 }, 0, 1);
-                            token.Serialize(new DataOutputStream(response.Body));
+                            string currentVersion = request.QueryParam(REPLICATE_VERSION_PARAM);
+                            SessionToken token = replicator.CheckForUpdate(currentVersion);
+                            await writeTokenFunc(token);
+                            break;
                         }
-                        break;
 
-                    // LUCENENET specific:
                     default:
                         if (Debugging.AssertsEnabled) Debugging.Assert(false, "Invalid ReplicationAction specified");
                         break;
@@ -185,8 +180,70 @@ public virtual void Perform(IReplicationRequest request, IReplicationResponse re
             }
             finally
             {
-                response.Flush();
+                await flushFunc();
             }
         }
+
+        // LUCENENET specific - copy method not used
+
+        /// <summary>
+        /// Executes the replication task.
+        /// </summary>
+        /// <exception cref="InvalidOperationException">required parameters are missing</exception>
+        public virtual void Perform(IReplicationRequest request, IReplicationResponse response)
+        {
+            ExecuteReplicationAsync(
+                request,
+                response,
+                stream => { stream.CopyTo(response.Body); return Task.CompletedTask; },
+                token =>
+                {
+                    if (token == null)
+                    {
+                        response.Body.Write(new byte[] { 0 }, 0, 1);
+                    }
+                    else
+                    {
+                        response.Body.Write(new byte[] { 1 }, 0, 1);
+                        token.Serialize(new DataOutputStream(response.Body));
+                    }
+                    return Task.CompletedTask;
+                },
+                () => { response.Body.Flush(); return Task.CompletedTask; }
+            ).ConfigureAwait(false).GetAwaiter().GetResult(); // keep sync behavior
+        }
+
+
+        /// <summary>
+        /// Executes the replication task asynchronously.
+        /// </summary>
+        /// <param name="request">The replication request containing action and parameters.</param>
+        /// <param name="response">The replication response used to send data back to the client.</param>
+        /// <param name="cancellationToken">A <see cref="CancellationToken"/> to observe while performing the replication.</param>
+        /// <exception cref="InvalidOperationException">Thrown when required parameters are missing or invalid.</exception>
+        public virtual Task PerformAsync(
+            IReplicationRequest request,
+            IReplicationResponse response,
+            CancellationToken cancellationToken = default)
+        {
+            return ExecuteReplicationAsync(
+                request,
+                response,
+                stream => stream.CopyToAsync(response.Body, 81920, cancellationToken),
+                async token =>
+                {
+                    if (token == null)
+                    {
+                        await response.Body.WriteAsync(new byte[] { 0 }, 0, 1, cancellationToken);
+                    }
+                    else
+                    {
+                        await response.Body.WriteAsync(new byte[] { 1 }, 0, 1, cancellationToken);
+                        await token.SerializeAsync(response.Body, cancellationToken);
+                    }
+                },
+                () => response.Body.FlushAsync(cancellationToken)
+            );
+        }
     }
 }

src/Lucene.Net.Replicator/SessionToken.cs

@@ -1,7 +1,11 @@
 using J2N.IO;
 using System.Collections.Generic;
+using System;
 using System.IO;
 using JCG = J2N.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using Lucene.Net.Support.IO;
 
 namespace Lucene.Net.Replicator
 {
@@ -112,6 +116,37 @@ public void Serialize(DataOutputStream writer)
             }
         }
 
+        /// <summary>
+        /// Asynchronously serializes the token's properties, including ID, version, and source files,
+        /// to the provided <see cref="Stream"/> for transmission or storage.
+        /// </summary>
+        /// <param name="output">The <see cref="Stream"/> to write the token data to.</param>
+        /// <param name="cancellationToken">A cancellation token to observe while writing and flushing the stream.</param>
+        /// <returns>A task representing the asynchronous serialization operation.</returns>
+        internal async Task SerializeAsync(Stream output, CancellationToken cancellationToken = default)
+        {
+            if (output is null)
+                throw new ArgumentNullException(nameof(output));
+
+            await output.WriteUTFAsync(Id, cancellationToken).ConfigureAwait(false);
+            await output.WriteUTFAsync(Version, cancellationToken).ConfigureAwait(false);
+            await output.WriteInt32BigEndianAsync(SourceFiles.Count, cancellationToken).ConfigureAwait(false);
+
+            foreach (var pair in SourceFiles)
+            {
+                await output.WriteUTFAsync(pair.Key, cancellationToken).ConfigureAwait(false);
+                await output.WriteInt32BigEndianAsync(pair.Value.Count, cancellationToken).ConfigureAwait(false);
+
+                foreach (var file in pair.Value)
+                {
+                    await output.WriteUTFAsync(file.FileName, cancellationToken).ConfigureAwait(false);
+                    await output.WriteInt64BigEndianAsync(file.Length, cancellationToken).ConfigureAwait(false);
+                }
+            }
+
+            await output.FlushAsync(cancellationToken).ConfigureAwait(false);
+        }
+
         public override string ToString()
         {
             return string.Format("id={0} version={1} files={2}", Id, Version, SourceFiles);

src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs

@@ -37,10 +37,5 @@ public interface IReplicationResponse
         /// The response content.
         /// </summary>
         Stream Body { get; }
-
-        /// <summary>
-        /// Flushes the reponse to the underlying response stream.
-        /// </summary>
-        void Flush();
     }
 }

src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs

@@ -1,4 +1,7 @@
 using System;
+using System.Threading;
+using System.Threading.Tasks;
+
 
 namespace Lucene.Net.Replicator.Http.Abstractions
 {
@@ -29,5 +32,15 @@ public interface IReplicationService
         /// </summary>
         /// <exception cref="InvalidOperationException">required parameters are missing</exception>
         void Perform(IReplicationRequest request, IReplicationResponse response);
+
+        /// <summary>
+        /// Executes the replication task asynchronously.
+        /// </summary>
+        /// <param name="request">The replication request.</param>
+        /// <param name="response">The replication response.</param>
+        /// <param name="cancellationToken">Optional cancellation token.</param>
+        /// <returns>A task representing the asynchronous operation.</returns>
+        Task PerformAsync(IReplicationRequest request, IReplicationResponse response, CancellationToken cancellationToken = default);
+
     }
 }

src/Lucene.Net.Tests.Replicator/Http/HttpReplicatorTest.cs

@@ -2,14 +2,20 @@
 using Lucene.Net.Index;
 using Lucene.Net.Support;
 using Lucene.Net.Util;
-using Microsoft.AspNetCore.TestHost;
 using NUnit.Framework;
 using System;
 using System.Collections.Generic;
 using System.Globalization;
 using System.IO;
 using Directory = Lucene.Net.Store.Directory;
 
+#if FEATURE_ASPNETCORE_TESTHOST
+using Microsoft.AspNetCore.TestHost;
+#else
+using Lucene.Net.Replicator.Net;
+#endif
+
+
 namespace Lucene.Net.Replicator.Http
 {
     /*
@@ -29,6 +35,14 @@ namespace Lucene.Net.Replicator.Http
      * limitations under the License.
      */
 
+    // Technically, the ConfigOption is only supported by ASP.NET Core
+    // so we just ignore the other option when running on HttpListener.
+    [TestFixture(IOOption.Synchronous, ConfigOption.StartupClass)]
+    [TestFixture(IOOption.Asynchronous, ConfigOption.StartupClass)]
+#if FEATURE_ASPNETCORE_TESTHOST
+    [TestFixture(IOOption.Synchronous, ConfigOption.Middleware)]
+    [TestFixture(IOOption.Asynchronous, ConfigOption.Middleware)]
+#endif
     public class HttpReplicatorTest : ReplicatorTestCase
     {
         private DirectoryInfo clientWorkDir;
@@ -45,16 +59,31 @@ public class HttpReplicatorTest : ReplicatorTestCase
 
         private MockErrorConfig mockErrorConfig;
 
-        private void StartServer()
+        private readonly bool useSynchronousIO;
+        private readonly bool useStartupClass;
+
+        public enum IOOption
         {
-            ReplicationService service = new ReplicationService(new Dictionary<string, IReplicator> { { "s1", serverReplicator } });
+            Synchronous,
+            Asynchronous,
+        }
 
-#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG
-            server = NewHttpServer(service, mockErrorConfig); // Call like this to use ReplicationServerMiddleware on the specific path /replicate/{shard?}/{action?}, but allow other paths to be served
-#else
-            server = NewHttpServer<ReplicationServlet>(service, mockErrorConfig); // Call like this to use ReplicationServlet as a Startup Class
-#endif
+        public enum ConfigOption
+        {
+            StartupClass,
+            Middleware
+        }
+
+        public HttpReplicatorTest(IOOption ioOption, ConfigOption configOption)
+        {
+            this.useSynchronousIO = ioOption == IOOption.Synchronous;
+            this.useStartupClass = configOption == ConfigOption.StartupClass;
+        }
 
+        private void StartServer()
+        {
+            ReplicationService service = new ReplicationService(new Dictionary<string, IReplicator> { { "s1", serverReplicator } });
+            server = NewHttpServer(service, mockErrorConfig, useSynchronousIO, useStartupClass);
             port = ServerPort(server);
             host = ServerHost(server);
         }
@@ -144,7 +173,7 @@ public void TestServerErrors()
                 mockErrorConfig.RespondWithError = false;
                 client.UpdateNow(); // now it should work
                 ReopenReader();
-                assertEquals(5, J2N.Numerics.Int32.Parse(reader.IndexCommit.UserData["ID"], 16));
+                assertEquals(5, int.Parse(reader.IndexCommit.UserData["ID"], NumberStyles.HexNumber));
 
                 client.Dispose();
             }