You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tinkerpop.apache.org by xi...@apache.org on 2026/03/12 17:06:34 UTC
(tinkerpop) 02/03: Clean up websocket connection and fix test failures
This is an automated email from the ASF dual-hosted git repository.
xiazcy pushed a commit to branch dotnet-http-connection
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
commit 704ad263430c35b3fbd30b78e32fd8f979e99049
Author: Yang Xia <xi...@gmail.com>
AuthorDate: Wed Mar 11 19:31:11 2026 -0700
Clean up websocket connection and fix test failures
---
.../src/Gremlin.Net/Driver/Connection.cs | 258 ++++-------
.../src/Gremlin.Net/Driver/ConnectionFactory.cs | 49 ---
.../src/Gremlin.Net/Driver/ConnectionPool.cs | 298 -------------
.../Gremlin.Net/Driver/ConnectionPoolSettings.cs | 146 ------
...ConnectionSettings.cs => ConnectionSettings.cs} | 2 +-
.../Driver/Exceptions/ConnectionClosedException.cs | 65 ---
.../Exceptions/ConnectionPoolBusyException.cs | 60 ---
.../Exceptions/NoConnectionAvailableException.cs | 56 ---
.../Exceptions/ServerUnavailableException.cs | 45 --
.../src/Gremlin.Net/Driver/GremlinClient.cs | 17 +-
.../src/Gremlin.Net/Driver/HttpConnection.cs | 183 --------
.../src/Gremlin.Net/Driver/IClientWebSocket.cs | 43 --
.../src/Gremlin.Net/Driver/IConnection.cs | 39 --
.../src/Gremlin.Net/Driver/IConnectionFactory.cs | 30 --
gremlin-dotnet/src/Gremlin.Net/Driver/Log.cs | 16 +-
.../Gremlin.Net/Driver/Messages/ResponseResult.cs | 56 ---
.../Gremlin.Net/Driver/Messages/ResponseStatus.cs | 72 ---
.../Driver/Messages/ResponseStatusCode.cs | 157 -------
.../src/Gremlin.Net/Driver/ProxyClientWebSocket.cs | 81 ----
.../src/Gremlin.Net/Driver/ProxyConnection.cs | 67 ---
.../Driver/Remote/DriverRemoteConnection.cs | 1 +
.../ResponseHandlerForSingleRequestMessage.cs | 65 ---
gremlin-dotnet/src/Gremlin.Net/Driver/Tokens.cs | 44 +-
.../src/Gremlin.Net/Driver/WebSocketConnection.cs | 157 -------
.../src/Gremlin.Net/Driver/WebSocketSettings.cs | 58 ---
.../Process/Traversal/GraphTraversal.cs | 6 +-
.../Process/Traversal/GraphTraversalSource.cs | 8 +-
.../Structure/IO/GraphBinary4/GraphBinaryWriter.cs | 2 +-
.../IO/GraphBinary4/ResponseMessageSerializer.cs | 1 -
.../Driver/ConnectionPoolTests.cs | 74 ----
.../Driver/GremlinClientAuthenticationTests.cs | 64 ---
.../GremlinClientBehaviorIntegrationTests.cs | 89 ----
.../Driver/GremlinClientTests.cs | 40 --
.../Driver/MessagesTests.cs | 24 -
.../Gherkin/CommonSteps.cs | 49 ++-
.../Gherkin/ScenarioData.cs | 123 +++---
.../DriverRemoteConnection/GraphTraversalTests.cs | 11 +-
.../GraphTraversalTransactionTests.cs | 6 +-
.../Driver/ConnectionPoolSettingsTests.cs | 70 ---
.../Driver/ConnectionPoolTests.cs | 394 -----------------
.../Gremlin.Net.UnitTest/Driver/ConnectionTests.cs | 488 +++++++++++----------
.../Driver/GremlinClientTests.cs | 2 +-
.../Driver/HttpConnectionTests.cs | 341 --------------
.../Driver/Messages/RequestMessageTests.cs | 110 +++++
.../GraphBinary4/RequestMessageSerializerTests.cs | 93 ++++
.../GraphBinary4/ResponseMessageSerializerTests.cs | 210 +++++++++
46 files changed, 894 insertions(+), 3376 deletions(-)
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
index 3d12fb1684..d7b5f119b6 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
@@ -22,210 +22,138 @@
#endregion
using System;
-using System.Collections.Concurrent;
using System.Collections.Generic;
+using System.IO;
+using System.IO.Compression;
+using System.Linq;
+using System.Net.Http;
+using System.Net.Http.Headers;
using System.Threading;
using System.Threading.Tasks;
using Gremlin.Net.Driver.Messages;
using Gremlin.Net.Process;
+using Gremlin.Net.Structure.IO;
namespace Gremlin.Net.Driver
{
- internal interface IResponseHandlerForSingleRequestMessage
+ /// <summary>
+ /// HTTP-based connection that sends requests via HTTP POST to Gremlin Server.
+ /// </summary>
+ internal class Connection : IDisposable
{
- void HandleReceived(ResponseMessage<List<object>> received);
- void Finalize(Dictionary<string, object> statusAttributes);
- void HandleFailure(Exception objException);
- void Cancel();
- }
+ private const string GraphBinaryMimeType = SerializationTokens.GraphBinary4MimeType;
- internal class Connection : IConnection
- {
- private readonly IMessageSerializer _messageSerializer;
+ private readonly HttpClient _httpClient;
private readonly Uri _uri;
- private readonly IWebSocketConnection _webSocketConnection;
- private readonly string? _username;
- private readonly string? _password;
- private readonly string? _sessionId;
-
- private readonly ConcurrentQueue<(RequestMessage msg, CancellationToken cancellationToken)> _writeQueue = new();
-
- private readonly ConcurrentDictionary<Guid, IResponseHandlerForSingleRequestMessage> _callbackByRequestId =
- new();
+ private readonly IMessageSerializer _serializer;
+ private readonly bool _enableCompression;
+ private readonly bool _enableUserAgentOnConnect;
+ private readonly bool _bulkResults;
+ // Interceptor slot reserved for future spec
+ // private readonly IReadOnlyList<Func<HttpRequestMessage, Task>> _interceptors;
- private readonly ConcurrentDictionary<Guid, CancellationTokenRegistration> _cancellationByRequestId = new();
- private int _connectionState = 0;
- private int _writeInProgress = 0;
- private const int Closed = 1;
-
- public Connection(IWebSocketConnection webSocketConnection, Uri uri, string? username, string? password,
- IMessageSerializer messageSerializer, string? sessionId)
+ public Connection(Uri uri, IMessageSerializer serializer,
+ ConnectionSettings settings)
{
_uri = uri;
- _username = username;
- _password = password;
- _sessionId = sessionId;
- _messageSerializer = messageSerializer;
- _webSocketConnection = webSocketConnection;
- }
+ _serializer = serializer;
+ _enableCompression = settings.EnableCompression;
+ _enableUserAgentOnConnect = settings.EnableUserAgentOnConnect;
+ _bulkResults = settings.BulkResults;
- public async Task ConnectAsync(CancellationToken cancellationToken)
+#if NET6_0_OR_GREATER
+ var handler = new SocketsHttpHandler
+ {
+ PooledConnectionIdleTimeout = settings.IdleConnectionTimeout,
+ MaxConnectionsPerServer = settings.MaxConnectionsPerServer,
+ ConnectTimeout = settings.ConnectionTimeout,
+ KeepAlivePingTimeout = settings.KeepAliveInterval,
+ };
+ _httpClient = new HttpClient(handler);
+#else
+ _httpClient = new HttpClient();
+ _httpClient.Timeout = settings.ConnectionTimeout;
+#endif
+ }
+
+ /// <summary>
+ /// Constructor that accepts a pre-configured HttpClient (for testing).
+ /// </summary>
+ internal Connection(Uri uri, IMessageSerializer serializer,
+ ConnectionSettings settings, HttpClient httpClient)
{
- await _webSocketConnection.ConnectAsync(_uri, cancellationToken).ConfigureAwait(false);
- BeginReceiving();
+ _uri = uri;
+ _serializer = serializer;
+ _enableCompression = settings.EnableCompression;
+ _enableUserAgentOnConnect = settings.EnableUserAgentOnConnect;
+ _bulkResults = settings.BulkResults;
+ _httpClient = httpClient;
}
- public int NrRequestsInFlight => _callbackByRequestId.Count;
-
- public bool IsOpen => _webSocketConnection.IsOpen && Volatile.Read(ref _connectionState) != Closed;
-
- public Task<ResultSet<T>> SubmitAsync<T>(RequestMessage requestMessage, CancellationToken cancellationToken)
+ public async Task<ResultSet<T>> SubmitAsync<T>(RequestMessage requestMessage,
+ CancellationToken cancellationToken = default)
{
- var receiver = new ResponseHandlerForSingleRequestMessage<T>();
- var requestId = Guid.NewGuid();
- _callbackByRequestId.GetOrAdd(requestId, receiver);
+ var requestBytes = await _serializer.SerializeMessageAsync(requestMessage, cancellationToken)
+ .ConfigureAwait(false);
- _cancellationByRequestId.GetOrAdd(requestId, cancellationToken.Register(() =>
- {
- if (_callbackByRequestId.TryRemove(requestId, out var responseHandler))
- {
- responseHandler.Cancel();
- }
- }));
- _writeQueue.Enqueue((requestMessage, cancellationToken));
- BeginSendingMessages();
- return receiver.Result;
- }
+ using var content = new ByteArrayContent(requestBytes);
+ content.Headers.ContentType = new MediaTypeHeaderValue(GraphBinaryMimeType);
- private void BeginReceiving()
- {
- var state = Volatile.Read(ref _connectionState);
- if (state == Closed) return;
- ReceiveMessagesAsync().Forget();
- }
+ using var httpRequest = new HttpRequestMessage(HttpMethod.Post, _uri);
+ httpRequest.Content = content;
+ httpRequest.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue(GraphBinaryMimeType));
- private async Task ReceiveMessagesAsync()
- {
- while (true)
+ if (_enableCompression)
{
- try
- {
- var received = await _webSocketConnection.ReceiveMessageAsync().ConfigureAwait(false);
- await HandleReceivedAsync(received).ConfigureAwait(false);
- }
- catch (Exception e)
- {
- await CloseConnectionBecauseOfFailureAsync(e).ConfigureAwait(false);
- break;
- }
+ httpRequest.Headers.AcceptEncoding.Add(new StringWithQualityHeaderValue("deflate"));
}
- }
- private async Task HandleReceivedAsync(byte[] received)
- {
- var receivedMsg = await _messageSerializer.DeserializeMessageAsync(received).ConfigureAwait(false);
- try
+ if (_enableUserAgentOnConnect)
{
- HandleReceivedMessage(receivedMsg);
+ httpRequest.Headers.TryAddWithoutValidation("User-Agent", Utils.UserAgent);
}
- catch (Exception e)
- {
- // Propagate failure to all pending handlers
- foreach (var cb in _callbackByRequestId.Values)
- {
- cb.HandleFailure(e);
- }
- _callbackByRequestId.Clear();
- DisposeCancellationRegistrations();
- }
- }
- private void HandleReceivedMessage(ResponseMessage<List<object>> receivedMsg)
- {
- if (receivedMsg.StatusCode != 200)
+ if (_bulkResults)
{
- throw new Exceptions.ResponseException(receivedMsg.StatusCode, receivedMsg.StatusMessage,
- receivedMsg.Exception);
+ httpRequest.Headers.Add("bulkResults", "true");
}
- // In the 4.0 format, all results arrive in a single response
- foreach (var handler in _callbackByRequestId.Values)
- {
- handler.HandleReceived(receivedMsg);
- handler.Finalize(new Dictionary<string, object>());
- }
- _callbackByRequestId.Clear();
- DisposeCancellationRegistrations();
- }
+ // Future: apply interceptors here
- private void BeginSendingMessages()
- {
- if (Interlocked.CompareExchange(ref _writeInProgress, 1, 0) != 0)
- return;
- SendMessagesFromQueueAsync().Forget();
- }
+ using var response = await _httpClient.SendAsync(httpRequest, cancellationToken)
+ .ConfigureAwait(false);
- private async Task SendMessagesFromQueueAsync()
- {
- while (_writeQueue.TryDequeue(out var msg))
- {
- try
- {
- await SendMessageAsync(msg.msg, msg.cancellationToken).ConfigureAwait(false);
- }
- catch (OperationCanceledException e) when (msg.cancellationToken == e.CancellationToken)
- {
- // Send was cancelled for this message
- }
- catch (Exception e)
- {
- await CloseConnectionBecauseOfFailureAsync(e).ConfigureAwait(false);
- break;
- }
- }
- Interlocked.CompareExchange(ref _writeInProgress, 0, 1);
+ var responseBytes = await ReadResponseBytesAsync(response).ConfigureAwait(false);
- if (!_writeQueue.IsEmpty && Interlocked.CompareExchange(ref _writeInProgress, 1, 0) == 0)
- {
- await SendMessagesFromQueueAsync().ConfigureAwait(false);
- }
- }
+ var responseMessage = await _serializer.DeserializeMessageAsync(responseBytes, cancellationToken)
+ .ConfigureAwait(false);
- private async Task CloseConnectionBecauseOfFailureAsync(Exception exception)
- {
- EmptyWriteQueue();
- await CloseAsync().ConfigureAwait(false);
- NotifyAboutConnectionFailure(exception);
+ return BuildResultSet<T>(responseMessage);
}
- private void EmptyWriteQueue()
+ private static async Task<byte[]> ReadResponseBytesAsync(HttpResponseMessage response)
{
- while (_writeQueue.TryDequeue(out _))
+ var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
+ if (response.Content.Headers.ContentEncoding.Contains("deflate"))
{
+ using var deflateStream = new DeflateStream(stream, CompressionMode.Decompress);
+ using var ms = new MemoryStream();
+ await deflateStream.CopyToAsync(ms).ConfigureAwait(false);
+ return ms.ToArray();
}
+ using var memoryStream = new MemoryStream();
+ await stream.CopyToAsync(memoryStream).ConfigureAwait(false);
+ return memoryStream.ToArray();
}
- private void NotifyAboutConnectionFailure(Exception exception)
+ private static ResultSet<T> BuildResultSet<T>(ResponseMessage<List<object>> responseMessage)
{
- foreach (var cb in _callbackByRequestId.Values)
+ var results = new List<T>();
+ foreach (var item in responseMessage.Result)
{
- cb.HandleFailure(exception);
+ results.Add((T)item);
}
- _callbackByRequestId.Clear();
- DisposeCancellationRegistrations();
- }
-
- private async Task SendMessageAsync(RequestMessage message, CancellationToken cancellationToken)
- {
- var serializedMsg = await _messageSerializer.SerializeMessageAsync(message, cancellationToken)
- .ConfigureAwait(false);
- await _webSocketConnection.SendMessageAsync(serializedMsg, cancellationToken).ConfigureAwait(false);
- }
-
- public async Task CloseAsync()
- {
- Interlocked.Exchange(ref _connectionState, Closed);
- await _webSocketConnection.CloseAsync().ConfigureAwait(false);
+ return new ResultSet<T>(results, new Dictionary<string, object>());
}
#region IDisposable Support
@@ -244,22 +172,12 @@ namespace Gremlin.Net.Driver
{
if (disposing)
{
- _webSocketConnection?.Dispose();
- DisposeCancellationRegistrations();
+ _httpClient?.Dispose();
}
_disposed = true;
}
}
- private void DisposeCancellationRegistrations()
- {
- foreach (var cancellation in _cancellationByRequestId.Values)
- {
- cancellation.Dispose();
- }
- _cancellationByRequestId.Clear();
- }
-
#endregion
}
}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionFactory.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionFactory.cs
deleted file mode 100644
index 4657172d47..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionFactory.cs
+++ /dev/null
@@ -1,49 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-namespace Gremlin.Net.Driver
-{
- internal class ConnectionFactory : IConnectionFactory
- {
- private readonly WebSocketSettings _webSocketSettings;
- private readonly GremlinServer _gremlinServer;
- private readonly string? _sessionId;
- private readonly IMessageSerializer _messageSerializer;
-
- public ConnectionFactory(GremlinServer gremlinServer, IMessageSerializer messageSerializer,
- WebSocketSettings webSocketSettings, string? sessionId)
- {
- _gremlinServer = gremlinServer;
- _messageSerializer = messageSerializer;
- _sessionId = sessionId;
- _webSocketSettings = webSocketSettings;
- }
-
- public IConnection CreateConnection()
- {
- return new Connection(
- new WebSocketConnection(ProxyClientWebSocket.CreateClientWebSocket(), _webSocketSettings),
- _gremlinServer.Uri, _gremlinServer.Username, _gremlinServer.Password, _messageSerializer, _sessionId);
- }
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs
deleted file mode 100644
index e4f4a72ae2..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPool.cs
+++ /dev/null
@@ -1,298 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
-using Gremlin.Net.Driver.Exceptions;
-using Gremlin.Net.Process;
-using Microsoft.Extensions.Logging;
-using Polly;
-
-namespace Gremlin.Net.Driver
-{
- internal class ConnectionPool : IDisposable
- {
- private const int ConnectionIndexOverflowLimit = int.MaxValue - 1000000;
-
- private readonly IConnectionFactory _connectionFactory;
- private readonly CopyOnWriteCollection<IConnection> _connections = new CopyOnWriteCollection<IConnection>();
-
- private readonly ConcurrentDictionary<IConnection, byte> _deadConnections =
- new ConcurrentDictionary<IConnection, byte>();
- private readonly ConnectionPoolSettings _settings;
- private readonly ILogger<ConnectionPool> _logger;
- private int _connectionIndex;
- private int _poolState;
- private const int PoolIdle = 0;
- private const int PoolPopulationInProgress = 1;
- private readonly CancellationTokenSource _cts = new CancellationTokenSource();
-
- public ConnectionPool(IConnectionFactory connectionFactory, ConnectionPoolSettings settings,
- ILogger<ConnectionPool> logger)
- {
- _connectionFactory = connectionFactory;
- _settings = settings;
- _logger = logger;
- ReplaceDeadConnectionsAsync().WaitUnwrap();
- }
-
- public int NrConnections => _connections.Count;
-
- public IConnection GetAvailableConnection()
- {
- var connection = Policy.Handle<ServerUnavailableException>()
- .WaitAndRetry(_settings.ReconnectionAttempts, ComputeRetrySleepDuration, onRetry:
- (_, _, retryCount, _) =>
- {
- _logger.CouldNotGetConnectionFromPool(retryCount, _settings.ReconnectionAttempts);
- })
- .Execute(GetConnectionFromPool);
-
- return ProxiedConnection(connection);
- }
-
- private bool IsHealthy => _deadConnections.IsEmpty && _connections.Count == _settings.PoolSize;
-
- private TimeSpan ComputeRetrySleepDuration(int retryAttemptNr)
- {
- return TimeSpan.FromTicks(_settings.ReconnectionBaseDelay.Ticks * (long) Math.Pow(2, retryAttemptNr));
- }
-
- /// <summary>
- /// Replaces dead connections.
- /// </summary>
- /// <returns>True if the pool was repaired, false if repairing was not necessary.</returns>
- private async Task<bool> EnsurePoolIsHealthyAsync()
- {
- if (IsHealthy) return false;
- var poolState = Interlocked.CompareExchange(ref _poolState, PoolPopulationInProgress, PoolIdle);
- if (poolState == PoolPopulationInProgress) return false;
- try
- {
- await ReplaceDeadConnectionsAsync().ConfigureAwait(false);
- }
- finally
- {
- // We need to remove the PoolPopulationInProgress flag again even if an exception occurred, so we don't block the pool population for ever
- Interlocked.CompareExchange(ref _poolState, PoolIdle, PoolPopulationInProgress);
- }
-
- return true;
- }
-
- private async Task ReplaceDeadConnectionsAsync()
- {
- RemoveDeadConnections();
-
- await FillPoolAsync().ConfigureAwait(false);
- }
-
- private void RemoveDeadConnections()
- {
- if (_deadConnections.IsEmpty) return;
-
- foreach (var deadConnection in _deadConnections.Keys)
- {
- if (_connections.TryRemove(deadConnection))
- {
- DefinitelyDestroyConnection(deadConnection);
- }
- }
-
- _deadConnections.Clear();
- }
-
- private async Task FillPoolAsync()
- {
- var nrConnectionsToCreate = _settings.PoolSize - _connections.Count;
- _logger.FillingPool(nrConnectionsToCreate);
- var connectionCreationTasks = new List<Task<IConnection>>(nrConnectionsToCreate);
- try
- {
- for (var i = 0; i < nrConnectionsToCreate; i++)
- {
- connectionCreationTasks.Add(CreateNewConnectionAsync());
- }
-
- var createdConnections = await Task.WhenAll(connectionCreationTasks).ConfigureAwait(false);
- _connections.AddRange(createdConnections);
- }
- catch (Exception)
- {
- // Dispose all connections that were already created
- foreach (var creationTask in connectionCreationTasks)
- {
- if (creationTask.IsCompleted)
- creationTask.Result?.Dispose();
- }
- throw;
- }
-
- if (_disposed)
- {
- await CloseAndRemoveAllConnectionsAsync().ConfigureAwait(false);
- }
- }
-
- private async Task<IConnection> CreateNewConnectionAsync()
- {
- var newConnection = _connectionFactory.CreateConnection();
- try
- {
- await newConnection.ConnectAsync(_cts.Token).ConfigureAwait(false);
- }
- catch (Exception)
- {
- // Dispose created connection if the connection establishment failed
- newConnection.Dispose();
- throw;
- }
- return newConnection;
- }
-
- private IConnection GetConnectionFromPool()
- {
- var connections = _connections.Snapshot;
- if (connections.Length < _settings.PoolSize)
- {
- TriggerReplacementOfDeadConnections();
- }
- if (connections.Length == 0) throw new ServerUnavailableException();
- return TryGetAvailableConnection(connections);
- }
-
- private IConnection TryGetAvailableConnection(IConnection[] connections)
- {
- var index = Interlocked.Increment(ref _connectionIndex);
- ProtectIndexFromOverflowing(index);
-
- var closedConnections = 0;
- for (var i = 0; i < connections.Length; i++)
- {
- var connection = connections[(index + i) % connections.Length];
- if (connection.NrRequestsInFlight >= _settings.MaxInProcessPerConnection) continue;
- if (!connection.IsOpen)
- {
- ReplaceConnection(connection);
- closedConnections++;
- continue;
- }
- return connection;
- }
-
- if (connections.Length > closedConnections)
- {
- throw new ConnectionPoolBusyException(_settings.PoolSize, _settings.MaxInProcessPerConnection);
- }
- else
- {
- throw new ServerUnavailableException();
- }
- }
-
- private void ProtectIndexFromOverflowing(int currentIndex)
- {
- if (currentIndex > ConnectionIndexOverflowLimit)
- Interlocked.Exchange(ref _connectionIndex, 0);
- }
-
- private void ReplaceConnection(IConnection connection)
- {
- RemoveConnectionFromPool(connection);
- TriggerReplacementOfDeadConnections();
- }
-
- private void RemoveConnectionFromPool(IConnection connection)
- {
- _logger.RemovingClosedConnectionFromPool();
- _deadConnections.TryAdd(connection, 0);
- }
-
- private void TriggerReplacementOfDeadConnections()
- {
- ReplaceClosedConnectionsAsync().Forget();
- }
-
- private async Task ReplaceClosedConnectionsAsync()
- {
- var poolWasPopulated = await EnsurePoolIsHealthyAsync().ConfigureAwait(false);
- // Another connection could have been removed already, check if another population is necessary
- if (poolWasPopulated && !_disposed)
- await ReplaceClosedConnectionsAsync().ConfigureAwait(false);
- }
-
- private IConnection ProxiedConnection(IConnection connection)
- {
- return new ProxyConnection(connection, ReplaceConnectionIfItWasClosed);
- }
-
- private void ReplaceConnectionIfItWasClosed(IConnection connection)
- {
- if (connection.IsOpen) return;
- ReplaceConnection(connection);
- }
-
- private async Task CloseAndRemoveAllConnectionsAsync()
- {
- foreach (var connection in _connections.RemoveAndGetAll())
- {
- await connection.CloseAsync().ConfigureAwait(false);
- DefinitelyDestroyConnection(connection);
- }
- }
-
- private void DefinitelyDestroyConnection(IConnection connection)
- {
- connection.Dispose();
- }
-
- #region IDisposable Support
-
- private bool _disposed;
-
- public void Dispose()
- {
- Dispose(true);
- GC.SuppressFinalize(this);
- }
-
- protected virtual void Dispose(bool disposing)
- {
- if (!_disposed)
- {
- if (disposing)
- {
- _cts.Cancel();
- CloseAndRemoveAllConnectionsAsync().WaitUnwrap();
- _cts.Dispose();
- }
- _disposed = true;
- }
- }
-
- #endregion
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPoolSettings.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPoolSettings.cs
deleted file mode 100644
index 0f49cf964b..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionPoolSettings.cs
+++ /dev/null
@@ -1,146 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using Gremlin.Net.Driver.Exceptions;
-
-namespace Gremlin.Net.Driver
-{
- /// <summary>
- /// Holds settings for the <see cref="ConnectionPool"/>.
- /// </summary>
- public class ConnectionPoolSettings
- {
- private int _poolSize = DefaultPoolSize;
- private int _maxInProcessPerConnection = DefaultMaxInProcessPerConnection;
- private int _reconnectionAttempts = DefaultReconnectionAttempts;
- private TimeSpan _reconnectionBaseDelay = DefaultReconnectionBaseDelay;
- private const int DefaultPoolSize = 8;
- private const int DefaultMaxInProcessPerConnection = 16;
- private const int DefaultReconnectionAttempts = 4;
- private static readonly TimeSpan DefaultReconnectionBaseDelay = TimeSpan.FromSeconds(1);
- internal const bool DefaultEnableUserAgentOnConnect = true;
-
- /// <summary>
- /// Gets or sets the size of the connection pool.
- /// </summary>
- /// <exception cref="ArgumentOutOfRangeException">The specified pool size is less than or equal to zero.</exception>
- /// <remarks>
- /// The default value is 8.
- /// </remarks>
- public int PoolSize
- {
- get => _poolSize;
- set
- {
- if (value <= 0)
- throw new ArgumentOutOfRangeException(nameof(PoolSize), $"{nameof(PoolSize)} must be > 0!");
- _poolSize = value;
- }
- }
-
- /// <summary>
- /// Gets or sets the maximum number of in-flight requests that can occur on a connection.
- /// </summary>
- /// <exception cref="ArgumentOutOfRangeException">The specified number is less than or equal to zero.</exception>
- /// <remarks>
- /// The default value is 16. A <see cref="ConnectionPoolBusyException" /> is thrown if this limit is reached
- /// on all connections when a new request comes in.
- /// </remarks>
- public int MaxInProcessPerConnection
- {
- get => _maxInProcessPerConnection;
- set
- {
- if (value <= 0)
- throw new ArgumentOutOfRangeException(nameof(MaxInProcessPerConnection),
- $"{nameof(MaxInProcessPerConnection)} must be > 0!");
- _maxInProcessPerConnection = value;
- }
- }
-
- /// <summary>
- /// Gets or sets the number of attempts to get an open connection from the pool to submit a request.
- /// </summary>
- /// <exception cref="ArgumentOutOfRangeException">The number of attempts specified is less than zero.</exception>
- /// <remarks>
- /// The driver always tries to reconnect to a server in the background after it has noticed that a
- /// connection is dead. This setting only specifies how often the driver will retry to get an open
- /// connection from its pool when no open connection is available to submit a request.
- /// These retries give the driver time to establish new connections to the server that might have been
- /// unavailable temporarily or that might have closed the connections, e.g., because they were idle for some
- /// time.
- ///
- /// The default value is 4. A <see cref="ServerUnavailableException" /> is thrown if the server can still
- /// not be reached after this many retry attempts.
- ///
- /// Setting this to zero means that the exception is thrown immediately when no open connection is available
- /// to submit a request. The driver will however still try to reconnect to the server in the background for
- /// subsequent requests.
- /// </remarks>
- public int ReconnectionAttempts
- {
- get => _reconnectionAttempts;
- set
- {
- if (value < 0)
- throw new ArgumentOutOfRangeException(nameof(ReconnectionAttempts),
- $"{ReconnectionAttempts} must be >= 0!");
- _reconnectionAttempts = value;
- }
- }
-
- /// <summary>
- /// Gets or sets the base delay used for the exponential backoff for the reconnection attempts.
- /// </summary>
- /// <exception cref="ArgumentOutOfRangeException">The specified delay is negative or zero.</exception>
- /// <remarks>
- /// The driver employs an exponential backoff strategy that uses this base delay for its reconnection
- /// attempts. With a base delay of 100 ms for example, retries happen after 100 ms, 200 ms, 400 ms, 800 ms,
- /// and so on, assuming that enough <see cref="ReconnectionAttempts"/> are configured.
- ///
- /// The default value is 1 second.
- /// </remarks>
- public TimeSpan ReconnectionBaseDelay
- {
- get => _reconnectionBaseDelay;
- set
- {
- if (value <= TimeSpan.Zero)
- throw new ArgumentOutOfRangeException(nameof(ReconnectionBaseDelay),
- $"{nameof(ReconnectionBaseDelay)} must be > 0!");
- _reconnectionBaseDelay = value;
- }
- }
-
- /// <summary>
- /// Gets or sets whether a connection pool will send a user agent during web socket handshakes
- /// </summary>
- /// <remarks>
- /// The default value is true. When enabled, user agents will only be sent during the web socket
- /// handshake. User agent follows the form:
- /// [Application Name] [GLV Name].[Version] [Language Runtime Version] [OS].[Version] [CPU Architecture]
- /// </remarks>
- public bool EnableUserAgentOnConnect { get; set; } = DefaultEnableUserAgentOnConnect;
- }
-}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/HttpConnectionSettings.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionSettings.cs
similarity index 98%
rename from gremlin-dotnet/src/Gremlin.Net/Driver/HttpConnectionSettings.cs
rename to gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionSettings.cs
index 2975c77adc..0d8df4a6a7 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/HttpConnectionSettings.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/ConnectionSettings.cs
@@ -28,7 +28,7 @@ namespace Gremlin.Net.Driver
/// <summary>
/// Holds settings for the HTTP connection.
/// </summary>
- public class HttpConnectionSettings
+ public class ConnectionSettings
{
/// <summary>
/// The default TCP connection timeout.
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ConnectionClosedException.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ConnectionClosedException.cs
deleted file mode 100644
index e03c7f69a3..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ConnectionClosedException.cs
+++ /dev/null
@@ -1,65 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Net.WebSockets;
-
-namespace Gremlin.Net.Driver.Exceptions
-{
- /// <summary>
- /// The exception that is thrown for requests that in-flight when the underlying websocket is closed by the server.
- /// Requests that receive this exception will be in a non-deterministic state and may need to be retried.
- /// </summary>
- /// <remarks>
- /// It is recommended to use the same request retry policy that is applied when a <see cref="System.Net.WebSockets.WebSocketException" />,
- /// <see cref="System.Net.Sockets.SocketException"/>, <see cref="System.IO.IOException" />.
- /// </remarks>
- public class ConnectionClosedException : Exception
- {
- /// <summary>
- /// Gets the well-known WebSocket close status code provided by the server.
- /// </summary>
- public WebSocketCloseStatus? Status { get; }
-
- /// <summary>
- /// Gets the Websocket closure description provided by the server.
- /// </summary>
- public string? Description { get; }
-
- /// <summary>
- /// Initializes a new instance of the <see cref="ConnectionClosedException" /> class.
- /// </summary>
- public ConnectionClosedException(WebSocketCloseStatus? status, string? description)
- : base(CreateMessage(status, description))
- {
- Status = status;
- Description = description;
- }
-
- private static string CreateMessage(WebSocketCloseStatus? status, string? description)
- {
- return $"Connection closed by server. CloseStatus: {status?.ToString() ?? "<none>"}, CloseDescription: {description ?? "<none>"}." +
- " Any in-progress requests on the connection will be in an unknown state, and may need to be retried.";
- }
- }
-}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ConnectionPoolBusyException.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ConnectionPoolBusyException.cs
deleted file mode 100644
index 6cd40ff7df..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ConnectionPoolBusyException.cs
+++ /dev/null
@@ -1,60 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-namespace Gremlin.Net.Driver.Exceptions
-{
- /// <summary>
- /// The exception that is thrown when all connections in the pool have reached their maximum number of in-flight
- /// requests.
- /// </summary>
- public class ConnectionPoolBusyException : NoConnectionAvailableException
- {
- /// <summary>
- /// Gets the size of the connection pool.
- /// </summary>
- public int PoolSize { get; }
-
- /// <summary>
- /// Gets the maximum number of in-flight requests that can occur on a connection.
- /// </summary>
- public int MaxInProcessPerConnection { get; }
-
- /// <summary>
- /// Initializes a new instance of the <see cref="ConnectionPoolBusyException" /> class.
- /// </summary>
- public ConnectionPoolBusyException(int poolSize, int maxInProcessPerConnection)
- : base(CreateMessage(poolSize, maxInProcessPerConnection))
- {
- PoolSize = poolSize;
- MaxInProcessPerConnection = maxInProcessPerConnection;
- }
-
- private static string CreateMessage(int poolSize, int maxInProcessPerConnection)
- {
- return $"All {poolSize} connections have reached their " +
- $"{nameof(ConnectionPoolSettings.MaxInProcessPerConnection)} limit of {maxInProcessPerConnection}." +
- $" Consider increasing either the {nameof(ConnectionPoolSettings.PoolSize)} or the " +
- $"{nameof(ConnectionPoolSettings.MaxInProcessPerConnection)} limit.";
- }
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/NoConnectionAvailableException.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/NoConnectionAvailableException.cs
deleted file mode 100644
index d62436c76d..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/NoConnectionAvailableException.cs
+++ /dev/null
@@ -1,56 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-
-namespace Gremlin.Net.Driver.Exceptions
-{
- /// <summary>
- /// The exception that is thrown when no connection is available to service a request.
- /// </summary>
- public class NoConnectionAvailableException : Exception
- {
- /// <summary>
- /// Initializes a new instance of the <see cref="NoConnectionAvailableException" /> class.
- /// </summary>
- public NoConnectionAvailableException()
- {
- }
-
- /// <summary>
- /// Initializes a new instance of the <see cref="NoConnectionAvailableException" /> class.
- /// </summary>
- public NoConnectionAvailableException(string message)
- : base(message)
- {
- }
-
- /// <summary>
- /// Initializes a new instance of the <see cref="NoConnectionAvailableException" /> class.
- /// </summary>
- public NoConnectionAvailableException(string message, Exception inner)
- : base(message, inner)
- {
- }
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ServerUnavailableException.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ServerUnavailableException.cs
deleted file mode 100644
index a4b467aab8..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Exceptions/ServerUnavailableException.cs
+++ /dev/null
@@ -1,45 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-namespace Gremlin.Net.Driver.Exceptions
-{
- /// <summary>
- /// The exception that is thrown when no connection is available to the <see cref="GremlinServer"/>.
- /// </summary>
- public class ServerUnavailableException : NoConnectionAvailableException
- {
- /// <summary>
- /// Initializes a new instance of the <see cref="ServerUnavailableException" /> class.
- /// </summary>
- public ServerUnavailableException()
- : base(CreateMessage())
- {
- }
-
- private static string CreateMessage()
- {
- return
- "No connection to the server available which most likely means that the server is completely unavailable.";
- }
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClient.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClient.cs
index 280afe5f03..6445ac6621 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClient.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClient.cs
@@ -36,7 +36,7 @@ namespace Gremlin.Net.Driver
/// </summary>
public class GremlinClient : IGremlinClient
{
- private readonly HttpConnection _httpConnection;
+ private readonly Connection _connection;
internal ILoggerFactory LoggerFactory { get; }
@@ -48,29 +48,32 @@ namespace Gremlin.Net.Driver
/// A <see cref="IMessageSerializer" /> instance to serialize messages sent to and received
/// from the server.
/// </param>
- /// <param name="connectionSettings">The <see cref="HttpConnectionSettings" /> for the HTTP connection.</param>
+ /// <param name="connectionSettings">The <see cref="ConnectionSettings" /> for the HTTP connection.</param>
/// <param name="loggerFactory">A factory to create loggers. If not provided, then nothing will be logged.</param>
// Interceptor slot reserved for future spec:
// IReadOnlyList<Func<HttpRequestMessage, Task>>? interceptors = null,
public GremlinClient(GremlinServer gremlinServer, IMessageSerializer? messageSerializer = null,
- HttpConnectionSettings? connectionSettings = null,
+ ConnectionSettings? connectionSettings = null,
ILoggerFactory? loggerFactory = null)
{
messageSerializer ??= new GraphBinaryMessageSerializer();
- connectionSettings ??= new HttpConnectionSettings();
+ connectionSettings ??= new ConnectionSettings();
LoggerFactory = loggerFactory ?? NullLoggerFactory.Instance;
- _httpConnection = new HttpConnection(
+ _connection = new Connection(
gremlinServer.Uri,
messageSerializer,
connectionSettings);
+
+ var logger = LoggerFactory.CreateLogger<GremlinClient>();
+ logger.InitializedHttpConnection(gremlinServer.Uri);
}
/// <inheritdoc />
public async Task<ResultSet<T>> SubmitAsync<T>(RequestMessage requestMessage,
CancellationToken cancellationToken = default)
{
- return await _httpConnection.SubmitAsync<T>(requestMessage, cancellationToken)
+ return await _connection.SubmitAsync<T>(requestMessage, cancellationToken)
.ConfigureAwait(false);
}
@@ -94,7 +97,7 @@ namespace Gremlin.Net.Driver
if (!_disposed)
{
if (disposing)
- _httpConnection?.Dispose();
+ _connection?.Dispose();
_disposed = true;
}
}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/HttpConnection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/HttpConnection.cs
deleted file mode 100644
index dd9c5274c8..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/HttpConnection.cs
+++ /dev/null
@@ -1,183 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.IO.Compression;
-using System.Linq;
-using System.Net.Http;
-using System.Net.Http.Headers;
-using System.Threading;
-using System.Threading.Tasks;
-using Gremlin.Net.Driver.Messages;
-using Gremlin.Net.Process;
-using Gremlin.Net.Structure.IO;
-
-namespace Gremlin.Net.Driver
-{
- /// <summary>
- /// HTTP-based connection that sends requests via HTTP POST to Gremlin Server.
- /// </summary>
- internal class HttpConnection : IDisposable
- {
- private const string GraphBinaryMimeType = SerializationTokens.GraphBinary4MimeType;
-
- private readonly HttpClient _httpClient;
- private readonly Uri _uri;
- private readonly IMessageSerializer _serializer;
- private readonly bool _enableCompression;
- private readonly bool _enableUserAgentOnConnect;
- private readonly bool _bulkResults;
- // Interceptor slot reserved for future spec
- // private readonly IReadOnlyList<Func<HttpRequestMessage, Task>> _interceptors;
-
- public HttpConnection(Uri uri, IMessageSerializer serializer,
- HttpConnectionSettings settings)
- {
- _uri = uri;
- _serializer = serializer;
- _enableCompression = settings.EnableCompression;
- _enableUserAgentOnConnect = settings.EnableUserAgentOnConnect;
- _bulkResults = settings.BulkResults;
-
-#if NET6_0_OR_GREATER
- var handler = new SocketsHttpHandler
- {
- PooledConnectionIdleTimeout = settings.IdleConnectionTimeout,
- MaxConnectionsPerServer = settings.MaxConnectionsPerServer,
- ConnectTimeout = settings.ConnectionTimeout,
- KeepAlivePingTimeout = settings.KeepAliveInterval,
- };
- _httpClient = new HttpClient(handler);
-#else
- _httpClient = new HttpClient();
- _httpClient.Timeout = settings.ConnectionTimeout;
-#endif
- }
-
- /// <summary>
- /// Constructor that accepts a pre-configured HttpClient (for testing).
- /// </summary>
- internal HttpConnection(Uri uri, IMessageSerializer serializer,
- HttpConnectionSettings settings, HttpClient httpClient)
- {
- _uri = uri;
- _serializer = serializer;
- _enableCompression = settings.EnableCompression;
- _enableUserAgentOnConnect = settings.EnableUserAgentOnConnect;
- _bulkResults = settings.BulkResults;
- _httpClient = httpClient;
- }
-
- public async Task<ResultSet<T>> SubmitAsync<T>(RequestMessage requestMessage,
- CancellationToken cancellationToken = default)
- {
- var requestBytes = await _serializer.SerializeMessageAsync(requestMessage, cancellationToken)
- .ConfigureAwait(false);
-
- using var content = new ByteArrayContent(requestBytes);
- content.Headers.ContentType = new MediaTypeHeaderValue(GraphBinaryMimeType);
-
- using var httpRequest = new HttpRequestMessage(HttpMethod.Post, _uri);
- httpRequest.Content = content;
- httpRequest.Headers.Accept.Add(new MediaTypeWithQualityHeaderValue(GraphBinaryMimeType));
-
- if (_enableCompression)
- {
- httpRequest.Headers.AcceptEncoding.Add(new StringWithQualityHeaderValue("deflate"));
- }
-
- if (_enableUserAgentOnConnect)
- {
- httpRequest.Headers.TryAddWithoutValidation("User-Agent", Utils.UserAgent);
- }
-
- if (_bulkResults)
- {
- httpRequest.Headers.Add("bulkResults", "true");
- }
-
- // Future: apply interceptors here
-
- using var response = await _httpClient.SendAsync(httpRequest, cancellationToken)
- .ConfigureAwait(false);
-
- var responseBytes = await ReadResponseBytesAsync(response).ConfigureAwait(false);
-
- var responseMessage = await _serializer.DeserializeMessageAsync(responseBytes, cancellationToken)
- .ConfigureAwait(false);
-
- return BuildResultSet<T>(responseMessage);
- }
-
- private static async Task<byte[]> ReadResponseBytesAsync(HttpResponseMessage response)
- {
- var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
- if (response.Content.Headers.ContentEncoding.Contains("deflate"))
- {
- using var deflateStream = new DeflateStream(stream, CompressionMode.Decompress);
- using var ms = new MemoryStream();
- await deflateStream.CopyToAsync(ms).ConfigureAwait(false);
- return ms.ToArray();
- }
- using var memoryStream = new MemoryStream();
- await stream.CopyToAsync(memoryStream).ConfigureAwait(false);
- return memoryStream.ToArray();
- }
-
- private static ResultSet<T> BuildResultSet<T>(ResponseMessage<List<object>> responseMessage)
- {
- var results = new List<T>();
- foreach (var item in responseMessage.Result)
- {
- results.Add((T)item);
- }
- return new ResultSet<T>(results, new Dictionary<string, object>());
- }
-
- #region IDisposable Support
-
- private bool _disposed;
-
- public void Dispose()
- {
- Dispose(true);
- GC.SuppressFinalize(this);
- }
-
- protected virtual void Dispose(bool disposing)
- {
- if (!_disposed)
- {
- if (disposing)
- {
- _httpClient?.Dispose();
- }
- _disposed = true;
- }
- }
-
- #endregion
- }
-}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/IClientWebSocket.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/IClientWebSocket.cs
deleted file mode 100644
index a14a6b35d3..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/IClientWebSocket.cs
+++ /dev/null
@@ -1,43 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Net.WebSockets;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace Gremlin.Net.Driver
-{
- internal interface IClientWebSocket : IDisposable
- {
- WebSocketState State { get; }
- ClientWebSocketOptions Options { get; }
- Task CloseAsync(WebSocketCloseStatus closeStatus, string statusDescription, CancellationToken cancellationToken);
- Task ConnectAsync(Uri uri, CancellationToken cancellationToken);
- Task<WebSocketReceiveResult> ReceiveAsync(ArraySegment<byte> buffer, CancellationToken cancellationToken);
- Task SendAsync(ArraySegment<byte> buffer, WebSocketMessageType messageType, bool endOfMessage, CancellationToken cancellationToken);
-#if NET6_0_OR_GREATER
- ValueTask SendAsync(ArraySegment<byte> buffer, WebSocketMessageType messageType, WebSocketMessageFlags messageFlags, CancellationToken cancellationToken);
-#endif
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/IConnection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/IConnection.cs
deleted file mode 100644
index 4043c894d6..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/IConnection.cs
+++ /dev/null
@@ -1,39 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using Gremlin.Net.Driver.Messages;
-
-namespace Gremlin.Net.Driver
-{
- internal interface IConnection : IDisposable
- {
- Task ConnectAsync(CancellationToken cancellationToken);
- Task<ResultSet<T>> SubmitAsync<T>(RequestMessage requestMessage, CancellationToken cancellationToken);
- int NrRequestsInFlight { get; }
- bool IsOpen { get; }
- Task CloseAsync();
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/IConnectionFactory.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/IConnectionFactory.cs
deleted file mode 100644
index 0c7ace286d..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/IConnectionFactory.cs
+++ /dev/null
@@ -1,30 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-namespace Gremlin.Net.Driver
-{
- internal interface IConnectionFactory
- {
- IConnection CreateConnection();
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Log.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Log.cs
index 3d5543a8c4..d5b17604b2 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Log.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Log.cs
@@ -29,18 +29,10 @@ namespace Gremlin.Net.Driver
{
internal static partial class Log
{
- [LoggerMessage(20000, LogLevel.Information, "Creating {nrConnections} new connections")]
- public static partial void FillingPool(this ILogger logger, int nrConnections);
-
- [LoggerMessage(20001, LogLevel.Warning,
- "Could not get a connection from the pool. Will try again. Retry {retryCount} of {maxRetries}")]
- public static partial void CouldNotGetConnectionFromPool(this ILogger logger, int retryCount, int maxRetries);
-
- [LoggerMessage(20002, LogLevel.Information,
- "A connection was closed. Removing it from the pool so it can be replaced.")]
- public static partial void RemovingClosedConnectionFromPool(this ILogger logger);
+ [LoggerMessage(20000, LogLevel.Information, "Initialized HTTP connections to {uri}")]
+ public static partial void InitializedHttpConnection(this ILogger logger, Uri uri);
- [LoggerMessage(20003, LogLevel.Debug, "Submitting GremlinLang {gremlinLang} for request: {requestId}")]
- public static partial void SubmittingGremlinLang(this ILogger logger, GremlinLang gremlinLang, Guid requestId);
+ [LoggerMessage(20001, LogLevel.Debug, "Submitting GremlinLang {gremlinLang}")]
+ public static partial void SubmittingGremlinLang(this ILogger logger, GremlinLang gremlinLang);
}
}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseResult.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseResult.cs
deleted file mode 100644
index 387893a574..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseResult.cs
+++ /dev/null
@@ -1,56 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System.Collections.Generic;
-
-namespace Gremlin.Net.Driver.Messages
-{
- /// <summary>
- /// Represents the result as a response to a <see cref="RequestMessage"/> sent as part of a
- /// <see cref="ResponseMessage{T}"/> by the server.
- /// </summary>
- /// <typeparam name="T">The type of the <see cref="Data"/>.</typeparam>
- public record ResponseResult<T>
- {
- /// <summary>
- /// Initializes a new instance of the <see cref="ResponseResult{T}" /> record.
- /// </summary>
- /// <param name="data">The data of this result.</param>
- /// <param name="meta">The (optional) meta data of this result.</param>
- public ResponseResult(T? data, Dictionary<string, object>? meta = null)
- {
- Data = data;
- Meta = meta ?? new Dictionary<string, object>();
- }
-
- /// <summary>
- /// Gets the data of this result.
- /// </summary>
- public T? Data { get; }
-
- /// <summary>
- /// Gets meta data of this result.
- /// </summary>
- public Dictionary<string, object> Meta { get; }
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseStatus.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseStatus.cs
deleted file mode 100644
index f7c43cd941..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseStatus.cs
+++ /dev/null
@@ -1,72 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System.Collections.Generic;
-using Gremlin.Net.Driver.Exceptions;
-
-namespace Gremlin.Net.Driver.Messages
-{
- /// <summary>
- /// Represents status information of a <see cref="ResponseMessage{T}"/>.
- /// </summary>
- public record ResponseStatus
- {
- /// <summary>
- /// Initializes a new instance of the <see cref="ResponseStatus" /> record.
- /// </summary>
- /// <param name="code">The <see cref="ResponseStatusCode" />.</param>
- /// <param name="attributes">The (optional) attributes with protocol-level information.</param>
- /// <param name="message">The (optional) message which is just a human-readable string usually associated with errors.</param>
- public ResponseStatus(ResponseStatusCode code, Dictionary<string, object>? attributes = null,
- string? message = null)
- {
- Code = code;
- Attributes = attributes ?? new Dictionary<string, object>();
- Message = message;
- }
-
- /// <summary>
- /// Gets the <see cref="ResponseStatusCode"/>.
- /// </summary>
- public ResponseStatusCode Code { get; }
-
- /// <summary>
- /// Gets the attributes <see cref="Dictionary{TKey,TValue}"/> with protocol-level information.
- /// </summary>
- public Dictionary<string, object> Attributes { get; }
-
- /// <summary>
- /// Gets the message which is just a human-readable string usually associated with errors.
- /// </summary>
- public string? Message { get; }
- }
-
- internal static class ResponseStatusExtensions
- {
- public static void ThrowIfStatusIndicatesError(this ResponseStatus status)
- {
- if (status.Code.IndicatesError())
- throw new ResponseException((int)status.Code, $"{status.Code}: {status.Message}", null);
- }
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseStatusCode.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseStatusCode.cs
deleted file mode 100644
index bd4f55d2b2..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseStatusCode.cs
+++ /dev/null
@@ -1,157 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-
-namespace Gremlin.Net.Driver.Messages
-{
- /// <summary>
- /// Represents the various status codes that Gremlin Server returns.
- /// </summary>
- public enum ResponseStatusCode
- {
- /// <summary>
- /// The server successfully processed a request to completion - there are no messages remaining in this
- /// stream.
- /// </summary>
- Success = 200,
-
- /// <summary>
- /// The server processed the request but there is no result to return (e.g. an Iterator with no elements)
- /// - there are no messages remaining in this stream.
- /// </summary>
- NoContent = 204,
-
- /// <summary>
- /// The server successfully returned some content, but there is more in the stream to arrive - wait for a
- /// SUCCESS to signify the end of the stream.
- /// </summary>
- PartialContent = 206,
-
- /// <summary>
- /// The request attempted to access resources that the requesting user did not have access to.
- /// </summary>
- Unauthorized = 401,
-
- /// <summary>
- /// The server could authenticate the request, but will not fulfill it. This is a general purpose code that
- /// would typically be returned if the request is authenticated but not authorized to do what it is doing.
- /// </summary>
- Forbidden = 403,
-
- /// <summary>
- /// A challenge from the server for the client to authenticate its request.
- /// </summary>
- Authenticate = 407,
-
- /// <summary>
- /// Indicates that too many requests have been sent in a given amount of time.
- /// </summary>
- TooManyRequests = 429,
-
- /// <summary>
- /// The request message contains objects that were not serializable on the client side.
- /// </summary>
- ClientSerializationError = 497,
-
- /// <summary>
- /// The request message was not properly formatted which means it could not be parsed at all or the "op" code
- /// was not recognized such that Gremlin Server could properly route it for processing. Check the message format
- /// and retry the request.
- /// </summary>
- MalformedRequest = 498,
-
- /// <summary>
- /// The request message was parseable, but the arguments supplied in the message were in conflict or incomplete.
- /// Check the message format and retry the request.
- /// </summary>
- InvalidRequestArguments = 499,
-
- /// <summary>
- /// A general server error occurred that prevented the request from being processed.
- /// </summary>
- ServerError = 500,
-
- /// <summary>
- /// A server error that is produced when the fail()-step is triggered.
- /// </summary>
- ServerErrorFailStep = 595,
-
- /// <summary>
- /// A server error that indicates that the client should retry the request. A graph will typically return this error
- /// when a transaction fails due to a locking exception or some other sort of concurrent modification. In other
- /// words, the request was likely valid but the state of the server at the particular time the request arrived
- /// could not be processed to success, but could be at a later moment.
- /// </summary>
- ServerErrorTemporary = 596,
-
- /// <summary>
- /// The request submitted for processing evaluated by the server with errors and could not be processed.
- /// Check the script or remote traversal submitted for errors or other problems and then resubmit.
- /// </summary>
- ServerEvaluationError = 597,
-
- /// <summary>
- /// The server exceeded one of the timeout settings for the request and could therefore only partially responded
- /// or did not respond at all.
- /// </summary>
- ServerTimeout = 598,
-
- /// <summary>
- /// The server was not capable of serializing an object that was returned from the script supplied on the request.
- /// Either transform the object into something Gremlin Server can process within the script or install mapper
- /// serialization classes to Gremlin Server.
- /// </summary>
- ServerSerializationError = 599
- }
-
- internal static class ResponseStatusCodeExtensions
- {
- public static bool IndicatesError(this ResponseStatusCode statusCode)
- {
- switch (statusCode)
- {
- case ResponseStatusCode.Success:
- case ResponseStatusCode.NoContent:
- case ResponseStatusCode.PartialContent:
- case ResponseStatusCode.Authenticate:
- return false;
- case ResponseStatusCode.Unauthorized:
- case ResponseStatusCode.Forbidden:
- case ResponseStatusCode.TooManyRequests:
- case ResponseStatusCode.ClientSerializationError:
- case ResponseStatusCode.MalformedRequest:
- case ResponseStatusCode.InvalidRequestArguments:
- case ResponseStatusCode.ServerError:
- case ResponseStatusCode.ServerErrorFailStep:
- case ResponseStatusCode.ServerErrorTemporary:
- case ResponseStatusCode.ServerEvaluationError:
- case ResponseStatusCode.ServerTimeout:
- case ResponseStatusCode.ServerSerializationError:
- return true;
- default:
- throw new ArgumentOutOfRangeException(nameof(statusCode), statusCode, null);
- }
- }
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ProxyClientWebSocket.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ProxyClientWebSocket.cs
deleted file mode 100644
index 508073a37c..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/ProxyClientWebSocket.cs
+++ /dev/null
@@ -1,81 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Net.WebSockets;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace Gremlin.Net.Driver
-{
- internal class ProxyClientWebSocket : IClientWebSocket
- {
- private readonly ClientWebSocket _client;
-
- public ProxyClientWebSocket(ClientWebSocket client)
- {
- _client = client;
- }
-
- public WebSocketState State => _client.State;
-
- public ClientWebSocketOptions Options => _client.Options;
-
- public static IClientWebSocket CreateClientWebSocket()
- {
- return new ProxyClientWebSocket(new ClientWebSocket());
- }
-
- public Task CloseAsync(WebSocketCloseStatus closeStatus, string statusDescription, CancellationToken cancellationToken)
- {
- return _client.CloseAsync(closeStatus, statusDescription, cancellationToken);
- }
-
- public Task ConnectAsync(Uri uri, CancellationToken cancellationToken)
- {
- return _client.ConnectAsync(uri, cancellationToken);
- }
-
- public void Dispose()
- {
- _client.Dispose();
- }
-
- public Task<WebSocketReceiveResult> ReceiveAsync(ArraySegment<byte> buffer, CancellationToken cancellationToken)
- {
- return _client.ReceiveAsync(buffer, cancellationToken);
- }
-
- public Task SendAsync(ArraySegment<byte> buffer, WebSocketMessageType messageType, bool endOfMessage, CancellationToken cancellationToken)
- {
- return _client.SendAsync(buffer, messageType, endOfMessage, cancellationToken);
- }
-
-#if NET6_0_OR_GREATER
- public ValueTask SendAsync(ArraySegment<byte> buffer, WebSocketMessageType messageType, WebSocketMessageFlags messageFlags, CancellationToken cancellationToken)
- {
- return _client.SendAsync(buffer, messageType, messageFlags, cancellationToken);
- }
-#endif
- }
-}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ProxyConnection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ProxyConnection.cs
deleted file mode 100644
index 99d6daee71..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/ProxyConnection.cs
+++ /dev/null
@@ -1,67 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-using Gremlin.Net.Driver.Messages;
-
-namespace Gremlin.Net.Driver
-{
- internal sealed class ProxyConnection : IConnection
- {
- public IConnection ProxiedConnection { get; }
- private readonly Action<IConnection> _releaseAction;
-
- public ProxyConnection(IConnection proxiedConnection, Action<IConnection> releaseAction)
- {
- ProxiedConnection = proxiedConnection;
- _releaseAction = releaseAction;
- }
-
- public async Task ConnectAsync(CancellationToken cancellationToken)
- {
- await ProxiedConnection.ConnectAsync(cancellationToken).ConfigureAwait(false);
- }
-
- public async Task<ResultSet<T>> SubmitAsync<T>(RequestMessage requestMessage,
- CancellationToken cancellationToken)
- {
- return await ProxiedConnection.SubmitAsync<T>(requestMessage, cancellationToken).ConfigureAwait(false);
- }
-
- public int NrRequestsInFlight => ProxiedConnection.NrRequestsInFlight;
-
- public bool IsOpen => ProxiedConnection.IsOpen;
-
- public async Task CloseAsync()
- {
- await ProxiedConnection.CloseAsync().ConfigureAwait(false);
- }
-
- public void Dispose()
- {
- _releaseAction(ProxiedConnection);
- }
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteConnection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteConnection.cs
index a30f3d18cb..c25cffe725 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteConnection.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteConnection.cs
@@ -111,6 +111,7 @@ namespace Gremlin.Net.Driver.Remote
public async Task<ITraversal<TStart, TEnd>> SubmitAsync<TStart, TEnd>(GremlinLang gremlinLang,
CancellationToken cancellationToken = default)
{
+ _logger.SubmittingGremlinLang(gremlinLang);
gremlinLang.AddG(_traversalSource);
var requestMsg = RequestMessage.Build(gremlinLang.GetGremlin())
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ResponseHandlerForSingleRequestMessage.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ResponseHandlerForSingleRequestMessage.cs
deleted file mode 100644
index a91feaf82f..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/ResponseHandlerForSingleRequestMessage.cs
+++ /dev/null
@@ -1,65 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Collections.Generic;
-using System.Threading.Tasks;
-using Gremlin.Net.Driver.Messages;
-
-namespace Gremlin.Net.Driver
-{
- internal class ResponseHandlerForSingleRequestMessage<T> : IResponseHandlerForSingleRequestMessage
- {
- public Task<ResultSet<T>> Result => _tcs.Task;
-
- private readonly TaskCompletionSource<ResultSet<T>> _tcs =
- new(TaskCreationOptions.RunContinuationsAsynchronously);
-
- private readonly List<T> _result = new();
-
- public void HandleReceived(ResponseMessage<List<object>> received)
- {
- foreach (var d in received.Result)
- {
- _result.Add((T) d);
- }
- }
-
- public void Finalize(Dictionary<string, object> statusAttributes)
- {
- var resultSet =
- new ResultSet<T>(_result, statusAttributes);
- _tcs.TrySetResult(resultSet);
- }
-
- public void HandleFailure(Exception objException)
- {
- _tcs.TrySetException(objException);
- }
-
- public void Cancel()
- {
- _tcs.SetCanceled();
- }
- }
-}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Tokens.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Tokens.cs
index 59627b82fa..ccd0c4b3f5 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Tokens.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Tokens.cs
@@ -40,31 +40,6 @@ namespace Gremlin.Net.Driver
/// </summary>
public const string RequestId = "requestId";
- /// <summary>
- /// Operation used by the client to authenticate itself.
- /// </summary>
- public const string OpsAuthentication = "authentication";
-
- /// <summary>
- /// Operation used to evaluate a Gremlin script provided as a string.
- /// </summary>
- public const string OpsEval = "eval";
-
- /// <summary>
- /// Operation used to get all the keys of all side-effects as produced by a previously executed Traversal.
- /// </summary>
- public const string OpsClose = "close";
-
- /// <summary>
- /// Default OpProcessor.
- /// </summary>
- public const string ProcessorTraversal = "traversal";
-
- /// <summary>
- /// Session OpProcessor.
- /// </summary>
- public const string ProcessorSession = "session";
-
/// <summary>
/// Argument name that allows the definition of the number of items each ResponseMessage should
/// contain from a particular result - overrides the resultIterationBatchSize server setting.
@@ -77,22 +52,11 @@ namespace Gremlin.Net.Driver
/// </summary>
public const string ArgsBindings = "bindings";
- /// <summary>
- /// Argument name that allows definition of alias names for Graph and TraversalSource objects on the
- /// remote system.
- /// </summary>
- public const string ArgsAliases = "aliases";
-
/// <summary>
/// Argument name that corresponds to the Gremlin to evaluate.
/// </summary>
public const string ArgsGremlin = "gremlin";
- /// <summary>
- /// Argument name that allows to define the id of session.
- /// </summary>
- public const string ArgsSession = "session";
-
/// <summary>
/// Argument name that allows a value that is a custom string that the user can pass to a server that
/// might accept it for purpose of identifying the kind of client it came from.
@@ -116,12 +80,6 @@ namespace Gremlin.Net.Driver
/// </summary>
public const string ArgMaterializeProperties = "materializeProperties";
- /// <summary>
- /// Argument name for the response to the server authentication challenge. This value is dependent on the SASL
- /// authentication mechanism required by the server.
- /// </summary>
- public const string ArgsSasl = "sasl";
-
/// <summary>
/// Argument name for the traversal source alias.
/// </summary>
@@ -132,4 +90,4 @@ namespace Gremlin.Net.Driver
/// </summary>
public const string ArgsBulkResults = "bulkResults";
}
-}
\ No newline at end of file
+}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/WebSocketConnection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/WebSocketConnection.cs
deleted file mode 100644
index 68fe91369f..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/WebSocketConnection.cs
+++ /dev/null
@@ -1,157 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using Gremlin.Net.Driver.Exceptions;
-using System;
-using System.IO;
-using System.Net.WebSockets;
-using System.Threading;
-using System.Threading.Tasks;
-using Gremlin.Net.Process;
-
-namespace Gremlin.Net.Driver
-{
- internal interface IWebSocketConnection : IDisposable
- {
- Task ConnectAsync(Uri uri, CancellationToken cancellationToken);
- Task CloseAsync();
- Task SendMessageAsync(byte[] message, CancellationToken cancellationToken);
-#if NET6_0_OR_GREATER
- Task SendMessageUncompressedAsync(byte[] message, CancellationToken cancellationToken);
-#endif
- Task<byte[]> ReceiveMessageAsync();
- bool IsOpen { get; }
- }
-
- internal class WebSocketConnection : IWebSocketConnection
- {
- private const int ReceiveBufferSize = 1024;
- private const WebSocketMessageType MessageType = WebSocketMessageType.Binary;
- private readonly IClientWebSocket _client;
- private const string userAgentHeaderName = "User-Agent";
-
- public WebSocketConnection(IClientWebSocket client, WebSocketSettings settings)
- {
- if (settings.EnableUserAgentOnConnect)
- {
- client.Options.SetRequestHeader(userAgentHeaderName, Utils.UserAgent);
- }
- _client = client;
-
-#if NET6_0_OR_GREATER
- if (settings.UseCompression)
- {
- _client.Options.DangerousDeflateOptions = settings.CompressionOptions;
- }
-#endif
- settings.WebSocketConfigurationCallback?.Invoke(_client.Options);
- }
-
- public async Task ConnectAsync(Uri uri, CancellationToken cancellationToken)
- {
- await _client.ConnectAsync(uri, cancellationToken).ConfigureAwait(false);
- }
-
- public async Task CloseAsync()
- {
- if (CloseAlreadyInitiated) return;
-
- try
- {
- await
- _client.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, CancellationToken.None)
- .ConfigureAwait(false);
- }
- catch (Exception)
- {
- // Swallow exceptions silently as there is nothing to do when closing fails
- }
- }
-
- private bool CloseAlreadyInitiated => _client.State == WebSocketState.Closed ||
- _client.State == WebSocketState.Aborted ||
- _client.State == WebSocketState.CloseSent;
-
-#if NET6_0_OR_GREATER
- public async Task SendMessageUncompressedAsync(byte[] message, CancellationToken cancellationToken)
- {
- await _client.SendAsync(new ArraySegment<byte>(message), MessageType,
- WebSocketMessageFlags.EndOfMessage | WebSocketMessageFlags.DisableCompression,
- cancellationToken)
- .ConfigureAwait(false);
- }
-#endif
-
- public async Task SendMessageAsync(byte[] message, CancellationToken cancellationToken)
- {
- await
- _client.SendAsync(new ArraySegment<byte>(message), MessageType, true, cancellationToken)
- .ConfigureAwait(false);
- }
-
- public async Task<byte[]> ReceiveMessageAsync()
- {
- using var ms = new MemoryStream();
- WebSocketReceiveResult received;
- var buffer = new byte[ReceiveBufferSize];
-
- do
- {
- var receiveBuffer = new ArraySegment<byte>(buffer);
- received = await _client.ReceiveAsync(receiveBuffer, CancellationToken.None).ConfigureAwait(false);
- if (received.MessageType == WebSocketMessageType.Close)
- {
- throw new ConnectionClosedException(received.CloseStatus, received.CloseStatusDescription);
- }
-
- ms.Write(receiveBuffer.Array!, receiveBuffer.Offset, received.Count);
- } while (!received.EndOfMessage);
-
- return ms.ToArray();
- }
-
- public bool IsOpen => _client.State == WebSocketState.Open;
-
- #region IDisposable Support
-
- private bool _disposed;
-
- public void Dispose()
- {
- Dispose(true);
- GC.SuppressFinalize(this);
- }
-
- protected virtual void Dispose(bool disposing)
- {
- if (!_disposed)
- {
- if (disposing)
- _client?.Dispose();
- _disposed = true;
- }
- }
-
- #endregion
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/WebSocketSettings.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/WebSocketSettings.cs
deleted file mode 100644
index 62c8ca0996..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/WebSocketSettings.cs
+++ /dev/null
@@ -1,58 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Net.WebSockets;
-
-namespace Gremlin.Net.Driver
-{
- /// <summary>
- /// Holds settings for WebSocket connections.
- /// </summary>
- internal class WebSocketSettings
- {
- // We currently only use this internally so we don't have to pass these values through the different layers
- // all the way down to the WebSocketConnection class. We can make this public however when we refactor the
- // configuration in general.
-
- /// <summary>
- /// Gets or sets the delegate that will be invoked with the <see cref="ClientWebSocketOptions" />
- /// object used to configure WebSocket connections.
- /// </summary>
- public Action<ClientWebSocketOptions>? WebSocketConfigurationCallback { get; set; }
-
- public bool EnableUserAgentOnConnect { get; set; }
-
-#if NET6_0_OR_GREATER
- /// <summary>
- /// Gets or sets whether compressions will be used. The default is true. (Only available since .NET 6.)
- /// </summary>
- public bool UseCompression { get; set; } = true;
-
- /// <summary>
- /// Gets or sets compression options. (Only available since .SET 6.)
- /// </summary>
- public WebSocketDeflateOptions CompressionOptions { get; set; } = new WebSocketDeflateOptions();
-#endif
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/GraphTraversal.cs b/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/GraphTraversal.cs
index 2ef46e8991..5d2e6dd503 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/GraphTraversal.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/GraphTraversal.cs
@@ -1459,10 +1459,10 @@ namespace Gremlin.Net.Process.Traversal
/// </summary>
public GraphTraversal<TStart, Edge> MergeE (IDictionary<object,object?>? m)
{
- if (m != null && m.TryGetValue(Direction.Out, out var outValue) && outValue is Vertex outV) {
+ if (m != null && m.ContainsKey(Direction.Out) && m[Direction.Out] is Vertex outV) {
m[Direction.Out] = outV.Id;
}
- if (m != null && m.TryGetValue(Direction.In, out var inValue) && inValue is Vertex inV) {
+ if (m != null && m.ContainsKey(Direction.In) && m[Direction.In] is Vertex inV) {
m[Direction.In] = inV.Id;
}
GremlinLang.AddStep("mergeE", m);
@@ -1809,7 +1809,7 @@ namespace Gremlin.Net.Process.Traversal
/// </summary>
public GraphTraversal<TStart, TEnd> Property(Cardinality cardinality, IDictionary<object, object> map)
{
- if (map == null) throw new ArgumentNullException(nameof(map));
+ if (map == null) return Wrap<TStart, TEnd>(this);
foreach (var entry in map)
{
diff --git a/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/GraphTraversalSource.cs b/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/GraphTraversalSource.cs
index 72b5ef5929..7759d8552c 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/GraphTraversalSource.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/GraphTraversalSource.cs
@@ -75,8 +75,8 @@ namespace Gremlin.Net.Process.Traversal
GremlinLang = gremlinLang;
}
- public GraphTraversalSource(ICollection<ITraversalStrategy> traversalStrategies, GremlinLang gremlinLang,
- IRemoteConnection connection)
+ public GraphTraversalSource(ICollection<ITraversalStrategy> traversalStrategies,
+ GremlinLang gremlinLang, IRemoteConnection connection)
: this(traversalStrategies.Where(strategy => strategy.GetType() != typeof(RemoteStrategy)).ToList(),
gremlinLang)
{
@@ -369,10 +369,10 @@ namespace Gremlin.Net.Process.Traversal
/// </summary>
public GraphTraversal<Edge, Edge> MergeE(IDictionary<object,object?>? m)
{
- if (m != null && m[Direction.Out] is Vertex outV) {
+ if (m != null && m.ContainsKey(Direction.Out) && m[Direction.Out] is Vertex outV) {
m[Direction.Out] = outV.Id;
}
- if (m != null && m[Direction.In] is Vertex inV) {
+ if (m != null && m.ContainsKey(Direction.In) && m[Direction.In] is Vertex inV) {
m[Direction.In] = inV.Id;
}
var traversal = new GraphTraversal<Edge, Edge>(TraversalStrategies, GremlinLang.Clone());
diff --git a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/GraphBinaryWriter.cs b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/GraphBinaryWriter.cs
index c5c4fc6a7d..5ad1d65853 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/GraphBinaryWriter.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/GraphBinaryWriter.cs
@@ -39,7 +39,7 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
/// <summary>
/// A <see cref="byte"/> representing the version of the GraphBinary v4 specification.
/// </summary>
- public const byte VersionByte = 0x84;
+ public const byte VersionByte = 0x81;
private static readonly byte[] UnspecifiedNullBytes = {DataType.UnspecifiedNull.TypeCode, 0x01};
diff --git a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/ResponseMessageSerializer.cs b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/ResponseMessageSerializer.cs
index 1f1f10bd49..78b05bd6d4 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/ResponseMessageSerializer.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/ResponseMessageSerializer.cs
@@ -28,7 +28,6 @@ using System.Threading.Tasks;
using Gremlin.Net.Driver.Exceptions;
using Gremlin.Net.Driver.Messages;
using Gremlin.Net.Process.Traversal;
-using Gremlin.Net.Structure.IO.GraphBinary4;
namespace Gremlin.Net.Structure.IO.GraphBinary4
{
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ConnectionPoolTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ConnectionPoolTests.cs
deleted file mode 100644
index dc840573da..0000000000
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/ConnectionPoolTests.cs
+++ /dev/null
@@ -1,74 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Collections.Generic;
-using System.Threading.Tasks;
-using Gremlin.Net.Driver;
-using Gremlin.Net.Driver.Exceptions;
-using Gremlin.Net.IntegrationTest.Util;
-using Xunit;
-
-namespace Gremlin.Net.IntegrationTest.Driver
-{
- public class ConnectionPoolTests
- {
- private readonly RequestMessageProvider _requestMessageProvider = new RequestMessageProvider();
- private static readonly string TestHost = ConfigProvider.Configuration["TestServerIpAddress"]!;
- private static readonly int TestPort = Convert.ToInt32(ConfigProvider.Configuration["TestServerPort"]);
-
- private async Task ExecuteMultipleLongRunningRequestsInParallel(IGremlinClient gremlinClient, int nrRequests,
- int requestRunningTimeInMs)
- {
- var tasks = new List<Task>(nrRequests);
- for (var i = 0; i < nrRequests; i++)
- {
- tasks.Add(gremlinClient.SubmitAsync(_requestMessageProvider.GetSleepMessage(requestRunningTimeInMs)));
- }
-
- await Task.WhenAll(tasks);
- }
-
- [Theory(Skip = "WebSocket connection pool test, not applicable to HTTP (Phase 2 cleanup)")]
- [InlineData(1)]
- [InlineData(2)]
- [InlineData(8)]
- public void ShouldCreateConfiguredNrConnections(int connectionPoolSize)
- {
- // Phase 2: test body removed — connection pool not applicable to HTTP
- }
-
- [Fact(Skip = "WebSocket connection pool test, not applicable to HTTP (Phase 2 cleanup)")]
- public async Task ShouldThrowConnectionPoolBusyExceptionWhenPoolIsBusy()
- {
- // Phase 2: test body removed — connection pool not applicable to HTTP
- await Task.CompletedTask;
- }
-
- private static GremlinClient CreateGremlinClient(int connectionPoolSize = 2, int maxInProcessPerConnection = 4)
- {
- var gremlinServer = new GremlinServer(TestHost, TestPort);
- return new GremlinClient(gremlinServer);
- }
- }
-}
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientAuthenticationTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientAuthenticationTests.cs
deleted file mode 100644
index eed43635be..0000000000
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientAuthenticationTests.cs
+++ /dev/null
@@ -1,64 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Threading.Tasks;
-using Gremlin.Net.Driver;
-using Gremlin.Net.Driver.Exceptions;
-using Gremlin.Net.IntegrationTest.Util;
-using Xunit;
-
-namespace Gremlin.Net.IntegrationTest.Driver
-{
- public class GremlinClientAuthenticationTests
- {
- private static readonly string TestHost = ConfigProvider.Configuration["TestServerIpAddress"]!;
- private static readonly int TestPort = Convert.ToInt32(ConfigProvider.Configuration["TestSecureServerPort"]);
- private readonly RequestMessageProvider _requestMessageProvider = new RequestMessageProvider();
-
- [Fact(Skip = "WebSocket authentication test (Phase 2 cleanup)")]
- public async Task ShouldThrowForMissingCredentials()
- {
- // Phase 2: test body removed — WebSocket SSL/auth not applicable to HTTP
- await Task.CompletedTask;
- }
-
- [Theory(Skip = "WebSocket authentication test (Phase 2 cleanup)")]
- [InlineData("unknownUser", "passwordDoesntMatter")]
- [InlineData("stephen", "wrongPassword")]
- public async Task ShouldThrowForWrongCredentials(string username, string password)
- {
- // Phase 2: test body removed — WebSocket SSL/auth not applicable to HTTP
- await Task.CompletedTask;
- }
-
- [Theory(Skip = "WebSocket authentication test (Phase 2 cleanup)")]
- [InlineData("'Hello' + 'World'", "HelloWorld")]
- public async Task ScriptShouldBeEvaluatedAndResultReturnedForCorrectCredentials(string requestMsg,
- string expectedResponse)
- {
- // Phase 2: test body removed — WebSocket SSL/auth not applicable to HTTP
- await Task.CompletedTask;
- }
- }
-}
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientBehaviorIntegrationTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientBehaviorIntegrationTests.cs
deleted file mode 100644
index a3c557e4fb..0000000000
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientBehaviorIntegrationTests.cs
+++ /dev/null
@@ -1,89 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Threading.Tasks;
-using Gremlin.Net.Driver;
-using Gremlin.Net.Driver.Messages;
-using Gremlin.Net.IntegrationTest.Util;
-using Gremlin.Net.Structure.IO.GraphBinary4;
-using Gremlin.Net.Structure.IO.GraphSON;
-using Xunit;
-
-namespace Gremlin.Net.IntegrationTest.Driver
-{
- public class GremlinClientBehaviorIntegrationTests
- {
- private static readonly string TestHost = ConfigProvider.Configuration["GremlinSocketServerIpAddress"]!;
-
- private static readonly SocketServerSettings Settings =
- SocketServerSettings.FromYaml(ConfigProvider.Configuration["GremlinSocketServerConfig"]);
-
- private static IMessageSerializer Serializer;
-
- public GremlinClientBehaviorIntegrationTests()
- {
- switch (Settings.Serializer)
- {
- case "GraphSONV2":
- Serializer = new GraphSON2MessageSerializer();
- break;
- case "GraphSONV3":
- Serializer = new GraphSON3MessageSerializer();
- break;
- case "GraphBinaryV1":
- default:
- Serializer = new GraphBinaryMessageSerializer();
- break;
- }
- }
-
- [Fact(Skip = "WebSocket behavior test using GremlinSocketServer (Phase 2 cleanup)")]
- public async Task ShouldTryCreateNewConnectionIfClosedByServer()
- {
- // Phase 2: test body removed — uses WebSocket-specific GremlinSocketServer
- await Task.CompletedTask;
- }
-
- [Fact(Skip = "WebSocket behavior test using GremlinSocketServer (Phase 2 cleanup)")]
- public async Task ShouldIncludeUserAgentInHandshakeRequest()
- {
- // Phase 2: test body removed — uses WebSocket-specific GremlinSocketServer
- await Task.CompletedTask;
- }
-
- [Fact(Skip = "WebSocket behavior test using GremlinSocketServer (Phase 2 cleanup)")]
- public async Task ShouldNotIncludeUserAgentInHandshakeRequestIfDisabled()
- {
- // Phase 2: test body removed — uses WebSocket-specific GremlinSocketServer
- await Task.CompletedTask;
- }
-
- [Fact(Skip = "WebSocket behavior test using GremlinSocketServer (Phase 2 cleanup)")]
- public async Task ShouldSendPerRequestSettingsToServer()
- {
- // Phase 2: test body removed — uses WebSocket-specific GremlinSocketServer
- await Task.CompletedTask;
- }
- }
-}
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs
index 27841bcee8..3fb4cd2e8a 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs
@@ -99,14 +99,6 @@ namespace Gremlin.Net.IntegrationTest.Driver
}
}
- [Fact(Skip = "Thread.sleep not available in GremlinLang (Phase 2 cleanup)")]
- public async Task ShouldSupportCancellation()
- {
- var gremlinServer = new GremlinServer(TestHost, TestPort);
- using var gremlinClient = new GremlinClient(gremlinServer);
- await Task.CompletedTask;
- }
-
[Fact]
public async Task ShouldReassembleResponseBatches()
{
@@ -122,16 +114,6 @@ namespace Gremlin.Net.IntegrationTest.Driver
}
}
- [Fact(Skip = "Thread.sleep not available in GremlinLang (Phase 2 cleanup)")]
- public async Task ShouldCorrectlyAssignResponsesToRequests()
- {
- var gremlinServer = new GremlinServer(TestHost, TestPort);
- using (var gremlinClient = new GremlinClient(gremlinServer))
- {
- await Task.CompletedTask;
- }
- }
-
[Fact]
public async Task ShouldReturnEnumerableResult()
{
@@ -197,28 +179,6 @@ namespace Gremlin.Net.IntegrationTest.Driver
}
}
- [Fact(Skip = "WebSocket-specific test, not applicable to HTTP connection (Phase 2 cleanup)")]
- public async Task ShouldConfigureWebSocketOptionsAsSpecified()
- {
- var gremlinServer = new GremlinServer(TestHost, TestPort);
- using (var gremlinClient = new GremlinClient(gremlinServer))
- {
- // Phase 2: test body removed — WebSocket options not applicable to HTTP
- await Task.CompletedTask;
- }
- }
-
- [Fact(Skip = "Session-based test, sessions not supported over HTTP (Phase 2 cleanup)")]
- public async Task ShouldSaveVariableBetweenRequestsInSession()
- {
- var gremlinServer = new GremlinServer(TestHost, TestPort);
- using (var gremlinClient = new GremlinClient(gremlinServer))
- {
- // Phase 2: test body removed — sessions not supported over HTTP
- await Task.CompletedTask;
- }
- }
-
[Fact]
public void ShouldLogWithProvidedLoggerFactory()
{
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/MessagesTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/MessagesTests.cs
index e3a7a2ec61..4873e75892 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/MessagesTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/MessagesTests.cs
@@ -37,13 +37,6 @@ namespace Gremlin.Net.IntegrationTest.Driver
private static readonly string TestHost = ConfigProvider.Configuration["TestServerIpAddress"]!;
private static readonly int TestPort = Convert.ToInt32(ConfigProvider.Configuration["TestServerPort"]);
- [Fact(Skip = "3.x alias API not available in 4.0 (Phase 2 cleanup)")]
- public async Task ShouldUseAliasForTraversalSource()
- {
- // Phase 2: 4.0 uses AddG() instead of aliases map
- await Task.CompletedTask;
- }
-
[Fact]
public async Task ShouldThrowForInvalidOperation()
{
@@ -61,23 +54,6 @@ namespace Gremlin.Net.IntegrationTest.Driver
}
}
- [Fact(Skip = "3.x processor API not available in 4.0 (Phase 2 cleanup)")]
- public async Task ShouldThrowForInvalidProcessor()
- {
- // Phase 2: Processor() method removed from 4.0 RequestMessage.Builder
- await Task.CompletedTask;
- }
-
- [Fact(Skip = "Thread.sleep not available in GremlinLang (Phase 2 cleanup)")]
- public async Task ShouldUseSpecifiedEvaluationTimeout()
- {
- var gremlinServer = new GremlinServer(TestHost, TestPort);
- using (var gremlinClient = new GremlinClient(gremlinServer))
- {
- await Task.CompletedTask;
- }
- }
-
[Fact]
public async Task ShouldThrowForUnsupportedLanguage()
{
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/CommonSteps.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/CommonSteps.cs
index 8112ff6957..0c9288c57f 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/CommonSteps.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/CommonSteps.cs
@@ -33,7 +33,7 @@ using Gherkin.Ast;
using Gremlin.Net.IntegrationTest.Gherkin.Attributes;
using Gremlin.Net.Process.Traversal;
using Gremlin.Net.Structure;
-using Gremlin.Net.Structure.IO.GraphSON;
+using Gremlin.Net.Structure.IO.GraphBinary4;
using Xunit;
using static Gremlin.Net.Process.Traversal.AnonymousTraversalSource;
@@ -53,13 +53,14 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
private static readonly JsonSerializerOptions JsonDeserializingOptions =
new() { PropertyNamingPolicy = JsonNamingPolicy.CamelCase };
- public static ScenarioData ScenarioData { get; set; } = new ScenarioData(new GraphSON3MessageSerializer());
+ public static ScenarioData ScenarioData { get; set; } = new ScenarioData(new GraphBinaryMessageSerializer());
private static readonly IDictionary<Regex, Func<string, string, object?>> Parsers =
new Dictionary<string, Func<string, string, object?>>
{
{@"str\[(.*)\]", (x, graphName) => x }, //returns the string value as is
{@"vp\[(.+)\]", ToVertexProperty},
+ {@"prop\[(.+)\]", ToProperty},
{@"dt\[(.+)\]", ToDateTime},
{@"uuid\[(.+)\]", ToUuid},
{@"d\[(.*)\]\.([bsilfdmn])", ToNumber},
@@ -288,7 +289,21 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
if (ordered)
{
- Assert.Equal(expected, _result!);
+ var expectedList = expected.ToList();
+ var resultList = _result!.ToList();
+ Assert.Equal(expectedList.Count, resultList.Count);
+ for (int i = 0; i < expectedList.Count; i++)
+ {
+ if (expectedList[i] is Property expectedProp && resultList[i] is Property resultProp)
+ {
+ Assert.Equal(expectedProp.Key, resultProp.Key);
+ Assert.Equal(expectedProp.Value, resultProp.Value);
+ }
+ else
+ {
+ Assert.Equal(expectedList[i], resultList[i]);
+ }
+ }
}
else
{
@@ -319,6 +334,21 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
}
Assert.True(expectedArrayContainsResultAsSet);
}
+ else if (resultItem is Property resultProperty)
+ {
+ var found = false;
+ foreach (var expectedItem in expectedArray)
+ {
+ if (expectedItem is Property expectedProperty &&
+ resultProperty.Key == expectedProperty.Key &&
+ Equals(resultProperty.Value, expectedProperty.Value))
+ {
+ found = true;
+ break;
+ }
+ }
+ Assert.True(found, $"Property {resultItem} not found in expected results");
+ }
else if (resultItem is double resultItemDouble &&
expectedArray.Select(e => e!.GetType()).Any(t => t == typeof(decimal)))
{
@@ -501,6 +531,19 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
return ScenarioData.GetByGraphName(graphName).VertexProperties[triplet];
}
+ private static object ToProperty(string value, string graphName)
+ {
+ var commaIndex = value.IndexOf(',');
+ if (commaIndex < 0)
+ {
+ return new Property(value, null);
+ }
+ var key = value.Substring(0, commaIndex);
+ var valueStr = value.Substring(commaIndex + 1);
+ var parsedValue = ParseValue(valueStr, graphName);
+ return new Property(key, parsedValue);
+ }
+
private static Path ToPath(string value, string graphName)
{
return new Path(new List<ISet<string>>(0),
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/ScenarioData.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/ScenarioData.cs
index ad42f54c8e..3fb96abff4 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/ScenarioData.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/ScenarioData.cs
@@ -109,7 +109,8 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
// Property name might not exist and C# doesn't support "null" keys in Dictionary
if (g.V().Count().Next() == g.V().Has("name").Count().Next())
{
- return g.V().Group<string, object>().By("name").By(__.Tail<Vertex>()).Next()!
+ return g.With("materializeProperties", "all").V().Group<string, object>()
+ .By("name").By(__.Tail<Vertex>()).Next()!
.ToDictionary(kv => kv.Key, kv => (Vertex) kv.Value);
}
else
@@ -122,81 +123,93 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
{
try
{
- IFunction lambda = Lambda.Groovy(
- "it.outVertex().value('name') + '-' + it.label() + '->' + it.inVertex().value('name')", 1);
-
- return g.E().Group<string, Edge>()
- .By(lambda)
+ // Use project() instead of a Groovy lambda to build the edge key string
+ // in the form "outV-label->inV". This avoids lambdas which are not supported
+ // over HTTP 4.0 with GraphBinary serialization.
+ var edgeMap = g.With("materializeProperties", "all").E()
+ .Group<IDictionary<string, object>, object>()
+ .By(__.Project<object>("o", "l", "i")
+ .By(__.OutV().Values<object>("name"))
+ .By(__.Label())
+ .By(__.InV().Values<object>("name")))
.By(__.Tail<object>())
.Next()!;
+
+ var edges = new Dictionary<string, Edge>();
+ foreach (var kv in edgeMap)
+ {
+ var key = kv.Key;
+ var outV = key["o"]?.ToString();
+ var label = key["l"]?.ToString();
+ var inV = key["i"]?.ToString();
+ edges[$"{outV}-{label}->{inV}"] = (Edge) kv.Value;
+ }
+ return edges;
}
catch (ResponseException)
{
// Property name might not exist
return EmptyEdges;
}
+ catch (KeyNotFoundException)
+ {
+ return EmptyEdges;
+ }
}
private static IDictionary<string, VertexProperty> GetVertexProperties(GraphTraversalSource g)
{
try
{
- /*
- * This closure will turn a VertexProperty into a triple string of the form:
- * "vertexName-propKey->propVal"
- *
- * It will also take care of wrapping propVal in the appropriate Numeric format. We must do this in
- * case a Vertex has multiple properties with the same key and number value but in different numeric
- * type spaces (rare, but possible, and presumably something we may want to write tests around).
- */
- var groovy = @"
- { vp ->
- def result = vp.element().value('name') + '-' + vp.key() + '->'
- def value = vp.value()
- def type = ''
- switch(value) {
- case { !(it instanceof Number) }:
- return result + value
- case Byte:
- type = 'b'
- break
- case Short:
- type = 's'
- break
- case Integer:
- type = 'i'
- break
- case Long:
- type = 'l'
- break
- case Float:
- type = 'f'
- break
- case Double:
- type = 'd'
- break
- case BigDecimal:
- type = 'm'
- break
- case BigInteger:
- type = 'n'
- break
- }
- return result + 'd[' + value + '].' + type
- }
- ";
-
- IFunction lambda = Lambda.Groovy(groovy, 1);
-
- return g.V().Properties<VertexProperty>().Group<string, VertexProperty>()
- .By(lambda)
+ // Use project() instead of a Groovy lambda to build the vertex property key
+ // string in the form "vertexName-propKey->propVal". This avoids lambdas which
+ // are not supported over HTTP 4.0 with GraphBinary serialization.
+ // The toy graphs only use String/Int/Float/Double types so we handle those here.
+ var vpMap = g.V().Properties<object>()
+ .Group<IDictionary<string, object>, object>()
+ .By(__.Project<object>("n", "k", "v")
+ .By(__.Element().Values<object>("name"))
+ .By(__.Key())
+ .By(__.Value<object>()))
.By(__.Tail<object>())
.Next()!;
+
+ var vps = new Dictionary<string, VertexProperty>();
+ foreach (var kv in vpMap)
+ {
+ var key = kv.Key;
+ var name = key["n"]?.ToString();
+ var k = key["k"]?.ToString();
+ var val = key["v"];
+ var valStr = FormatVertexPropertyValue(k!, val);
+ vps[$"{name}-{k}->{valStr}"] = (VertexProperty) kv.Value;
+ }
+ return vps;
}
catch (ResponseException)
{
return EmptyVertexProperties;
}
+ catch (KeyNotFoundException)
+ {
+ return EmptyVertexProperties;
+ }
+ }
+
+ /// <summary>
+ /// Formats a vertex property value for use as a lookup key, wrapping numeric values
+ /// in the Gherkin d[value].type notation.
+ /// </summary>
+ private static string FormatVertexPropertyValue(string propertyKey, object? value)
+ {
+ // The toy graphs use known property keys with specific numeric types.
+ // Match the same logic as the Python GLV's _get_v_keys helper.
+ return propertyKey switch
+ {
+ "weight" => $"d[{value}].d",
+ "age" or "since" or "skill" => $"d[{value}].i",
+ _ => value?.ToString() ?? ""
+ };
}
public void Dispose()
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs
index a09618b383..87260ae446 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTests.cs
@@ -212,12 +212,6 @@ namespace Gremlin.Net.IntegrationTest.Process.Traversal.DriverRemoteConnection
var countWithStrategy = g.WithStrategies(new OptionsStrategy(options)).V().Count().Next();
Assert.Equal(6, countWithStrategy);
-
- var responseException = Assert.Throws<ResponseException>(() =>
- g.With("y").With("x", "test").With(Tokens.ArgsEvalTimeout, 10).Inject(1)
- .SideEffect(Lambda.Groovy("Thread.sleep(10000)")).Iterate());
- Assert.Equal(598, responseException.StatusCode);
-
}
[Fact]
@@ -269,7 +263,8 @@ namespace Gremlin.Net.IntegrationTest.Process.Traversal.DriverRemoteConnection
Assert.True(await g.V().Promise(t => t.HasNext(), CancellationToken.None));
}
- [Fact]
+ // TODO: update after tx is implemented in 4.0
+ [Fact (Skip = "transaction-related tests are disabled until Tx implemented")]
public async Task ShouldThrowExceptionOnCommitWhenGraphNotSupportTx()
{
var connection = _connectionFactory.CreateRemoteConnection();
@@ -279,7 +274,7 @@ namespace Gremlin.Net.IntegrationTest.Process.Traversal.DriverRemoteConnection
Assert.Equal("ServerError: Graph does not support transactions", exception.Message);
}
- [Fact]
+ [Fact (Skip = "transaction-related tests are disabled until Tx implemented")]
public async Task ShouldThrowExceptionOnRollbackWhenGraphNotSupportTx()
{
var connection = _connectionFactory.CreateRemoteConnection();
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTransactionTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTransactionTests.cs
index 6cfd4ad7db..f57f7b48fa 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTransactionTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Process/Traversal/DriverRemoteConnection/GraphTraversalTransactionTests.cs
@@ -33,7 +33,8 @@ namespace Gremlin.Net.IntegrationTest.Process.Traversal.DriverRemoteConnection
{
private readonly IRemoteConnection _connection = new RemoteConnectionFactory().CreateRemoteConnection("gtx");
- [Fact]
+ // TODO: update after tx is implemented in 4.0
+ [Fact(Skip = "transaction-related tests are disabled until Tx implemented")]
public async Task ShouldSupportRemoteTransactionsCommit()
{
var g = AnonymousTraversalSource.Traversal().With(_connection);
@@ -57,7 +58,8 @@ namespace Gremlin.Net.IntegrationTest.Process.Traversal.DriverRemoteConnection
Assert.Equal(2, count);
}
- [Fact]
+ // TODO: update after tx is implemented in 4.0
+ [Fact(Skip = "transaction-related tests are disabled until Tx implemented")]
public async Task ShouldSupportRemoteTransactionsRollback()
{
var g = AnonymousTraversalSource.Traversal().With(_connection);
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionPoolSettingsTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionPoolSettingsTests.cs
deleted file mode 100644
index 6eff93efea..0000000000
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionPoolSettingsTests.cs
+++ /dev/null
@@ -1,70 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using Gremlin.Net.Driver;
-using Xunit;
-
-namespace Gremlin.Net.UnitTest.Driver
-{
- public class ConnectionPoolSettingsTests
- {
- [Theory]
- [InlineData(0)]
- [InlineData(-1)]
- [InlineData(-100)]
- public void ShouldThrowForInvalidPoolSize(int invalidPoolSize)
- {
- Assert.Throws<ArgumentOutOfRangeException>(() => new ConnectionPoolSettings {PoolSize = invalidPoolSize});
- }
-
- [Theory]
- [InlineData(0)]
- [InlineData(-1)]
- [InlineData(-100)]
- public void ShouldThrowForInvalidMaxInProcessPerConnection(int invalidMaxInProcessPerConnection)
- {
- Assert.Throws<ArgumentOutOfRangeException>(() => new ConnectionPoolSettings
- {MaxInProcessPerConnection = invalidMaxInProcessPerConnection});
- }
-
- [Theory]
- [InlineData(-1)]
- [InlineData(-100)]
- public void ShouldThrowForInvalidReconnectionAttempts(int reconnectionAttempts)
- {
- Assert.Throws<ArgumentOutOfRangeException>(() => new ConnectionPoolSettings
- {ReconnectionAttempts = reconnectionAttempts});
- }
-
- [Theory]
- [InlineData(-1)]
- [InlineData(-1000)]
- [InlineData(0)]
- public void ShouldThrowForInvalidReconnectionBaseDelay(int baseDelayInMs)
- {
- Assert.Throws<ArgumentOutOfRangeException>(() => new ConnectionPoolSettings
- {ReconnectionBaseDelay = TimeSpan.FromMilliseconds(baseDelayInMs)});
- }
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionPoolTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionPoolTests.cs
deleted file mode 100644
index 3708a2ee20..0000000000
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionPoolTests.cs
+++ /dev/null
@@ -1,394 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Collections.Generic;
-using System.Threading;
-using System.Threading.Tasks;
-using Gremlin.Net.Driver;
-using Gremlin.Net.Driver.Exceptions;
-using Gremlin.Net.Driver.Messages;
-using Microsoft.Extensions.Logging.Abstractions;
-using NSubstitute;
-using NSubstitute.ExceptionExtensions;
-using NSubstitute.Extensions;
-using Xunit;
-
-namespace Gremlin.Net.UnitTest.Driver
-{
- public class ConnectionPoolTests
- {
- [Theory]
- [InlineData(1)]
- [InlineData(2)]
- [InlineData(10)]
- public void ShouldEstablishConfiguredNrConnections(int poolSize)
- {
- var mockedConnectionFactory = Substitute.For<IConnectionFactory>();
- var mockedConnection = Substitute.For<IConnection>();
- mockedConnectionFactory.CreateConnection().Returns(mockedConnection);
- var pool = CreateConnectionPool(mockedConnectionFactory, poolSize);
-
- Assert.Equal(poolSize, pool.NrConnections);
- mockedConnectionFactory.Received(poolSize).CreateConnection();
- mockedConnection.Received(poolSize).ConnectAsync(Arg.Any<CancellationToken>());
- }
-
- [Fact]
- public void GetAvailableConnectionShouldReturnFirstOpenConnection()
- {
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- var openConnectionToReturn = OpenConnection;
- fakeConnectionFactory.CreateConnection().Returns(_ => ClosedConnection, _ => ClosedConnection,
- _ => openConnectionToReturn);
- var pool = CreateConnectionPool(fakeConnectionFactory, 3);
-
- var returnedConnection = pool.GetAvailableConnection();
-
- Assert.Equal(openConnectionToReturn, ((ProxyConnection) returnedConnection).ProxiedConnection);
- }
-
- [Fact]
- public void GetAvailableConnectionShouldThrowIfAllConnectionsAreClosed()
- {
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- fakeConnectionFactory.CreateConnection().Returns(_ => ClosedConnection);
- var pool = CreateConnectionPool(fakeConnectionFactory);
-
- Assert.Throws<ServerUnavailableException>(() => pool.GetAvailableConnection());
- }
-
- [Fact]
- public void GetAvailableConnectionShouldEmptyPoolIfServerUnavailable()
- {
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- fakeConnectionFactory.CreateConnection()
- .Returns(_ => ClosedConnection, _ => ClosedConnection, _ => ClosedConnection);
- var pool = CreateConnectionPool(fakeConnectionFactory, 3);
- fakeConnectionFactory.Configure().CreateConnection().Returns(_ => CannotConnectConnection);
-
- Assert.Throws<ServerUnavailableException>(() => pool.GetAvailableConnection());
-
- Assert.Equal(0, pool.NrConnections);
- }
-
- [Fact]
- public void GetAvailableConnectionShouldEventuallyRefillPoolIfEmpty()
- {
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- fakeConnectionFactory.CreateConnection()
- .Returns(_ => ClosedConnection, _ => ClosedConnection, _ => ClosedConnection);
- var pool = CreateConnectionPool(fakeConnectionFactory, 3);
- fakeConnectionFactory.Configure().CreateConnection().Returns(_ => CannotConnectConnection);
- Assert.Throws<ServerUnavailableException>(() => pool.GetAvailableConnection());
- // Pool is now empty
- Assert.Equal(0, pool.NrConnections);
- fakeConnectionFactory.Configure().CreateConnection().Returns(_ => OpenConnection);
-
- pool.GetAvailableConnection();
-
- AssertNrOpenConnections(pool, 3);
- }
-
- [Fact]
- public void GetAvailableConnectionsShouldEventuallyFillUpPoolIfNotFull()
- {
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- fakeConnectionFactory.CreateConnection()
- .Returns(_ => ClosedConnection, _ => ClosedConnection, _ => OpenConnection);
- var pool = CreateConnectionPool(fakeConnectionFactory, 3);
- fakeConnectionFactory.Configure().CreateConnection().Returns(_ => CannotConnectConnection);
- pool.GetAvailableConnection();
- pool.GetAvailableConnection();
- // Pool is now just partially filled
- Assert.Equal(1, pool.NrConnections);
-
- fakeConnectionFactory.Configure().CreateConnection().Returns(_ => OpenConnection);
- pool.GetAvailableConnection();
-
- AssertNrOpenConnections(pool, 3);
- }
-
- [Fact]
- public void GetAvailableConnectionShouldReplaceClosedConnections()
- {
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- fakeConnectionFactory.CreateConnection()
- .Returns(_ => ClosedConnection, _ => ClosedConnection, _ => OpenConnection);
- var pool = CreateConnectionPool(fakeConnectionFactory, 3);
- fakeConnectionFactory.Configure().CreateConnection().Returns(_ => OpenConnection);
- var nrCreatedConnections = pool.NrConnections;
-
- pool.GetAvailableConnection();
- pool.GetAvailableConnection();
- pool.GetAvailableConnection();
-
- AssertNrOpenConnections(pool, nrCreatedConnections);
- }
-
- private static void AssertNrOpenConnections(ConnectionPool connectionPool, int expectedNrConnections)
- {
- for (var i = 0; i < expectedNrConnections; i++)
- {
- var connection = connectionPool.GetAvailableConnection();
- Assert.True(connection.IsOpen);
- }
- Assert.Equal(expectedNrConnections, connectionPool.NrConnections);
- }
-
- [Fact]
- public async Task ShouldNotCreateMoreConnectionsThanConfiguredForParallelRequests()
- {
- var mockedConnectionFactory = Substitute.For<IConnectionFactory>();
- mockedConnectionFactory.CreateConnection()
- .Returns(_ => ClosedConnection, _ => ClosedConnection, _ => OpenConnection);
- var pool = CreateConnectionPool(mockedConnectionFactory, 3);
- mockedConnectionFactory.Configure().CreateConnection().Returns(_ => OpenConnection);
- var nrCreatedConnections = pool.NrConnections;
- var getConnectionTasks = new List<Task<IConnection>>();
-
- for (var i = 0; i < 100; i++)
- {
- getConnectionTasks.Add(Task.Run(() => pool.GetAvailableConnection()));
- }
- await Task.WhenAll(getConnectionTasks);
-
- await Task.Delay(TimeSpan.FromMilliseconds(100));
- Assert.Equal(nrCreatedConnections, pool.NrConnections);
- }
-
- [Fact]
- public async Task ShouldReplaceConnectionClosedDuringSubmit()
- {
- var mockedConnectionFactory = Substitute.For<IConnectionFactory>();
- var fakedConnection = Substitute.For<IConnection>();
- fakedConnection.IsOpen.Returns(true);
- mockedConnectionFactory.CreateConnection().Returns(fakedConnection);
- var pool = CreateConnectionPool(mockedConnectionFactory, 1);
- var returnedConnection = pool.GetAvailableConnection();
- fakedConnection.IsOpen.Returns(false);
- mockedConnectionFactory.CreateConnection().Returns(_ => OpenConnection);
-
- await returnedConnection.SubmitAsync<bool>(RequestMessage.Build(string.Empty).Create(),
- CancellationToken.None);
- returnedConnection.Dispose();
-
- Assert.Equal(1, pool.NrConnections);
- Assert.True(pool.GetAvailableConnection().IsOpen);
- }
-
- [Fact]
- public void ShouldWaitForHostToBecomeAvailable()
- {
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- fakeConnectionFactory.CreateConnection().Returns(_ => ClosedConnection);
- var pool = CreateConnectionPool(fakeConnectionFactory, 1);
- fakeConnectionFactory.Configure().CreateConnection().Returns(_ => OpenConnection);
- var nrCreatedConnections = pool.NrConnections;
-
- var connection = pool.GetAvailableConnection();
-
- AssertNrOpenConnections(pool, nrCreatedConnections);
- Assert.True(connection.IsOpen);
- }
-
- [Theory]
- [InlineData(0)]
- [InlineData(1)]
- [InlineData(2)]
- public void ShouldPerformConfiguredNrReconnectionAttemptsForUnavailableServer(int nrAttempts)
- {
- var mockedConnectionFactory = Substitute.For<IConnectionFactory>();
- mockedConnectionFactory.CreateConnection().Returns(_ => ClosedConnection);
- var pool = CreateConnectionPool(mockedConnectionFactory, 1, nrAttempts);
-
- Assert.ThrowsAny<Exception>(() => pool.GetAvailableConnection());
-
- mockedConnectionFactory.Received(nrAttempts + 2).CreateConnection();
- // 2 additional calls are expected: 1 for the initial creation of the pool and 1 when the connection should
- // be returned and none is open for the first attempt (before any retries)
- }
-
- [Fact]
- public void ShouldThrowAfterWaitingTooLongForUnavailableServer()
- {
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- fakeConnectionFactory.CreateConnection().Returns(_ => ClosedConnection);
- var pool = CreateConnectionPool(fakeConnectionFactory, 1);
-
- Assert.Throws<ServerUnavailableException>(() => pool.GetAvailableConnection());
- }
-
- [Fact]
- public async Task ShouldNotLeakConnectionsIfDisposeIsCalledWhilePoolIsPopulating()
- {
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- fakeConnectionFactory.CreateConnection().Returns(_ => ClosedConnection);
- var pool = CreateConnectionPool(fakeConnectionFactory, 1);
- var mockedConnectionToBeDisposed = Substitute.For<IConnection>();
- var poolWasDisposedSignal = new SemaphoreSlim(0, 1);
- mockedConnectionToBeDisposed.ConnectAsync(Arg.Any<CancellationToken>())
- .Returns(x => poolWasDisposedSignal.WaitAsync(CancellationToken.None));
- var connectionWasSuccessfullyDisposed = new SemaphoreSlim(0, 1);
- mockedConnectionToBeDisposed.When(x => x.Dispose())
- .Do(_ => connectionWasSuccessfullyDisposed.Release());
- // We don't use the `CancellationToken` here as the connection should also be disposed if it did not
- // react on the cancellation. This can happen if the task is cancelled just before `ConnectAsync` returns.
- fakeConnectionFactory.Configure().CreateConnection().Returns(mockedConnectionToBeDisposed);
- try
- {
- pool.GetAvailableConnection();
- }
- catch (ServerUnavailableException)
- {
- // expected as the pool only contains a closed connection at this point
- }
-
- pool.Dispose();
- poolWasDisposedSignal.Release();
-
- await connectionWasSuccessfullyDisposed.WaitAsync(TimeSpan.FromSeconds(2));
- Assert.Equal(0, pool.NrConnections);
- await mockedConnectionToBeDisposed.Received(1).ConnectAsync(Arg.Any<CancellationToken>());
- mockedConnectionToBeDisposed.Received(1).Dispose();
- }
-
- [Fact]
- public async Task DisposeShouldCancelConnectionEstablishment()
- {
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- fakeConnectionFactory.CreateConnection().Returns(_ => ClosedConnection);
- var pool = CreateConnectionPool(fakeConnectionFactory, 1, 0);
- var mockedConnectionToBeDisposed = Substitute.For<IConnection>();
- mockedConnectionToBeDisposed.ConnectAsync(Arg.Any<CancellationToken>())
- .Returns(x => Task.Delay(-1, (CancellationToken)x[0]));
- var connectionWasSuccessfullyDisposed = new SemaphoreSlim(0, 1);
- mockedConnectionToBeDisposed.When(x => x.Dispose())
- .Do(_ => connectionWasSuccessfullyDisposed.Release());
- fakeConnectionFactory.Configure().CreateConnection().Returns(mockedConnectionToBeDisposed);
- try
- {
- pool.GetAvailableConnection();
- }
- catch (ServerUnavailableException)
- {
- // expected as the pool only contains a closed connection at this point
- }
-
- pool.Dispose();
-
- await connectionWasSuccessfullyDisposed.WaitAsync(TimeSpan.FromSeconds(2));
- Assert.Equal(0, pool.NrConnections);
- await mockedConnectionToBeDisposed.Received(1).ConnectAsync(Arg.Any<CancellationToken>());
- mockedConnectionToBeDisposed.Received(1).Dispose();
- }
-
- [Fact]
- public async Task ConnectionsEstablishedInParallelShouldAllBeDisposedIfOneThrowsDuringCreation()
- {
- // This test unfortunately needs a lot of knowledge about the inner working of the ConnectionPool to
- // adequately test that connections established in parallel will all be disposed if one throws an
- // exception.
-
- // First create a pool with only closed connections that we can then let the pool replace:
- var fakeConnectionFactory = Substitute.For<IConnectionFactory>();
- fakeConnectionFactory.CreateConnection()
- .Returns(_ => ClosedConnection, // We need to do it like this as we use a dictionary of dead connections in
- _ => ClosedConnection, // ConnectionPool and the three connections need to be different objects
- _ => ClosedConnection); // for this to work.
- var pool = CreateConnectionPool(fakeConnectionFactory, 3, 0);
- var startEstablishingProblematicConnections = new SemaphoreSlim(0, 1);
- // Let the pool get one connection that is so slow to open that the pool will afterwards try to create two
- // more connections in parallel.
- var fakedSlowToEstablishConnection = Substitute.For<IConnection>();
- fakedSlowToEstablishConnection.ConnectAsync(Arg.Any<CancellationToken>())
- .Returns(async _ => await startEstablishingProblematicConnections.WaitAsync());
- fakeConnectionFactory.Configure().CreateConnection().Returns(fakedSlowToEstablishConnection);
- // Trigger replacement of closed connections
- try
- {
- pool.GetAvailableConnection();
- }
- catch (ServerUnavailableException)
- {
- // expected as the pool only contain closed connections at this point
- }
-
- var fakedOpenConnection = OpenConnection;
- var fakedCannotConnectConnection = CannotConnectConnection;
- fakeConnectionFactory.Configure().CreateConnection()
- .Returns(fakedOpenConnection, fakedCannotConnectConnection);
- // Let the slow to establish connection finish so the pool can try to establish the other two connections
- startEstablishingProblematicConnections.Release();
- await Task.Delay(TimeSpan.FromSeconds(1));
-
- // Verify that the pool tried to establish both connections and then also disposed both, even though one throw an exception
- await fakedOpenConnection.Received(1).ConnectAsync(Arg.Any<CancellationToken>());
- fakedOpenConnection.Received(1).Dispose();
- await fakedCannotConnectConnection.Received(1).ConnectAsync(Arg.Any<CancellationToken>());
- fakedCannotConnectConnection.Received(1).Dispose();
- }
-
- private static IConnection OpenConnection
- {
- get
- {
- var fakedConnection = Substitute.For<IConnection>();
- fakedConnection.IsOpen.Returns(true);
- return fakedConnection;
- }
- }
-
- private static IConnection ClosedConnection
- {
- get
- {
- var fakedConnection = Substitute.For<IConnection>();
- fakedConnection.IsOpen.Returns(false);
- return fakedConnection;
- }
- }
-
- private static IConnection CannotConnectConnection
- {
- get
- {
- var fakedConnection = Substitute.For<IConnection>();
- fakedConnection.IsOpen.Returns(false);
- fakedConnection.ConnectAsync(Arg.Any<CancellationToken>())
- .Throws(new Exception("Cannot connect to server."));
- return fakedConnection;
- }
- }
-
- private static ConnectionPool CreateConnectionPool(IConnectionFactory connectionFactory, int poolSize = 2,
- int reconnectionAttempts = 1)
- {
- return new ConnectionPool(connectionFactory, new ConnectionPoolSettings
- {
- PoolSize = poolSize, ReconnectionAttempts = reconnectionAttempts,
- ReconnectionBaseDelay = TimeSpan.FromMilliseconds(10) // let the tests execute fast
- }, NullLogger<ConnectionPool>.Instance);
- }
- }
-}
\ No newline at end of file
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionTests.cs
index db42ff2891..1528a37567 100644
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionTests.cs
@@ -1,4 +1,4 @@
-#region License
+#region License
/*
* Licensed to the Apache Software Foundation (ASF) under one
@@ -23,315 +23,319 @@
using System;
using System.Collections.Generic;
-using System.Net.WebSockets;
+using System.IO;
+using System.IO.Compression;
+using System.Linq;
+using System.Net;
+using System.Net.Http;
+using System.Net.Http.Headers;
using System.Threading;
using System.Threading.Tasks;
using Gremlin.Net.Driver;
-using Gremlin.Net.Driver.Exceptions;
using Gremlin.Net.Driver.Messages;
-using Gremlin.Net.Structure.IO.GraphBinary4;
+using Gremlin.Net.Structure.IO;
using NSubstitute;
-using NSubstitute.ExceptionExtensions;
using Xunit;
namespace Gremlin.Net.UnitTest.Driver
{
public class ConnectionTests
{
+ private static readonly Uri TestUri = new Uri("http://localhost:8182/gremlin");
+
+ /// <summary>
+ /// Creates a mock HttpMessageHandler that captures the request and returns a canned response.
+ /// </summary>
+ private static (HttpClient httpClient, MockHandler handler) CreateMockHttpClient(
+ byte[]? responseBytes = null, string? contentEncoding = null)
+ {
+ var handler = new MockHandler(responseBytes ?? BuildMinimalResponseBytes(), contentEncoding);
+ var httpClient = new HttpClient(handler);
+ return (httpClient, handler);
+ }
+
+ /// <summary>
+ /// Builds a minimal valid 4.0 GraphBinary response: version + non-bulked + marker + status 200 + null msg + null exc.
+ /// </summary>
+ private static byte[] BuildMinimalResponseBytes()
+ {
+ using var ms = new MemoryStream();
+ ms.WriteByte(0x81); // version
+ ms.WriteByte(0x00); // non-bulked
+ ms.WriteByte(0xFD); // marker type code
+ ms.WriteByte(0x00); // marker value
+ WriteInt(ms, 200); // status code
+ ms.WriteByte(0x01); // null status message
+ ms.WriteByte(0x01); // null exception
+ return ms.ToArray();
+ }
+
+ private static void WriteInt(Stream stream, int value)
+ {
+ var bytes = BitConverter.GetBytes(value);
+ if (BitConverter.IsLittleEndian)
+ Array.Reverse(bytes);
+ stream.Write(bytes, 0, 4);
+ }
+
[Fact]
- public async Task ShouldHandleCloseMessageAfterConnectAsync()
+ public async Task ShouldSetContentTypeHeader()
{
- var mockedClientWebSocket = Substitute.For<IClientWebSocket>();
- mockedClientWebSocket.ReceiveAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<CancellationToken>()).Returns(
- new WebSocketReceiveResult(0, WebSocketMessageType.Close, true, WebSocketCloseStatus.MessageTooBig,
- "Message is too large"));
- mockedClientWebSocket.Options.Returns(new ClientWebSocket().Options);
- var uri = new Uri("wss://localhost:8182");
- var connection = GetConnection(mockedClientWebSocket, uri: uri);
-
- await connection.ConnectAsync(CancellationToken.None);
-
- Assert.False(connection.IsOpen);
- Assert.Equal(0, connection.NrRequestsInFlight);
- await mockedClientWebSocket.Received(1).ConnectAsync(uri, Arg.Any<CancellationToken>());
- await mockedClientWebSocket.Received(1)
- .ReceiveAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<CancellationToken>());
- await mockedClientWebSocket.Received(1).CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty,
- Arg.Any<CancellationToken>());
+ var (httpClient, handler) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings();
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
+
+ await connection.SubmitAsync<object>(CreateTestRequest());
+
+ Assert.NotNull(handler.CapturedRequest);
+ Assert.Equal(SerializationTokens.GraphBinary4MimeType,
+ handler.CapturedRequest!.Content!.Headers.ContentType!.MediaType);
}
[Fact]
- public async Task ShouldThrowIfClosedMessageReceivedWithValidPropertiesAsync()
+ public async Task ShouldSetAcceptHeader()
{
- var mockedClientWebSocket = Substitute.For<IClientWebSocket>();
- mockedClientWebSocket.Options.Returns(new ClientWebSocket().Options);
+ var (httpClient, handler) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings();
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
- var webSocketConnection = new WebSocketConnection(
- mockedClientWebSocket,
- new WebSocketSettings());
+ await connection.SubmitAsync<object>(CreateTestRequest());
- // Test all known close statuses
- foreach (Enum closeStatus in Enum.GetValues(typeof(WebSocketCloseStatus)))
- {
- mockedClientWebSocket
- .ReceiveAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<CancellationToken>())
- .Returns(new WebSocketReceiveResult(0, WebSocketMessageType.Close, true, (WebSocketCloseStatus)closeStatus, closeStatus.ToString()));
+ Assert.NotNull(handler.CapturedRequest);
+ Assert.Contains(handler.CapturedRequest!.Headers.Accept,
+ h => h.MediaType == SerializationTokens.GraphBinary4MimeType);
+ }
- await AssertExpectedConnectionClosedException((WebSocketCloseStatus?)closeStatus, closeStatus.ToString(), () => webSocketConnection.ReceiveMessageAsync());
- }
+ [Fact]
+ public async Task ShouldSendPostRequest()
+ {
+ var (httpClient, handler) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings();
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
- // Test null/empty close property values as well.
- mockedClientWebSocket
- .ReceiveAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<CancellationToken>())
- .Returns(new WebSocketReceiveResult(0, WebSocketMessageType.Close, true, null, null));
- await AssertExpectedConnectionClosedException(null, null, () => webSocketConnection.ReceiveMessageAsync());
+ await connection.SubmitAsync<object>(CreateTestRequest());
- mockedClientWebSocket
- .ReceiveAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<CancellationToken>())
- .Returns(new WebSocketReceiveResult(0, WebSocketMessageType.Close, true, null, string.Empty));
- await AssertExpectedConnectionClosedException(null, string.Empty, () => webSocketConnection.ReceiveMessageAsync());
+ Assert.NotNull(handler.CapturedRequest);
+ Assert.Equal(HttpMethod.Post, handler.CapturedRequest!.Method);
}
[Fact]
- public async Task ShouldThrowOnSubmitRequestIfWebSocketIsNotOpen()
+ public async Task ShouldSendToCorrectUri()
{
- // This is to test that race bugs don't get introduced which would cause submitted requests to hang if the
- // connection is being closed/aborted or is not connected. In these cases, WebSocket.SendAsync should throw for the underlying
- // websocket is not open and the caller should be notified of that failure.
- var mockedClientWebSocket = Substitute.For<IClientWebSocket>();
+ var (httpClient, handler) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings();
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
- mockedClientWebSocket.Options.Returns(new ClientWebSocket().Options);
+ await connection.SubmitAsync<object>(CreateTestRequest());
- var connection = GetConnection(mockedClientWebSocket);
- var request = RequestMessage.Build("gremlin").Create();
+ Assert.NotNull(handler.CapturedRequest);
+ Assert.Equal(TestUri, handler.CapturedRequest!.RequestUri);
+ }
- // Simulate the SendAsync exception behavior if the underlying websocket is closed (see reference https://docs.microsoft.com/en-us/dotnet/api/system.net.websockets.clientwebsocket.sendasync?view=net-6.0)
- mockedClientWebSocket
- .SendAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<WebSocketMessageType>(), Arg.Any<bool>(),
- Arg.Any<CancellationToken>())
- .ThrowsAsync(new ObjectDisposedException(nameof(ClientWebSocket), "Socket closed"));
+ [Fact]
+ public async Task ShouldSetAcceptEncodingWhenCompressionEnabled()
+ {
+ var (httpClient, handler) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings { EnableCompression = true };
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
- // Test various closing/closed WebSocketStates with SubmitAsync.
- mockedClientWebSocket.State.Returns(WebSocketState.Closed);
- await Assert.ThrowsAsync<ObjectDisposedException>(() => connection.SubmitAsync<dynamic>(request, CancellationToken.None));
+ await connection.SubmitAsync<object>(CreateTestRequest());
- mockedClientWebSocket.State.Returns(WebSocketState.CloseSent);
- await Assert.ThrowsAsync<ObjectDisposedException>(() => connection.SubmitAsync<dynamic>(request, CancellationToken.None));
+ Assert.NotNull(handler.CapturedRequest);
+ Assert.Contains(handler.CapturedRequest!.Headers.AcceptEncoding,
+ e => e.Value == "deflate");
+ }
- mockedClientWebSocket.State.Returns(WebSocketState.CloseReceived);
- await Assert.ThrowsAsync<ObjectDisposedException>(() => connection.SubmitAsync<dynamic>(request, CancellationToken.None));
+ [Fact]
+ public async Task ShouldNotSetAcceptEncodingWhenCompressionDisabled()
+ {
+ var (httpClient, handler) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings { EnableCompression = false };
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
- mockedClientWebSocket.State.Returns(WebSocketState.Aborted);
- await Assert.ThrowsAsync<ObjectDisposedException>(() => connection.SubmitAsync<dynamic>(request, CancellationToken.None));
+ await connection.SubmitAsync<object>(CreateTestRequest());
- // Simulate SendAsync exception behavior if underlying websocket is not connected.
- mockedClientWebSocket
- .SendAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<WebSocketMessageType>(), Arg.Any<bool>(),
- Arg.Any<CancellationToken>()).ThrowsAsync(new InvalidOperationException("Socket not connected"));
- mockedClientWebSocket.State.Returns(WebSocketState.Connecting);
- await Assert.ThrowsAsync<InvalidOperationException>(() => connection.SubmitAsync<dynamic>(request, CancellationToken.None));
+ Assert.NotNull(handler.CapturedRequest);
+ Assert.DoesNotContain(handler.CapturedRequest!.Headers.AcceptEncoding,
+ e => e.Value == "deflate");
}
[Fact]
- public async Task ShouldHandleCloseMessageForInFlightRequestsAsync()
+ public async Task ShouldSetUserAgentWhenEnabled()
{
- // Tests that in-flight requests will get notified if a connection close message is received.
- var uri = new Uri("wss://localhost:8182");
- var closeResult = new WebSocketReceiveResult(0, WebSocketMessageType.Close, true, WebSocketCloseStatus.EndpointUnavailable, "Server shutdown");
-
- var receiveSemaphore = new SemaphoreSlim(0, 1);
- var mockedClientWebSocket = Substitute.For<IClientWebSocket>();
- mockedClientWebSocket.ReceiveAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<CancellationToken>())
- .ReturnsForAnyArgs(async x =>
- {
- await receiveSemaphore.WaitAsync();
- mockedClientWebSocket.State.Returns(WebSocketState.CloseReceived);
- return closeResult;
- });
- mockedClientWebSocket.State.Returns(WebSocketState.Open);
- mockedClientWebSocket.Options.Returns(new ClientWebSocket().Options);
-
- var connection = GetConnection(mockedClientWebSocket, uri: uri);
- await connection.ConnectAsync(CancellationToken.None);
-
- // Create two in-flight requests that will block on waiting for a response.
- var requestMsg1 = RequestMessage.Build("gremlin").Create();
- var requestMsg2 = RequestMessage.Build("gremlin").Create();
- Task request1 = connection.SubmitAsync<dynamic>(requestMsg1, CancellationToken.None);
- Task request2 = connection.SubmitAsync<dynamic>(requestMsg2, CancellationToken.None);
-
- // Confirm the requests are in-flight.
- Assert.Equal(2, connection.NrRequestsInFlight);
-
- // Release the connection close message.
- receiveSemaphore.Release();
-
- // Assert that both requests get notified with the closed exception.
- await AssertExpectedConnectionClosedException(closeResult.CloseStatus, closeResult.CloseStatusDescription, () => request1);
- await AssertExpectedConnectionClosedException(closeResult.CloseStatus, closeResult.CloseStatusDescription, () => request2);
-
- // delay for NotifyAboutConnectionFailure running in another thread
- var runs = 0;
- while (connection.NrRequestsInFlight == 2 && ++runs < 5)
- {
- await Task.Delay(5);
- }
+ var (httpClient, handler) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings { EnableUserAgentOnConnect = true };
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
+
+ await connection.SubmitAsync<object>(CreateTestRequest());
- Assert.False(connection.IsOpen);
- Assert.Equal(0, connection.NrRequestsInFlight);
- await mockedClientWebSocket.Received(1).ConnectAsync(uri, Arg.Any<CancellationToken>());
- await mockedClientWebSocket.Received(1)
- .ReceiveAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<CancellationToken>());
- await mockedClientWebSocket.Received(1).CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty,
- Arg.Any<CancellationToken>());
+ Assert.NotNull(handler.CapturedRequest);
+ Assert.True(handler.CapturedRequest!.Headers.Contains("User-Agent"));
}
[Fact]
- public async Task ShouldProperlyHandleCancellationForSubmitAsync()
+ public async Task ShouldNotSetUserAgentWhenDisabled()
{
- var mockedClientWebSocket = Substitute.For<IClientWebSocket>();
- mockedClientWebSocket.Options.Returns(new ClientWebSocket().Options);
- var connection = GetConnection(mockedClientWebSocket);
- var cts = new CancellationTokenSource();
-
- var task = connection.SubmitAsync<object>(RequestMessage.Build(string.Empty).Create(),
- cts.Token);
- cts.Cancel();
-
- await Assert.ThrowsAsync<TaskCanceledException>(async () => await task);
- Assert.True(task.IsCanceled);
- await mockedClientWebSocket.Received(1).SendAsync(Arg.Any<ArraySegment<byte>>(),
- Arg.Any<WebSocketMessageType>(), Arg.Any<bool>(), cts.Token);
- await mockedClientWebSocket.DidNotReceive().CloseAsync(Arg.Any<WebSocketCloseStatus>(), Arg.Any<string>(),
- Arg.Any<CancellationToken>());
- await mockedClientWebSocket.DidNotReceive().ReceiveAsync(Arg.Any<ArraySegment<byte>>(), cts.Token);
+ var (httpClient, handler) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings { EnableUserAgentOnConnect = false };
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
+
+ await connection.SubmitAsync<object>(CreateTestRequest());
+
+ Assert.NotNull(handler.CapturedRequest);
+ Assert.False(handler.CapturedRequest!.Headers.Contains("User-Agent"));
}
[Fact]
- public async Task ShouldProperlyHandleCancellationForSubmitAsyncIfAlreadyCancelled()
+ public async Task ShouldSetBulkResultsHeaderWhenEnabled()
{
- var mockedClientWebSocket = Substitute.For<IClientWebSocket>();
- mockedClientWebSocket.Options.Returns(new ClientWebSocket().Options);
- var connection = GetConnection(mockedClientWebSocket);
- var token = new CancellationToken(canceled: true);
-
- var task = connection.SubmitAsync<object>(RequestMessage.Build(string.Empty).Create(),
- token);
-
- await Assert.ThrowsAsync<TaskCanceledException>(async () => await task);
- Assert.True(task.IsCanceled);
- await mockedClientWebSocket.DidNotReceive().SendAsync(Arg.Any<ArraySegment<byte>>(),
- Arg.Any<WebSocketMessageType>(), Arg.Any<bool>(), Arg.Any<CancellationToken>());
- await mockedClientWebSocket.DidNotReceive().CloseAsync(Arg.Any<WebSocketCloseStatus>(), Arg.Any<string>(),
- Arg.Any<CancellationToken>());
- await mockedClientWebSocket.DidNotReceive().ReceiveAsync(Arg.Any<ArraySegment<byte>>(), token);
+ var (httpClient, handler) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings { BulkResults = true };
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
+
+ await connection.SubmitAsync<object>(CreateTestRequest());
+
+ Assert.NotNull(handler.CapturedRequest);
+ Assert.True(handler.CapturedRequest!.Headers.Contains("bulkResults"));
+ Assert.Equal("true", handler.CapturedRequest.Headers.GetValues("bulkResults").First());
}
[Fact]
- public async Task ShouldContinueSubmittingOtherMessagesIfOneIsCancelled()
+ public async Task ShouldNotSetBulkResultsHeaderWhenDisabled()
{
- var mockedClientWebSocket = Substitute.For<IClientWebSocket>();
- mockedClientWebSocket.Options.Returns(new ClientWebSocket().Options);
- var tcs = new TaskCompletionSource();
- mockedClientWebSocket.SendAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<WebSocketMessageType>(),
- Arg.Any<bool>(), Arg.Any<CancellationToken>()).Returns(tcs.Task);
- var connection = GetConnection(mockedClientWebSocket);
- var cts = new CancellationTokenSource();
-
- var taskToCancel = connection.SubmitAsync<object>(RequestMessage.Build(string.Empty).Create(),
- cts.Token);
- var taskToComplete = connection.SubmitAsync<object>(RequestMessage.Build(string.Empty).Create(),
- CancellationToken.None);
- cts.Cancel();
- var taskToComplete2 = connection.SubmitAsync<object>(RequestMessage.Build(string.Empty).Create(),
- CancellationToken.None);
- tcs.TrySetResult();
-
- await Assert.ThrowsAsync<TaskCanceledException>(async () => await taskToCancel);
- Assert.True(taskToCancel.IsCanceled);
- await Task.Delay(TimeSpan.FromMilliseconds(400)); // wait a bit to let the messages being sent
- await mockedClientWebSocket.Received(3).SendAsync(Arg.Any<ArraySegment<byte>>(),
- Arg.Any<WebSocketMessageType>(), Arg.Any<bool>(), Arg.Any<CancellationToken>());
+ var (httpClient, handler) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings { BulkResults = false };
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
+
+ await connection.SubmitAsync<object>(CreateTestRequest());
+
+ Assert.NotNull(handler.CapturedRequest);
+ Assert.False(handler.CapturedRequest!.Headers.Contains("bulkResults"));
}
[Fact]
- public async Task ShouldContinueSubmittingOtherMessagesIfOneIsAlreadyCancelled()
+ public async Task ShouldDecompressDeflateResponse()
{
- var mockedClientWebSocket = Substitute.For<IClientWebSocket>();
- mockedClientWebSocket.Options.Returns(new ClientWebSocket().Options);
- var cancelledToken = new CancellationToken(canceled: true);
- mockedClientWebSocket.SendAsync(Arg.Any<ArraySegment<byte>>(), Arg.Any<WebSocketMessageType>(),
- Arg.Any<bool>(), cancelledToken)
- .Throws(new TaskCanceledException(null, null, cancelledToken));
- var messageToSend = RequestMessage.Build(string.Empty).Create();
- var fakeMessageSerializer = Substitute.For<IMessageSerializer>();
- var bytesToSend = new byte[] { 1, 2, 3 };
- fakeMessageSerializer.SerializeMessageAsync(messageToSend, Arg.Any<CancellationToken>())
- .Returns(bytesToSend);
- var connection = GetConnection(mockedClientWebSocket, fakeMessageSerializer);
-
- var taskToCancel = connection.SubmitAsync<object>(RequestMessage.Build(string.Empty).Create(),
- cancelledToken);
- var taskToComplete = connection.SubmitAsync<object>(messageToSend, CancellationToken.None);
-
- await Assert.ThrowsAsync<TaskCanceledException>(async () => await taskToCancel);
- Assert.True(taskToCancel.IsCanceled);
- await mockedClientWebSocket.Received(1).SendAsync(bytesToSend, Arg.Any<WebSocketMessageType>(),
- Arg.Any<bool>(), Arg.Any<CancellationToken>());
+ // Compress the minimal response bytes with deflate
+ var originalBytes = BuildMinimalResponseBytes();
+ byte[] compressedBytes;
+ using (var compressedStream = new MemoryStream())
+ {
+ using (var deflateStream = new DeflateStream(compressedStream, CompressionMode.Compress, true))
+ {
+ deflateStream.Write(originalBytes, 0, originalBytes.Length);
+ }
+ compressedBytes = compressedStream.ToArray();
+ }
+
+ var (httpClient, handler) = CreateMockHttpClient(compressedBytes, "deflate");
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings { EnableCompression = true };
+ using var connection = new Connection(TestUri, serializer, settings, httpClient);
+
+ // Should not throw — decompression should work
+ var result = await connection.SubmitAsync<object>(CreateTestRequest());
+
+ Assert.NotNull(result);
}
[Fact]
- public async Task ShouldNotProcessReceivedMessageForCancelledRequest()
+ public void ShouldDisposeWithoutError()
{
- var fakeMessageSerializer = Substitute.For<IMessageSerializer>();
- var receivedBytes = new byte[] { 1, 2, 3 };
- var messageToCancel = RequestMessage.Build(string.Empty).Create();
- var receivedMessage = new ResponseMessage<List<object>>(false,
- new List<object>(), 200, null, null);
- fakeMessageSerializer.DeserializeMessageAsync(receivedBytes, Arg.Any<CancellationToken>())
- .Returns(receivedMessage);
- var fakeWebSocketConnection = Substitute.For<IWebSocketConnection>();
- var receiveTaskCompletionSource = new TaskCompletionSource<byte[]>();
- fakeWebSocketConnection.ReceiveMessageAsync().Returns(receiveTaskCompletionSource.Task);
- var connection = GetConnection(fakeWebSocketConnection, fakeMessageSerializer);
- await connection.ConnectAsync(CancellationToken.None);
- var cts = new CancellationTokenSource();
-
- var submitTask = connection.SubmitAsync<object>(messageToCancel, cts.Token);
- cts.Cancel();
- receiveTaskCompletionSource.SetResult(receivedBytes);
- await Assert.ThrowsAsync<TaskCanceledException>(() => submitTask);
-
- Assert.Equal(0, connection.NrRequestsInFlight);
+ var (httpClient, _) = CreateMockHttpClient();
+ var serializer = CreateMockSerializer();
+ var settings = new ConnectionSettings();
+ var connection = new Connection(TestUri, serializer, settings, httpClient);
+
+ connection.Dispose();
+ // Double dispose should not throw
+ connection.Dispose();
}
- private static Connection GetConnection(IClientWebSocket clientWebSocket,
- IMessageSerializer? messageSerializer = null, Uri? uri = null)
+ private static RequestMessage CreateTestRequest()
{
- return GetConnection(new WebSocketConnection(clientWebSocket, new WebSocketSettings()), messageSerializer,
- uri);
+ return RequestMessage.Build("g.V()").AddG("g").Create();
}
- private static Connection GetConnection(IWebSocketConnection webSocketConnection,
- IMessageSerializer? messageSerializer = null, Uri? uri = null)
+ private static IMessageSerializer CreateMockSerializer()
{
- uri ??= new Uri("wss://localhost:8182");
- messageSerializer ??= new GraphBinaryMessageSerializer();
- return new Connection(
- webSocketConnection,
- uri: uri,
- username: "user",
- password: "password",
- messageSerializer: messageSerializer,
- sessionId: null);
+ var serializer = Substitute.For<IMessageSerializer>();
+ serializer.SerializeMessageAsync(Arg.Any<RequestMessage>(), Arg.Any<CancellationToken>())
+ .Returns(Task.FromResult(new byte[] { 0x81 }));
+ serializer.DeserializeMessageAsync(Arg.Any<byte[]>(), Arg.Any<CancellationToken>())
+ .Returns(Task.FromResult(
+ new ResponseMessage<List<object>>(false, new List<object>(), 200, null, null)));
+ return serializer;
}
- private static async Task AssertExpectedConnectionClosedException(WebSocketCloseStatus? expectedCloseStatus,
- string? expectedCloseDescription, Func<Task> func)
+ /// <summary>
+ /// A test HttpMessageHandler that captures the request and returns a canned response.
+ /// </summary>
+ private class MockHandler : HttpMessageHandler
{
- var exception = await Assert.ThrowsAsync<ConnectionClosedException>(func);
- Assert.Equal(expectedCloseStatus, exception.Status);
- Assert.Equal(expectedCloseDescription, exception.Description);
+ private readonly byte[] _responseBytes;
+ private readonly string? _contentEncoding;
+
+ public HttpRequestMessage? CapturedRequest { get; private set; }
+
+ public MockHandler(byte[] responseBytes, string? contentEncoding = null)
+ {
+ _responseBytes = responseBytes;
+ _contentEncoding = contentEncoding;
+ }
+
+ protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request,
+ CancellationToken cancellationToken)
+ {
+ // Clone the request headers before the original request is disposed
+ CapturedRequest = CloneRequest(request);
+
+ var response = new HttpResponseMessage(HttpStatusCode.OK)
+ {
+ Content = new ByteArrayContent(_responseBytes)
+ };
+ response.Content.Headers.ContentType =
+ new MediaTypeHeaderValue(SerializationTokens.GraphBinary4MimeType);
+ if (_contentEncoding != null)
+ {
+ response.Content.Headers.ContentEncoding.Add(_contentEncoding);
+ }
+ return Task.FromResult(response);
+ }
+
+ private static HttpRequestMessage CloneRequest(HttpRequestMessage original)
+ {
+ var clone = new HttpRequestMessage(original.Method, original.RequestUri);
+
+ // Copy headers
+ foreach (var header in original.Headers)
+ {
+ clone.Headers.TryAddWithoutValidation(header.Key, header.Value);
+ }
+
+ // Copy content headers by creating a dummy content
+ if (original.Content != null)
+ {
+ var contentBytes = original.Content.ReadAsByteArrayAsync().Result;
+ clone.Content = new ByteArrayContent(contentBytes);
+ foreach (var header in original.Content.Headers)
+ {
+ clone.Content.Headers.TryAddWithoutValidation(header.Key, header.Value);
+ }
+ }
+
+ return clone;
+ }
}
}
}
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/GremlinClientTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/GremlinClientTests.cs
index c5ba915976..f31c33d2a4 100644
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/GremlinClientTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/GremlinClientTests.cs
@@ -47,7 +47,7 @@ namespace Gremlin.Net.UnitTest.Driver
[Fact]
public void ShouldCreateClientWithCustomSettings()
{
- var settings = new HttpConnectionSettings
+ var settings = new ConnectionSettings
{
EnableCompression = true,
BulkResults = true
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/HttpConnectionTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/HttpConnectionTests.cs
deleted file mode 100644
index e5c7f17a50..0000000000
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/HttpConnectionTests.cs
+++ /dev/null
@@ -1,341 +0,0 @@
-#region License
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-#endregion
-
-using System;
-using System.Collections.Generic;
-using System.IO;
-using System.IO.Compression;
-using System.Linq;
-using System.Net;
-using System.Net.Http;
-using System.Net.Http.Headers;
-using System.Threading;
-using System.Threading.Tasks;
-using Gremlin.Net.Driver;
-using Gremlin.Net.Driver.Messages;
-using Gremlin.Net.Structure.IO;
-using NSubstitute;
-using Xunit;
-
-namespace Gremlin.Net.UnitTest.Driver
-{
- public class HttpConnectionTests
- {
- private static readonly Uri TestUri = new Uri("http://localhost:8182/gremlin");
-
- /// <summary>
- /// Creates a mock HttpMessageHandler that captures the request and returns a canned response.
- /// </summary>
- private static (HttpClient httpClient, MockHandler handler) CreateMockHttpClient(
- byte[]? responseBytes = null, string? contentEncoding = null)
- {
- var handler = new MockHandler(responseBytes ?? BuildMinimalResponseBytes(), contentEncoding);
- var httpClient = new HttpClient(handler);
- return (httpClient, handler);
- }
-
- /// <summary>
- /// Builds a minimal valid 4.0 GraphBinary response: version + non-bulked + marker + status 200 + null msg + null exc.
- /// </summary>
- private static byte[] BuildMinimalResponseBytes()
- {
- using var ms = new MemoryStream();
- ms.WriteByte(0x81); // version
- ms.WriteByte(0x00); // non-bulked
- ms.WriteByte(0xFD); // marker type code
- ms.WriteByte(0x00); // marker value
- WriteInt(ms, 200); // status code
- ms.WriteByte(0x01); // null status message
- ms.WriteByte(0x01); // null exception
- return ms.ToArray();
- }
-
- private static void WriteInt(Stream stream, int value)
- {
- var bytes = BitConverter.GetBytes(value);
- if (BitConverter.IsLittleEndian)
- Array.Reverse(bytes);
- stream.Write(bytes, 0, 4);
- }
-
- [Fact]
- public async Task ShouldSetContentTypeHeader()
- {
- var (httpClient, handler) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings();
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(handler.CapturedRequest);
- Assert.Equal(SerializationTokens.GraphBinary4MimeType,
- handler.CapturedRequest!.Content!.Headers.ContentType!.MediaType);
- }
-
- [Fact]
- public async Task ShouldSetAcceptHeader()
- {
- var (httpClient, handler) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings();
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(handler.CapturedRequest);
- Assert.Contains(handler.CapturedRequest!.Headers.Accept,
- h => h.MediaType == SerializationTokens.GraphBinary4MimeType);
- }
-
- [Fact]
- public async Task ShouldSendPostRequest()
- {
- var (httpClient, handler) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings();
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(handler.CapturedRequest);
- Assert.Equal(HttpMethod.Post, handler.CapturedRequest!.Method);
- }
-
- [Fact]
- public async Task ShouldSendToCorrectUri()
- {
- var (httpClient, handler) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings();
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(handler.CapturedRequest);
- Assert.Equal(TestUri, handler.CapturedRequest!.RequestUri);
- }
-
- [Fact]
- public async Task ShouldSetAcceptEncodingWhenCompressionEnabled()
- {
- var (httpClient, handler) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings { EnableCompression = true };
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(handler.CapturedRequest);
- Assert.Contains(handler.CapturedRequest!.Headers.AcceptEncoding,
- e => e.Value == "deflate");
- }
-
- [Fact]
- public async Task ShouldNotSetAcceptEncodingWhenCompressionDisabled()
- {
- var (httpClient, handler) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings { EnableCompression = false };
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(handler.CapturedRequest);
- Assert.DoesNotContain(handler.CapturedRequest!.Headers.AcceptEncoding,
- e => e.Value == "deflate");
- }
-
- [Fact]
- public async Task ShouldSetUserAgentWhenEnabled()
- {
- var (httpClient, handler) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings { EnableUserAgentOnConnect = true };
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(handler.CapturedRequest);
- Assert.True(handler.CapturedRequest!.Headers.Contains("User-Agent"));
- }
-
- [Fact]
- public async Task ShouldNotSetUserAgentWhenDisabled()
- {
- var (httpClient, handler) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings { EnableUserAgentOnConnect = false };
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(handler.CapturedRequest);
- Assert.False(handler.CapturedRequest!.Headers.Contains("User-Agent"));
- }
-
- [Fact]
- public async Task ShouldSetBulkResultsHeaderWhenEnabled()
- {
- var (httpClient, handler) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings { BulkResults = true };
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(handler.CapturedRequest);
- Assert.True(handler.CapturedRequest!.Headers.Contains("bulkResults"));
- Assert.Equal("true", handler.CapturedRequest.Headers.GetValues("bulkResults").First());
- }
-
- [Fact]
- public async Task ShouldNotSetBulkResultsHeaderWhenDisabled()
- {
- var (httpClient, handler) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings { BulkResults = false };
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(handler.CapturedRequest);
- Assert.False(handler.CapturedRequest!.Headers.Contains("bulkResults"));
- }
-
- [Fact]
- public async Task ShouldDecompressDeflateResponse()
- {
- // Compress the minimal response bytes with deflate
- var originalBytes = BuildMinimalResponseBytes();
- byte[] compressedBytes;
- using (var compressedStream = new MemoryStream())
- {
- using (var deflateStream = new DeflateStream(compressedStream, CompressionMode.Compress, true))
- {
- deflateStream.Write(originalBytes, 0, originalBytes.Length);
- }
- compressedBytes = compressedStream.ToArray();
- }
-
- var (httpClient, handler) = CreateMockHttpClient(compressedBytes, "deflate");
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings { EnableCompression = true };
- using var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- // Should not throw — decompression should work
- var result = await connection.SubmitAsync<object>(CreateTestRequest());
-
- Assert.NotNull(result);
- }
-
- [Fact]
- public void ShouldDisposeWithoutError()
- {
- var (httpClient, _) = CreateMockHttpClient();
- var serializer = CreateMockSerializer();
- var settings = new HttpConnectionSettings();
- var connection = new HttpConnection(TestUri, serializer, settings, httpClient);
-
- connection.Dispose();
- // Double dispose should not throw
- connection.Dispose();
- }
-
- private static RequestMessage CreateTestRequest()
- {
- return RequestMessage.Build("g.V()").AddG("g").Create();
- }
-
- private static IMessageSerializer CreateMockSerializer()
- {
- var serializer = Substitute.For<IMessageSerializer>();
- serializer.SerializeMessageAsync(Arg.Any<RequestMessage>(), Arg.Any<CancellationToken>())
- .Returns(Task.FromResult(new byte[] { 0x81 }));
- serializer.DeserializeMessageAsync(Arg.Any<byte[]>(), Arg.Any<CancellationToken>())
- .Returns(Task.FromResult(
- new ResponseMessage<List<object>>(false, new List<object>(), 200, null, null)));
- return serializer;
- }
-
- /// <summary>
- /// A test HttpMessageHandler that captures the request and returns a canned response.
- /// </summary>
- private class MockHandler : HttpMessageHandler
- {
- private readonly byte[] _responseBytes;
- private readonly string? _contentEncoding;
-
- public HttpRequestMessage? CapturedRequest { get; private set; }
-
- public MockHandler(byte[] responseBytes, string? contentEncoding = null)
- {
- _responseBytes = responseBytes;
- _contentEncoding = contentEncoding;
- }
-
- protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request,
- CancellationToken cancellationToken)
- {
- // Clone the request headers before the original request is disposed
- CapturedRequest = CloneRequest(request);
-
- var response = new HttpResponseMessage(HttpStatusCode.OK)
- {
- Content = new ByteArrayContent(_responseBytes)
- };
- response.Content.Headers.ContentType =
- new MediaTypeHeaderValue(SerializationTokens.GraphBinary4MimeType);
- if (_contentEncoding != null)
- {
- response.Content.Headers.ContentEncoding.Add(_contentEncoding);
- }
- return Task.FromResult(response);
- }
-
- private static HttpRequestMessage CloneRequest(HttpRequestMessage original)
- {
- var clone = new HttpRequestMessage(original.Method, original.RequestUri);
-
- // Copy headers
- foreach (var header in original.Headers)
- {
- clone.Headers.TryAddWithoutValidation(header.Key, header.Value);
- }
-
- // Copy content headers by creating a dummy content
- if (original.Content != null)
- {
- var contentBytes = original.Content.ReadAsByteArrayAsync().Result;
- clone.Content = new ByteArrayContent(contentBytes);
- foreach (var header in original.Content.Headers)
- {
- clone.Content.Headers.TryAddWithoutValidation(header.Key, header.Value);
- }
- }
-
- return clone;
- }
- }
- }
-}
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/Messages/RequestMessageTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/Messages/RequestMessageTests.cs
new file mode 100644
index 0000000000..a7412cf609
--- /dev/null
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/Messages/RequestMessageTests.cs
@@ -0,0 +1,110 @@
+#region License
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#endregion
+
+using System.Collections.Generic;
+using Gremlin.Net.Driver;
+using Gremlin.Net.Driver.Messages;
+using Xunit;
+
+namespace Gremlin.Net.UnitTest.Driver.Messages
+{
+ public class RequestMessageTests
+ {
+ [Fact]
+ public void ShouldSetGremlinProperty()
+ {
+ var msg = RequestMessage.Build("g.V()").Create();
+
+ Assert.Equal("g.V()", msg.Gremlin);
+ }
+
+ [Fact]
+ public void ShouldSetDefaultLanguageField()
+ {
+ var msg = RequestMessage.Build("g.V()").Create();
+
+ Assert.Equal("gremlin-lang", msg.Fields[Tokens.ArgsLanguage]);
+ }
+
+ [Fact]
+ public void ShouldSetGField()
+ {
+ var msg = RequestMessage.Build("g.V()").AddG("g").Create();
+
+ Assert.Equal("g", msg.Fields[Tokens.ArgsG]);
+ }
+
+ [Fact]
+ public void ShouldSetBindings()
+ {
+ var msg = RequestMessage.Build("g.V(_0)")
+ .AddBinding("_0", 1)
+ .Create();
+
+ var bindings = (Dictionary<string, object>)msg.Fields[Tokens.ArgsBindings];
+ Assert.Equal(1, bindings["_0"]);
+ }
+
+ [Fact]
+ public void ShouldSetMultipleBindings()
+ {
+ var bindings = new Dictionary<string, object> { { "_0", 1 }, { "_1", "test" } };
+ var msg = RequestMessage.Build("g.V(_0).has(_1)")
+ .AddBindings(bindings)
+ .Create();
+
+ var resultBindings = (Dictionary<string, object>)msg.Fields[Tokens.ArgsBindings];
+ Assert.Equal(1, resultBindings["_0"]);
+ Assert.Equal("test", resultBindings["_1"]);
+ }
+
+ [Fact]
+ public void ShouldSetAdditionalField()
+ {
+ var msg = RequestMessage.Build("g.V()")
+ .AddField(Tokens.ArgsEvalTimeout, 5000L)
+ .Create();
+
+ Assert.Equal(5000L, msg.Fields[Tokens.ArgsEvalTimeout]);
+ }
+
+ [Fact]
+ public void ShouldReportHasField()
+ {
+ var builder = RequestMessage.Build("g.V()")
+ .AddField(Tokens.ArgsBulkResults, "true");
+
+ Assert.True(builder.HasField(Tokens.ArgsBulkResults));
+ Assert.False(builder.HasField("nonexistent"));
+ }
+
+ [Fact]
+ public void ShouldIncludeEmptyBindingsWhenNoneAdded()
+ {
+ var msg = RequestMessage.Build("g.V()").Create();
+
+ var bindings = (Dictionary<string, object>)msg.Fields[Tokens.ArgsBindings];
+ Assert.Empty(bindings);
+ }
+ }
+}
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/RequestMessageSerializerTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/RequestMessageSerializerTests.cs
new file mode 100644
index 0000000000..16a27d73e2
--- /dev/null
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/RequestMessageSerializerTests.cs
@@ -0,0 +1,93 @@
+#region License
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#endregion
+
+using System.IO;
+using System.Threading.Tasks;
+using Gremlin.Net.Driver.Messages;
+using Gremlin.Net.Structure.IO.GraphBinary4;
+using Xunit;
+
+namespace Gremlin.Net.UnitTest.Structure.IO.GraphBinary4
+{
+ public class RequestMessageSerializerTests
+ {
+ [Fact]
+ public async Task ShouldWriteVersionByteFirst()
+ {
+ var msg = RequestMessage.Build("g.V()").AddG("g").Create();
+ var serializer = new RequestMessageSerializer();
+ var writer = new GraphBinaryWriter();
+ using var stream = new MemoryStream();
+
+ await serializer.WriteValueAsync(msg, stream, writer);
+
+ var bytes = stream.ToArray();
+ Assert.Equal(0x81, bytes[0]);
+ }
+
+ [Fact]
+ public async Task ShouldNotIncludeMimePrefix()
+ {
+ var msg = RequestMessage.Build("g.V()").AddG("g").Create();
+ var serializer = new RequestMessageSerializer();
+ var writer = new GraphBinaryWriter();
+ using var stream = new MemoryStream();
+
+ await serializer.WriteValueAsync(msg, stream, writer);
+
+ var bytes = stream.ToArray();
+ // First byte is version byte (MSB set), not a MIME length
+ Assert.True(bytes[0] >> 7 == 1);
+ }
+
+ [Fact]
+ public async Task ShouldSerializeFieldsAndGremlin()
+ {
+ var msg = RequestMessage.Build("g.V()").AddG("g").Create();
+ var serializer = new RequestMessageSerializer();
+ var writer = new GraphBinaryWriter();
+ using var stream = new MemoryStream();
+
+ await serializer.WriteValueAsync(msg, stream, writer);
+
+ var bytes = stream.ToArray();
+ // Should have more than just the version byte
+ Assert.True(bytes.Length > 1);
+ // Version byte
+ Assert.Equal(0x81, bytes[0]);
+ }
+
+ [Fact]
+ public async Task ShouldSerializeViaGraphBinaryMessageSerializer()
+ {
+ var msg = RequestMessage.Build("g.V()").AddG("g").Create();
+ var serializer = new GraphBinaryMessageSerializer();
+
+ var bytes = await serializer.SerializeMessageAsync(msg);
+
+ // First byte is version byte, no MIME prefix
+ Assert.Equal(0x81, bytes[0]);
+ Assert.True(bytes.Length > 1);
+ }
+ }
+}
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/ResponseMessageSerializerTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/ResponseMessageSerializerTests.cs
new file mode 100644
index 0000000000..a4e4669304
--- /dev/null
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/ResponseMessageSerializerTests.cs
@@ -0,0 +1,210 @@
+#region License
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#endregion
+
+using System.IO;
+using System.Threading.Tasks;
+using Gremlin.Net.Driver.Exceptions;
+using Gremlin.Net.Process.Traversal;
+using Gremlin.Net.Structure.IO.GraphBinary4;
+using Xunit;
+
+namespace Gremlin.Net.UnitTest.Structure.IO.GraphBinary4
+{
+ public class ResponseMessageSerializerTests
+ {
+ [Fact]
+ public async Task ShouldDeserializeNonBulkedResponse()
+ {
+ // Build a response: version(0x81) + bulked(0x00) + int value 42 + marker + status footer
+ using var stream = new MemoryStream();
+ var writer = new GraphBinaryWriter();
+
+ // Version byte
+ await stream.WriteByteAsync(0x81);
+ // Bulked = false
+ await stream.WriteByteAsync(0x00);
+ // Fully-qualified int value: type_code(0x01) + value_flag(0x00) + int32(42)
+ await stream.WriteByteAsync(0x01); // DataType.Int
+ await stream.WriteByteAsync(0x00); // value_flag = not null
+ await stream.WriteIntAsync(42);
+ // Marker: type_code(0xFD) + value_flag(0x00) + value(0x00)
+ await stream.WriteByteAsync(0xFD);
+ await stream.WriteByteAsync(0x00); // value_flag
+ await stream.WriteByteAsync(0x00); // marker value
+ // Status footer: status code 200
+ await stream.WriteIntAsync(200);
+ // Nullable status message: null (value_flag = 1)
+ await stream.WriteByteAsync(0x01);
+ // Nullable exception: null (value_flag = 1)
+ await stream.WriteByteAsync(0x01);
+
+ stream.Position = 0;
+ var reader = new GraphBinaryReader();
+ var serializer = new ResponseMessageSerializer();
+
+ var result = await serializer.ReadValueAsync(stream, reader);
+
+ Assert.False(result.Bulked);
+ Assert.Single(result.Result);
+ Assert.Equal(42, result.Result[0]);
+ Assert.Equal(200, result.StatusCode);
+ Assert.Null(result.StatusMessage);
+ Assert.Null(result.Exception);
+ }
+
+ [Fact]
+ public async Task ShouldDeserializeBulkedResponseWithTraverserWrapping()
+ {
+ using var stream = new MemoryStream();
+
+ // Version byte
+ await stream.WriteByteAsync(0x81);
+ // Bulked = true
+ await stream.WriteByteAsync(0x01);
+ // Fully-qualified string value: type_code(0x03) + value_flag(0x00) + string "hello"
+ await stream.WriteByteAsync(0x03); // DataType.String
+ await stream.WriteByteAsync(0x00); // value_flag
+ var helloBytes = System.Text.Encoding.UTF8.GetBytes("hello");
+ await stream.WriteIntAsync(helloBytes.Length);
+ await stream.WriteAsync(helloBytes);
+ // Bulk count as fully-qualified Long: type_code(0x02) + value_flag(0x00) + long(3)
+ await stream.WriteByteAsync(0x02); // DataType.Long
+ await stream.WriteByteAsync(0x00); // value_flag = not null
+ await stream.WriteLongAsync(3);
+ // Marker: type_code(0xFD) + value_flag(0x00) + value(0x00)
+ await stream.WriteByteAsync(0xFD);
+ await stream.WriteByteAsync(0x00); // value_flag
+ await stream.WriteByteAsync(0x00); // marker value
+ // Status footer
+ await stream.WriteIntAsync(200);
+ await stream.WriteByteAsync(0x01); // null message
+ await stream.WriteByteAsync(0x01); // null exception
+
+ stream.Position = 0;
+ var reader = new GraphBinaryReader();
+ var serializer = new ResponseMessageSerializer();
+
+ var result = await serializer.ReadValueAsync(stream, reader);
+
+ Assert.True(result.Bulked);
+ Assert.Single(result.Result);
+ var traverser = Assert.IsType<Traverser>(result.Result[0]);
+ Assert.Equal("hello", (string)traverser.Object);
+ Assert.Equal(3L, traverser.Bulk);
+ }
+
+ [Fact]
+ public async Task ShouldThrowResponseExceptionOnErrorStatus()
+ {
+ using var stream = new MemoryStream();
+
+ // Version byte
+ await stream.WriteByteAsync(0x81);
+ // Bulked = false
+ await stream.WriteByteAsync(0x00);
+ // No result data — go straight to marker: type_code(0xFD) + value_flag(0x00) + value(0x00)
+ await stream.WriteByteAsync(0xFD);
+ await stream.WriteByteAsync(0x00); // value_flag
+ await stream.WriteByteAsync(0x00); // marker value
+ // Status footer: status code 500
+ await stream.WriteIntAsync(500);
+ // Status message: "Server error"
+ await stream.WriteByteAsync(0x00); // value_flag = not null
+ var msgBytes = System.Text.Encoding.UTF8.GetBytes("Server error");
+ await stream.WriteIntAsync(msgBytes.Length);
+ await stream.WriteAsync(msgBytes);
+ // Exception: "java.lang.RuntimeException"
+ await stream.WriteByteAsync(0x00); // value_flag = not null
+ var excBytes = System.Text.Encoding.UTF8.GetBytes("java.lang.RuntimeException");
+ await stream.WriteIntAsync(excBytes.Length);
+ await stream.WriteAsync(excBytes);
+
+ stream.Position = 0;
+ var reader = new GraphBinaryReader();
+ var serializer = new ResponseMessageSerializer();
+
+ var ex = await Assert.ThrowsAsync<ResponseException>(
+ () => serializer.ReadValueAsync(stream, reader));
+
+ Assert.Equal(500, ex.StatusCode);
+ Assert.Equal("Server error", ex.Message);
+ Assert.Equal("java.lang.RuntimeException", ex.ServerException);
+ }
+
+ [Fact]
+ public async Task ShouldThrowOnInvalidVersionByte()
+ {
+ using var stream = new MemoryStream();
+ // Invalid version byte (MSB not set)
+ await stream.WriteByteAsync(0x01);
+
+ stream.Position = 0;
+ var reader = new GraphBinaryReader();
+ var serializer = new ResponseMessageSerializer();
+
+ await Assert.ThrowsAsync<IOException>(
+ () => serializer.ReadValueAsync(stream, reader));
+ }
+
+ [Fact]
+ public async Task ShouldDeserializeMultipleNonBulkedResults()
+ {
+ using var stream = new MemoryStream();
+
+ // Version + non-bulked
+ await stream.WriteByteAsync(0x81);
+ await stream.WriteByteAsync(0x00);
+ // First result: int 1
+ await stream.WriteByteAsync(0x01);
+ await stream.WriteByteAsync(0x00);
+ await stream.WriteIntAsync(1);
+ // Second result: int 2
+ await stream.WriteByteAsync(0x01);
+ await stream.WriteByteAsync(0x00);
+ await stream.WriteIntAsync(2);
+ // Third result: int 3
+ await stream.WriteByteAsync(0x01);
+ await stream.WriteByteAsync(0x00);
+ await stream.WriteIntAsync(3);
+ // Marker: type_code(0xFD) + value_flag(0x00) + value(0x00)
+ await stream.WriteByteAsync(0xFD);
+ await stream.WriteByteAsync(0x00); // value_flag
+ await stream.WriteByteAsync(0x00); // marker value
+ // Status footer: 200, null, null
+ await stream.WriteIntAsync(200);
+ await stream.WriteByteAsync(0x01);
+ await stream.WriteByteAsync(0x01);
+
+ stream.Position = 0;
+ var reader = new GraphBinaryReader();
+ var serializer = new ResponseMessageSerializer();
+
+ var result = await serializer.ReadValueAsync(stream, reader);
+
+ Assert.Equal(3, result.Result.Count);
+ Assert.Equal(1, result.Result[0]);
+ Assert.Equal(2, result.Result[1]);
+ Assert.Equal(3, result.Result[2]);
+ }
+ }
+}