You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by fl...@apache.org on 2021/04/14 12:24:20 UTC
[tinkerpop] branch TINKERPOP-2019 created (now 93ab05c)
This is an automated email from the ASF dual-hosted git repository.
florianhockmann pushed a change to branch TINKERPOP-2019
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git.
at 93ab05c TINKERPOP-2019 Cancel WS operations before closing
This branch includes the following new commits:
new 93ab05c TINKERPOP-2019 Cancel WS operations before closing
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[tinkerpop] 01/01: TINKERPOP-2019 Cancel WS operations before
closing
Posted by fl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
florianhockmann pushed a commit to branch TINKERPOP-2019
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 93ab05c8d53d9573822b9012d18410c24ed469c7
Author: Florian Hockmann <fh...@florian-hockmann.de>
AuthorDate: Wed Apr 14 14:23:19 2021 +0200
TINKERPOP-2019 Cancel WS operations before closing
This should avoid rare race conditions where we apparently still call
ReceiveAsync on a Websocket that we have already closed which results in
an InvalidOperationException.
---
gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs | 13 ++++++++-----
.../src/Gremlin.Net/Driver/WebSocketConnection.cs | 13 +++++++------
2 files changed, 15 insertions(+), 11 deletions(-)
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
index 7436c3d..06fd176 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
@@ -54,6 +54,7 @@ namespace Gremlin.Net.Driver
private readonly string _sessionId;
private readonly bool _sessionEnabled;
private readonly ConcurrentQueue<RequestMessage> _writeQueue = new ConcurrentQueue<RequestMessage>();
+ private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
private readonly ConcurrentDictionary<Guid, IResponseHandlerForSingleRequestMessage> _callbackByRequestId =
new ConcurrentDictionary<Guid, IResponseHandlerForSingleRequestMessage>();
@@ -81,7 +82,7 @@ namespace Gremlin.Net.Driver
public async Task ConnectAsync()
{
- await _webSocketConnection.ConnectAsync(_uri).ConfigureAwait(false);
+ await _webSocketConnection.ConnectAsync(_uri, _cancellationTokenSource.Token).ConfigureAwait(false);
BeginReceiving();
}
@@ -107,11 +108,11 @@ namespace Gremlin.Net.Driver
private async Task ReceiveMessagesAsync()
{
- while (true)
+ while (!_cancellationTokenSource.IsCancellationRequested)
{
try
{
- var received = await _webSocketConnection.ReceiveMessageAsync().ConfigureAwait(false);
+ var received = await _webSocketConnection.ReceiveMessageAsync(_cancellationTokenSource.Token).ConfigureAwait(false);
Parse(received);
}
catch (Exception e)
@@ -198,6 +199,7 @@ namespace Gremlin.Net.Driver
{
while (_writeQueue.TryDequeue(out var msg))
{
+ _cancellationTokenSource.Token.ThrowIfCancellationRequested();
try
{
await SendMessageAsync(msg).ConfigureAwait(false);
@@ -249,7 +251,7 @@ namespace Gremlin.Net.Driver
}
var graphsonMsg = _graphSONWriter.WriteObject(message);
var serializedMsg = _messageSerializer.SerializeMessage(graphsonMsg);
- await _webSocketConnection.SendMessageAsync(serializedMsg).ConfigureAwait(false);
+ await _webSocketConnection.SendMessageAsync(serializedMsg, _cancellationTokenSource.Token).ConfigureAwait(false);
}
private RequestMessage RebuildSessionMessage(RequestMessage message)
@@ -272,11 +274,12 @@ namespace Gremlin.Net.Driver
public async Task CloseAsync()
{
Interlocked.Exchange(ref _connectionState, Closed);
-
+
if (_sessionEnabled)
{
await CloseSession().ConfigureAwait(false);
}
+ _cancellationTokenSource.Cancel();
await _webSocketConnection.CloseAsync().ConfigureAwait(false);
}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/WebSocketConnection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/WebSocketConnection.cs
index 1ea02dd..881b404 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/WebSocketConnection.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/WebSocketConnection.cs
@@ -41,9 +41,9 @@ namespace Gremlin.Net.Driver
webSocketConfiguration?.Invoke(_client.Options);
}
- public async Task ConnectAsync(Uri uri)
+ public async Task ConnectAsync(Uri uri, CancellationToken cancellationToken)
{
- await _client.ConnectAsync(uri, CancellationToken.None).ConfigureAwait(false);
+ await _client.ConnectAsync(uri, cancellationToken).ConfigureAwait(false);
}
public async Task CloseAsync()
@@ -66,14 +66,14 @@ namespace Gremlin.Net.Driver
_client.State == WebSocketState.Aborted ||
_client.State == WebSocketState.CloseSent;
- public async Task SendMessageAsync(byte[] message)
+ public async Task SendMessageAsync(byte[] message, CancellationToken cancellationToken)
{
await
- _client.SendAsync(new ArraySegment<byte>(message), MessageType, true, CancellationToken.None)
+ _client.SendAsync(new ArraySegment<byte>(message), MessageType, true, cancellationToken)
.ConfigureAwait(false);
}
- public async Task<byte[]> ReceiveMessageAsync()
+ public async Task<byte[]> ReceiveMessageAsync(CancellationToken cancellationToken)
{
using (var ms = new MemoryStream())
{
@@ -82,8 +82,9 @@ namespace Gremlin.Net.Driver
do
{
+ cancellationToken.ThrowIfCancellationRequested();
var receiveBuffer = new ArraySegment<byte>(buffer);
- received = await _client.ReceiveAsync(receiveBuffer, CancellationToken.None).ConfigureAwait(false);
+ received = await _client.ReceiveAsync(receiveBuffer, cancellationToken).ConfigureAwait(false);
ms.Write(receiveBuffer.Array, receiveBuffer.Offset, received.Count);
} while (!received.EndOfMessage);