From 8f28cd7ce5bfdf946282650bc17f1fcf8cfd39c3 Mon Sep 17 00:00:00 2001 From: NehanPathan Date: Tue, 26 Aug 2025 15:42:56 +0530 Subject: [PATCH 01/13] Add async PerformAsync and FlushAsync methods with CancellationToken support --- .../Http/ReplicationService.cs | 68 +++++++++++++++++++ src/Lucene.Net.Replicator/SessionToken.cs | 29 ++++++++ .../Http/Abstractions/IReplicationResponse.cs | 9 +++ .../Http/Abstractions/IReplicationService.cs | 13 ++++ .../AspNetCoreReplicationResponse.cs | 18 +++++ 5 files changed, 137 insertions(+) diff --git a/src/Lucene.Net.Replicator/Http/ReplicationService.cs b/src/Lucene.Net.Replicator/Http/ReplicationService.cs index ae2426a8f4..72b6796e87 100644 --- a/src/Lucene.Net.Replicator/Http/ReplicationService.cs +++ b/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 { @@ -188,5 +191,70 @@ public virtual void Perform(IReplicationRequest request, IReplicationResponse re response.Flush(); } } + + /// + /// Executes the replication task asynchronously. + /// + /// The replication request containing action and parameters. + /// The replication response used to send data back to the client. + /// A to observe while performing the replication. + /// Thrown when required parameters are missing or invalid. + public virtual async Task PerformAsync(IReplicationRequest request, IReplicationResponse response, CancellationToken cancellationToken = default) + { + 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]); + + 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)) + await stream.CopyToAsync(response.Body, 81920, cancellationToken); + break; + + case ReplicationAction.RELEASE: + replicator.Release(ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM)); + break; + + case ReplicationAction.UPDATE: + string currentVersion = request.QueryParam(REPLICATE_VERSION_PARAM); + SessionToken token = replicator.CheckForUpdate(currentVersion); + if (token is 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); + } + break; + + default: + if (Debugging.AssertsEnabled) Debugging.Assert(false, "Invalid ReplicationAction specified"); + break; + } + } + catch (Exception) + { + response.StatusCode = (int)HttpStatusCode.InternalServerError; + } + finally + { + await response.FlushAsync(cancellationToken); + } + } + } } diff --git a/src/Lucene.Net.Replicator/SessionToken.cs b/src/Lucene.Net.Replicator/SessionToken.cs index fd17649d73..a4e3023df7 100644 --- a/src/Lucene.Net.Replicator/SessionToken.cs +++ b/src/Lucene.Net.Replicator/SessionToken.cs @@ -2,6 +2,8 @@ using System.Collections.Generic; using System.IO; using JCG = J2N.Collections.Generic; +using System.Threading; +using System.Threading.Tasks; namespace Lucene.Net.Replicator { @@ -112,6 +114,33 @@ public void Serialize(DataOutputStream writer) } } + /// + /// Asynchronously serialize the token data for communication between server and client. + /// + /// The to write the token data to. + /// A cancellation token to observe while waiting for the flush to complete. + /// A task representing the asynchronous operation. + public async Task SerializeAsync(Stream output, CancellationToken cancellationToken = default) + { + using var writer = new DataOutputStream(output); + writer.WriteUTF(Id); + writer.WriteUTF(Version); + writer.WriteInt32(SourceFiles.Count); + + foreach (var pair in SourceFiles) + { + writer.WriteUTF(pair.Key); + writer.WriteInt32(pair.Value.Count); + foreach (var file in pair.Value) + { + writer.WriteUTF(file.FileName); + writer.WriteInt64(file.Length); + } + } + + await output.FlushAsync(cancellationToken); + } + public override string ToString() { return string.Format("id={0} version={1} files={2}", Id, Version, SourceFiles); diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs index e55b37996b..72e4233967 100644 --- a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs +++ b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs @@ -1,4 +1,6 @@ using System.IO; +using System.Threading; +using System.Threading.Tasks; namespace Lucene.Net.Replicator.Http.Abstractions { @@ -42,5 +44,12 @@ public interface IReplicationResponse /// Flushes the reponse to the underlying response stream. /// void Flush(); + + /// + /// Flushes the response to the underlying response stream asynchronously. + /// + /// Optional cancellation token. + /// A task representing the asynchronous operation. + Task FlushAsync(CancellationToken cancellationToken = default); } } diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs index c1e67e22b8..78ca9bf565 100644 --- a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs +++ b/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 /// /// required parameters are missing void Perform(IReplicationRequest request, IReplicationResponse response); + + /// + /// Executes the replication task asynchronously. + /// + /// The replication request. + /// The replication response. + /// Optional cancellation token. + /// A task representing the asynchronous operation. + Task PerformAsync(IReplicationRequest request, IReplicationResponse response, CancellationToken cancellationToken = default); + } } diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs index 4808f7592b..b4a97f2d20 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs @@ -1,6 +1,8 @@ using Lucene.Net.Replicator.Http.Abstractions; using Microsoft.AspNetCore.Http; using System.IO; +using System.Threading; +using System.Threading.Tasks; namespace Lucene.Net.Replicator.AspNetCore { @@ -68,5 +70,21 @@ public void Flush() { response.Body.Flush(); } + + /// + /// Flushes the response to the underlying response stream asynchronously. + /// + /// Optional cancellation token. + /// A task representing the asynchronous operation. + /// + /// This simply calls on the . + /// + public async Task FlushAsync(CancellationToken cancellationToken = default) + { + if (response.Body.CanWrite) + { + await response.Body.FlushAsync(cancellationToken); + } + } } } From 3e1068a7c3df94a8db9c99bce1db4488f6cba5d8 Mon Sep 17 00:00:00 2001 From: NehanPathan Date: Thu, 28 Aug 2025 01:08:14 +0530 Subject: [PATCH 02/13] Refactor replication service and stream extensions for fully async I/O, updating PerformAsync, SessionToken, and ASP.NET Core response handling. --- .../Http/ReplicationService.cs | 166 ++++++++---------- src/Lucene.Net.Replicator/SessionToken.cs | 30 ++-- src/Lucene.Net/Support/IO/StreamExtensions.cs | 152 ++++++++++++++++ .../AspNetCoreReplicationResponse.cs | 5 +- 4 files changed, 248 insertions(+), 105 deletions(-) diff --git a/src/Lucene.Net.Replicator/Http/ReplicationService.cs b/src/Lucene.Net.Replicator/Http/ReplicationService.cs index 72b6796e87..c1c4f801f8 100644 --- a/src/Lucene.Net.Replicator/Http/ReplicationService.cs +++ b/src/Lucene.Net.Replicator/Http/ReplicationService.cs @@ -121,62 +121,53 @@ private static string ExtractRequestParam(IReplicationRequest request, string pa return param; } - // LUCENENET specific - copy method not used - - /// - /// Executes the replication task. - /// - /// required parameters are missing - 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 copyStreamFunc, + Func writeTokenFunc, + Func 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; @@ -184,14 +175,44 @@ public virtual void Perform(IReplicationRequest request, IReplicationResponse re } catch (Exception) { - response.StatusCode = (int)HttpStatusCode.InternalServerError; // propagate the failure + response.StatusCode = (int)HttpStatusCode.InternalServerError; } finally { - response.Flush(); + await flushFunc(); } } + // LUCENENET specific - copy method not used + + /// + /// Executes the replication task. + /// + /// required parameters are missing + 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.Flush(); return Task.CompletedTask; } + ).GetAwaiter().GetResult(); // // keep sync behavior + } + + /// /// Executes the replication task asynchronously. /// @@ -199,62 +220,29 @@ public virtual void Perform(IReplicationRequest request, IReplicationResponse re /// The replication response used to send data back to the client. /// A to observe while performing the replication. /// Thrown when required parameters are missing or invalid. - public virtual async Task PerformAsync(IReplicationRequest request, IReplicationResponse response, CancellationToken cancellationToken = default) + public virtual Task PerformAsync( + IReplicationRequest request, + IReplicationResponse response, + CancellationToken cancellationToken = default) { - 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]); - - try - { - switch (action) + return ExecuteReplicationAsync( + request, + response, + stream => stream.CopyToAsync(response.Body, 81920, cancellationToken), + async token => { - 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)) - await stream.CopyToAsync(response.Body, 81920, cancellationToken); - break; - - case ReplicationAction.RELEASE: - replicator.Release(ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM)); - break; - - case ReplicationAction.UPDATE: - string currentVersion = request.QueryParam(REPLICATE_VERSION_PARAM); - SessionToken token = replicator.CheckForUpdate(currentVersion); - if (token is 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); - } - break; - - default: - if (Debugging.AssertsEnabled) Debugging.Assert(false, "Invalid ReplicationAction specified"); - break; - } - } - catch (Exception) - { - response.StatusCode = (int)HttpStatusCode.InternalServerError; - } - finally - { - await response.FlushAsync(cancellationToken); - } + 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.FlushAsync(cancellationToken) + ); } - } } diff --git a/src/Lucene.Net.Replicator/SessionToken.cs b/src/Lucene.Net.Replicator/SessionToken.cs index a4e3023df7..bce5ce3e7f 100644 --- a/src/Lucene.Net.Replicator/SessionToken.cs +++ b/src/Lucene.Net.Replicator/SessionToken.cs @@ -1,9 +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 { @@ -115,30 +117,34 @@ public void Serialize(DataOutputStream writer) } /// - /// Asynchronously serialize the token data for communication between server and client. + /// Asynchronously serializes the token's properties, including ID, version, and source files, + /// to the provided for transmission or storage. /// /// The to write the token data to. - /// A cancellation token to observe while waiting for the flush to complete. - /// A task representing the asynchronous operation. + /// A cancellation token to observe while writing and flushing the stream. + /// A task representing the asynchronous serialization operation. public async Task SerializeAsync(Stream output, CancellationToken cancellationToken = default) { - using var writer = new DataOutputStream(output); - writer.WriteUTF(Id); - writer.WriteUTF(Version); - writer.WriteInt32(SourceFiles.Count); + 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.WriteInt32Async(SourceFiles.Count, cancellationToken).ConfigureAwait(false); foreach (var pair in SourceFiles) { - writer.WriteUTF(pair.Key); - writer.WriteInt32(pair.Value.Count); + await output.WriteUTFAsync(pair.Key, cancellationToken).ConfigureAwait(false); + await output.WriteInt32Async(pair.Value.Count, cancellationToken).ConfigureAwait(false); + foreach (var file in pair.Value) { - writer.WriteUTF(file.FileName); - writer.WriteInt64(file.Length); + await output.WriteUTFAsync(file.FileName, cancellationToken).ConfigureAwait(false); + await output.WriteInt64Async(file.Length, cancellationToken).ConfigureAwait(false); } } - await output.FlushAsync(cancellationToken); + await output.FlushAsync(cancellationToken).ConfigureAwait(false); } public override string ToString() diff --git a/src/Lucene.Net/Support/IO/StreamExtensions.cs b/src/Lucene.Net/Support/IO/StreamExtensions.cs index 513d127b88..6d09395ea5 100644 --- a/src/Lucene.Net/Support/IO/StreamExtensions.cs +++ b/src/Lucene.Net/Support/IO/StreamExtensions.cs @@ -3,6 +3,9 @@ using System; using System.IO; using System.Runtime.CompilerServices; +using System.Text; +using System.Threading; +using System.Threading.Tasks; namespace Lucene.Net.Support.IO { @@ -210,5 +213,154 @@ public static long ReadInt64(this Stream stream) buff[6] << 16 | buff[7] << 24); return (long)((ulong)hi) << 32 | lo; } + + //async versions of the above methods + public static async Task WriteInt32Async(this Stream output, int value, CancellationToken cancellationToken = default) + { + if (output is null) + throw new ArgumentNullException(nameof(output)); + + byte[] buff = new byte[4]; + buff[0] = (byte)(value >> 24); + buff[1] = (byte)(value >> 16); + buff[2] = (byte)(value >> 8); + buff[3] = (byte)value; + + await output.WriteAsync(buff, 0, buff.Length, cancellationToken).ConfigureAwait(false); + } + + public static async Task WriteInt64Async(this Stream output, long value, CancellationToken cancellationToken = default) + { + if (output is null) + throw new ArgumentNullException(nameof(output)); + + byte[] buff = new byte[8]; + buff[0] = (byte)(value >> 56); + buff[1] = (byte)(value >> 48); + buff[2] = (byte)(value >> 40); + buff[3] = (byte)(value >> 32); + buff[4] = (byte)(value >> 24); + buff[5] = (byte)(value >> 16); + buff[6] = (byte)(value >> 8); + buff[7] = (byte)value; + + await output.WriteAsync(buff, 0, buff.Length, cancellationToken).ConfigureAwait(false); + } + + public static async Task WriteUTFAsync(this Stream output, string value, CancellationToken cancellationToken = default) + { + if (output is null) + throw new ArgumentNullException(nameof(output)); + if (value is null) + throw new ArgumentNullException(nameof(value)); + + long utfCount = CountUTFBytes(value); + if (utfCount > ushort.MaxValue) + throw new EncoderFallbackException("Encoded string too long."); + + byte[] buffer = new byte[(int)utfCount + 2]; + int offset = 0; + offset = WriteInt16ToBuffer((int)utfCount, buffer, offset); + offset = WriteUTFBytesToBuffer(value, (int)utfCount, buffer, offset); + + await output.WriteAsync(buffer, 0, offset, cancellationToken).ConfigureAwait(false); + } + + public static async Task ReadInt32Async(this Stream input, CancellationToken cancellationToken = default) + { + if (input is null) + throw new ArgumentNullException(nameof(input)); + + byte[] buff = new byte[4]; + int read = await input.ReadAsync(buff, 0, buff.Length, cancellationToken).ConfigureAwait(false); + if (read < buff.Length) throw new EndOfStreamException(); + + return (buff[0] << 24) | (buff[1] << 16) | (buff[2] << 8) | buff[3]; + } + + public static async Task ReadInt64Async(this Stream input, CancellationToken cancellationToken = default) + { + if (input is null) + throw new ArgumentNullException(nameof(input)); + + byte[] buff = new byte[8]; + int read = await input.ReadAsync(buff, 0, buff.Length, cancellationToken).ConfigureAwait(false); + if (read < buff.Length) throw new EndOfStreamException(); + + return ((long)buff[0] << 56) | + ((long)buff[1] << 48) | + ((long)buff[2] << 40) | + ((long)buff[3] << 32) | + ((long)buff[4] << 24) | + ((long)buff[5] << 16) | + ((long)buff[6] << 8) | + buff[7]; + } + + public static async Task ReadUTFAsync(this Stream input, CancellationToken cancellationToken = default) + { + if (input is null) + throw new ArgumentNullException(nameof(input)); + + byte[] lenBuff = new byte[2]; + int readLen = await input.ReadAsync(lenBuff, 0, lenBuff.Length, cancellationToken).ConfigureAwait(false); + if (readLen < lenBuff.Length) throw new EndOfStreamException(); + + int length = (lenBuff[0] << 8) | lenBuff[1]; + byte[] buffer = new byte[length]; + int read = await input.ReadAsync(buffer, 0, length, cancellationToken).ConfigureAwait(false); + if (read < length) + throw new EndOfStreamException("Unexpected end of stream while reading UTF string."); + + return Encoding.UTF8.GetString(buffer); + } + + // ======================== + // Helper methods for UTF + // ======================== + private static long CountUTFBytes(string value) + { + long utfCount = 0; + foreach (char ch in value) + { + if (ch > 0 && ch <= 127) + utfCount++; + else if (ch <= 2047) + utfCount += 2; + else + utfCount += 3; + } + return utfCount; + } + + private static int WriteInt16ToBuffer(int value, byte[] buffer, int offset) + { + buffer[offset++] = (byte)(value >> 8); + buffer[offset++] = (byte)value; + return offset; + } + + private static int WriteUTFBytesToBuffer(string value, long count, byte[] buffer, int offset) + { + foreach (char ch in value) + { + if (ch > 0 && ch <= 127) + { + buffer[offset++] = (byte)ch; + } + else if (ch <= 2047) + { + buffer[offset++] = (byte)(0xc0 | (0x1f & (ch >> 6))); + buffer[offset++] = (byte)(0x80 | (0x3f & ch)); + } + else + { + buffer[offset++] = (byte)(0xe0 | (0x0f & (ch >> 12))); + buffer[offset++] = (byte)(0x80 | (0x3f & (ch >> 6))); + buffer[offset++] = (byte)(0x80 | (0x3f & ch)); + } + } + return offset; + } } } diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs index b4a97f2d20..90cb172b88 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs @@ -81,10 +81,7 @@ public void Flush() /// public async Task FlushAsync(CancellationToken cancellationToken = default) { - if (response.Body.CanWrite) - { - await response.Body.FlushAsync(cancellationToken); - } + await response.Body.FlushAsync(cancellationToken); } } } From 406940f39252d193430b58e7015d04937ad5d559 Mon Sep 17 00:00:00 2001 From: NehanPathan Date: Fri, 29 Aug 2025 06:57:14 +0530 Subject: [PATCH 03/13] refactor: rename BigEndian methods and use ArrayPool for buffer allocations in StreamExtensions --- src/Lucene.Net.Replicator/SessionToken.cs | 10 +-- src/Lucene.Net/Support/IO/StreamExtensions.cs | 64 +++++++++++++------ 2 files changed, 49 insertions(+), 25 deletions(-) diff --git a/src/Lucene.Net.Replicator/SessionToken.cs b/src/Lucene.Net.Replicator/SessionToken.cs index bce5ce3e7f..d8024f4d2e 100644 --- a/src/Lucene.Net.Replicator/SessionToken.cs +++ b/src/Lucene.Net.Replicator/SessionToken.cs @@ -117,30 +117,30 @@ public void Serialize(DataOutputStream writer) } /// - /// Asynchronously serializes the token's properties, including ID, version, and source files, + /// Asynchronously serializes the token's properties, including ID, version, and source files, /// to the provided for transmission or storage. /// /// The to write the token data to. /// A cancellation token to observe while writing and flushing the stream. /// A task representing the asynchronous serialization operation. - public async Task SerializeAsync(Stream output, CancellationToken cancellationToken = default) + 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.WriteInt32Async(SourceFiles.Count, 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.WriteInt32Async(pair.Value.Count, 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.WriteInt64Async(file.Length, cancellationToken).ConfigureAwait(false); + await output.WriteInt64BigEndianAsync(file.Length, cancellationToken).ConfigureAwait(false); } } diff --git a/src/Lucene.Net/Support/IO/StreamExtensions.cs b/src/Lucene.Net/Support/IO/StreamExtensions.cs index 6d09395ea5..080a9e7b2d 100644 --- a/src/Lucene.Net/Support/IO/StreamExtensions.cs +++ b/src/Lucene.Net/Support/IO/StreamExtensions.cs @@ -1,6 +1,7 @@ using J2N.IO; using Lucene.Net.Support.Threading; using System; +using System.Buffers; using System.IO; using System.Runtime.CompilerServices; using System.Text; @@ -215,7 +216,7 @@ public static long ReadInt64(this Stream stream) } //async versions of the above methods - public static async Task WriteInt32Async(this Stream output, int value, CancellationToken cancellationToken = default) + public static async Task WriteInt32BigEndianAsync(this Stream output, int value, CancellationToken cancellationToken = default) { if (output is null) throw new ArgumentNullException(nameof(output)); @@ -229,7 +230,7 @@ public static async Task WriteInt32Async(this Stream output, int value, Cancella await output.WriteAsync(buff, 0, buff.Length, cancellationToken).ConfigureAwait(false); } - public static async Task WriteInt64Async(this Stream output, long value, CancellationToken cancellationToken = default) + public static async Task WriteInt64BigEndianAsync(this Stream output, long value, CancellationToken cancellationToken = default) { if (output is null) throw new ArgumentNullException(nameof(output)); @@ -258,15 +259,21 @@ public static async Task WriteUTFAsync(this Stream output, string value, Cancell if (utfCount > ushort.MaxValue) throw new EncoderFallbackException("Encoded string too long."); - byte[] buffer = new byte[(int)utfCount + 2]; - int offset = 0; - offset = WriteInt16ToBuffer((int)utfCount, buffer, offset); - offset = WriteUTFBytesToBuffer(value, (int)utfCount, buffer, offset); + byte[] buffer = ArrayPool.Shared.Rent((int)utfCount + 2); + try + { + int offset = 0; + offset = WriteInt16BigEndianToBuffer((int)utfCount, buffer, offset); + offset = WriteUTFBytesToBuffer(value, (int)utfCount, buffer, offset); - await output.WriteAsync(buffer, 0, offset, cancellationToken).ConfigureAwait(false); + await output.WriteAsync(buffer, 0, offset, cancellationToken).ConfigureAwait(false); + } + finally + { + ArrayPool.Shared.Return(buffer); + } } - - public static async Task ReadInt32Async(this Stream input, CancellationToken cancellationToken = default) + public static async Task ReadInt32BigEndianAsync(this Stream input, CancellationToken cancellationToken = default) { if (input is null) throw new ArgumentNullException(nameof(input)); @@ -278,7 +285,7 @@ public static async Task ReadInt32Async(this Stream input, CancellationToke return (buff[0] << 24) | (buff[1] << 16) | (buff[2] << 8) | buff[3]; } - public static async Task ReadInt64Async(this Stream input, CancellationToken cancellationToken = default) + public static async Task ReadInt64BigEndianAsync(this Stream input, CancellationToken cancellationToken = default) { if (input is null) throw new ArgumentNullException(nameof(input)); @@ -302,19 +309,36 @@ public static async Task ReadUTFAsync(this Stream input, CancellationTok if (input is null) throw new ArgumentNullException(nameof(input)); - byte[] lenBuff = new byte[2]; - int readLen = await input.ReadAsync(lenBuff, 0, lenBuff.Length, cancellationToken).ConfigureAwait(false); - if (readLen < lenBuff.Length) throw new EndOfStreamException(); + byte[] lenBuff = ArrayPool.Shared.Rent(2); + try + { + int readLen = await input.ReadAsync(lenBuff, 0, 2, cancellationToken).ConfigureAwait(false); + if (readLen < 2) + throw new EndOfStreamException("Unexpected end of stream while reading UTF length."); - int length = (lenBuff[0] << 8) | lenBuff[1]; - byte[] buffer = new byte[length]; - int read = await input.ReadAsync(buffer, 0, length, cancellationToken).ConfigureAwait(false); - if (read < length) - throw new EndOfStreamException("Unexpected end of stream while reading UTF string."); + int length = (lenBuff[0] << 8) | lenBuff[1]; - return Encoding.UTF8.GetString(buffer); + byte[] buffer = ArrayPool.Shared.Rent(length); + try + { + int read = await input.ReadAsync(buffer, 0, length, cancellationToken).ConfigureAwait(false); + if (read < length) + throw new EndOfStreamException("Unexpected end of stream while reading UTF string."); + + return Encoding.UTF8.GetString(buffer, 0, length); + } + finally + { + ArrayPool.Shared.Return(buffer); + } + } + finally + { + ArrayPool.Shared.Return(lenBuff); + } } + // ======================== // Helper methods for UTF // ======================== @@ -333,7 +357,7 @@ private static long CountUTFBytes(string value) return utfCount; } - private static int WriteInt16ToBuffer(int value, byte[] buffer, int offset) + private static int WriteInt16BigEndianToBuffer(int value, byte[] buffer, int offset) { buffer[offset++] = (byte)(value >> 8); buffer[offset++] = (byte)value; From 0276e102c4198c055f4d4b7c2ee00658673475d6 Mon Sep 17 00:00:00 2001 From: NehanPathan Date: Sat, 30 Aug 2025 00:12:58 +0530 Subject: [PATCH 04/13] feat: Add async read/write support in ReplicationServlet and corresponding J2N-based async tests in TestStreamExtensions --- .../Http/ReplicationServlet.cs | 31 +++--- .../Support/IO/TestStreamExtensions.cs | 95 +++++++++++++++++++ .../AspNetCoreReplicationServiceExtentions.cs | 18 ++++ 3 files changed, 130 insertions(+), 14 deletions(-) diff --git a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs index 8146470a66..6ef6fd2cd1 100644 --- a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs +++ b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs @@ -52,14 +52,16 @@ public void Configure(IApplicationBuilder app, IReplicationService service, Repl { // LUCENENET: This is to allow synchronous IO to happen for these requests. // LUCENENET TODO: Allow async operations from Replicator. - var syncIoFeature = context.Features.Get(); - if (syncIoFeature != null) - { - syncIoFeature.AllowSynchronousIO = true; - } + // var syncIoFeature = context.Features.Get(); + // if (syncIoFeature != null) + // { + // syncIoFeature.AllowSynchronousIO = false; + // } + + // await Task.Yield(); + // service.Perform(context.Request, context.Response); + await service.PerformAsync(context.Request, context.Response, context.RequestAborted); - await Task.Yield(); - service.Perform(context.Request, context.Response); }); } } @@ -84,14 +86,15 @@ public async Task InvokeAsync(HttpContext context) { // LUCENENET: This is to allow synchronous IO to happen for these requests. // LUCENENET TODO: Allow async operations from Replicator. - var syncIoFeature = context.Features.Get(); - if (syncIoFeature != null) - { - syncIoFeature.AllowSynchronousIO = true; - } + // var syncIoFeature = context.Features.Get(); + // if (syncIoFeature != null) + // { + // syncIoFeature.AllowSynchronousIO = false; + // } - await Task.Yield(); - service.Perform(context.Request, context.Response); + // await Task.Yield(); + // service.Perform(context.Request, context.Response); + await service.PerformAsync(context.Request, context.Response, context.RequestAborted); // This is a terminating endpoint. Do not call the next delegate/middleware in the pipeline. } diff --git a/src/Lucene.Net.Tests/Support/IO/TestStreamExtensions.cs b/src/Lucene.Net.Tests/Support/IO/TestStreamExtensions.cs index 597f1ed4e4..f66482458b 100644 --- a/src/Lucene.Net.Tests/Support/IO/TestStreamExtensions.cs +++ b/src/Lucene.Net.Tests/Support/IO/TestStreamExtensions.cs @@ -8,6 +8,7 @@ using System.Diagnostics.CodeAnalysis; using System.IO; using System.Text; +using System.Threading.Tasks; namespace Lucene.Net.Support.IO { @@ -33,6 +34,8 @@ public class TestStreamExtensions : LuceneTestCase { private Stream stream; + private static readonly string unihw = "\u0048\u0065\u006C\u006C\u006F\u0020\u0057\u006F\u0072\u006C\u0064"; + private const string fileString = "Test_All_Tests\nTest_java_io_BufferedInputStream\nTest_java_io_BufferedOutputStream\nTest_java_io_ByteArrayInputStream\nTest_java_io_ByteArrayOutputStream\nTest_DataInputStream\n"; [Test] @@ -115,6 +118,98 @@ public void TestReadInt64() Assert.AreEqual(9875645283333L, stream.ReadInt64(), "Incorrect long read"); } + // Additional async tests + + [Test] + // LUCENENET note: adapted from test_writeInt() + public async Task TestWriteInt32BigEndianAsync() + { + await stream.WriteInt32BigEndianAsync(9087589); + // Reset the stream so we can read back + ResetStreamForReading(); + int c = await stream.ReadInt32BigEndianAsync(); + Assert.AreEqual(9087589, c, "Incorrect int written (async)"); + } + + [Test] + // LUCENENET note: adapted from test_writeLong() + public async Task TestWriteInt64BigEndianAsync() + { + await stream.WriteInt64BigEndianAsync(908755555456L); + // Reset the stream so we can read back + ResetStreamForReading(); + long c = await stream.ReadInt64BigEndianAsync(); + Assert.AreEqual(908755555456L, c, "Incorrect long written (async)"); + } + + [Test] + // LUCENENET note: adapted from test_writeUTF() + public async Task TestWriteUTFAsync() + { + await stream.WriteUTFAsync(unihw); + // Reset the stream so we can read back + ResetStreamForReading(); + string result = await stream.ReadUTFAsync(); + Assert.AreEqual(unihw, result, "Incorrect string written (async)"); + } + + [Test] + // LUCENENET note: adapted from test_readInt() + public async Task TestReadInt32BigEndianAsync() + { + await stream.WriteInt32BigEndianAsync(768347202); + // Reset the stream so we can read back + ResetStreamForReading(); + int result = await stream.ReadInt32BigEndianAsync(); + Assert.AreEqual(768347202, result, "Incorrect int read (async)"); + } + + [Test] + // LUCENENET note: adapted from test_readLong() + public async Task TestReadInt64BigEndianAsync() + { + await stream.WriteInt64BigEndianAsync(9875645283333L); + // Reset the stream so we can read back + ResetStreamForReading(); + long result = await stream.ReadInt64BigEndianAsync(); + Assert.AreEqual(9875645283333L, result, "Incorrect long read (async)"); + } + + [Test] + // LUCENENET note: adapted from test_readUTF() + public async Task TestReadUTFAsync() + { + await stream.WriteUTFAsync(unihw); + + // Check that the length was written correctly (UTF length + 2 bytes for length header) + long expectedStreamLength = CalculateExpectedUTFStreamLength(unihw); + Assert.AreEqual(expectedStreamLength, stream.Length, "Failed to write string in UTF format"); + + // Reset and read the string + ResetStreamForReading(); + string result = await stream.ReadUTFAsync(); + Assert.AreEqual(unihw, result, "Incorrect string read (async)"); + } + + /// + /// Helper method to calculate expected UTF stream length for validation + /// Matches DataOutput.writeUTF() spec (Java) + /// + private long CalculateExpectedUTFStreamLength(string value) + { + long utfCount = 0; + foreach (char ch in value) + { + if (ch > 0 && ch <= 127) + utfCount++; + else if (ch <= 2047) + utfCount += 2; + else + utfCount += 3; + } + return utfCount + 2; // +2 for the 2-byte length header + } + private void ResetStreamForReading() // LUCENENET - was "OpenDataInputStream" in Harmony tests { // LUCENENET specific - in the Harmony tests, there were separate streams diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs index b79a8545ee..86966fe552 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs @@ -1,6 +1,8 @@ using Lucene.Net.Replicator.Http; using Lucene.Net.Replicator.Http.Abstractions; using Microsoft.AspNetCore.Http; +using System.Threading; +using System.Threading.Tasks; namespace Lucene.Net.Replicator.AspNetCore { @@ -27,9 +29,25 @@ public static class AspNetCoreReplicationServiceExtentions /// /// Extension method that mirrors the signature of using AspNetCore as implementation. /// + // public static void Perform(this IReplicationService self, HttpRequest request, HttpResponse response) + // { + // self.Perform(new AspNetCoreReplicationRequest(request), new AspNetCoreReplicationResponse(response)); + // } public static void Perform(this IReplicationService self, HttpRequest request, HttpResponse response) { self.Perform(new AspNetCoreReplicationRequest(request), new AspNetCoreReplicationResponse(response)); } + + public static async Task PerformAsync( + this IReplicationService self, + HttpRequest request, + HttpResponse response, + CancellationToken cancellationToken = default) + { + await self.PerformAsync( + new AspNetCoreReplicationRequest(request), + new AspNetCoreReplicationResponse(response), + cancellationToken); + } } } From 841545793de5afb1451d7c26efc92bc58bcc358c Mon Sep 17 00:00:00 2001 From: NehanPathan Date: Sat, 30 Aug 2025 00:18:48 +0530 Subject: [PATCH 05/13] Add XML Summary For PerformAsync and remove dupilcate comment code --- .../AspNetCoreReplicationServiceExtentions.cs | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs index 86966fe552..ce87aa0f84 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs @@ -29,20 +29,18 @@ public static class AspNetCoreReplicationServiceExtentions /// /// Extension method that mirrors the signature of using AspNetCore as implementation. /// - // public static void Perform(this IReplicationService self, HttpRequest request, HttpResponse response) - // { - // self.Perform(new AspNetCoreReplicationRequest(request), new AspNetCoreReplicationResponse(response)); - // } public static void Perform(this IReplicationService self, HttpRequest request, HttpResponse response) { self.Perform(new AspNetCoreReplicationRequest(request), new AspNetCoreReplicationResponse(response)); } - + /// + /// Extension method that mirrors the signature of using AspNetCore as implementation. + /// public static async Task PerformAsync( - this IReplicationService self, - HttpRequest request, - HttpResponse response, - CancellationToken cancellationToken = default) + this IReplicationService self, + HttpRequest request, + HttpResponse response, + CancellationToken cancellationToken = default) { await self.PerformAsync( new AspNetCoreReplicationRequest(request), From 4cace1faff98b65b849f78331620c1492baf3920 Mon Sep 17 00:00:00 2001 From: Shad Storhaug Date: Sun, 31 Aug 2025 00:47:54 +0700 Subject: [PATCH 06/13] 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). --- .../Http/HttpReplicatorTest.cs | 37 ++++- .../Http/ReplicationServlet.cs | 38 ++--- .../ReplicatorTestCase.cs | 154 +++++++++++------- .../Http/SynchronousReplicationServlet.cs | 104 ++++++++++++ 4 files changed, 242 insertions(+), 91 deletions(-) create mode 100644 src/Lucene.Net.Tests.Replicator/Support/Http/SynchronousReplicationServlet.cs diff --git a/src/Lucene.Net.Tests.Replicator/Http/HttpReplicatorTest.cs b/src/Lucene.Net.Tests.Replicator/Http/HttpReplicatorTest.cs index a5106e1979..6ff0dcacc2 100644 --- a/src/Lucene.Net.Tests.Replicator/Http/HttpReplicatorTest.cs +++ b/src/Lucene.Net.Tests.Replicator/Http/HttpReplicatorTest.cs @@ -29,6 +29,12 @@ namespace Lucene.Net.Replicator.Http * limitations under the License. */ + [TestFixture(IOOption.Synchronous, ConfigOption.StartupClass)] + [TestFixture(IOOption.Asynchronous, ConfigOption.StartupClass)] +#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG + [TestFixture(IOOption.Synchronous, ConfigOption.Middleware)] + [TestFixture(IOOption.Asynchronous, ConfigOption.Middleware)] +#endif public class HttpReplicatorTest : ReplicatorTestCase { private DirectoryInfo clientWorkDir; @@ -45,16 +51,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 { { "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(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 { { "s1", serverReplicator } }); + server = NewHttpServer(service, mockErrorConfig, useSynchronousIO, useStartupClass); port = ServerPort(server); host = ServerHost(server); } @@ -144,7 +165,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(); } diff --git a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs index 6ef6fd2cd1..2b6bedec99 100644 --- a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs +++ b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs @@ -1,14 +1,13 @@ using Lucene.Net.Replicator.AspNetCore; using Lucene.Net.Replicator.Http.Abstractions; using Microsoft.AspNetCore.Builder; -using Microsoft.AspNetCore.Http.Features; using System; using System.Threading.Tasks; #if FEATURE_ASPNETCORE_ENDPOINT_CONFIG +using Microsoft.AspNetCore.Routing; using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Http; -using Microsoft.AspNetCore.Routing; #endif namespace Lucene.Net.Replicator.Http @@ -31,7 +30,7 @@ namespace Lucene.Net.Replicator.Http */ // ********************** Option 1: Use a Startup Class ******************************************** - // The startup class must define all middleware (app.Use) before the terminating endpoint (app.Run) + // The startup class must define all middleware (app.Use) before the terminating endpoint (app.Run). public class ReplicationServlet { @@ -50,18 +49,11 @@ public void Configure(IApplicationBuilder app, IReplicationService service, Repl app.Run(async (context) => { - // LUCENENET: This is to allow synchronous IO to happen for these requests. - // LUCENENET TODO: Allow async operations from Replicator. - // var syncIoFeature = context.Features.Get(); - // if (syncIoFeature != null) - // { - // syncIoFeature.AllowSynchronousIO = false; - // } - - // await Task.Yield(); - // service.Perform(context.Request, context.Response); + // LUCENENET: Although the async/await pattern doesn't exist in Java Lucene, this is the recommended + // approach for modern .NET development. await service.PerformAsync(context.Request, context.Response, context.RequestAborted); + // This is a terminating endpoint. Do not call the next delegate/middleware in the pipeline. }); } } @@ -84,28 +76,20 @@ public ReplicationServiceMiddleware(RequestDelegate next, IReplicationService se public async Task InvokeAsync(HttpContext context) { - // LUCENENET: This is to allow synchronous IO to happen for these requests. - // LUCENENET TODO: Allow async operations from Replicator. - // var syncIoFeature = context.Features.Get(); - // if (syncIoFeature != null) - // { - // syncIoFeature.AllowSynchronousIO = false; - // } - - // await Task.Yield(); - // service.Perform(context.Request, context.Response); - await service.PerformAsync(context.Request, context.Response, context.RequestAborted); + // LUCENENET: Although the async/await pattern doesn't exist in Java Lucene, this is the recommended + // approach for modern .NET development. + await service.PerformAsync(context.Request, context.Response, context.RequestAborted); // This is a terminating endpoint. Do not call the next delegate/middleware in the pipeline. } } - public static class ReplicationServiceRouteBuilderExtensions + public static partial class ReplicationServiceRouteBuilderExtensions { - public static IEndpointConventionBuilder MapReplicator(this IEndpointRouteBuilder endpoints, string pattern) + public static IEndpointConventionBuilder MapReplicator(this IEndpointRouteBuilder endpoints, string pattern) where TReplicationServiceMiddleware : class { var pipeline = endpoints.CreateApplicationBuilder() - .UseMiddleware() + .UseMiddleware() .Build(); return endpoints diff --git a/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs b/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs index fe0c7d60bd..1cfe9712fe 100644 --- a/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs +++ b/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs @@ -3,6 +3,7 @@ using Lucene.Net.Util; using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.Features; using Microsoft.AspNetCore.TestHost; using Microsoft.Extensions.DependencyInjection; using System; @@ -10,6 +11,7 @@ #if FEATURE_ASPNETCORE_ENDPOINT_CONFIG using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Routing; #endif namespace Lucene.Net.Replicator @@ -34,73 +36,88 @@ namespace Lucene.Net.Replicator [SuppressCodecs("Lucene3x")] public class ReplicatorTestCase : LuceneTestCase { - -#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG - /// - /// Call this overload to use to host . - /// - /// The that will be registered as singleton. - /// The that will be registered as singleton. - /// A configured instance. - public static TestServer NewHttpServer(IReplicationService service, MockErrorConfig mockErrorConfig) + public static TestServer NewHttpServer(IReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) { - var builder = new WebHostBuilder() - .ConfigureServices(container => + if (useStartupClass) + { + var builder = new WebHostBuilder() + .ConfigureServices(container => + { + container.AddSingleton(service); + container.AddSingleton(mockErrorConfig); + }); + + if (useSynchronousIO) { - container.AddRouting(); - container.AddSingleton(service); - container.AddSingleton(mockErrorConfig); - container.AddSingleton(); - container.AddSingleton(); - }) - .Configure(app => + builder.UseStartup(); + } + else { - app.UseRouting(); - - // Middleware so we can mock a server exception and toggle the exception on and off. - app.UseMiddleware(); + builder.UseStartup(); + } - app.UseEndpoints(endpoints => + var server = new TestServer(builder); + server.BaseAddress = new Uri("http://localhost" + ReplicationService.REPLICATION_CONTEXT); + return server; + } + else + { +#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG + var builder = new WebHostBuilder() + .ConfigureServices(container => + { + container.AddRouting(); + container.AddSingleton(service); + container.AddSingleton(mockErrorConfig); + if (useSynchronousIO) + { + container.AddSingleton(); + container.AddSingleton(); + } + else + { + container.AddSingleton(); + } + container.AddSingleton(); + }) + .Configure(app => { - // This is to define the endpoint for Replicator. - // All URLs with the pattern /replicate/{shard?}/{action?} terminate here and any middleware that - // is expected to run for Replicator must be registered before this call. - endpoints.MapReplicator(ReplicationService.REPLICATION_CONTEXT + "/{shard?}/{action?}"); + app.UseRouting(); - endpoints.MapGet("/{controller?}/{action?}/{id?}", async context => + // Middleware so we can mock a server exception and toggle the exception on and off. + app.UseMiddleware(); + + if (useSynchronousIO) + { + app.UseMiddleware(); + } + + app.UseEndpoints(endpoints => { - // This is just to demonstrate allowing requests to other services/controllers in the same - // application. This isn't required, but is allowed. - await context.Response.WriteAsync("Hello World!"); + // This is to define the endpoint for Replicator. + // All URLs with the pattern /replicate/{shard?}/{action?} terminate here and any middleware that + // is expected to run for Replicator must be registered before this call. + string pattern = ReplicationService.REPLICATION_CONTEXT + "/{shard?}/{action?}"; + if (useSynchronousIO) + endpoints.MapReplicator(pattern); + else + endpoints.MapReplicator(pattern); + + endpoints.MapGet("/{controller?}/{action?}/{id?}", async context => + { + // This is just to demonstrate allowing requests to other services/controllers in the same + // application. This isn't required, but is allowed. + await context.Response.WriteAsync("Hello World!"); + }); }); }); - }); - var server = new TestServer(builder); - return server; - } + var server = new TestServer(builder); + return server; #else - /// - /// Call this overload to use as the Startup Class. - /// - /// The type of startup class. - /// The that will be registered as singleton. - /// The that will be registered as singleton. - /// A configured instance. - public static TestServer NewHttpServer(IReplicationService service, MockErrorConfig mockErrorConfig) where TStartUp : class - { - var builder = new WebHostBuilder() - .ConfigureServices(container => - { - container.AddSingleton(service); - container.AddSingleton(mockErrorConfig); - }) - .UseStartup(); - - var server = new TestServer(builder); - server.BaseAddress = new Uri("http://localhost" + ReplicationService.REPLICATION_CONTEXT); - return server; - } + throw new PlatformNotSupportedException("Endpoint configuration is not supported prior to .NET 5.0"); #endif + } + } /// /// Returns a 's port. @@ -164,5 +181,30 @@ public class MockErrorConfig { public bool RespondWithError { get; set; } = false; } + + public class EnableSynchronousIOMiddleware + { + private readonly RequestDelegate next; + + public EnableSynchronousIOMiddleware(RequestDelegate next) + { + this.next = next ?? throw new ArgumentNullException(nameof(next)); + } + + public async Task InvokeAsync(HttpContext context) + { + // LUCENENET: This is to allow synchronous IO to happen for these requests. + // Note that in a real-world app this would be set in the configuration, not + // per HTTP request. However, this setting is not recommended in modern production + // applications. + var syncIoFeature = context.Features.Get(); + if (syncIoFeature != null) + { + syncIoFeature.AllowSynchronousIO = true; + } + + await next(context); + } + } } } diff --git a/src/Lucene.Net.Tests.Replicator/Support/Http/SynchronousReplicationServlet.cs b/src/Lucene.Net.Tests.Replicator/Support/Http/SynchronousReplicationServlet.cs new file mode 100644 index 0000000000..ff66f2f326 --- /dev/null +++ b/src/Lucene.Net.Tests.Replicator/Support/Http/SynchronousReplicationServlet.cs @@ -0,0 +1,104 @@ +using Lucene.Net.Replicator.AspNetCore; +using Lucene.Net.Replicator.Http.Abstractions; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Http.Features; +using System; +using System.Threading.Tasks; + +#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http; +#endif + +namespace Lucene.Net.Replicator.Http +{ + /* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + // IMPORTANT: These tests are primarily to test the synchronous methods that are a direct port from Java. + // However, modern ASP.NET development is primarily done asynchronously, so it is recommended to follow + // the patterns in the ReplicationServlet.cs class rather than this one in most cases. + + // ********************** Option 1: Use a Startup Class ******************************************** + // The startup class must define all middleware (app.Use) before the terminating endpoint (app.Run) + + public class SynchronousReplicationServlet + { + public void Configure(IApplicationBuilder app, IReplicationService service, ReplicatorTestCase.MockErrorConfig mockErrorConfig) + { + // Middleware to throw an exception conditionally from our test server. + app.Use(async (context, next) => + { + if (mockErrorConfig.RespondWithError) + { + throw new ReplicatorTestCase.HttpResponseException(); + } + + await next(); + }); + app.Use(async (context, next) => + { + // LUCENENET: This is to allow synchronous IO to happen for these requests. + var syncIoFeature = context.Features.Get(); + if (syncIoFeature != null) + { + syncIoFeature.AllowSynchronousIO = true; + } + + await next(); + }); + + + app.Run(async (context) => + { + await Task.Yield(); + service.Perform(context.Request, context.Response); + + // This is a terminating endpoint. Do not call the next delegate/middleware in the pipeline. + + }); + } + } + + // ********************** Option 2: Use Middleware with Endpoint Routing ******************************* + // Running ReplicationService as middleware allows registering other URL patterns so other services + // (such as controllers or razor pages) can be served from the same application. + +#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG // Only available in .NET 5+ + public class SynchronousReplicationServiceMiddleware + { + private readonly RequestDelegate next; + private readonly IReplicationService service; + + public SynchronousReplicationServiceMiddleware(RequestDelegate next, IReplicationService service) + { + this.next = next ?? throw new ArgumentNullException(nameof(next)); + this.service = service ?? throw new ArgumentNullException(nameof(service)); + } + + public async Task InvokeAsync(HttpContext context) + { + // NOTE: SynchronousIO enabled by middleware + + await Task.Yield(); + service.Perform(context.Request, context.Response); + + // This is a terminating endpoint. Do not call the next delegate/middleware in the pipeline. + } + } +#endif +} From a52d05696cdaa2d7af152cf5290098ba3870b7a5 Mon Sep 17 00:00:00 2001 From: Shad Storhaug Date: Sun, 31 Aug 2025 03:45:07 +0700 Subject: [PATCH 07/13] 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. --- .build/dependencies.props | 8 +- Directory.Build.targets | 2 +- .../Http/HttpReplicatorTest.cs | 12 +- .../Http/ReplicationServlet.cs | 11 +- .../Lucene.Net.Tests.Replicator.csproj | 7 +- .../ReplicatorTestCase.cs | 143 ++++++++++-------- .../Http/SynchronousReplicationServlet.cs | 11 +- .../Net/HttpListenerReplicationRequest.cs | 37 +++++ .../Net/HttpListenerReplicationResponse.cs | 49 ++++++ .../Support/Net/TestServer.cs | 140 +++++++++++++++++ 10 files changed, 328 insertions(+), 92 deletions(-) create mode 100644 src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationRequest.cs create mode 100644 src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs create mode 100644 src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs diff --git a/.build/dependencies.props b/.build/dependencies.props index f3c8dda3fa..1620ceff1b 100644 --- a/.build/dependencies.props +++ b/.build/dependencies.props @@ -1,4 +1,4 @@ - + $(DefineConstants);FEATURE_ASPNETCORE_TESTHOST $(DefineConstants);FEATURE_UTF8_TOUTF16 @@ -56,7 +57,6 @@ - $(DefineConstants);FEATURE_ASPNETCORE_ENDPOINT_CONFIG $(DefineConstants);FEATURE_READONLYSET diff --git a/src/Lucene.Net.Tests.Replicator/Http/HttpReplicatorTest.cs b/src/Lucene.Net.Tests.Replicator/Http/HttpReplicatorTest.cs index 6ff0dcacc2..3466007883 100644 --- a/src/Lucene.Net.Tests.Replicator/Http/HttpReplicatorTest.cs +++ b/src/Lucene.Net.Tests.Replicator/Http/HttpReplicatorTest.cs @@ -2,7 +2,6 @@ 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; @@ -10,6 +9,13 @@ 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,9 +35,11 @@ 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_ENDPOINT_CONFIG +#if FEATURE_ASPNETCORE_TESTHOST [TestFixture(IOOption.Synchronous, ConfigOption.Middleware)] [TestFixture(IOOption.Asynchronous, ConfigOption.Middleware)] #endif diff --git a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs index 2b6bedec99..e9775a63f7 100644 --- a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs +++ b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs @@ -1,14 +1,12 @@ +#if FEATURE_ASPNETCORE_TESTHOST using Lucene.Net.Replicator.AspNetCore; using Lucene.Net.Replicator.Http.Abstractions; using Microsoft.AspNetCore.Builder; -using System; -using System.Threading.Tasks; - -#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG using Microsoft.AspNetCore.Routing; using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Http; -#endif +using System; +using System.Threading.Tasks; namespace Lucene.Net.Replicator.Http { @@ -62,7 +60,6 @@ public void Configure(IApplicationBuilder app, IReplicationService service, Repl // Running ReplicationService as middleware allows registering other URL patterns so other services // (such as controllers or razor pages) can be served from the same application. -#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG // Only available in .NET 5+ public class ReplicationServiceMiddleware { private readonly RequestDelegate next; @@ -97,5 +94,5 @@ public static IEndpointConventionBuilder MapReplicator - - - - - + + diff --git a/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs b/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs index 1cfe9712fe..4a80cdbb78 100644 --- a/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs +++ b/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs @@ -1,19 +1,22 @@ using Lucene.Net.Replicator.Http; using Lucene.Net.Replicator.Http.Abstractions; using Lucene.Net.Util; -using Microsoft.AspNetCore.Hosting; -using Microsoft.AspNetCore.Http; -using Microsoft.AspNetCore.Http.Features; -using Microsoft.AspNetCore.TestHost; -using Microsoft.Extensions.DependencyInjection; using System; using System.Threading.Tasks; -#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG +#if FEATURE_ASPNETCORE_TESTHOST using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Http.Features; using Microsoft.AspNetCore.Routing; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.DependencyInjection; +#else +using Lucene.Net.Replicator.Net; #endif + namespace Lucene.Net.Replicator { /* @@ -34,7 +37,49 @@ namespace Lucene.Net.Replicator */ [SuppressCodecs("Lucene3x")] - public class ReplicatorTestCase : LuceneTestCase + public partial class ReplicatorTestCase : LuceneTestCase + { + // LUCENENET: Moved NewHttpServer() implementations into partial classes for different server stacks + + /// + /// Returns a 's port. + /// + public static int ServerPort(TestServer server) + { + return server.BaseAddress.Port; + } + + /// + /// Returns a 's host. + /// + public static string ServerHost(TestServer server) + { + return server.BaseAddress.Host; + } + + /// + /// Stops the given HTTP Server instance. + /// + public static void StopHttpServer(TestServer server) + { + server.Dispose(); + } + + public class HttpResponseException : Exception + { + public int Status { get; set; } = 500; + + public object Value { get; set; } + } + + public class MockErrorConfig + { + public bool RespondWithError { get; set; } = false; + } + } + +#if FEATURE_ASPNETCORE_TESTHOST + public partial class ReplicatorTestCase { public static TestServer NewHttpServer(IReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) { @@ -62,7 +107,6 @@ public static TestServer NewHttpServer(IReplicationService service, MockErrorCon } else { -#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG var builder = new WebHostBuilder() .ConfigureServices(container => { @@ -113,41 +157,32 @@ public static TestServer NewHttpServer(IReplicationService service, MockErrorCon }); var server = new TestServer(builder); return server; -#else - throw new PlatformNotSupportedException("Endpoint configuration is not supported prior to .NET 5.0"); -#endif } } - /// - /// Returns a 's port. - /// - public static int ServerPort(TestServer server) - { - return server.BaseAddress.Port; - } - - /// - /// Returns a 's host. - /// - public static string ServerHost(TestServer server) + public class EnableSynchronousIOMiddleware { - return server.BaseAddress.Host; - } + private readonly RequestDelegate next; - /// - /// Stops the given HTTP Server instance. - /// - public static void StopHttpServer(TestServer server) - { - server.Dispose(); - } + public EnableSynchronousIOMiddleware(RequestDelegate next) + { + this.next = next ?? throw new ArgumentNullException(nameof(next)); + } - public class HttpResponseException : Exception - { - public int Status { get; set; } = 500; + public async Task InvokeAsync(HttpContext context) + { + // LUCENENET: This is to allow synchronous IO to happen for these requests. + // Note that in a real-world app this would be set in the configuration, not + // per HTTP request. However, this setting is not recommended in modern production + // applications. + var syncIoFeature = context.Features.Get(); + if (syncIoFeature != null) + { + syncIoFeature.AllowSynchronousIO = true; + } - public object Value { get; set; } + await next(context); + } } public class MockErrorMiddleware @@ -176,35 +211,15 @@ public async Task InvokeAsync(HttpContext context) await next(context); } } - - public class MockErrorConfig - { - public bool RespondWithError { get; set; } = false; - } - - public class EnableSynchronousIOMiddleware + } +#else + public partial class ReplicatorTestCase + { + // LUCENENET: This uses HttpListener from System.Net to test the service where ASP.NET Core is not supported. + public static TestServer NewHttpServer(IReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) { - private readonly RequestDelegate next; - - public EnableSynchronousIOMiddleware(RequestDelegate next) - { - this.next = next ?? throw new ArgumentNullException(nameof(next)); - } - - public async Task InvokeAsync(HttpContext context) - { - // LUCENENET: This is to allow synchronous IO to happen for these requests. - // Note that in a real-world app this would be set in the configuration, not - // per HTTP request. However, this setting is not recommended in modern production - // applications. - var syncIoFeature = context.Features.Get(); - if (syncIoFeature != null) - { - syncIoFeature.AllowSynchronousIO = true; - } - - await next(context); - } + return new TestServer(service, mockErrorConfig, useSynchronousIO); } } +#endif } diff --git a/src/Lucene.Net.Tests.Replicator/Support/Http/SynchronousReplicationServlet.cs b/src/Lucene.Net.Tests.Replicator/Support/Http/SynchronousReplicationServlet.cs index ff66f2f326..9efbe56726 100644 --- a/src/Lucene.Net.Tests.Replicator/Support/Http/SynchronousReplicationServlet.cs +++ b/src/Lucene.Net.Tests.Replicator/Support/Http/SynchronousReplicationServlet.cs @@ -1,15 +1,13 @@ +#if FEATURE_ASPNETCORE_TESTHOST using Lucene.Net.Replicator.AspNetCore; using Lucene.Net.Replicator.Http.Abstractions; using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.Http; using Microsoft.AspNetCore.Http.Features; using System; using System.Threading.Tasks; -#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG -using Microsoft.AspNetCore.Hosting; -using Microsoft.AspNetCore.Http; -#endif - namespace Lucene.Net.Replicator.Http { /* @@ -78,7 +76,6 @@ public void Configure(IApplicationBuilder app, IReplicationService service, Repl // Running ReplicationService as middleware allows registering other URL patterns so other services // (such as controllers or razor pages) can be served from the same application. -#if FEATURE_ASPNETCORE_ENDPOINT_CONFIG // Only available in .NET 5+ public class SynchronousReplicationServiceMiddleware { private readonly RequestDelegate next; @@ -100,5 +97,5 @@ public async Task InvokeAsync(HttpContext context) // This is a terminating endpoint. Do not call the next delegate/middleware in the pipeline. } } -#endif } +#endif diff --git a/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationRequest.cs b/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationRequest.cs new file mode 100644 index 0000000000..be1d772c79 --- /dev/null +++ b/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationRequest.cs @@ -0,0 +1,37 @@ +using Lucene.Net.Replicator.Http.Abstractions; +using System.Net; + +namespace Lucene.Net.Replicator.Net +{ + /* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + /// + /// A concrete implementation of for supporting + /// . + /// + public class HttpListenerReplicationRequest : IReplicationRequest + { + private readonly HttpListenerRequest _request; + + public HttpListenerReplicationRequest(HttpListenerRequest request) => _request = request; + + public string Path => _request.Url.AbsolutePath; + + public string QueryParam(string name) => _request.QueryString[name]; + } +} diff --git a/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs b/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs new file mode 100644 index 0000000000..9cc1280a4e --- /dev/null +++ b/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs @@ -0,0 +1,49 @@ +using Lucene.Net.Replicator.Http.Abstractions; +using System.IO; +using System.Net; +using System.Threading; +using System.Threading.Tasks; + +namespace Lucene.Net.Replicator.Net +{ + /* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + /// + /// A concrete implementation of for supporting + /// . + /// + public class HttpListenerReplicationResponse : IReplicationResponse + { + private readonly HttpListenerResponse _response; + + public HttpListenerReplicationResponse(HttpListenerResponse response) => _response = response; + + public int StatusCode + { + get => _response.StatusCode; + set => _response.StatusCode = value; + } + + public Stream Body => _response.OutputStream; + + public void Flush() => _response.OutputStream.Flush(); + + public Task FlushAsync(CancellationToken cancellationToken = default) => + _response.OutputStream.FlushAsync(cancellationToken); + } +} diff --git a/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs b/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs new file mode 100644 index 0000000000..695fab54cf --- /dev/null +++ b/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs @@ -0,0 +1,140 @@ +using Lucene.Net.Replicator.Http.Abstractions; +using System; +using System.Net; +using System.Net.Http; +using System.Net.Sockets; +using System.Text; +using System.Threading; +using System.Threading.Tasks; + +namespace Lucene.Net.Replicator.Net +{ + /* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + /// + /// A simple -based test server with an API modeled + /// after Microsoft.AspNetCore.TestHost.TestServer so we can easily swap this + /// implementation in its place. + /// + public class TestServer : IDisposable + { + private readonly HttpListener _listener; + private readonly IReplicationService _service; + private readonly ReplicatorTestCase.MockErrorConfig _mockErrorConfig; + private readonly bool _useSynchronousIO; + private readonly CancellationTokenSource _cancellationTokenSource = new(); + private readonly Task _serverTask; + + public Uri BaseAddress { get; } + + public TestServer( + IReplicationService service, + ReplicatorTestCase.MockErrorConfig mockErrorConfig, + bool useSynchronousIO, + string prefix = "http://localhost:0/") + { + _service = service ?? throw new ArgumentNullException(nameof(service)); + _mockErrorConfig = mockErrorConfig ?? throw new ArgumentNullException(nameof(mockErrorConfig)); + _useSynchronousIO = useSynchronousIO; + + // Auto-select a free port if prefix ends with 0 + if (prefix.EndsWith("0/", StringComparison.Ordinal)) + { + int port = GetFreePort(); + prefix = prefix.Replace("0/", port + "/"); + } + + _listener = new HttpListener(); + _listener.Prefixes.Add(prefix); + BaseAddress = new Uri(prefix); + _listener.Start(); + + // Start listening loop + _serverTask = Task.Run(() => ListenLoopAsync(_cancellationTokenSource.Token)); + } + + private async Task ListenLoopAsync(CancellationToken token) + { + try + { + while (!token.IsCancellationRequested) + { + var context = await _listener.GetContextAsync(); + + // Handle each request in background + _ = Task.Run(async () => + { + var request = new HttpListenerReplicationRequest(context.Request); + var response = new HttpListenerReplicationResponse(context.Response); + + try + { + // Simulate test error condition + if (_mockErrorConfig.RespondWithError) + { + throw new ReplicatorTestCase.HttpResponseException(); + } + + if (_useSynchronousIO) + { + _service.Perform(request, response); + } + else + { + await _service.PerformAsync(request, response, token); + } + } + catch + { + response.StatusCode = 500; + byte[] errorBytes = Encoding.UTF8.GetBytes("Internal Server Error"); + await response.Body.WriteAsync(errorBytes, 0, errorBytes.Length, token); + } + finally + { + await response.FlushAsync(token); + context.Response.Close(); + } + }, token); + } + } + catch (ObjectDisposedException) { } + catch (HttpListenerException) { } + } + + public void Dispose() + { + _cancellationTokenSource.Cancel(); + _listener.Stop(); + _listener.Close(); + try { _serverTask.Wait(); } catch { } + _cancellationTokenSource.Dispose(); + } + + private static int GetFreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + int port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + public HttpMessageHandler CreateHandler() => new HttpClientHandler(); + } +} From df8526638d2272ad2382f5015ee45e80bcac011d Mon Sep 17 00:00:00 2001 From: Shad Storhaug Date: Sun, 31 Aug 2025 11:27:48 +0700 Subject: [PATCH 08/13] Lucene.Net.Tests.Replicator: Only reference Lucene.Net.Replicator.AspNetCore on .NET Core 8 or higher --- .../Lucene.Net.Tests.Replicator.csproj | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/Lucene.Net.Tests.Replicator/Lucene.Net.Tests.Replicator.csproj b/src/Lucene.Net.Tests.Replicator/Lucene.Net.Tests.Replicator.csproj index 3393c89de4..bd802e42e2 100644 --- a/src/Lucene.Net.Tests.Replicator/Lucene.Net.Tests.Replicator.csproj +++ b/src/Lucene.Net.Tests.Replicator/Lucene.Net.Tests.Replicator.csproj @@ -52,14 +52,15 @@ $(SetTargetFramework) - - $(SetTargetFramework) - $(SetTargetFramework) + + + + From e057d332a717b9e2c6010c578a83f565291a41f8 Mon Sep 17 00:00:00 2001 From: Shad Storhaug Date: Sun, 31 Aug 2025 11:28:59 +0700 Subject: [PATCH 09/13] Lucene.Net.Replicator.AspNetCore: Dropped support for all target framworks but net8.0. --- .../Lucene.Net.Tests.Replicator.csproj | 2 +- .../Lucene.Net.Replicator.AspNetCore.csproj | 4 +--- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/src/Lucene.Net.Tests.Replicator/Lucene.Net.Tests.Replicator.csproj b/src/Lucene.Net.Tests.Replicator/Lucene.Net.Tests.Replicator.csproj index bd802e42e2..053cb5b896 100644 --- a/src/Lucene.Net.Tests.Replicator/Lucene.Net.Tests.Replicator.csproj +++ b/src/Lucene.Net.Tests.Replicator/Lucene.Net.Tests.Replicator.csproj @@ -57,7 +57,7 @@ - + diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/Lucene.Net.Replicator.AspNetCore.csproj b/src/dotnet/Lucene.Net.Replicator.AspNetCore/Lucene.Net.Replicator.AspNetCore.csproj index cf6e041633..8ef445f505 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/Lucene.Net.Replicator.AspNetCore.csproj +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/Lucene.Net.Replicator.AspNetCore.csproj @@ -25,7 +25,7 @@ - net8.0;netstandard2.1;netstandard2.0;net462 + net8.0 Lucene.Net.Replicator.AspNetCore AspNetCore integration of Lucene.Net.Replicator for the Lucene.Net full-text search engine library from The Apache Software Foundation. @@ -34,8 +34,6 @@ $(NoWarn);1591;1573 - - From 4a06a5db6b4190416faa6965ee1d80f33bae1863 Mon Sep 17 00:00:00 2001 From: Shad Storhaug Date: Sun, 31 Aug 2025 13:11:07 +0700 Subject: [PATCH 10/13] 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. --- .build/dependencies.props | 3 +-- .../Lucene.Net.Replicator.AspNetCore.csproj | 3 --- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/.build/dependencies.props b/.build/dependencies.props index 1620ceff1b..4c2e671329 100644 --- a/.build/dependencies.props +++ b/.build/dependencies.props @@ -39,7 +39,7 @@ [2.1.0, 3.0.0) 1.0.9 - 2.1.1 + 2.3.0 8.0.19 2.9.8 2.6.1 @@ -76,7 +76,6 @@ 4.3.0 4.3.0 5.0.0 - 8.0.0 6.0.10 4.3.1 6.1.0 diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/Lucene.Net.Replicator.AspNetCore.csproj b/src/dotnet/Lucene.Net.Replicator.AspNetCore/Lucene.Net.Replicator.AspNetCore.csproj index 8ef445f505..1595be29b6 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/Lucene.Net.Replicator.AspNetCore.csproj +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/Lucene.Net.Replicator.AspNetCore.csproj @@ -40,9 +40,6 @@ - - - From 349621f595a0fe2d396dc537ae1b6c018e301c13 Mon Sep 17 00:00:00 2001 From: NehanPathan Date: Mon, 1 Sep 2025 23:26:01 +0530 Subject: [PATCH 11/13] Add IAsyncReplicationServer/IAsyncReplicationResponse, update ReplicationService, Middleware, and TestServer to support async replication with conditional sync/async DI. --- .../Http/ReplicationService.cs | 33 +++++++++++--- .../Abstractions/IAsyncReplicationResponse.cs | 40 +++++++++++++++++ .../Abstractions/IAsyncReplicationService.cs | 39 +++++++++++++++++ .../Abstractions/IBaseReplicationResponse.cs | 43 +++++++++++++++++++ .../Http/Abstractions/IReplicationResponse.cs | 19 +------- .../Http/Abstractions/IReplicationService.cs | 10 ----- .../Http/ReplicationServlet.cs | 6 +-- .../ReplicatorTestCase.cs | 13 +++--- .../Net/HttpListenerReplicationResponse.cs | 2 +- .../Support/Net/TestServer.cs | 16 ++++--- .../AspNetCoreReplicationResponse.cs | 2 +- .../AspNetCoreReplicationServiceExtentions.cs | 2 +- 12 files changed, 176 insertions(+), 49 deletions(-) create mode 100644 src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationResponse.cs create mode 100644 src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationService.cs create mode 100644 src/Lucene.Net.Replicator/Support/Http/Abstractions/IBaseReplicationResponse.cs diff --git a/src/Lucene.Net.Replicator/Http/ReplicationService.cs b/src/Lucene.Net.Replicator/Http/ReplicationService.cs index c1c4f801f8..7fbace4788 100644 --- a/src/Lucene.Net.Replicator/Http/ReplicationService.cs +++ b/src/Lucene.Net.Replicator/Http/ReplicationService.cs @@ -49,7 +49,7 @@ namespace Lucene.Net.Replicator.Http /// /// @lucene.experimental /// - public class ReplicationService : IReplicationService // LUCENENET specific: added interface so we can mock easier. + public class ReplicationService : IReplicationService, IAsyncReplicationService // LUCENENET specific: added interface so we can mock easier. { /// /// Actions supported by the . @@ -121,13 +121,14 @@ private static string ExtractRequestParam(IReplicationRequest request, string pa return param; } - // method to avoid code duplication in sync and async Perform methods - private async Task ExecuteReplicationAsync( + // Shared internal logic (generic on response) + private async Task ExecuteReplicationInternal( IReplicationRequest request, - IReplicationResponse response, + TResponse response, Func copyStreamFunc, Func writeTokenFunc, Func flushFunc) + where TResponse : IBaseReplicationResponse { string[] pathElements = GetPathElements(request); if (pathElements.Length != 2) @@ -183,6 +184,28 @@ private async Task ExecuteReplicationAsync( } } + // For sync + private async Task ExecuteReplicationAsync( + IReplicationRequest request, + IReplicationResponse response, + Func copyStreamFunc, + Func writeTokenFunc, + Func flushFunc) + { + await ExecuteReplicationInternal(request, response, copyStreamFunc, writeTokenFunc, flushFunc); + } + + // For async + private async Task ExecuteReplicationAsync( + IReplicationRequest request, + IAsyncReplicationResponse response, + Func copyStreamFunc, + Func writeTokenFunc, + Func flushFunc) + { + await ExecuteReplicationInternal(request, response, copyStreamFunc, writeTokenFunc, flushFunc); + } + // LUCENENET specific - copy method not used /// @@ -222,7 +245,7 @@ public virtual void Perform(IReplicationRequest request, IReplicationResponse re /// Thrown when required parameters are missing or invalid. public virtual Task PerformAsync( IReplicationRequest request, - IReplicationResponse response, + IAsyncReplicationResponse response, CancellationToken cancellationToken = default) { return ExecuteReplicationAsync( diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationResponse.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationResponse.cs new file mode 100644 index 0000000000..9a3097086c --- /dev/null +++ b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationResponse.cs @@ -0,0 +1,40 @@ +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Lucene.Net.Replicator.Http.Abstractions +{ + /* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + /// + /// Abstraction for remote replication response, allows easy integration into any hosting frameworks. + /// + /// + /// .NET Specific Abstraction + /// + //Note: LUCENENET specific + public interface IAsyncReplicationResponse: IBaseReplicationResponse + { + /// + /// Flushes the response to the underlying response stream asynchronously. + /// + /// Optional cancellation token. + /// A task representing the asynchronous operation. + Task FlushAsync(CancellationToken cancellationToken = default); + } +} diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationService.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationService.cs new file mode 100644 index 0000000000..2bdc6a27ca --- /dev/null +++ b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationService.cs @@ -0,0 +1,39 @@ +using System; +using System.Threading; +using System.Threading.Tasks; + + +namespace Lucene.Net.Replicator.Http.Abstractions +{ + /* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + /// + /// Contract for a replication service. + /// + public interface IAsyncReplicationService + { + /// + /// Executes the replication task asynchronously. + /// + /// The replication request. + /// The replication response. + /// Optional cancellation token. + /// A task representing the asynchronous operation. + Task PerformAsync(IReplicationRequest request, IAsyncReplicationResponse response, CancellationToken cancellationToken = default); + } +} diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IBaseReplicationResponse.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IBaseReplicationResponse.cs new file mode 100644 index 0000000000..d1d9b71a67 --- /dev/null +++ b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IBaseReplicationResponse.cs @@ -0,0 +1,43 @@ +using System.IO; +using System.Threading; +using System.Threading.Tasks; + +namespace Lucene.Net.Replicator.Http.Abstractions +{ + /* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + + /// + /// Abstraction for remote replication response, allows easy integration into any hosting frameworks. + /// + /// + /// .NET Specific Abstraction + /// + //Note: LUCENENET specific + public interface IBaseReplicationResponse + { + /// + /// Gets or sets the http status code of the response. + /// + int StatusCode { get; set; } + + /// + /// The response content. + /// + Stream Body { get; } + } +} diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs index 72e4233967..24067be687 100644 --- a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs +++ b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs @@ -28,28 +28,11 @@ namespace Lucene.Net.Replicator.Http.Abstractions /// .NET Specific Abstraction /// //Note: LUCENENET specific - public interface IReplicationResponse + public interface IReplicationResponse: IBaseReplicationResponse { - /// - /// Gets or sets the http status code of the response. - /// - int StatusCode { get; set; } - - /// - /// The response content. - /// - Stream Body { get; } - /// /// Flushes the reponse to the underlying response stream. /// void Flush(); - - /// - /// Flushes the response to the underlying response stream asynchronously. - /// - /// Optional cancellation token. - /// A task representing the asynchronous operation. - Task FlushAsync(CancellationToken cancellationToken = default); } } diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs index 78ca9bf565..4ad14aef79 100644 --- a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs +++ b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs @@ -32,15 +32,5 @@ public interface IReplicationService /// /// required parameters are missing void Perform(IReplicationRequest request, IReplicationResponse response); - - /// - /// Executes the replication task asynchronously. - /// - /// The replication request. - /// The replication response. - /// Optional cancellation token. - /// A task representing the asynchronous operation. - Task PerformAsync(IReplicationRequest request, IReplicationResponse response, CancellationToken cancellationToken = default); - } } diff --git a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs index e9775a63f7..be271dc4bf 100644 --- a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs +++ b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs @@ -32,7 +32,7 @@ namespace Lucene.Net.Replicator.Http public class ReplicationServlet { - public void Configure(IApplicationBuilder app, IReplicationService service, ReplicatorTestCase.MockErrorConfig mockErrorConfig) + public void Configure(IApplicationBuilder app, IAsyncReplicationService service, ReplicatorTestCase.MockErrorConfig mockErrorConfig) { // Middleware to throw an exception conditionally from our test server. app.Use(async (context, next) => @@ -63,9 +63,9 @@ public void Configure(IApplicationBuilder app, IReplicationService service, Repl public class ReplicationServiceMiddleware { private readonly RequestDelegate next; - private readonly IReplicationService service; + private readonly IAsyncReplicationService service; - public ReplicationServiceMiddleware(RequestDelegate next, IReplicationService service) + public ReplicationServiceMiddleware(RequestDelegate next, IAsyncReplicationService service) { this.next = next ?? throw new ArgumentNullException(nameof(next)); this.service = service ?? throw new ArgumentNullException(nameof(service)); diff --git a/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs b/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs index 4a80cdbb78..191fc00180 100644 --- a/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs +++ b/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs @@ -81,15 +81,17 @@ public class MockErrorConfig #if FEATURE_ASPNETCORE_TESTHOST public partial class ReplicatorTestCase { - public static TestServer NewHttpServer(IReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) + public static TestServer NewHttpServer(ReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) { if (useStartupClass) { var builder = new WebHostBuilder() .ConfigureServices(container => { - container.AddSingleton(service); - container.AddSingleton(mockErrorConfig); + // Register concrete services + container.AddSingleton(service); + container.AddSingleton(service); + container.AddSingleton(mockErrorConfig); }); if (useSynchronousIO) @@ -111,7 +113,8 @@ public static TestServer NewHttpServer(IReplicationService service, MockErrorCon .ConfigureServices(container => { container.AddRouting(); - container.AddSingleton(service); + container.AddSingleton(service); + container.AddSingleton(service); container.AddSingleton(mockErrorConfig); if (useSynchronousIO) { @@ -216,7 +219,7 @@ public async Task InvokeAsync(HttpContext context) public partial class ReplicatorTestCase { // LUCENENET: This uses HttpListener from System.Net to test the service where ASP.NET Core is not supported. - public static TestServer NewHttpServer(IReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) + public static TestServer NewHttpServer(ReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) { return new TestServer(service, mockErrorConfig, useSynchronousIO); } diff --git a/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs b/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs index 9cc1280a4e..85d9d42019 100644 --- a/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs +++ b/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs @@ -27,7 +27,7 @@ namespace Lucene.Net.Replicator.Net /// A concrete implementation of for supporting /// . /// - public class HttpListenerReplicationResponse : IReplicationResponse + public class HttpListenerReplicationResponse : IReplicationResponse, IAsyncReplicationResponse { private readonly HttpListenerResponse _response; diff --git a/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs b/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs index 695fab54cf..668833f0f5 100644 --- a/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs +++ b/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs @@ -6,6 +6,7 @@ using System.Text; using System.Threading; using System.Threading.Tasks; +using Lucene.Net.Replicator.Http; namespace Lucene.Net.Replicator.Net { @@ -34,7 +35,9 @@ namespace Lucene.Net.Replicator.Net public class TestServer : IDisposable { private readonly HttpListener _listener; - private readonly IReplicationService _service; + private readonly IAsyncReplicationService _asyncService; + private readonly IReplicationService _syncService; + private readonly ReplicatorTestCase.MockErrorConfig _mockErrorConfig; private readonly bool _useSynchronousIO; private readonly CancellationTokenSource _cancellationTokenSource = new(); @@ -43,12 +46,15 @@ public class TestServer : IDisposable public Uri BaseAddress { get; } public TestServer( - IReplicationService service, + ReplicationService service, // concrete type implements both sync + async ReplicatorTestCase.MockErrorConfig mockErrorConfig, bool useSynchronousIO, string prefix = "http://localhost:0/") { - _service = service ?? throw new ArgumentNullException(nameof(service)); + if (service == null) throw new ArgumentNullException(nameof(service)); + _asyncService = service; + _syncService = service; + _mockErrorConfig = mockErrorConfig ?? throw new ArgumentNullException(nameof(mockErrorConfig)); _useSynchronousIO = useSynchronousIO; @@ -92,11 +98,11 @@ private async Task ListenLoopAsync(CancellationToken token) if (_useSynchronousIO) { - _service.Perform(request, response); + _syncService.Perform(request, response); } else { - await _service.PerformAsync(request, response, token); + await _asyncService.PerformAsync(request, response, token); } } catch diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs index 90cb172b88..347615fd0e 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs @@ -30,7 +30,7 @@ namespace Lucene.Net.Replicator.AspNetCore /// .NET Specific Implementation of the Lucene Replicator using AspNetCore /// //Note: LUCENENET specific - public class AspNetCoreReplicationResponse : IReplicationResponse + public class AspNetCoreReplicationResponse : IReplicationResponse, IAsyncReplicationResponse { private readonly HttpResponse response; diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs index ce87aa0f84..30cb3c74a3 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs @@ -37,7 +37,7 @@ public static void Perform(this IReplicationService self, HttpRequest request, H /// Extension method that mirrors the signature of using AspNetCore as implementation. /// public static async Task PerformAsync( - this IReplicationService self, + this IAsyncReplicationService self, HttpRequest request, HttpResponse response, CancellationToken cancellationToken = default) From 8dcdc0dba0a865ac5701fe3ccaa874967c161fcc Mon Sep 17 00:00:00 2001 From: NehanPathan Date: Tue, 2 Sep 2025 06:22:43 +0530 Subject: [PATCH 12/13] Revert "Add IAsyncReplicationServer/IAsyncReplicationResponse, update ReplicationService, Middleware, and TestServer to support async replication with conditional sync/async DI." This reverts commit 349621f595a0fe2d396dc537ae1b6c018e301c13. --- .../Http/ReplicationService.cs | 33 +++----------- .../Abstractions/IAsyncReplicationResponse.cs | 40 ----------------- .../Abstractions/IAsyncReplicationService.cs | 39 ----------------- .../Abstractions/IBaseReplicationResponse.cs | 43 ------------------- .../Http/Abstractions/IReplicationResponse.cs | 19 +++++++- .../Http/Abstractions/IReplicationService.cs | 10 +++++ .../Http/ReplicationServlet.cs | 6 +-- .../ReplicatorTestCase.cs | 13 +++--- .../Net/HttpListenerReplicationResponse.cs | 2 +- .../Support/Net/TestServer.cs | 16 +++---- .../AspNetCoreReplicationResponse.cs | 2 +- .../AspNetCoreReplicationServiceExtentions.cs | 2 +- 12 files changed, 49 insertions(+), 176 deletions(-) delete mode 100644 src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationResponse.cs delete mode 100644 src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationService.cs delete mode 100644 src/Lucene.Net.Replicator/Support/Http/Abstractions/IBaseReplicationResponse.cs diff --git a/src/Lucene.Net.Replicator/Http/ReplicationService.cs b/src/Lucene.Net.Replicator/Http/ReplicationService.cs index 7fbace4788..c1c4f801f8 100644 --- a/src/Lucene.Net.Replicator/Http/ReplicationService.cs +++ b/src/Lucene.Net.Replicator/Http/ReplicationService.cs @@ -49,7 +49,7 @@ namespace Lucene.Net.Replicator.Http /// /// @lucene.experimental /// - public class ReplicationService : IReplicationService, IAsyncReplicationService // LUCENENET specific: added interface so we can mock easier. + public class ReplicationService : IReplicationService // LUCENENET specific: added interface so we can mock easier. { /// /// Actions supported by the . @@ -121,14 +121,13 @@ private static string ExtractRequestParam(IReplicationRequest request, string pa return param; } - // Shared internal logic (generic on response) - private async Task ExecuteReplicationInternal( + // method to avoid code duplication in sync and async Perform methods + private async Task ExecuteReplicationAsync( IReplicationRequest request, - TResponse response, + IReplicationResponse response, Func copyStreamFunc, Func writeTokenFunc, Func flushFunc) - where TResponse : IBaseReplicationResponse { string[] pathElements = GetPathElements(request); if (pathElements.Length != 2) @@ -184,28 +183,6 @@ private async Task ExecuteReplicationInternal( } } - // For sync - private async Task ExecuteReplicationAsync( - IReplicationRequest request, - IReplicationResponse response, - Func copyStreamFunc, - Func writeTokenFunc, - Func flushFunc) - { - await ExecuteReplicationInternal(request, response, copyStreamFunc, writeTokenFunc, flushFunc); - } - - // For async - private async Task ExecuteReplicationAsync( - IReplicationRequest request, - IAsyncReplicationResponse response, - Func copyStreamFunc, - Func writeTokenFunc, - Func flushFunc) - { - await ExecuteReplicationInternal(request, response, copyStreamFunc, writeTokenFunc, flushFunc); - } - // LUCENENET specific - copy method not used /// @@ -245,7 +222,7 @@ public virtual void Perform(IReplicationRequest request, IReplicationResponse re /// Thrown when required parameters are missing or invalid. public virtual Task PerformAsync( IReplicationRequest request, - IAsyncReplicationResponse response, + IReplicationResponse response, CancellationToken cancellationToken = default) { return ExecuteReplicationAsync( diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationResponse.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationResponse.cs deleted file mode 100644 index 9a3097086c..0000000000 --- a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationResponse.cs +++ /dev/null @@ -1,40 +0,0 @@ -using System.IO; -using System.Threading; -using System.Threading.Tasks; - -namespace Lucene.Net.Replicator.Http.Abstractions -{ - /* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - - /// - /// Abstraction for remote replication response, allows easy integration into any hosting frameworks. - /// - /// - /// .NET Specific Abstraction - /// - //Note: LUCENENET specific - public interface IAsyncReplicationResponse: IBaseReplicationResponse - { - /// - /// Flushes the response to the underlying response stream asynchronously. - /// - /// Optional cancellation token. - /// A task representing the asynchronous operation. - Task FlushAsync(CancellationToken cancellationToken = default); - } -} diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationService.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationService.cs deleted file mode 100644 index 2bdc6a27ca..0000000000 --- a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IAsyncReplicationService.cs +++ /dev/null @@ -1,39 +0,0 @@ -using System; -using System.Threading; -using System.Threading.Tasks; - - -namespace Lucene.Net.Replicator.Http.Abstractions -{ - /* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - - /// - /// Contract for a replication service. - /// - public interface IAsyncReplicationService - { - /// - /// Executes the replication task asynchronously. - /// - /// The replication request. - /// The replication response. - /// Optional cancellation token. - /// A task representing the asynchronous operation. - Task PerformAsync(IReplicationRequest request, IAsyncReplicationResponse response, CancellationToken cancellationToken = default); - } -} diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IBaseReplicationResponse.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IBaseReplicationResponse.cs deleted file mode 100644 index d1d9b71a67..0000000000 --- a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IBaseReplicationResponse.cs +++ /dev/null @@ -1,43 +0,0 @@ -using System.IO; -using System.Threading; -using System.Threading.Tasks; - -namespace Lucene.Net.Replicator.Http.Abstractions -{ - /* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - - /// - /// Abstraction for remote replication response, allows easy integration into any hosting frameworks. - /// - /// - /// .NET Specific Abstraction - /// - //Note: LUCENENET specific - public interface IBaseReplicationResponse - { - /// - /// Gets or sets the http status code of the response. - /// - int StatusCode { get; set; } - - /// - /// The response content. - /// - Stream Body { get; } - } -} diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs index 24067be687..72e4233967 100644 --- a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs +++ b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs @@ -28,11 +28,28 @@ namespace Lucene.Net.Replicator.Http.Abstractions /// .NET Specific Abstraction /// //Note: LUCENENET specific - public interface IReplicationResponse: IBaseReplicationResponse + public interface IReplicationResponse { + /// + /// Gets or sets the http status code of the response. + /// + int StatusCode { get; set; } + + /// + /// The response content. + /// + Stream Body { get; } + /// /// Flushes the reponse to the underlying response stream. /// void Flush(); + + /// + /// Flushes the response to the underlying response stream asynchronously. + /// + /// Optional cancellation token. + /// A task representing the asynchronous operation. + Task FlushAsync(CancellationToken cancellationToken = default); } } diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs index 4ad14aef79..78ca9bf565 100644 --- a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs +++ b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationService.cs @@ -32,5 +32,15 @@ public interface IReplicationService /// /// required parameters are missing void Perform(IReplicationRequest request, IReplicationResponse response); + + /// + /// Executes the replication task asynchronously. + /// + /// The replication request. + /// The replication response. + /// Optional cancellation token. + /// A task representing the asynchronous operation. + Task PerformAsync(IReplicationRequest request, IReplicationResponse response, CancellationToken cancellationToken = default); + } } diff --git a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs index be271dc4bf..e9775a63f7 100644 --- a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs +++ b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs @@ -32,7 +32,7 @@ namespace Lucene.Net.Replicator.Http public class ReplicationServlet { - public void Configure(IApplicationBuilder app, IAsyncReplicationService service, ReplicatorTestCase.MockErrorConfig mockErrorConfig) + public void Configure(IApplicationBuilder app, IReplicationService service, ReplicatorTestCase.MockErrorConfig mockErrorConfig) { // Middleware to throw an exception conditionally from our test server. app.Use(async (context, next) => @@ -63,9 +63,9 @@ public void Configure(IApplicationBuilder app, IAsyncReplicationService service, public class ReplicationServiceMiddleware { private readonly RequestDelegate next; - private readonly IAsyncReplicationService service; + private readonly IReplicationService service; - public ReplicationServiceMiddleware(RequestDelegate next, IAsyncReplicationService service) + public ReplicationServiceMiddleware(RequestDelegate next, IReplicationService service) { this.next = next ?? throw new ArgumentNullException(nameof(next)); this.service = service ?? throw new ArgumentNullException(nameof(service)); diff --git a/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs b/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs index 191fc00180..4a80cdbb78 100644 --- a/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs +++ b/src/Lucene.Net.Tests.Replicator/ReplicatorTestCase.cs @@ -81,17 +81,15 @@ public class MockErrorConfig #if FEATURE_ASPNETCORE_TESTHOST public partial class ReplicatorTestCase { - public static TestServer NewHttpServer(ReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) + public static TestServer NewHttpServer(IReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) { if (useStartupClass) { var builder = new WebHostBuilder() .ConfigureServices(container => { - // Register concrete services - container.AddSingleton(service); - container.AddSingleton(service); - container.AddSingleton(mockErrorConfig); + container.AddSingleton(service); + container.AddSingleton(mockErrorConfig); }); if (useSynchronousIO) @@ -113,8 +111,7 @@ public static TestServer NewHttpServer(ReplicationService service, MockErrorCon .ConfigureServices(container => { container.AddRouting(); - container.AddSingleton(service); - container.AddSingleton(service); + container.AddSingleton(service); container.AddSingleton(mockErrorConfig); if (useSynchronousIO) { @@ -219,7 +216,7 @@ public async Task InvokeAsync(HttpContext context) public partial class ReplicatorTestCase { // LUCENENET: This uses HttpListener from System.Net to test the service where ASP.NET Core is not supported. - public static TestServer NewHttpServer(ReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) + public static TestServer NewHttpServer(IReplicationService service, MockErrorConfig mockErrorConfig, bool useSynchronousIO, bool useStartupClass) { return new TestServer(service, mockErrorConfig, useSynchronousIO); } diff --git a/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs b/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs index 85d9d42019..9cc1280a4e 100644 --- a/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs +++ b/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs @@ -27,7 +27,7 @@ namespace Lucene.Net.Replicator.Net /// A concrete implementation of for supporting /// . /// - public class HttpListenerReplicationResponse : IReplicationResponse, IAsyncReplicationResponse + public class HttpListenerReplicationResponse : IReplicationResponse { private readonly HttpListenerResponse _response; diff --git a/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs b/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs index 668833f0f5..695fab54cf 100644 --- a/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs +++ b/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs @@ -6,7 +6,6 @@ using System.Text; using System.Threading; using System.Threading.Tasks; -using Lucene.Net.Replicator.Http; namespace Lucene.Net.Replicator.Net { @@ -35,9 +34,7 @@ namespace Lucene.Net.Replicator.Net public class TestServer : IDisposable { private readonly HttpListener _listener; - private readonly IAsyncReplicationService _asyncService; - private readonly IReplicationService _syncService; - + private readonly IReplicationService _service; private readonly ReplicatorTestCase.MockErrorConfig _mockErrorConfig; private readonly bool _useSynchronousIO; private readonly CancellationTokenSource _cancellationTokenSource = new(); @@ -46,15 +43,12 @@ public class TestServer : IDisposable public Uri BaseAddress { get; } public TestServer( - ReplicationService service, // concrete type implements both sync + async + IReplicationService service, ReplicatorTestCase.MockErrorConfig mockErrorConfig, bool useSynchronousIO, string prefix = "http://localhost:0/") { - if (service == null) throw new ArgumentNullException(nameof(service)); - _asyncService = service; - _syncService = service; - + _service = service ?? throw new ArgumentNullException(nameof(service)); _mockErrorConfig = mockErrorConfig ?? throw new ArgumentNullException(nameof(mockErrorConfig)); _useSynchronousIO = useSynchronousIO; @@ -98,11 +92,11 @@ private async Task ListenLoopAsync(CancellationToken token) if (_useSynchronousIO) { - _syncService.Perform(request, response); + _service.Perform(request, response); } else { - await _asyncService.PerformAsync(request, response, token); + await _service.PerformAsync(request, response, token); } } catch diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs index 347615fd0e..90cb172b88 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs @@ -30,7 +30,7 @@ namespace Lucene.Net.Replicator.AspNetCore /// .NET Specific Implementation of the Lucene Replicator using AspNetCore /// //Note: LUCENENET specific - public class AspNetCoreReplicationResponse : IReplicationResponse, IAsyncReplicationResponse + public class AspNetCoreReplicationResponse : IReplicationResponse { private readonly HttpResponse response; diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs index 30cb3c74a3..ce87aa0f84 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationServiceExtentions.cs @@ -37,7 +37,7 @@ public static void Perform(this IReplicationService self, HttpRequest request, H /// Extension method that mirrors the signature of using AspNetCore as implementation. /// public static async Task PerformAsync( - this IAsyncReplicationService self, + this IReplicationService self, HttpRequest request, HttpResponse response, CancellationToken cancellationToken = default) From 869ed3733824ddaffec6191226ef47c44cdc5d10 Mon Sep 17 00:00:00 2001 From: NehanPathan Date: Tue, 2 Sep 2025 07:16:57 +0530 Subject: [PATCH 13/13] Revert async interfaces, unify IReplicationService with Perform/PerformAsync, update ReplicationServlet, Middleware, and TestServer to use single interface --- .../Http/ReplicationService.cs | 9 ++++--- .../Http/Abstractions/IReplicationResponse.cs | 14 ---------- .../Http/ReplicationServlet.cs | 2 +- .../Net/HttpListenerReplicationResponse.cs | 5 ---- .../Support/Net/TestServer.cs | 2 +- .../AspNetCoreReplicationResponse.cs | 26 ------------------- 6 files changed, 7 insertions(+), 51 deletions(-) diff --git a/src/Lucene.Net.Replicator/Http/ReplicationService.cs b/src/Lucene.Net.Replicator/Http/ReplicationService.cs index c1c4f801f8..ccdaba7e92 100644 --- a/src/Lucene.Net.Replicator/Http/ReplicationService.cs +++ b/src/Lucene.Net.Replicator/Http/ReplicationService.cs @@ -139,6 +139,7 @@ private async Task ExecuteReplicationAsync( 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) @@ -175,7 +176,7 @@ private async Task ExecuteReplicationAsync( } catch (Exception) { - response.StatusCode = (int)HttpStatusCode.InternalServerError; + response.StatusCode = (int)HttpStatusCode.InternalServerError; // propagate the failure } finally { @@ -208,8 +209,8 @@ public virtual void Perform(IReplicationRequest request, IReplicationResponse re } return Task.CompletedTask; }, - () => { response.Flush(); return Task.CompletedTask; } - ).GetAwaiter().GetResult(); // // keep sync behavior + () => { response.Body.Flush(); return Task.CompletedTask; } + ).ConfigureAwait(false).GetAwaiter().GetResult(); // keep sync behavior } @@ -241,7 +242,7 @@ public virtual Task PerformAsync( await token.SerializeAsync(response.Body, cancellationToken); } }, - () => response.FlushAsync(cancellationToken) + () => response.Body.FlushAsync(cancellationToken) ); } } diff --git a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs index 72e4233967..2d035d34af 100644 --- a/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs +++ b/src/Lucene.Net.Replicator/Support/Http/Abstractions/IReplicationResponse.cs @@ -1,6 +1,4 @@ using System.IO; -using System.Threading; -using System.Threading.Tasks; namespace Lucene.Net.Replicator.Http.Abstractions { @@ -39,17 +37,5 @@ public interface IReplicationResponse /// The response content. /// Stream Body { get; } - - /// - /// Flushes the reponse to the underlying response stream. - /// - void Flush(); - - /// - /// Flushes the response to the underlying response stream asynchronously. - /// - /// Optional cancellation token. - /// A task representing the asynchronous operation. - Task FlushAsync(CancellationToken cancellationToken = default); } } diff --git a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs index e9775a63f7..0aa4dc33a1 100644 --- a/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs +++ b/src/Lucene.Net.Tests.Replicator/Http/ReplicationServlet.cs @@ -2,9 +2,9 @@ using Lucene.Net.Replicator.AspNetCore; using Lucene.Net.Replicator.Http.Abstractions; using Microsoft.AspNetCore.Builder; -using Microsoft.AspNetCore.Routing; using Microsoft.AspNetCore.Hosting; using Microsoft.AspNetCore.Http; +using Microsoft.AspNetCore.Routing; using System; using System.Threading.Tasks; diff --git a/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs b/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs index 9cc1280a4e..91c3a456dd 100644 --- a/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs +++ b/src/Lucene.Net.Tests.Replicator/Support/Net/HttpListenerReplicationResponse.cs @@ -40,10 +40,5 @@ public int StatusCode } public Stream Body => _response.OutputStream; - - public void Flush() => _response.OutputStream.Flush(); - - public Task FlushAsync(CancellationToken cancellationToken = default) => - _response.OutputStream.FlushAsync(cancellationToken); } } diff --git a/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs b/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs index 695fab54cf..1da5d84ee0 100644 --- a/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs +++ b/src/Lucene.Net.Tests.Replicator/Support/Net/TestServer.cs @@ -107,7 +107,7 @@ private async Task ListenLoopAsync(CancellationToken token) } finally { - await response.FlushAsync(token); + await response.Body.FlushAsync(token); context.Response.Close(); } }, token); diff --git a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs index 90cb172b88..5141a89489 100644 --- a/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs +++ b/src/dotnet/Lucene.Net.Replicator.AspNetCore/AspNetCoreReplicationResponse.cs @@ -1,8 +1,6 @@ using Lucene.Net.Replicator.Http.Abstractions; using Microsoft.AspNetCore.Http; using System.IO; -using System.Threading; -using System.Threading.Tasks; namespace Lucene.Net.Replicator.AspNetCore { @@ -59,29 +57,5 @@ public int StatusCode /// This simply returns the . /// public Stream Body => response.Body; - - /// - /// Flushes the reponse to the underlying response stream. - /// - /// - /// This simply calls on the . - /// - public void Flush() - { - response.Body.Flush(); - } - - /// - /// Flushes the response to the underlying response stream asynchronously. - /// - /// Optional cancellation token. - /// A task representing the asynchronous operation. - /// - /// This simply calls on the . - /// - public async Task FlushAsync(CancellationToken cancellationToken = default) - { - await response.Body.FlushAsync(cancellationToken); - } } }