Skip to content

Commit 3e1068a

Browse files
NehanPathanNehanPathan
committed
Refactor replication service and stream extensions for fully async I/O, updating PerformAsync, SessionToken, and ASP.NET Core response handling.
1 parent 8f28cd7 commit 3e1068a

4 files changed

Lines changed: 248 additions & 105 deletions

File tree

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

Lines changed: 77 additions & 89 deletions
Original file line numberDiff line numberDiff line change
@@ -121,140 +121,128 @@ private static string ExtractRequestParam(IReplicationRequest request, string pa
121121
return param;
122122
}
123123

124-
// LUCENENET specific - copy method not used
125-
126-
/// <summary>
127-
/// Executes the replication task.
128-
/// </summary>
129-
/// <exception cref="InvalidOperationException">required parameters are missing</exception>
130-
public virtual void Perform(IReplicationRequest request, IReplicationResponse response)
124+
// method to avoid code duplication in sync and async Perform methods
125+
private async Task ExecuteReplicationAsync(
126+
IReplicationRequest request,
127+
IReplicationResponse response,
128+
Func<Stream, Task> copyStreamFunc,
129+
Func<SessionToken, Task> writeTokenFunc,
130+
Func<Task> flushFunc)
131131
{
132132
string[] pathElements = GetPathElements(request);
133133
if (pathElements.Length != 2)
134-
{
135134
throw ServletException.Create("invalid path, must contain shard ID and action, e.g. */s1/update");
136-
}
137135

138136
if (!Enum.TryParse(pathElements[ACTION_IDX], true, out ReplicationAction action))
139-
{
140137
throw ServletException.Create("Unsupported action provided: " + pathElements[ACTION_IDX]);
141-
}
142138

143139
if (!replicators.TryGetValue(pathElements[SHARD_IDX], out IReplicator replicator))
144-
{
145140
throw ServletException.Create("unrecognized shard ID " + pathElements[SHARD_IDX]);
146-
}
147141

148-
// SOLR-8933 Don't close this stream.
149142
try
150143
{
151144
switch (action)
152145
{
153146
case ReplicationAction.OBTAIN:
154-
string sessionId = ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM);
155-
string fileName = ExtractRequestParam(request, REPLICATE_FILENAME_PARAM);
156-
string source = ExtractRequestParam(request, REPLICATE_SOURCE_PARAM);
157-
using (Stream stream = replicator.ObtainFile(sessionId, source, fileName))
158-
stream.CopyTo(response.Body);
159-
break;
147+
{
148+
string sessionId = ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM);
149+
string fileName = ExtractRequestParam(request, REPLICATE_FILENAME_PARAM);
150+
string source = ExtractRequestParam(request, REPLICATE_SOURCE_PARAM);
160151

161-
case ReplicationAction.RELEASE:
162-
replicator.Release(ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM));
163-
break;
152+
using (Stream stream = replicator.ObtainFile(sessionId, source, fileName))
153+
await copyStreamFunc(stream);
154+
break;
155+
}
164156

165-
case ReplicationAction.UPDATE:
166-
string currentVersion = request.QueryParam(REPLICATE_VERSION_PARAM);
167-
SessionToken token = replicator.CheckForUpdate(currentVersion);
168-
if (token is null)
157+
case ReplicationAction.RELEASE:
169158
{
170-
response.Body.Write(new byte[] { 0 }, 0, 1); // marker for null token
159+
replicator.Release(ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM));
160+
break;
171161
}
172-
else
162+
163+
case ReplicationAction.UPDATE:
173164
{
174-
response.Body.Write(new byte[] { 1 }, 0, 1);
175-
token.Serialize(new DataOutputStream(response.Body));
165+
string currentVersion = request.QueryParam(REPLICATE_VERSION_PARAM);
166+
SessionToken token = replicator.CheckForUpdate(currentVersion);
167+
await writeTokenFunc(token);
168+
break;
176169
}
177-
break;
178170

179-
// LUCENENET specific:
180171
default:
181172
if (Debugging.AssertsEnabled) Debugging.Assert(false, "Invalid ReplicationAction specified");
182173
break;
183174
}
184175
}
185176
catch (Exception)
186177
{
187-
response.StatusCode = (int)HttpStatusCode.InternalServerError; // propagate the failure
178+
response.StatusCode = (int)HttpStatusCode.InternalServerError;
188179
}
189180
finally
190181
{
191-
response.Flush();
182+
await flushFunc();
192183
}
193184
}
194185

186+
// LUCENENET specific - copy method not used
187+
188+
/// <summary>
189+
/// Executes the replication task.
190+
/// </summary>
191+
/// <exception cref="InvalidOperationException">required parameters are missing</exception>
192+
public virtual void Perform(IReplicationRequest request, IReplicationResponse response)
193+
{
194+
ExecuteReplicationAsync(
195+
request,
196+
response,
197+
stream => { stream.CopyTo(response.Body); return Task.CompletedTask; },
198+
token =>
199+
{
200+
if (token == null)
201+
{
202+
response.Body.Write(new byte[] { 0 }, 0, 1);
203+
}
204+
else
205+
{
206+
response.Body.Write(new byte[] { 1 }, 0, 1);
207+
token.Serialize(new DataOutputStream(response.Body));
208+
}
209+
return Task.CompletedTask;
210+
},
211+
() => { response.Flush(); return Task.CompletedTask; }
212+
).GetAwaiter().GetResult(); // // keep sync behavior
213+
}
214+
215+
195216
/// <summary>
196217
/// Executes the replication task asynchronously.
197218
/// </summary>
198219
/// <param name="request">The replication request containing action and parameters.</param>
199220
/// <param name="response">The replication response used to send data back to the client.</param>
200221
/// <param name="cancellationToken">A <see cref="CancellationToken"/> to observe while performing the replication.</param>
201222
/// <exception cref="InvalidOperationException">Thrown when required parameters are missing or invalid.</exception>
202-
public virtual async Task PerformAsync(IReplicationRequest request, IReplicationResponse response, CancellationToken cancellationToken = default)
223+
public virtual Task PerformAsync(
224+
IReplicationRequest request,
225+
IReplicationResponse response,
226+
CancellationToken cancellationToken = default)
203227
{
204-
string[] pathElements = GetPathElements(request);
205-
if (pathElements.Length != 2)
206-
throw ServletException.Create("invalid path, must contain shard ID and action, e.g. */s1/update");
207-
208-
if (!Enum.TryParse(pathElements[ACTION_IDX], true, out ReplicationAction action))
209-
throw ServletException.Create("Unsupported action provided: " + pathElements[ACTION_IDX]);
210-
211-
if (!replicators.TryGetValue(pathElements[SHARD_IDX], out IReplicator replicator))
212-
throw ServletException.Create("unrecognized shard ID " + pathElements[SHARD_IDX]);
213-
214-
try
215-
{
216-
switch (action)
228+
return ExecuteReplicationAsync(
229+
request,
230+
response,
231+
stream => stream.CopyToAsync(response.Body, 81920, cancellationToken),
232+
async token =>
217233
{
218-
case ReplicationAction.OBTAIN:
219-
string sessionId = ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM);
220-
string fileName = ExtractRequestParam(request, REPLICATE_FILENAME_PARAM);
221-
string source = ExtractRequestParam(request, REPLICATE_SOURCE_PARAM);
222-
using (Stream stream = replicator.ObtainFile(sessionId, source, fileName))
223-
await stream.CopyToAsync(response.Body, 81920, cancellationToken);
224-
break;
225-
226-
case ReplicationAction.RELEASE:
227-
replicator.Release(ExtractRequestParam(request, REPLICATE_SESSION_ID_PARAM));
228-
break;
229-
230-
case ReplicationAction.UPDATE:
231-
string currentVersion = request.QueryParam(REPLICATE_VERSION_PARAM);
232-
SessionToken token = replicator.CheckForUpdate(currentVersion);
233-
if (token is null)
234-
{
235-
await response.Body.WriteAsync(new byte[] { 0 }, 0, 1, cancellationToken);
236-
}
237-
else
238-
{
239-
await response.Body.WriteAsync(new byte[] { 1 }, 0, 1, cancellationToken);
240-
await token.SerializeAsync(response.Body, cancellationToken);
241-
}
242-
break;
243-
244-
default:
245-
if (Debugging.AssertsEnabled) Debugging.Assert(false, "Invalid ReplicationAction specified");
246-
break;
247-
}
248-
}
249-
catch (Exception)
250-
{
251-
response.StatusCode = (int)HttpStatusCode.InternalServerError;
252-
}
253-
finally
254-
{
255-
await response.FlushAsync(cancellationToken);
256-
}
234+
if (token == null)
235+
{
236+
await response.Body.WriteAsync(new byte[] { 0 }, 0, 1, cancellationToken);
237+
}
238+
else
239+
{
240+
await response.Body.WriteAsync(new byte[] { 1 }, 0, 1, cancellationToken);
241+
await token.SerializeAsync(response.Body, cancellationToken);
242+
}
243+
},
244+
() => response.FlushAsync(cancellationToken)
245+
);
257246
}
258-
259247
}
260248
}

‎src/Lucene.Net.Replicator/SessionToken.cs‎

Lines changed: 18 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,11 @@
11
using J2N.IO;
22
using System.Collections.Generic;
3+
using System;
34
using System.IO;
45
using JCG = J2N.Collections.Generic;
56
using System.Threading;
67
using System.Threading.Tasks;
8+
using Lucene.Net.Support.IO;
79

810
namespace Lucene.Net.Replicator
911
{
@@ -115,30 +117,34 @@ public void Serialize(DataOutputStream writer)
115117
}
116118

117119
/// <summary>
118-
/// Asynchronously serialize the token data for communication between server and client.
120+
/// Asynchronously serializes the token's properties, including ID, version, and source files,
121+
/// to the provided <see cref="Stream"/> for transmission or storage.
119122
/// </summary>
120123
/// <param name="output">The <see cref="Stream"/> to write the token data to.</param>
121-
/// <param name="cancellationToken">A cancellation token to observe while waiting for the flush to complete.</param>
122-
/// <returns>A task representing the asynchronous operation.</returns>
124+
/// <param name="cancellationToken">A cancellation token to observe while writing and flushing the stream.</param>
125+
/// <returns>A task representing the asynchronous serialization operation.</returns>
123126
public async Task SerializeAsync(Stream output, CancellationToken cancellationToken = default)
124127
{
125-
using var writer = new DataOutputStream(output);
126-
writer.WriteUTF(Id);
127-
writer.WriteUTF(Version);
128-
writer.WriteInt32(SourceFiles.Count);
128+
if (output is null)
129+
throw new ArgumentNullException(nameof(output));
130+
131+
await output.WriteUTFAsync(Id, cancellationToken).ConfigureAwait(false);
132+
await output.WriteUTFAsync(Version, cancellationToken).ConfigureAwait(false);
133+
await output.WriteInt32Async(SourceFiles.Count, cancellationToken).ConfigureAwait(false);
129134

130135
foreach (var pair in SourceFiles)
131136
{
132-
writer.WriteUTF(pair.Key);
133-
writer.WriteInt32(pair.Value.Count);
137+
await output.WriteUTFAsync(pair.Key, cancellationToken).ConfigureAwait(false);
138+
await output.WriteInt32Async(pair.Value.Count, cancellationToken).ConfigureAwait(false);
139+
134140
foreach (var file in pair.Value)
135141
{
136-
writer.WriteUTF(file.FileName);
137-
writer.WriteInt64(file.Length);
142+
await output.WriteUTFAsync(file.FileName, cancellationToken).ConfigureAwait(false);
143+
await output.WriteInt64Async(file.Length, cancellationToken).ConfigureAwait(false);
138144
}
139145
}
140146

141-
await output.FlushAsync(cancellationToken);
147+
await output.FlushAsync(cancellationToken).ConfigureAwait(false);
142148
}
143149

144150
public override string ToString()

0 commit comments

Comments
 (0)