NehanPathan commented on code in PR #1170:
URL: https://github.com/apache/lucenenet/pull/1170#discussion_r2314748496
##########
src/Lucene.Net.Replicator/Http/ReplicationService.cs:
##########
@@ -118,75 +121,151 @@ private static string
ExtractRequestParam(IReplicationRequest request, string pa
return param;
}
- // LUCENENET specific - copy method not used
-
- /// <summary>
- /// Executes the replication task.
- /// </summary>
- /// <exception cref="InvalidOperationException">required parameters
are missing</exception>
- public virtual void Perform(IReplicationRequest request,
IReplicationResponse response)
+ // Shared internal logic (generic on response)
+ private async Task ExecuteReplicationInternal<TResponse>(
+ IReplicationRequest request,
+ TResponse response,
+ Func<Stream, Task> copyStreamFunc,
+ Func<SessionToken, Task> writeTokenFunc,
+ Func<Task> flushFunc)
+ where TResponse : IBaseReplicationResponse
{
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;
}
}
catch (Exception)
{
- response.StatusCode = (int)HttpStatusCode.InternalServerError;
// propagate the failure
+ response.StatusCode = (int)HttpStatusCode.InternalServerError;
}
finally
{
- response.Flush();
+ await flushFunc();
}
}
+
+ // For sync
+ private async Task ExecuteReplicationAsync(
+ IReplicationRequest request,
+ IReplicationResponse response,
+ Func<Stream, Task> copyStreamFunc,
+ Func<SessionToken, Task> writeTokenFunc,
+ Func<Task> flushFunc)
+ {
+ await ExecuteReplicationInternal(request, response,
copyStreamFunc, writeTokenFunc, flushFunc);
+ }
+
+ // For async
+ private async Task ExecuteReplicationAsync(
+ IReplicationRequest request,
+ IAsyncReplicationResponse response,
+ Func<Stream, Task> copyStreamFunc,
+ Func<SessionToken, Task> writeTokenFunc,
+ Func<Task> flushFunc)
+ {
+ await ExecuteReplicationInternal(request, response,
copyStreamFunc, writeTokenFunc, flushFunc);
+ }
+
+ // LUCENENET specific - copy method not used
+
+ /// <summary>
+ /// Executes the replication task.
+ /// </summary>
+ /// <exception cref="InvalidOperationException">required parameters
are missing</exception>
+ public virtual void Perform(IReplicationRequest request,
IReplicationResponse response)
+ {
+ ExecuteReplicationAsync(
+ request,
+ response,
+ stream => { stream.CopyTo(response.Body); return
Task.CompletedTask; },
+ token =>
+ {
+ if (token == null)
+ {
+ response.Body.Write(new byte[] { 0 }, 0, 1);
+ }
+ else
+ {
+ response.Body.Write(new byte[] { 1 }, 0, 1);
+ token.Serialize(new DataOutputStream(response.Body));
+ }
+ return Task.CompletedTask;
+ },
+ () => { response.Flush(); return Task.CompletedTask; }
+ ).GetAwaiter().GetResult(); // // keep sync behavior
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscr...@lucenenet.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
