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_r243807446
##########
File path: gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
##########
@@ -55,87 +71,75 @@ 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);
+ ReceiveNext();
}
- 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 ReceiveNext()
{
- 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();
-
- if (status.Code == ResponseStatusCode.Authenticate)
+ var state = Volatile.Read(ref _connectionState);
+ if (state == Closed) return;
+ _webSocketConnection.ReceiveMessageAsync().ContinueWith(
Review comment:
I think we can do something similar to the send portion.
Otherwise, faulted tasks at this level will be unhandled. We should invoke
all the pending callbacks if there is a single receiving failure (network
related).
----------------------------------------------------------------
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
