jorgebay commented on a change in pull request #1016: Add request pipelining
and a fixed ConnectionPool size TINKERPOP-1775 and TINKERPOP-1774
URL: https://github.com/apache/tinkerpop/pull/1016#discussion_r244571763
##########
File path: gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
##########
@@ -55,87 +71,86 @@ internal class Connection : IConnection
_webSocketConnection = new
WebSocketConnection(webSocketConfiguration);
}
- public async Task<ResultSet<T>> SubmitAsync<T>(RequestMessage
requestMessage)
- {
- await SendAsync(requestMessage).ConfigureAwait(false);
- return await ReceiveAsync<T>().ConfigureAwait(false);
- }
-
public async Task ConnectAsync()
{
await
_webSocketConnection.ConnectAsync(_uri).ConfigureAwait(false);
+ BeginReceiving();
}
- public async Task CloseAsync()
- {
- await _webSocketConnection.CloseAsync().ConfigureAwait(false);
- }
+ public int NrRequestsInFlight => _callbackByRequestId.Count;
public bool IsOpen => _webSocketConnection.IsOpen;
- private async Task SendAsync(RequestMessage message)
+ public Task<ResultSet<T>> SubmitAsync<T>(RequestMessage requestMessage)
{
- var graphsonMsg = _graphSONWriter.WriteObject(message);
- var serializedMsg =
_messageSerializer.SerializeMessage(graphsonMsg);
- await
_webSocketConnection.SendMessageAsync(serializedMsg).ConfigureAwait(false);
+ var receiver = new
ResponseHandlerForSingleRequestMessage<T>(_graphSONReader);
+ _callbackByRequestId.GetOrAdd(requestMessage.RequestId, receiver);
+ _writeQueue.Enqueue(requestMessage);
+ BeginSendingMessages();
+ return receiver.Result;
}
- private async Task<ResultSet<T>> ReceiveAsync<T>()
+ private void BeginReceiving()
{
- ResultSet<T> resultSet = null;
- Dictionary<string, object> statusAttributes = null;
- ResponseStatus status;
- IAggregator aggregator = null;
- var isAggregatingSideEffects = false;
- var result = new List<T>();
- do
- {
- var received = await
_webSocketConnection.ReceiveMessageAsync().ConfigureAwait(false);
- var receivedMsg =
_messageSerializer.DeserializeMessage<ResponseMessage<JToken>>(received);
-
- status = receivedMsg.Status;
- status.ThrowIfStatusIndicatesError();
+ var state = Volatile.Read(ref _connectionState);
+ if (state == Closed) return;
+ ReceiveMessagesAsync().Forget();
+ }
- if (status.Code == ResponseStatusCode.Authenticate)
+ private async Task ReceiveMessagesAsync()
+ {
+ while (true)
+ {
+ try
{
- await AuthenticateAsync().ConfigureAwait(false);
+ var received = await
_webSocketConnection.ReceiveMessageAsync().ConfigureAwait(false);
+ Parse(received);
}
- else if (status.Code != ResponseStatusCode.NoContent)
+ catch (Exception e)
{
- var receivedData = typeof(T) == typeof(JToken)
- ? new[] { receivedMsg.Result.Data }
- : _graphSONReader.ToObject(receivedMsg.Result.Data);
-
- foreach (var d in receivedData)
- if
(receivedMsg.Result.Meta.ContainsKey(Tokens.ArgsSideEffectKey))
- {
- if (aggregator == null)
- aggregator =
- new AggregatorFactory().GetAggregatorFor(
- (string)
receivedMsg.Result.Meta[Tokens.ArgsAggregateTo]);
- aggregator.Add(d);
- isAggregatingSideEffects = true;
- }
- else
- {
- result.Add(d);
- }
+ await
CloseConnectionBecauseOfFailureAsync(e).ConfigureAwait(false);
+ break;
}
+ }
+ }
- if (status.Code == ResponseStatusCode.Success || status.Code
== ResponseStatusCode.NoContent)
- {
- statusAttributes = receivedMsg.Status.Attributes;
- }
+ private void Parse(byte[] received)
+ {
+ var receivedMsg =
_messageSerializer.DeserializeMessage<ResponseMessage<JToken>>(received);
+
+ try
+ {
+ TryParseResponseMessage(receivedMsg);
+ }
+ catch (Exception e)
+ {
+ _callbackByRequestId[receivedMsg.RequestId].HandleFailure(e);
Review comment:
NIT, we can avoid the second read access to the dictionary by using:
```C#
if (_callbackByRequestId.TryRemove(receivedMsg.RequestId, out var
responseHandler))
{
responseHandler.HandleFailure(e);
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
