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/05/06 22:42:02 UTC
(tinkerpop) branch master updated: Support Streaming in .NET (#3403)
This is an automated email from the ASF dual-hosted git repository.
xiazcy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/tinkerpop.git
The following commit(s) were added to refs/heads/master by this push:
new 30129ad373 Support Streaming in .NET (#3403)
30129ad373 is described below
commit 30129ad3733650ba363caf0ef4a06e71bc6ed971
Author: Yang Xia <55...@users.noreply.github.com>
AuthorDate: Wed May 6 15:41:56 2026 -0700
Support Streaming in .NET (#3403)
Added streaming deserialization in `gremlin-dotnet`, moved minimal target to `.net8.0`.
Assisted-by: Kiro:claude-opus-4-6
---
CHANGELOG.asciidoc | 2 +
docs/src/reference/gremlin-variants.asciidoc | 6 +-
docs/src/upgrade/release-4.x.x.asciidoc | 56 ++
.../src/Gremlin.Net/Driver/Connection.cs | 194 +++---
.../Gremlin.Net/Driver/GremlinClientExtensions.cs | 29 +-
.../src/Gremlin.Net/Driver/IMessageSerializer.cs | 12 +-
.../Gremlin.Net/Driver/Messages/ResponseMessage.cs | 75 ---
.../Driver/Remote/DriverRemoteTraversal.cs | 10 +-
gremlin-dotnet/src/Gremlin.Net/Driver/ResultSet.cs | 121 +++-
.../Gremlin.Net/Driver/StreamingResponseContext.cs | 71 +++
gremlin-dotnet/src/Gremlin.Net/Gremlin.Net.csproj | 13 +-
.../Process/Traversal/DefaultTraversal.cs | 252 ++++----
.../Gremlin.Net/Process/Traversal/ITraversal.cs | 39 +-
gremlin-dotnet/src/Gremlin.Net/Process/Utils.cs | 21 +
.../GraphBinary4/GraphBinary4MessageSerializer.cs | 8 +-
...eMessageSerializer.cs => ResponseSerializer.cs} | 65 +-
.../Structure/IO/GraphBinary4/StreamExtensions.cs | 87 +--
.../IO/GraphBinary4/Types/SingleTypeSerializer.cs | 4 +-
.../MessageSerializerBenchmarks.cs | 29 +-
.../Driver/GremlinClientTests.cs | 30 +-
.../Driver/MessagesTests.cs | 12 +-
.../Gherkin/CommonSteps.cs | 54 +-
.../Gremlin.Net.IntegrationTest/Gherkin/Gremlin.cs | 4 +-
.../Gremlin.Net.UnitTest/Driver/ConnectionTests.cs | 140 ++++-
.../Driver/DriverRemoteConnectionTests.cs | 5 +-
.../Gremlin.Net.UnitTest/Driver/ResultSetTests.cs | 328 ++++++++++
.../Process/Traversal/TestTraversal.cs | 19 +-
.../Process/Traversal/TestTraversalStrategy.cs | 4 +-
.../Process/Traversal/TraversalTests.cs | 141 ++++-
.../GraphBinary4MessageSerializerTests.cs | 8 +-
...rializerTests.cs => ResponseSerializerTests.cs} | 102 +++-
.../IO/GraphBinary4/StreamExtensionsTests.cs | 382 ++++++++++++
.../StreamingDeserializerRoundTripTests.cs | 669 +++++++++++++++++++++
33 files changed, 2456 insertions(+), 536 deletions(-)
diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index f895ea2657..193d9c5587 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -67,6 +67,8 @@ image::https://raw.githubusercontent.com/apache/tinkerpop/master/docs/static/ima
* Replace `WebSocket` with `HTTP` (non-streaming) in `gremlin-dotnet`.
* Added `MimeType` to `IMessageSerializer` and split client option to allow separate request and response serialization in `gremlin-dotnet`.
* Added `RequestInterceptor` to `gremlin-dotnet` with `auth` reference implementations.
+* Added streaming deserialization to `gremlin-dotnet`, results are now deserialized incrementally from the HTTP response stream via `IAsyncEnumerable<T>`.
+* Target framework for `gremlin-dotnet` updated to `net8.0`.
* Bumped minimum Node.js version for `gremlin-javascript` from 20 to 22.
* Removed `mimeType` connection option in `gremlin-javascript` in favor of direct use of `reader` and `writer` options.
* Bumped ANTLR to 4.13.2
diff --git a/docs/src/reference/gremlin-variants.asciidoc b/docs/src/reference/gremlin-variants.asciidoc
index ee12ef6409..ee94b0ecd2 100644
--- a/docs/src/reference/gremlin-variants.asciidoc
+++ b/docs/src/reference/gremlin-variants.asciidoc
@@ -2053,13 +2053,14 @@ anchor:gremlin-DotNet[]
== Gremlin.Net
image:gremlin-dotnet-logo.png[width=371,float=right] Apache TinkerPop's Gremlin.Net implements Gremlin within the C#
-language. It targets .NET Standard and can therefore be used on different operating systems and with different .NET
-frameworks, such as .NET Framework and link:https://www.microsoft.com/net/core[.NET Core]. Since the C# syntax is very
+language. It targets .NET 8.0 and can therefore be used on different operating systems. Since the C# syntax is very
similar to that of Java, it should be easy to switch between Gremlin-Java and Gremlin.Net. The only major syntactical
difference is that all method names in Gremlin.Net use PascalCase as opposed to camelCase in Gremlin-Java in order
to comply with .NET conventions.
The Gremlin.Net driver communicates with Gremlin Server over HTTP using the GraphBinary 4.0 serialization format.
+Response deserialization is streaming — results are yielded incrementally as they arrive from the server, rather than
+buffered entirely before processing.
[source,powershell]
nuget install Gremlin.Net
@@ -2075,6 +2076,7 @@ The following table outlines recommended runtime versions by the release in whic
|3.6.0 |.NET Standard 2.0; .NET 6.0
|3.7.0 |.NET Standard 2.0; .NET6.0
|3.8.0 |.NET Standard 2.0; .NET6.0
+|4.0.0 |.NET 8.0
|===
[[gremlin-dotnet-connecting]]
diff --git a/docs/src/upgrade/release-4.x.x.asciidoc b/docs/src/upgrade/release-4.x.x.asciidoc
index 547f153b10..f632a61c77 100644
--- a/docs/src/upgrade/release-4.x.x.asciidoc
+++ b/docs/src/upgrade/release-4.x.x.asciidoc
@@ -215,6 +215,62 @@ Connection(url, traversal_source, executor, pool,
Custom transport implementations are no longer supported. The driver uses `AiohttpHTTPTransport` directly.
+==== Runtime Upgrades
+
+The minimum target framework is now `net8.0` (previously `netstandard2.0;net6.0`). .NET 6 reached end-of-life in
+November 2024; .NET 8 is the current LTS release.
+
+==== .NET HTTP Streaming Response Support
+
+The .NET driver (`Gremlin.Net`) now uses streaming deserialization, results are deserialized incrementally from the
+HTTP response stream as bytes arrive over the network, rather than buffering the entire response before processing.
+This reduces memory usage and latency for large result sets.
+
+The traversal API (`Next()`, `ToList()`, `HasNext()`, `ToSet()`, `Iterate()`) is unchanged — these methods block
+internally on the async stream, so existing traversal code works without modification.
+
+`ResultSet<T>` no longer implements `IReadOnlyCollection<T>`. It now implements `IAsyncEnumerable<T>`, backed by a
+`Channel<T>` that receives items from a background deserialization task. Users who access `ResultSet<T>` directly
+via `GremlinClient.SubmitAsync<T>()` will need to update their code:
+
+[source,csharp]
+----
+// 3.x — ResultSet was IReadOnlyCollection<T>
+var resultSet = await client.SubmitAsync<int>("g.inject(1,2,3)");
+var count = resultSet.Count; // no longer available
+foreach (var item in resultSet) { } // no longer available
+
+// 4.0 — ResultSet is IAsyncEnumerable<T>
+var resultSet = await client.SubmitAsync<int>("g.inject(1,2,3)");
+var list = await resultSet.ToListAsync(); // materialize if you need Count
+await foreach (var item in resultSet) { } // async iteration
+----
+
+The `IMessageSerializer` interface has also changed: `DeserializeMessageAsync` now accepts a `Stream` and returns
+`IAsyncEnumerable<object>` instead of accepting `byte[]` and returning `Task<ResponseMessage<List<object>>>`.
+Custom serializer implementations will need to be updated. The `ResponseMessage<T>` type has been removed.
+
+With streaming, server errors (`ResponseException`) are now thrown during result iteration rather than during
+`SubmitAsync`. Code that catches `ResponseException` around `SubmitAsync` should be updated to catch around the
+result consumption instead:
+
+[source,csharp]
+----
+// 3.x — exception thrown during SubmitAsync
+try {
+ var resultSet = await client.SubmitAsync<int>("invalid");
+} catch (ResponseException ex) { ... }
+
+// 4.0 — exception thrown during iteration
+var resultSet = await client.SubmitAsync<int>("invalid");
+try {
+ var list = await resultSet.ToListAsync();
+} catch (ResponseException ex) { ... }
+----
+
+The traversal API is not affected — `Next()`, `ToList()`, etc. still throw `ResponseException` directly since they
+block on the async stream internally.
+
=== Upgrading for Providers
==== Graph System Providers
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
index 415513918c..e0b0e30b78 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Connection.cs
@@ -25,10 +25,10 @@ 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.Channels;
using System.Threading.Tasks;
using Gremlin.Net.Driver.Messages;
using Gremlin.Net.Process;
@@ -77,7 +77,6 @@ namespace Gremlin.Net.Driver
_settings = settings;
_interceptors = interceptors ?? Array.Empty<Func<HttpRequestContext, Task>>();
-#if NET6_0_OR_GREATER
var handler = new SocketsHttpHandler
{
PooledConnectionIdleTimeout = settings.IdleConnectionTimeout,
@@ -93,10 +92,6 @@ namespace Gremlin.Net.Driver
};
}
_httpClient = new HttpClient(handler);
-#else
- _httpClient = new HttpClient();
- _httpClient.Timeout = settings.ConnectionTimeout;
-#endif
}
/// <summary>
@@ -115,10 +110,17 @@ namespace Gremlin.Net.Driver
_interceptors = interceptors ?? Array.Empty<Func<HttpRequestContext, Task>>();
}
+ /// <summary>
+ /// Submits a <see cref="RequestMessage"/> to the server and returns a streaming
+ /// <see cref="ResultSet{T}"/> whose background task owns the HTTP response lifetime.
+ /// </summary>
+ /// <typeparam name="T">The type of the expected result elements.</typeparam>
+ /// <param name="requestMessage">The request to send.</param>
+ /// <param name="cancellationToken">The token to cancel the operation.</param>
+ /// <returns>A <see cref="ResultSet{T}"/> that streams results as they arrive.</returns>
public async Task<ResultSet<T>> SubmitAsync<T>(RequestMessage requestMessage,
CancellationToken cancellationToken = default)
{
- // Build HttpRequestContext with default headers
var headers = new Dictionary<string, string>();
headers["Accept"] = _responseSerializer.MimeType;
@@ -140,7 +142,6 @@ namespace Gremlin.Net.Driver
object body;
if (_requestSerializer != null)
{
- // Default path: serialize before interceptors
var requestBytes = await _requestSerializer.SerializeMessageAsync(requestMessage, cancellationToken)
.ConfigureAwait(false);
body = requestBytes;
@@ -148,103 +149,146 @@ namespace Gremlin.Net.Driver
}
else
{
- // Interceptor-managed path: body is RequestMessage, interceptor must serialize
body = requestMessage;
}
var context = new HttpRequestContext("POST", _uri, headers, body);
- // Apply interceptors in order
foreach (var interceptor in _interceptors)
{
await interceptor(context).ConfigureAwait(false);
}
- // Convert HttpRequestContext to HttpRequestMessage
- using var httpRequest = new HttpRequestMessage(new HttpMethod(context.Method), context.Uri);
-
- if (context.Body is byte[] bodyBytes)
- {
- httpRequest.Content = new ByteArrayContent(bodyBytes);
- }
- else if (context.Body is HttpContent httpContent)
- {
- httpRequest.Content = httpContent;
- }
- else
- {
- throw new InvalidOperationException(
- "Request body must be byte[] or HttpContent after all interceptors complete, " +
- "but found " + (context.Body?.GetType().Name ?? "null") +
- ". Either provide a requestSerializer or add an interceptor " +
- "that serializes the RequestMessage.");
- }
-
- foreach (var header in context.Headers)
+ // The HttpResponseMessage is NOT disposed here — ownership transfers to
+ // StreamingResponseContext via the background task.
+ HttpResponseMessage response;
+ using (var httpRequest = new HttpRequestMessage(new HttpMethod(context.Method), context.Uri))
{
- // Content-Type must be set on the content headers, not the request headers
- if (string.Equals(header.Key, "Content-Type", StringComparison.OrdinalIgnoreCase))
+ if (context.Body is byte[] bodyBytes)
{
- httpRequest.Content.Headers.ContentType = new MediaTypeHeaderValue(header.Value);
+ httpRequest.Content = new ByteArrayContent(bodyBytes);
+ }
+ else if (context.Body is HttpContent httpContent)
+ {
+ httpRequest.Content = httpContent;
}
else
{
- httpRequest.Headers.TryAddWithoutValidation(header.Key, header.Value);
+ throw new InvalidOperationException(
+ "Request body must be byte[] or HttpContent after all interceptors complete, " +
+ "but found " + (context.Body?.GetType().Name ?? "null") +
+ ". Either provide a requestSerializer or add an interceptor " +
+ "that serializes the RequestMessage.");
+ }
+
+ foreach (var header in context.Headers)
+ {
+ if (string.Equals(header.Key, "Content-Type", StringComparison.OrdinalIgnoreCase))
+ {
+ httpRequest.Content.Headers.ContentType = new MediaTypeHeaderValue(header.Value);
+ }
+ else
+ {
+ httpRequest.Headers.TryAddWithoutValidation(header.Key, header.Value);
+ }
}
- }
- using var response = await _httpClient.SendAsync(httpRequest, cancellationToken)
- .ConfigureAwait(false);
+ response = await _httpClient.SendAsync(httpRequest,
+ HttpCompletionOption.ResponseHeadersRead, cancellationToken)
+ .ConfigureAwait(false);
+ }
- // If the HTTP status indicates an error and the response is not GraphBinary,
- // attempt to read the body as a text error message. The server may respond with
- // a JSON body containing an error field, or plain text. Only when the Content-Type
- // is GraphBinary do we fall through to normal deserialization so the status footer
- // in the GB4 response can surface the application-level error.
if (!response.IsSuccessStatusCode &&
response.Content.Headers.ContentType?.MediaType != _responseSerializer.MimeType)
{
- var errorBody = await response.Content.ReadAsStringAsync().ConfigureAwait(false);
+ using (response)
+ {
+ var errorBody = await response.Content.ReadAsStringAsync().ConfigureAwait(false);
- // Try to extract the "message" field from a JSON error response
- var errorMessage = TryExtractJsonError(errorBody)
- ?? $"Gremlin Server returned HTTP {(int)response.StatusCode}: {errorBody}";
+ // Try to extract the "message" field from a JSON error response
+ var errorMessage = TryExtractJsonError(errorBody)
+ ?? $"Gremlin Server returned HTTP {(int)response.StatusCode}: {errorBody}";
- throw new HttpRequestException(errorMessage);
+ throw new HttpRequestException(errorMessage);
+ }
}
- var responseBytes = await ReadResponseBytesAsync(response).ConfigureAwait(false);
+ StreamingResponseContext? streamingContext = null;
+ CancellationTokenSource? disposeCts = null;
+ CancellationTokenSource? linkedCts = null;
+ try
+ {
+ var contentStream = await response.Content.ReadAsStreamAsync()
+ .ConfigureAwait(false);
+ DeflateStream? deflateStream = null;
+ if (response.Content.Headers.ContentEncoding.Contains("deflate"))
+ {
+ deflateStream = new DeflateStream(contentStream, CompressionMode.Decompress);
+ }
+ streamingContext = new StreamingResponseContext(
+ response, contentStream, deflateStream);
- var responseMessage = await _responseSerializer.DeserializeMessageAsync(responseBytes, cancellationToken)
- .ConfigureAwait(false);
+ var resultStream = _responseSerializer.DeserializeMessageAsync(
+ streamingContext.Stream, cancellationToken);
- return BuildResultSet<T>(responseMessage);
- }
+ var channel = Channel.CreateUnbounded<object>(
+ new UnboundedChannelOptions { SingleWriter = true });
+ disposeCts = new CancellationTokenSource();
+ linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
+ cancellationToken, disposeCts.Token);
- private static async Task<byte[]> ReadResponseBytesAsync(HttpResponseMessage response)
- {
- using var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
- if (response.Content.Headers.ContentEncoding.Contains("deflate"))
+ var capturedLinkedCts = linkedCts;
+ var capturedStreamingContext = streamingContext;
+
+ var backgroundTask = Task.Run(async () =>
+ {
+ // Note: ResponseException from the status footer is propagated after
+ // all result items have been yielded, so consumers will see all results
+ // before the exception if the status code is non-200.
+ try
+ {
+ await foreach (var item in resultStream
+ .WithCancellation(capturedLinkedCts.Token).ConfigureAwait(false))
+ {
+ await channel.Writer.WriteAsync(item, capturedLinkedCts.Token)
+ .ConfigureAwait(false);
+ }
+ channel.Writer.Complete();
+ }
+ catch (Exception ex)
+ {
+ channel.Writer.Complete(ex);
+ }
+ finally
+ {
+ capturedLinkedCts.Dispose();
+ capturedStreamingContext.Dispose();
+ }
+ }, CancellationToken.None);
+
+ // Ownership transferred to background task — prevent catch block from
+ // double-disposing.
+ linkedCts = null;
+ streamingContext = null;
+
+ return new ResultSet<T>(channel.Reader, disposeCts, backgroundTask);
+ }
+ catch
{
- using var deflateStream = new DeflateStream(stream, CompressionMode.Decompress);
- using var ms = new MemoryStream();
- await deflateStream.CopyToAsync(ms).ConfigureAwait(false);
- return ms.ToArray();
+ linkedCts?.Dispose();
+ // If streamingContext was not yet created, the response is not yet owned
+ // by it and must be disposed separately.
+ if (streamingContext == null)
+ {
+ response.Dispose();
+ }
+ else
+ {
+ streamingContext.Dispose();
+ }
+ disposeCts?.Dispose();
+ throw;
}
- 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 = typeof(T) == typeof(Traverser)
- ? responseMessage.Result
- .Select(item => item is Traverser ? item : (object)new Traverser(item))
- .Cast<T>().ToList()
- : responseMessage.Result.Cast<T>().ToList();
-
- return new ResultSet<T>(results, new Dictionary<string, object>());
}
/// <summary>
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClientExtensions.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClientExtensions.cs
index 176def707f..66927c5b17 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClientExtensions.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/GremlinClientExtensions.cs
@@ -22,7 +22,6 @@
#endregion
using System.Collections.Generic;
-using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Gremlin.Net.Driver.Messages;
@@ -57,9 +56,13 @@ namespace Gremlin.Net.Driver
Dictionary<string, object>? bindings = null,
CancellationToken cancellationToken = default)
{
- var resultCollection = await gremlinClient.SubmitAsync<T>(requestScript, bindings, cancellationToken)
+ await using var resultSet = await gremlinClient.SubmitAsync<T>(requestScript, bindings, cancellationToken)
.ConfigureAwait(false);
- return resultCollection.FirstOrDefault();
+ await foreach (var item in resultSet.WithCancellation(cancellationToken).ConfigureAwait(false))
+ {
+ return item;
+ }
+ return default;
}
/// <summary>
@@ -81,9 +84,13 @@ namespace Gremlin.Net.Driver
public static async Task<T?> SubmitWithSingleResultAsync<T>(this IGremlinClient gremlinClient,
RequestMessage requestMessage, CancellationToken cancellationToken = default)
{
- var resultCollection =
+ await using var resultSet =
await gremlinClient.SubmitAsync<T>(requestMessage, cancellationToken).ConfigureAwait(false);
- return resultCollection.FirstOrDefault();
+ await foreach (var item in resultSet.WithCancellation(cancellationToken).ConfigureAwait(false))
+ {
+ return item;
+ }
+ return default;
}
/// <summary>
@@ -102,7 +109,11 @@ namespace Gremlin.Net.Driver
public static async Task SubmitAsync(this IGremlinClient gremlinClient, string requestScript,
Dictionary<string, object>? bindings = null, CancellationToken cancellationToken = default)
{
- await gremlinClient.SubmitAsync<object>(requestScript, bindings, cancellationToken).ConfigureAwait(false);
+ await using var resultSet = await gremlinClient.SubmitAsync<object>(requestScript, bindings, cancellationToken).ConfigureAwait(false);
+ // Drain the stream to ensure the background task completes and resources are released.
+ await foreach (var _ in resultSet.WithCancellation(cancellationToken).ConfigureAwait(false))
+ {
+ }
}
/// <summary>
@@ -120,7 +131,11 @@ namespace Gremlin.Net.Driver
public static async Task SubmitAsync(this IGremlinClient gremlinClient, RequestMessage requestMessage,
CancellationToken cancellationToken = default)
{
- await gremlinClient.SubmitAsync<object>(requestMessage, cancellationToken).ConfigureAwait(false);
+ await using var resultSet = await gremlinClient.SubmitAsync<object>(requestMessage, cancellationToken).ConfigureAwait(false);
+ // Drain the stream to ensure the background task completes and resources are released.
+ await foreach (var _ in resultSet.WithCancellation(cancellationToken).ConfigureAwait(false))
+ {
+ }
}
/// <summary>
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/IMessageSerializer.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/IMessageSerializer.cs
index 81fb5fd6b1..9abb59dc66 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/IMessageSerializer.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/IMessageSerializer.cs
@@ -22,6 +22,7 @@
#endregion
using System.Collections.Generic;
+using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Gremlin.Net.Driver.Messages;
@@ -50,12 +51,15 @@ namespace Gremlin.Net.Driver
CancellationToken cancellationToken = default);
/// <summary>
- /// Deserializes a <see cref="ResponseMessage{T}"/> from a byte array.
+ /// Deserializes result objects from a <see cref="Stream"/>, yielding each
+ /// object as it is deserialized. The stream is expected to contain a
+ /// GraphBinary 4.0 response including the version byte, result data, marker,
+ /// and status footer.
/// </summary>
- /// <param name="message">The serialized message to deserialize.</param>
+ /// <param name="stream">The response stream to deserialize from.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
- /// <returns>The deserialized <see cref="ResponseMessage{T}"/>.</returns>
- Task<ResponseMessage<List<object>>> DeserializeMessageAsync(byte[] message,
+ /// <returns>An async sequence of deserialized result objects.</returns>
+ IAsyncEnumerable<object> DeserializeMessageAsync(Stream stream,
CancellationToken cancellationToken = default);
}
}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseMessage.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseMessage.cs
deleted file mode 100644
index f05e5c56a1..0000000000
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Messages/ResponseMessage.cs
+++ /dev/null
@@ -1,75 +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.Messages
-{
- /// <summary>
- /// The 4.0 response message returned from the server.
- /// </summary>
- /// <typeparam name="T">The type of the result data.</typeparam>
- public record ResponseMessage<T>
- {
- /// <summary>
- /// Initializes a new instance of the <see cref="ResponseMessage{T}" /> record.
- /// </summary>
- /// <param name="bulked">Whether the response contains bulked results.</param>
- /// <param name="result">The deserialized result data.</param>
- /// <param name="statusCode">The status code from the GraphBinary response status (after the end-of-stream marker).</param>
- /// <param name="statusMessage">Optional status message from the GraphBinary response status.</param>
- /// <param name="exception">Optional exception string from the GraphBinary response status.</param>
- public ResponseMessage(bool bulked, T result, int statusCode,
- string? statusMessage, string? exception)
- {
- Bulked = bulked;
- Result = result;
- StatusCode = statusCode;
- StatusMessage = statusMessage;
- Exception = exception;
- }
-
- /// <summary>
- /// Gets whether the response contains bulked results.
- /// </summary>
- public bool Bulked { get; }
-
- /// <summary>
- /// Gets the deserialized result data.
- /// </summary>
- public T Result { get; }
-
- /// <summary>
- /// Gets the status code from the GraphBinary response status (200 = success).
- /// </summary>
- public int StatusCode { get; }
-
- /// <summary>
- /// Gets the optional status message from the GraphBinary response status.
- /// </summary>
- public string? StatusMessage { get; }
-
- /// <summary>
- /// Gets the optional exception string from the GraphBinary response status.
- /// </summary>
- public string? Exception { get; }
- }
-}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteTraversal.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteTraversal.cs
index 9cebc8180e..e4ff99563a 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteTraversal.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/Remote/DriverRemoteTraversal.cs
@@ -22,18 +22,22 @@
#endregion
using System;
-using System.Collections.Generic;
using Gremlin.Net.Process.Traversal;
namespace Gremlin.Net.Driver.Remote
{
+ /// <summary>
+ /// A traversal returned from a remote Gremlin Server submission, wrapping a
+ /// <see cref="ResultSet{T}"/> of <see cref="Traverser"/> instances.
+ /// </summary>
internal class DriverRemoteTraversal<TStart, TEnd> : DefaultTraversal<TStart, TEnd>
{
- public DriverRemoteTraversal(IEnumerable<Traverser> traversers)
+ public DriverRemoteTraversal(ResultSet<Traverser> resultSet)
{
- Traversers = traversers;
+ Traversers = resultSet;
}
+ /// <inheritdoc />
public override GremlinLang GremlinLang => throw new NotSupportedException("Remote traversals do not have GremlinLang");
}
}
\ No newline at end of file
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/ResultSet.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/ResultSet.cs
index 83a5184ba3..710066cce0 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Driver/ResultSet.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/ResultSet.cs
@@ -21,49 +21,128 @@
#endregion
-using System.Collections;
+using System;
using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
namespace Gremlin.Net.Driver
{
/// <summary>
- /// A ResultSet is returned from the submission of a Gremlin script to the server and represents the results
- /// provided by the server. ResultSet includes enumerable data and status attributes.
+ /// A streaming ResultSet returned from the submission of a Gremlin request.
+ /// Implements <see cref="IAsyncEnumerable{T}"/>; so consumers can iterate results as they
+ /// arrive from the server via <c>await foreach</c>.
+ /// Single-consumer only — calling GetAsyncEnumerator a second time throws
+ /// InvalidOperationException.
/// </summary>
/// <typeparam name="T">Type of the result elements</typeparam>
- public sealed class ResultSet<T> : IReadOnlyCollection<T>
+ public sealed class ResultSet<T> : IAsyncEnumerable<T>, IAsyncDisposable, IDisposable
{
- private readonly IReadOnlyCollection<T> _data;
+ private readonly ChannelReader<object> _channelReader;
+ private readonly CancellationTokenSource _disposeCts;
+ private readonly Task _backgroundTask;
+ // int rather than bool because Interlocked.CompareExchange does not support bool
+ private int _enumerated;
+ private int _disposed;
/// <summary>
- /// Gets or sets the status attributes from the gremlin response
+ /// Initializes a new <see cref="ResultSet{T}"/> backed by a channel, a cancellation
+ /// source for disposal, and a background task that writes items into the channel.
/// </summary>
- public IReadOnlyDictionary<string, object> StatusAttributes { get; }
+ internal ResultSet(ChannelReader<object> channelReader,
+ CancellationTokenSource disposeCts, Task backgroundTask)
+ {
+ _channelReader = channelReader;
+ _disposeCts = disposeCts;
+ _backgroundTask = backgroundTask;
+ }
+
+ /// <inheritdoc />
+ public IAsyncEnumerator<T> GetAsyncEnumerator(
+ CancellationToken cancellationToken = default)
+ {
+ if (Interlocked.CompareExchange(ref _enumerated, 1, 0) != 0)
+ {
+ throw new InvalidOperationException(
+ "ResultSet can only be enumerated once. Use ToListAsync() if you need " +
+ "to iterate the results multiple times.");
+ }
+
+ return EnumerateChannel(cancellationToken);
+ }
+
+ private async IAsyncEnumerator<T> EnumerateChannel(CancellationToken cancellationToken)
+ {
+ using var linked = CancellationTokenSource.CreateLinkedTokenSource(
+ cancellationToken, _disposeCts.Token);
+
+ await foreach (var item in _channelReader.ReadAllAsync(linked.Token)
+ .ConfigureAwait(false))
+ {
+ yield return (T)item;
+ }
+ }
/// <summary>
- /// Initializes a new instance of the ResultSet class for the specified data and status attributes.
+ /// Materializes all results into a List<T>.
+ /// This is a convenience method for consumers that need the complete collection.
+ /// Consumes the stream — cannot be called after GetAsyncEnumerator.
/// </summary>
- /// <param name="data"></param>
- /// <param name="attributes"></param>
- public ResultSet(IReadOnlyCollection<T> data, IReadOnlyDictionary<string, object> attributes)
+ public async Task<List<T>> ToListAsync(CancellationToken cancellationToken = default)
{
- _data = data;
- this.StatusAttributes = attributes;
+ var results = new List<T>();
+ await foreach (var item in this.WithCancellation(cancellationToken)
+ .ConfigureAwait(false))
+ {
+ results.Add(item);
+ }
+ return results;
}
/// <inheritdoc />
- public IEnumerator<T> GetEnumerator()
+ public async ValueTask DisposeAsync()
{
- return _data.GetEnumerator();
+ if (Interlocked.CompareExchange(ref _disposed, 1, 0) != 0)
+ {
+ return;
+ }
+
+ _disposeCts.Cancel();
+ try
+ {
+ await _backgroundTask.ConfigureAwait(false);
+ }
+ catch
+ {
+ // Background task may throw OperationCanceledException or other
+ // exceptions — swallow them during disposal.
+ }
+ _disposeCts.Dispose();
}
- /// <inheritdoc />
- IEnumerator IEnumerable.GetEnumerator()
+ /// <summary>
+ /// Synchronous disposal fallback. Prefer <see cref="DisposeAsync"/> in async contexts
+ /// to avoid blocking the calling thread.
+ /// </summary>
+ public void Dispose()
{
- return _data.GetEnumerator();
- }
+ if (Interlocked.CompareExchange(ref _disposed, 1, 0) != 0)
+ {
+ return;
+ }
- /// <inheritdoc />
- public int Count => _data.Count;
+ _disposeCts.Cancel();
+ try
+ {
+ _backgroundTask.GetAwaiter().GetResult();
+ }
+ catch
+ {
+ // Background task may throw OperationCanceledException or other
+ // exceptions — swallow them during disposal.
+ }
+ _disposeCts.Dispose();
+ }
}
}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Driver/StreamingResponseContext.cs b/gremlin-dotnet/src/Gremlin.Net/Driver/StreamingResponseContext.cs
new file mode 100644
index 0000000000..9d40e1f39f
--- /dev/null
+++ b/gremlin-dotnet/src/Gremlin.Net/Driver/StreamingResponseContext.cs
@@ -0,0 +1,71 @@
+#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.IO;
+using System.IO.Compression;
+using System.Net.Http;
+
+namespace Gremlin.Net.Driver
+{
+ /// <summary>
+ /// Holds the HTTP response resources that must remain alive during streaming
+ /// deserialization. Disposed when the stream is fully consumed, faulted, or cancelled.
+ /// </summary>
+ internal sealed class StreamingResponseContext : IDisposable
+ {
+ private readonly HttpResponseMessage _response;
+ private readonly Stream _contentStream;
+ private readonly DeflateStream? _deflateStream;
+
+ /// <summary>
+ /// Gets the stream to read from — the decompression stream if present,
+ /// otherwise the raw content stream.
+ /// </summary>
+ public Stream Stream => (Stream?)_deflateStream ?? _contentStream;
+
+ /// <summary>
+ /// Initializes a new instance of the <see cref="StreamingResponseContext"/> class.
+ /// </summary>
+ /// <param name="response">The HTTP response message.</param>
+ /// <param name="contentStream">The raw content stream from the response.</param>
+ /// <param name="deflateStream">An optional deflate decompression stream wrapping the content stream.</param>
+ public StreamingResponseContext(HttpResponseMessage response, Stream contentStream,
+ DeflateStream? deflateStream = null)
+ {
+ _response = response;
+ _contentStream = contentStream;
+ _deflateStream = deflateStream;
+ }
+
+ /// <summary>
+ /// Disposes the decompression stream (if any), the content stream, and the HTTP response.
+ /// </summary>
+ public void Dispose()
+ {
+ _deflateStream?.Dispose();
+ _contentStream.Dispose();
+ _response.Dispose();
+ }
+ }
+}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Gremlin.Net.csproj b/gremlin-dotnet/src/Gremlin.Net/Gremlin.Net.csproj
index c1c53ae69d..2d85e673b3 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Gremlin.Net.csproj
+++ b/gremlin-dotnet/src/Gremlin.Net/Gremlin.Net.csproj
@@ -18,10 +18,10 @@ limitations under the License.
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup Label="Build">
- <TargetFrameworks>netstandard2.0;net6.0</TargetFrameworks>
+ <TargetFrameworks>net8.0</TargetFrameworks>
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
<GenerateDocumentationFile>true</GenerateDocumentationFile>
- <LangVersion>9</LangVersion>
+ <LangVersion>12</LangVersion>
<Nullable>enable</Nullable>
</PropertyGroup>
@@ -77,15 +77,6 @@ NOTE that versions suffixed with "-rc" are considered release candidates (i.e. p
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="8.0.0" PrivateAssets="All" />
<PackageReference Include="Polly" Version="8.5.1" />
</ItemGroup>
-
- <ItemGroup Condition="'$(TargetFramework)' == 'netstandard2.0'">
- <PackageReference Include="Microsoft.CSharp" Version="4.7.0" />
- <PackageReference Include="Nullable" Version="1.3.1">
- <PrivateAssets>all</PrivateAssets>
- <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
- </PackageReference>
- <PackageReference Include="System.Text.Json" Version="8.0.5" />
- </ItemGroup>
<ItemGroup>
<None Include="../../LICENSE" Pack="true" PackagePath="" />
diff --git a/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/DefaultTraversal.cs b/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/DefaultTraversal.cs
index 27d8b4d5ef..5bcd7ff9a6 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/DefaultTraversal.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/DefaultTraversal.cs
@@ -36,13 +36,15 @@ namespace Gremlin.Net.Process.Traversal
/// </summary>
public abstract class DefaultTraversal<TStart, TEnd> : ITraversal<TStart, TEnd>
{
- private IEnumerator<Traverser>? _traverserEnumerator;
+ private IAsyncEnumerator<Traverser>? _traverserEnumerator;
+ private bool _nextAvailable;
+ private bool _fetchedNext;
/// <summary>
/// Gets the <see cref="Traversal.GremlinLang" /> representation of this traversal.
/// </summary>
public abstract GremlinLang GremlinLang { get; }
-
+
/// <summary>
/// Determines if this traversal was spawned anonymously or not.
/// </summary>
@@ -51,73 +53,162 @@ namespace Gremlin.Net.Process.Traversal
/// <summary>
/// Gets or sets the <see cref="Traverser" />'s of this traversal that hold the results of the traversal.
/// </summary>
- public IEnumerable<Traverser>? Traversers { get; set; }
-
- ITraversal ITraversal.Iterate()
- {
- return Iterate();
- }
+ public IAsyncEnumerable<Traverser>? Traversers { get; set; }
/// <summary>
/// Gets or sets the <see cref="ITraversalStrategy" /> strategies of this traversal.
/// </summary>
protected ICollection<ITraversalStrategy> TraversalStrategies { get; set; } = new List<ITraversalStrategy>();
- private IEnumerator<Traverser> TraverserEnumerator
+ private IAsyncEnumerator<Traverser> TraverserEnumerator
=> _traverserEnumerator ??= GetTraverserEnumerator();
- private bool _nextAvailable;
- private bool _fetchedNext;
+ /// <inheritdoc />
+ public TEnd? Current => GetCurrent<TEnd>();
/// <inheritdoc />
- public void Dispose()
- {
- }
+ public object? CurrentObject => GetValue();
/// <inheritdoc />
- public bool MoveNext()
+ public async ValueTask<bool> MoveNextAsync()
{
- var more = MoveNextInternal();
+ var more = await MoveNextInternalAsync().ConfigureAwait(false);
_fetchedNext = false;
return more;
}
-
- private bool MoveNextInternal()
+
+ private async ValueTask<bool> MoveNextInternalAsync()
{
if (_fetchedNext) return _nextAvailable;
if (!_nextAvailable || TraverserEnumerator.Current?.Bulk == 0)
{
- _nextAvailable = TraverserEnumerator.MoveNext();
+ _nextAvailable = await TraverserEnumerator.MoveNextAsync()
+ .ConfigureAwait(false);
}
if (!_nextAvailable) return false;
-
+
var currentTraverser = TraverserEnumerator.Current;
if (currentTraverser?.Bulk >= 1)
{
currentTraverser.Bulk--;
}
-
return true;
}
+ /// <inheritdoc />
+ public bool HasNext()
+ {
+ var more = MoveNextInternalAsync().AsTask().WaitUnwrap();
+ _fetchedNext = true;
+ return more;
+ }
+
/// <summary>
- /// Reset is not supported.
+ /// Gets the next result from the traversal.
/// </summary>
- /// <exception cref="NotSupportedException">Thrown always as this operation is not supported.</exception>
- public void Reset()
+ /// <returns>The result.</returns>
+ public TEnd? Next()
{
- throw new NotSupportedException();
+ MoveNextAsync().AsTask().WaitUnwrap();
+ return Current;
}
- /// <inheritdoc />
- public TEnd? Current => GetCurrent<TEnd>();
+ /// <summary>
+ /// Gets the next n-number of results from the traversal.
+ /// </summary>
+ /// <param name="amount">The number of results to get.</param>
+ /// <returns>The n-results.</returns>
+ public IEnumerable<TEnd?> Next(int amount)
+ {
+ for (var i = 0; i < amount; i++)
+ yield return Next();
+ }
- object? IEnumerator.Current => GetCurrent();
+ /// <summary>
+ /// Puts all the results into a <see cref="IList{T}" />.
+ /// </summary>
+ /// <returns>The results in a list.</returns>
+ public IList<TEnd?> ToList()
+ {
+ var objs = new List<TEnd?>();
+ while (MoveNextAsync().AsTask().WaitUnwrap())
+ objs.Add(Current);
+ return objs;
+ }
+
+ /// <summary>
+ /// Puts all the results into a <see cref="HashSet{T}" />.
+ /// </summary>
+ /// <returns>The results in a set.</returns>
+ public ISet<TEnd?> ToSet()
+ {
+ var objs = new HashSet<TEnd?>();
+ while (MoveNextAsync().AsTask().WaitUnwrap())
+ objs.Add(Current);
+ return objs;
+ }
+
+ /// <summary>
+ /// Iterates all <see cref="Traverser" /> instances in the traversal.
+ /// </summary>
+ /// <returns>The fully drained traversal.</returns>
+ public ITraversal<TStart, TEnd> Iterate()
+ {
+ GremlinLang.AddStep("discard");
+ IterateAsync().WaitUnwrap();
+ return this;
+ }
+
+ /// <summary>
+ /// Iterates all <see cref="Traverser" /> instances in the traversal asynchronously.
+ /// </summary>
+ /// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
+ /// <returns>The fully drained traversal.</returns>
+ public async Task<ITraversal<TStart, TEnd>> IterateAsync(
+ CancellationToken cancellationToken = default)
+ {
+ while (await MoveNextAsync().ConfigureAwait(false)) { }
+ return this;
+ }
+
+ ITraversal ITraversal.Iterate() => Iterate();
+
+ /// <summary>
+ /// Gets the next <see cref="Traverser" />.
+ /// </summary>
+ /// <returns>The next <see cref="Traverser" />.</returns>
+ public Traverser NextTraverser()
+ {
+ TraverserEnumerator.MoveNextAsync().AsTask().WaitUnwrap();
+ return TraverserEnumerator.Current;
+ }
+
+ /// <summary>
+ /// Starts a promise to execute a function on the current traversal that will be completed in the future.
+ /// </summary>
+ /// <typeparam name="TReturn">The return type of the <paramref name="callback" />.</typeparam>
+ /// <param name="callback">The function to execute on the current traversal.</param>
+ /// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
+ /// <returns>The result of the executed <paramref name="callback" />.</returns>
+ public async Task<TReturn> Promise<TReturn>(Func<ITraversal<TStart, TEnd>, TReturn> callback,
+ CancellationToken cancellationToken = default)
+ {
+ await ApplyStrategiesAsync(cancellationToken).ConfigureAwait(false);
+ return callback(this);
+ }
+
+ /// <summary>
+ /// Disposes the async enumerator used by this traversal.
+ /// </summary>
+ public ValueTask DisposeAsync()
+ {
+ return _traverserEnumerator?.DisposeAsync() ?? ValueTask.CompletedTask;
+ }
private TReturn? GetCurrent<TReturn>()
{
- var value = GetCurrent();
+ var value = GetValue();
var returnType = typeof(TReturn);
if (value == null || value.GetType() == returnType)
{
@@ -127,14 +218,13 @@ namespace Gremlin.Net.Process.Traversal
return (TReturn) GetValue(returnType, value);
}
- private object? GetCurrent()
+ private object? GetValue()
{
- // Use dynamic to object to prevent runtime dynamic conversion evaluation
return TraverserEnumerator.Current?.Object;
}
/// <summary>
- /// Gets the value, converting to the expected type when necessary and supported.
+ /// Converts a value to the expected type when necessary and supported.
/// </summary>
[return: NotNullIfNotNull("value")]
private static object? GetValue(Type type, object? value)
@@ -170,16 +260,16 @@ namespace Gremlin.Net.Process.Traversal
return value;
}
- private IEnumerator<Traverser> GetTraverserEnumerator()
+ private IAsyncEnumerator<Traverser> GetTraverserEnumerator()
{
if (Traversers == null)
ApplyStrategies();
if (Traversers == null)
{
throw new InvalidOperationException(
- $"Cannot enumerate the traversal as there are no {nameof(Traversers)}. Maybe a strategy needs to be added.");
+ $"Cannot enumerate the traversal as there are no {nameof(Traversers)}.");
}
- return Traversers.GetEnumerator();
+ return Traversers.GetAsyncEnumerator();
}
private void ApplyStrategies()
@@ -193,95 +283,5 @@ namespace Gremlin.Net.Process.Traversal
foreach (var strategy in TraversalStrategies)
await strategy.ApplyAsync(this, cancellationToken).ConfigureAwait(false);
}
-
- /// <inheritdoc />
- public bool HasNext()
- {
- var more = MoveNextInternal();
- _fetchedNext = true;
- return more;
- }
-
- /// <summary>
- /// Gets the next result from the traversal.
- /// </summary>
- /// <returns>The result.</returns>
- public TEnd? Next()
- {
- MoveNext();
- return Current;
- }
-
- /// <summary>
- /// Gets the next n-number of results from the traversal.
- /// </summary>
- /// <param name="amount">The number of results to get.</param>
- /// <returns>The n-results.</returns>
- public IEnumerable<TEnd?> Next(int amount)
- {
- for (var i = 0; i < amount; i++)
- yield return Next();
- }
-
- /// <summary>
- /// Iterates all <see cref="Traverser" /> instances in the traversal.
- /// </summary>
- /// <returns>The fully drained traversal.</returns>
- public ITraversal<TStart, TEnd> Iterate()
- {
- GremlinLang.AddStep("discard");
- while (MoveNext())
- {
- }
- return this;
- }
-
- /// <summary>
- /// Gets the next <see cref="Traverser" />.
- /// </summary>
- /// <returns>The next <see cref="Traverser" />.</returns>
- public Traverser NextTraverser()
- {
- TraverserEnumerator.MoveNext();
- return TraverserEnumerator.Current;
- }
-
- /// <summary>
- /// Puts all the results into a <see cref="List{T}" />.
- /// </summary>
- /// <returns>The results in a list.</returns>
- public IList<TEnd?> ToList()
- {
- var objs = new List<TEnd?>();
- while (MoveNext())
- objs.Add(Current);
- return objs;
- }
-
- /// <summary>
- /// Puts all the results into a <see cref="HashSet{T}" />.
- /// </summary>
- /// <returns>The results in a set.</returns>
- public ISet<TEnd?> ToSet()
- {
- var objs = new HashSet<TEnd?>();
- while (MoveNext())
- objs.Add(Current);
- return objs;
- }
-
- /// <summary>
- /// Starts a promise to execute a function on the current traversal that will be completed in the future.
- /// </summary>
- /// <typeparam name="TReturn">The return type of the <paramref name="callback" />.</typeparam>
- /// <param name="callback">The function to execute on the current traversal.</param>
- /// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
- /// <returns>The result of the executed <paramref name="callback" />.</returns>
- public async Task<TReturn> Promise<TReturn>(Func<ITraversal<TStart, TEnd>, TReturn> callback,
- CancellationToken cancellationToken = default)
- {
- await ApplyStrategiesAsync(cancellationToken).ConfigureAwait(false);
- return callback(this);
- }
}
-}
\ No newline at end of file
+}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/ITraversal.cs b/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/ITraversal.cs
index cbddf30969..b4999336cd 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/ITraversal.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Process/Traversal/ITraversal.cs
@@ -22,7 +22,6 @@
#endregion
using System;
-using System.Collections;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
@@ -30,25 +29,36 @@ using System.Threading.Tasks;
namespace Gremlin.Net.Process.Traversal
{
/// <summary>
- /// Represents the basic information for a walk over a graph.
+ /// Represents the basic information for a walk over a graph.
/// </summary>
/// <seealso cref="ITraversal{SType, EType}"/>
- public interface ITraversal: IEnumerator
+ public interface ITraversal : IAsyncDisposable
{
/// <summary>
/// Gets the <see cref="Traversal.GremlinLang" /> representation of this traversal.
/// </summary>
GremlinLang GremlinLang { get; }
-
+
/// <summary>
/// Determines if this traversal was spawned anonymously or not.
/// </summary>
- bool IsAnonymous { get; }
+ bool IsAnonymous { get; }
/// <summary>
/// Gets or sets the <see cref="Traverser" />'s of this traversal that hold the results of the traversal.
/// </summary>
- IEnumerable<Traverser>? Traversers { get; set; }
+ IAsyncEnumerable<Traverser>? Traversers { get; set; }
+
+ /// <summary>
+ /// Gets the current result as an untyped object, without type conversion.
+ /// This is useful when the caller does not know the specific TEnd type.
+ /// </summary>
+ object? CurrentObject { get; }
+
+ /// <summary>
+ /// Advances the traversal to the next result asynchronously (used internally).
+ /// </summary>
+ ValueTask<bool> MoveNextAsync();
/// <summary>
/// Iterates all <see cref="Traverser" /> instances in the traversal.
@@ -60,8 +70,19 @@ namespace Gremlin.Net.Process.Traversal
/// <summary>
/// A traversal represents a directed walk over a graph.
/// </summary>
- public interface ITraversal<TStart, TEnd> : ITraversal, IEnumerator<TEnd?>
+ public interface ITraversal<TStart, TEnd> : ITraversal
{
+ /// <summary>
+ /// Gets the current result of the traversal.
+ /// </summary>
+ TEnd? Current { get; }
+
+ /// <summary>
+ /// Advances the traversal to the next result asynchronously (used internally).
+ /// </summary>
+ /// <returns>True if there is a next result, false otherwise.</returns>
+ new ValueTask<bool> MoveNextAsync();
+
/// <summary>
/// Gets the next result from the traversal.
/// </summary>
@@ -100,7 +121,7 @@ namespace Gremlin.Net.Process.Traversal
IList<TEnd?> ToList();
/// <summary>
- /// Puts all the results into a <see cref="ISet{T}" />.
+ /// Puts all the results into a <see cref="HashSet{T}" />.
/// </summary>
/// <returns>The results in a set.</returns>
ISet<TEnd?> ToSet();
@@ -115,4 +136,4 @@ namespace Gremlin.Net.Process.Traversal
Task<TReturn> Promise<TReturn>(Func<ITraversal<TStart, TEnd>, TReturn> callback,
CancellationToken cancellationToken = default);
}
-}
\ No newline at end of file
+}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Process/Utils.cs b/gremlin-dotnet/src/Gremlin.Net/Process/Utils.cs
index ce0c2b6abf..9fb2077c02 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Process/Utils.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Process/Utils.cs
@@ -60,6 +60,27 @@ namespace Gremlin.Net.Process
throw;
}
}
+
+ /// <summary>
+ /// Waits the completion of the provided Task and returns the result.
+ /// When an AggregateException is thrown, the inner exception is thrown.
+ /// </summary>
+ public static T WaitUnwrap<T>(this Task<T> task)
+ {
+ try
+ {
+ task.Wait();
+ return task.Result;
+ }
+ catch (AggregateException ex)
+ {
+ if (ex.InnerException != null)
+ {
+ ExceptionDispatchInfo.Capture(ex.InnerException).Throw();
+ }
+ throw;
+ }
+ }
/// <summary>
/// Designed for Tasks that were started but the result should not be awaited upon (fire and forget).
diff --git a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/GraphBinary4MessageSerializer.cs b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/GraphBinary4MessageSerializer.cs
index 214f32be6e..706ab47dd0 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/GraphBinary4MessageSerializer.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/GraphBinary4MessageSerializer.cs
@@ -39,7 +39,7 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
private readonly GraphBinaryReader _reader;
private readonly GraphBinaryWriter _writer;
private readonly RequestMessageSerializer _requestSerializer = new RequestMessageSerializer();
- private readonly ResponseMessageSerializer _responseSerializer = new ResponseMessageSerializer();
+ private readonly ResponseSerializer _responseSerializer = new ResponseSerializer();
/// <inheritdoc />
public string MimeType => SerializationTokens.GraphBinary4MimeType;
@@ -66,12 +66,10 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
}
/// <inheritdoc />
- public async Task<ResponseMessage<List<object>>> DeserializeMessageAsync(byte[] message,
+ public IAsyncEnumerable<object> DeserializeMessageAsync(Stream stream,
CancellationToken cancellationToken = default)
{
- using var stream = new MemoryStream(message);
- return await _responseSerializer.ReadValueAsync(stream, _reader, cancellationToken)
- .ConfigureAwait(false);
+ return _responseSerializer.ReadStreamingAsync(stream, _reader, cancellationToken);
}
}
}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/ResponseMessageSerializer.cs b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/ResponseSerializer.cs
similarity index 51%
rename from gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/ResponseMessageSerializer.cs
rename to gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/ResponseSerializer.cs
index 78b05bd6d4..295705c8e3 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/ResponseMessageSerializer.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/ResponseSerializer.cs
@@ -23,46 +23,57 @@
using System.Collections.Generic;
using System.IO;
+using System.Runtime.CompilerServices;
using System.Threading;
-using System.Threading.Tasks;
using Gremlin.Net.Driver.Exceptions;
-using Gremlin.Net.Driver.Messages;
using Gremlin.Net.Process.Traversal;
namespace Gremlin.Net.Structure.IO.GraphBinary4
{
/// <summary>
- /// Deserializes a <see cref="ResponseMessage{T}"/> from the GraphBinary 4.0 streaming format.
+ /// Deserializes a streaming GraphBinary 4.0 response, yielding each result object
+ /// as it is deserialized from the stream. After the Marker, reads the status
+ /// footer and throws <see cref="ResponseException"/> if the status code is not 200.
/// </summary>
- public class ResponseMessageSerializer
+ public class ResponseSerializer
{
/// <summary>
- /// Reads a response message from the stream in the 4.0 format:
- /// version byte + bulked flag + result data + marker + status footer.
+ /// Reads a streaming GraphBinary 4.0 response, yielding each result object
+ /// as it is deserialized from the stream. After the Marker, reads the status
+ /// footer and throws <see cref="ResponseException"/> if the status code is not 200.
/// </summary>
- /// <param name="stream">The GraphBinary data to parse.</param>
+ /// <param name="stream">The stream to read the GraphBinary response from.</param>
/// <param name="reader">A <see cref="GraphBinaryReader"/> that can be used to read nested values.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
- /// <returns>The read response message.</returns>
- public async Task<ResponseMessage<List<object>>> ReadValueAsync(MemoryStream stream,
- GraphBinaryReader reader, CancellationToken cancellationToken = default)
+ /// <returns>An async enumerable of deserialized result objects.</returns>
+ public async IAsyncEnumerable<object> ReadStreamingAsync(Stream stream,
+ GraphBinaryReader reader,
+ [EnumeratorCancellation] CancellationToken cancellationToken = default)
{
+ // Wrap in BufferedStream for efficient reads (8192 bytes, like Go's bufio).
+ // Intentionally not disposed — disposing BufferedStream would also close the
+ // underlying stream, which is owned by StreamingResponseContext.
+ var buffered = new BufferedStream(stream, 8192);
+
// 1. Version byte — validate MSB is set
- var version = await stream.ReadByteAsync(cancellationToken).ConfigureAwait(false) & 0xFF;
+ var version = await buffered.ReadByteAsync(cancellationToken).ConfigureAwait(false) & 0xFF;
if (version >> 7 != 1)
{
- throw new IOException("The most significant bit should be set according to the format");
+ throw new IOException(
+ "The most significant bit should be set according to the format");
}
// 2. Bulked flag
- var bulkedByte = await stream.ReadByteAsync(cancellationToken).ConfigureAwait(false);
+ var bulkedByte = await buffered.ReadByteAsync(cancellationToken).ConfigureAwait(false);
var bulked = bulkedByte == 0x01;
- // 3. Read result data until Marker
- var results = new List<object>();
+ // 3. Read result data until Marker, yielding each result
while (true)
{
- var value = await reader.ReadAsync(stream, cancellationToken).ConfigureAwait(false);
+ cancellationToken.ThrowIfCancellationRequested();
+
+ var value = await reader.ReadAsync(buffered, cancellationToken)
+ .ConfigureAwait(false);
if (value is Marker)
{
@@ -71,36 +82,28 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
if (bulked)
{
- // Read the fully-qualified Long bulk count following the value
- var bulkCount = (long)(await reader.ReadAsync(stream, cancellationToken)
+ var bulkCount = (long)(await reader.ReadAsync(buffered, cancellationToken)
.ConfigureAwait(false))!;
- results.Add(new Traverser(value, bulkCount));
+ yield return new Traverser(value, bulkCount);
}
else
{
- results.Add(value!);
+ yield return value!;
}
}
- // 4. Status footer: status code (Int value)
+ // 4. Status footer
var statusCode = (int)await reader.ReadNonNullableValueAsync<int>(
- stream, cancellationToken).ConfigureAwait(false);
-
- // 5. Nullable status message (value_flag + String value)
+ buffered, cancellationToken).ConfigureAwait(false);
var statusMessage = (string?)await reader.ReadNullableValueAsync<string>(
- stream, cancellationToken).ConfigureAwait(false);
-
- // 6. Nullable exception (value_flag + String value)
+ buffered, cancellationToken).ConfigureAwait(false);
var exception = (string?)await reader.ReadNullableValueAsync<string>(
- stream, cancellationToken).ConfigureAwait(false);
+ buffered, cancellationToken).ConfigureAwait(false);
if (statusCode != 200)
{
throw new ResponseException(statusCode, statusMessage, exception);
}
-
- return new ResponseMessage<List<object>>(bulked, results, statusCode,
- statusMessage, exception);
}
}
}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/StreamExtensions.cs b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/StreamExtensions.cs
index e85767289f..8094146409 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/StreamExtensions.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/StreamExtensions.cs
@@ -22,6 +22,7 @@
#endregion
using System;
+using System.Buffers.Binary;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
@@ -52,10 +53,16 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
/// <param name="stream">The <see cref="Stream"/> to read from.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
/// <returns>The read <see cref="byte"/>.</returns>
- public static async Task<byte> ReadByteAsync(this Stream stream, CancellationToken cancellationToken = default)
+ public static async ValueTask<byte> ReadByteAsync(this Stream stream,
+ CancellationToken cancellationToken = default)
{
var readBuffer = new byte[1];
- await stream.ReadAsync(readBuffer, 0, 1, cancellationToken).ConfigureAwait(false);
+ var bytesRead = await stream.ReadAsync(readBuffer.AsMemory(0, 1), cancellationToken)
+ .ConfigureAwait(false);
+ if (bytesRead == 0)
+ {
+ throw new IOException("Unexpected end of stream");
+ }
return readBuffer[0];
}
@@ -77,7 +84,8 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
/// <param name="stream">The <see cref="Stream"/> to read from.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
/// <returns>The read <see cref="sbyte"/>.</returns>
- public static async Task<sbyte> ReadSByteAsync(this Stream stream, CancellationToken cancellationToken = default)
+ public static async ValueTask<sbyte> ReadSByteAsync(this Stream stream,
+ CancellationToken cancellationToken = default)
{
return (sbyte)await stream.ReadByteAsync(cancellationToken).ConfigureAwait(false);
}
@@ -91,8 +99,9 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
public static async Task WriteIntAsync(this Stream stream, int value,
CancellationToken cancellationToken = default)
{
- var bytes = BitConverter.GetBytes(value);
- await stream.WriteAsync(new[] { bytes[3], bytes[2], bytes[1], bytes[0] }, 0, 4, cancellationToken).ConfigureAwait(false);
+ var bytes = new byte[4];
+ BinaryPrimitives.WriteInt32BigEndian(bytes, value);
+ await stream.WriteAsync(bytes, 0, 4, cancellationToken).ConfigureAwait(false);
}
/// <summary>
@@ -101,11 +110,12 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
/// <param name="stream">The <see cref="Stream"/> to read from.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
/// <returns>The read <see cref="int"/>.</returns>
- public static async Task<int> ReadIntAsync(this Stream stream, CancellationToken cancellationToken = default)
+ public static async ValueTask<int> ReadIntAsync(this Stream stream,
+ CancellationToken cancellationToken = default)
{
var bytes = new byte[4];
- await stream.ReadAsync(bytes, 0, 4, cancellationToken).ConfigureAwait(false);
- return BitConverter.ToInt32(new []{bytes[3], bytes[2], bytes[1], bytes[0]}, 0);
+ await stream.ReadExactlyAsync(bytes, 0, 4, cancellationToken).ConfigureAwait(false);
+ return BinaryPrimitives.ReadInt32BigEndian(bytes);
}
/// <summary>
@@ -117,10 +127,9 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
public static async Task WriteLongAsync(this Stream stream, long value,
CancellationToken cancellationToken = default)
{
- var bytes = BitConverter.GetBytes(value);
- await stream
- .WriteAsync(new[] { bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0] }, 0,
- 8, cancellationToken).ConfigureAwait(false);
+ var bytes = new byte[8];
+ BinaryPrimitives.WriteInt64BigEndian(bytes, value);
+ await stream.WriteAsync(bytes, 0, 8, cancellationToken).ConfigureAwait(false);
}
/// <summary>
@@ -129,12 +138,12 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
/// <param name="stream">The <see cref="Stream"/> to read from.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
/// <returns>The read <see cref="long"/>.</returns>
- public static async Task<long> ReadLongAsync(this Stream stream, CancellationToken cancellationToken = default)
+ public static async ValueTask<long> ReadLongAsync(this Stream stream,
+ CancellationToken cancellationToken = default)
{
var bytes = new byte[8];
- await stream.ReadAsync(bytes, 0, 8, cancellationToken).ConfigureAwait(false);
- return BitConverter.ToInt64(
- new[] {bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]}, 0);
+ await stream.ReadExactlyAsync(bytes, 0, 8, cancellationToken).ConfigureAwait(false);
+ return BinaryPrimitives.ReadInt64BigEndian(bytes);
}
/// <summary>
@@ -146,9 +155,9 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
public static async Task WriteFloatAsync(this Stream stream, float value,
CancellationToken cancellationToken = default)
{
- var bytes = BitConverter.GetBytes(value);
- await stream.WriteAsync(new[] { bytes[3], bytes[2], bytes[1], bytes[0] }, 0, 4, cancellationToken)
- .ConfigureAwait(false);
+ var bytes = new byte[4];
+ BinaryPrimitives.WriteSingleBigEndian(bytes, value);
+ await stream.WriteAsync(bytes, 0, 4, cancellationToken).ConfigureAwait(false);
}
/// <summary>
@@ -157,12 +166,12 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
/// <param name="stream">The <see cref="Stream"/> to read from.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
/// <returns>The read <see cref="float"/>.</returns>
- public static async Task<float> ReadFloatAsync(this Stream stream,
+ public static async ValueTask<float> ReadFloatAsync(this Stream stream,
CancellationToken cancellationToken = default)
{
var bytes = new byte[4];
- await stream.ReadAsync(bytes, 0, 4, cancellationToken).ConfigureAwait(false);
- return BitConverter.ToSingle(new []{bytes[3], bytes[2], bytes[1], bytes[0]}, 0);
+ await stream.ReadExactlyAsync(bytes, 0, 4, cancellationToken).ConfigureAwait(false);
+ return BinaryPrimitives.ReadSingleBigEndian(bytes);
}
/// <summary>
@@ -174,10 +183,9 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
public static async Task WriteDoubleAsync(this Stream stream, double value,
CancellationToken cancellationToken = default)
{
- var bytes = BitConverter.GetBytes(value);
- await stream
- .WriteAsync(new[] {bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]}, 0,
- 8, cancellationToken).ConfigureAwait(false);
+ var bytes = new byte[8];
+ BinaryPrimitives.WriteDoubleBigEndian(bytes, value);
+ await stream.WriteAsync(bytes, 0, 8, cancellationToken).ConfigureAwait(false);
}
/// <summary>
@@ -186,13 +194,12 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
/// <param name="stream">The <see cref="Stream"/> to read from.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
/// <returns>The read <see cref="double"/>.</returns>
- public static async Task<double> ReadDoubleAsync(this Stream stream,
+ public static async ValueTask<double> ReadDoubleAsync(this Stream stream,
CancellationToken cancellationToken = default)
{
var bytes = new byte[8];
- await stream.ReadAsync(bytes, 0, 8, cancellationToken).ConfigureAwait(false);
- return BitConverter.ToDouble(
- new[] {bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]}, 0);
+ await stream.ReadExactlyAsync(bytes, 0, 8, cancellationToken).ConfigureAwait(false);
+ return BinaryPrimitives.ReadDoubleBigEndian(bytes);
}
/// <summary>
@@ -204,8 +211,9 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
public static async Task WriteShortAsync(this Stream stream, short value,
CancellationToken cancellationToken = default)
{
- var bytes = BitConverter.GetBytes(value);
- await stream.WriteAsync(new[] { bytes[1], bytes[0] }, 0, 2, cancellationToken).ConfigureAwait(false);
+ var bytes = new byte[2];
+ BinaryPrimitives.WriteInt16BigEndian(bytes, value);
+ await stream.WriteAsync(bytes, 0, 2, cancellationToken).ConfigureAwait(false);
}
/// <summary>
@@ -214,12 +222,12 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
/// <param name="stream">The <see cref="Stream"/> to read from.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
/// <returns>The read <see cref="short"/>.</returns>
- public static async Task<short> ReadShortAsync(this Stream stream,
+ public static async ValueTask<short> ReadShortAsync(this Stream stream,
CancellationToken cancellationToken = default)
{
var bytes = new byte[2];
- await stream.ReadAsync(bytes, 0, 2, cancellationToken).ConfigureAwait(false);
- return BitConverter.ToInt16(new []{bytes[1], bytes[0]}, 0);
+ await stream.ReadExactlyAsync(bytes, 0, 2, cancellationToken).ConfigureAwait(false);
+ return BinaryPrimitives.ReadInt16BigEndian(bytes);
}
/// <summary>
@@ -240,7 +248,8 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
/// <param name="stream">The <see cref="Stream"/> to read from.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
/// <returns>The read <see cref="bool"/>.</returns>
- public static async Task<bool> ReadBoolAsync(this Stream stream, CancellationToken cancellationToken = default)
+ public static async ValueTask<bool> ReadBoolAsync(this Stream stream,
+ CancellationToken cancellationToken = default)
{
var b = await stream.ReadByteAsync(cancellationToken).ConfigureAwait(false);
return b switch
@@ -270,12 +279,12 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4
/// <param name="count">The number of bytes to read.</param>
/// <param name="cancellationToken">The token to cancel the operation. The default value is None.</param>
/// <returns>The read <see cref="T:byte[]"/>.</returns>
- public static async Task<byte[]> ReadAsync(this Stream stream, int count,
+ public static async ValueTask<byte[]> ReadAsync(this Stream stream, int count,
CancellationToken cancellationToken = default)
{
var buffer = new byte[count];
- await stream.ReadAsync(buffer, 0, count, cancellationToken).ConfigureAwait(false);
+ await stream.ReadExactlyAsync(buffer, 0, count, cancellationToken).ConfigureAwait(false);
return buffer;
}
}
-}
\ No newline at end of file
+}
diff --git a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/Types/SingleTypeSerializer.cs b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/Types/SingleTypeSerializer.cs
index 6164581be5..4f7c2238b7 100644
--- a/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/Types/SingleTypeSerializer.cs
+++ b/gremlin-dotnet/src/Gremlin.Net/Structure/IO/GraphBinary4/Types/SingleTypeSerializer.cs
@@ -106,10 +106,10 @@ namespace Gremlin.Net.Structure.IO.GraphBinary4.Types
public class SingleTypeSerializer<T> : SimpleTypeSerializer<T>
{
private readonly Func<T, Stream, CancellationToken, Task> _writeFunc;
- private readonly Func<Stream, CancellationToken, Task<T>> _readFunc;
+ private readonly Func<Stream, CancellationToken, ValueTask<T>> _readFunc;
internal SingleTypeSerializer(DataType dataType, Func<T, Stream, CancellationToken, Task> writeFunc,
- Func<Stream, CancellationToken, Task<T>> readFunc)
+ Func<Stream, CancellationToken, ValueTask<T>> readFunc)
: base(dataType)
{
_writeFunc = writeFunc;
diff --git a/gremlin-dotnet/test/Gremlin.Net.Benchmarks/MessageSerializerBenchmarks.cs b/gremlin-dotnet/test/Gremlin.Net.Benchmarks/MessageSerializerBenchmarks.cs
index 25c442722a..dfbf0082db 100644
--- a/gremlin-dotnet/test/Gremlin.Net.Benchmarks/MessageSerializerBenchmarks.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.Benchmarks/MessageSerializerBenchmarks.cs
@@ -23,6 +23,7 @@
using System;
using System.Collections.Generic;
+using System.IO;
using System.Threading.Tasks;
using BenchmarkDotNet.Attributes;
using Gremlin.Net.Driver;
@@ -81,13 +82,29 @@ namespace Gremlin.Net.Benchmarks
.ConfigureAwait(false);
[Benchmark]
- public async Task<ResponseMessage<List<object>>> TestReadSmallBinary() =>
- await BinaryMessageSerializer.DeserializeMessageAsync(TestMessages.SmallBinaryResponseMessageBytes)
- .ConfigureAwait(false);
+ public async Task<List<object>> TestReadSmallBinary()
+ {
+ using var stream = new MemoryStream(TestMessages.SmallBinaryResponseMessageBytes);
+ var results = new List<object>();
+ await foreach (var item in BinaryMessageSerializer.DeserializeMessageAsync(stream)
+ .ConfigureAwait(false))
+ {
+ results.Add(item);
+ }
+ return results;
+ }
[Benchmark]
- public async Task<ResponseMessage<List<object>>> TestReadBigBinary() =>
- await BinaryMessageSerializer.DeserializeMessageAsync(TestMessages.BigBinaryResponseMessageBytes)
- .ConfigureAwait(false);
+ public async Task<List<object>> TestReadBigBinary()
+ {
+ using var stream = new MemoryStream(TestMessages.BigBinaryResponseMessageBytes);
+ var results = new List<object>();
+ await foreach (var item in BinaryMessageSerializer.DeserializeMessageAsync(stream)
+ .ConfigureAwait(false))
+ {
+ results.Add(item);
+ }
+ return results;
+ }
}
}
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs
index a91dfe8aab..a1fb59efc7 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/GremlinClientTests.cs
@@ -63,8 +63,9 @@ namespace Gremlin.Net.IntegrationTest.Driver
var requestMsg = "g.inject(1,2,3,4,5,6,7,8,9,10)";
var response = await gremlinClient.SubmitAsync<int>(requestMsg);
+ var results = await response.ToListAsync();
- Assert.Equal(10, response.Count);
+ Assert.Equal(10, results.Count);
}
}
@@ -92,7 +93,11 @@ namespace Gremlin.Net.IntegrationTest.Driver
var requestMsg = "invalid";
var exception =
- await Assert.ThrowsAsync<ResponseException>(() => gremlinClient.SubmitAsync(requestMsg));
+ await Assert.ThrowsAsync<ResponseException>(async () =>
+ {
+ var resultSet = await gremlinClient.SubmitAsync<object>(requestMsg);
+ await resultSet.ToListAsync();
+ });
Assert.Equal(typeof(ResponseException), exception.GetType());
Assert.Contains("Failed to interpret Gremlin query", exception.Message);
@@ -110,7 +115,7 @@ namespace Gremlin.Net.IntegrationTest.Driver
{
var response = await gremlinClient.SubmitAsync<int>(requestMessage);
- Assert.Equal(expectedResult, response);
+ Assert.Equal(expectedResult, await response.ToListAsync());
}
}
@@ -125,21 +130,10 @@ namespace Gremlin.Net.IntegrationTest.Driver
var response = await gremlinClient.SubmitAsync<int>(requestMsg);
- Assert.Equal(expectedResult, response);
+ Assert.Equal(expectedResult, await response.ToListAsync());
}
}
- [Fact]
- public async Task ShouldReturnResponseAttributes()
- {
- var gremlinServer = new GremlinServer(TestHost, TestPort);
- using var gremlinClient = new GremlinClient(gremlinServer);
- var requestMsg = _requestMessageProvider.GetDummyMessage();
- var resultSet = await gremlinClient.SubmitAsync<int>(requestMsg);
-
- Assert.NotNull(resultSet.StatusAttributes);
- }
-
[Fact]
public async Task ShouldThrowOnExecutionOfSimpleInvalidScript()
{
@@ -148,7 +142,11 @@ namespace Gremlin.Net.IntegrationTest.Driver
{
var invalidRequestScript = "invalid";
- await Assert.ThrowsAsync<ResponseException>(() => gremlinClient.SubmitAsync(invalidRequestScript));
+ await Assert.ThrowsAsync<ResponseException>(async () =>
+ {
+ var resultSet = await gremlinClient.SubmitAsync<object>(invalidRequestScript);
+ await resultSet.ToListAsync();
+ });
}
}
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/MessagesTests.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/MessagesTests.cs
index 082610cd43..7cad366f68 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/MessagesTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Driver/MessagesTests.cs
@@ -47,7 +47,11 @@ namespace Gremlin.Net.IntegrationTest.Driver
var requestMsg = RequestMessage.Build(ivalidOperationName).Create();
var thrownException =
- await Assert.ThrowsAsync<ResponseException>(() => gremlinClient.SubmitAsync<dynamic>(requestMsg));
+ await Assert.ThrowsAsync<ResponseException>(async () =>
+ {
+ var resultSet = await gremlinClient.SubmitAsync<dynamic>(requestMsg);
+ await resultSet.ToListAsync();
+ });
Assert.Contains("Failed to interpret Gremlin query", thrownException.Message);
Assert.Contains(ivalidOperationName, thrownException.Message);
@@ -67,7 +71,11 @@ namespace Gremlin.Net.IntegrationTest.Driver
.Create();
var thrownException =
- await Assert.ThrowsAsync<ResponseException>(() => gremlinClient.SubmitAsync(requestMsg));
+ await Assert.ThrowsAsync<ResponseException>(async () =>
+ {
+ var resultSet = await gremlinClient.SubmitAsync<object>(requestMsg);
+ await resultSet.ToListAsync();
+ });
Assert.Contains("not an available GremlinScript", thrownException.Message);
Assert.Contains(unknownLanguage, thrownException.Message);
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/CommonSteps.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/CommonSteps.cs
index 4921813393..790e412dcf 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/CommonSteps.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/CommonSteps.cs
@@ -29,6 +29,7 @@ using System.Linq;
using System.Numerics;
using System.Text.Json;
using System.Text.RegularExpressions;
+using System.Threading.Tasks;
using Gherkin.Ast;
using Gremlin.Net.IntegrationTest.Gherkin.Attributes;
using Gremlin.Net.Process.Traversal;
@@ -178,19 +179,20 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
{
throw new InvalidOperationException("Traversal should be set before iterating");
}
- ITraversal t = _traversal;
var list = new List<object>();
try
{
- while (t.MoveNext())
+ // Use the base ITraversal interface's MoveNextAsync() and CurrentObject
+ // to avoid type conversion issues with ITraversal<TStart, TEnd>.Current
+ while (_traversal.MoveNextAsync().AsTask().GetAwaiter().GetResult())
{
- list.Add(t.Current);
+ list.Add(_traversal.CurrentObject!);
}
_result = list.ToArray();
}
catch (Exception ex)
{
- _error = ex;
+ _error = UnwrapException(ex);
}
}
@@ -204,8 +206,8 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
try
{
- _traversal.MoveNext();
- var result = _traversal.Current;
+ _traversal.MoveNextAsync().AsTask().GetAwaiter().GetResult();
+ var result = _traversal.CurrentObject;
switch (result)
{
case null:
@@ -221,7 +223,7 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
}
catch (Exception ex)
{
- _error = ex;
+ _error = UnwrapException(ex);
}
}
@@ -239,15 +241,21 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
{
Assert.NotNull(_error);
+ // Unwrap AggregateException if the real exception is nested inside.
+ // With streaming deserialization, exceptions from the channel/async pipeline
+ // may be wrapped in AggregateException depending on how the async task is
+ // awaited (e.g., Task.Wait() vs GetAwaiter().GetResult()).
+ var errorMessage = UnwrapException(_error).Message;
+
switch (comparison) {
case "containing":
- Assert.Contains(expectedMessage.ToUpper(), _error.Message.ToUpper());
+ Assert.Contains(expectedMessage.ToUpper(), errorMessage.ToUpper());
break;
case "starting":
- Assert.StartsWith(expectedMessage.ToUpper(), _error.Message.ToUpper());
+ Assert.StartsWith(expectedMessage.ToUpper(), errorMessage.ToUpper());
break;
case "ending":
- Assert.EndsWith(expectedMessage.ToUpper(), _error.Message.ToUpper());
+ Assert.EndsWith(expectedMessage.ToUpper(), errorMessage.ToUpper());
break;
default:
throw new NotSupportedException(
@@ -258,6 +266,27 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
_error = null;
}
+ /// <summary>
+ /// Unwraps an exception to find the innermost meaningful exception.
+ /// With streaming deserialization, a ResponseException thrown in the background
+ /// deserialization task may be wrapped in AggregateException when the consuming
+ /// code uses Task.Wait() or similar synchronous blocking patterns.
+ /// </summary>
+ private static Exception UnwrapException(Exception ex)
+ {
+ while (true)
+ {
+ switch (ex)
+ {
+ case AggregateException agg when agg.InnerExceptions.Count == 1:
+ ex = agg.InnerExceptions[0];
+ continue;
+ default:
+ return ex;
+ }
+ }
+ }
+
[Then("the result should be a tree with a structure of")]
public void AssertTreeStructure(string expectedTree)
@@ -394,11 +423,10 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
traversalText = traversalText.Substring(1, traversalText.Length - 2);
}
- var traversal =
- Gremlin.UseTraversal(ScenarioData.CurrentScenario!.Name, _g, _parameters, _sideEffects);
+ var traversal = (ITraversal) Gremlin.UseTraversal(ScenarioData.CurrentScenario!.Name, _g, _parameters, _sideEffects);
var count = 0;
- while (traversal.MoveNext())
+ while (traversal.MoveNextAsync().AsTask().GetAwaiter().GetResult())
{
count++;
}
diff --git a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/Gremlin.cs b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/Gremlin.cs
index bf88cfccfa..448457ee97 100644
--- a/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/Gremlin.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.IntegrationTest/Gherkin/Gremlin.cs
@@ -946,9 +946,9 @@ namespace Gremlin.Net.IntegrationTest.Gherkin
{"g_addV_propertyXsingle_mapX", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV().Property(Cardinality.Single, new Dictionary<object, object> {{ "name", "foo" }, { "age", 42 }}), (g,p) =>g.V().Has("name", "foo")}},
{"g_V_hasXname_fooX_propertyXname_setXbarX_age_43X", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV().Property(Cardinality.Single, "name", "foo").Property("age", 42), (g,p) =>g.V().Has("name", "foo").Property(new Dictionary<object, object> {{ "name", CardinalityValue.Set("bar") }, { "age", 43 }}), (g,p) =>g.V().Has("name", "foo"), (g,p) =>g.V().Has("name", "bar"), (g,p) =>g.V().Has("age", 43), (g,p) =>g.V().Has("age", 42)}},
{"g_V_hasXname_fooX_propertyXset_name_bar_age_singleX43XX", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV().Property(Cardinality.Single, "name", "foo").Property("age", 42), (g,p) =>g.V().Has("name", "foo").Property(Cardinality.Set, new Dictionary<object, object> {{ "name", "bar" }, { "age", CardinalityValue.Single(43) }}), (g,p) =>g.V().Has("name", "foo"), (g,p) =>g.V().Has("name", "bar"), (g,p) =>g.V().Has("age", 43), (g,p) [...]
- {"g_addV_propertyXnullX", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV((string) "person").Property(null), (g,p) =>g.V().HasLabel("person").Values<object>()}},
+ {"g_addV_propertyXnullX", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV((string) "person").Property((IDictionary<object, object>) null), (g,p) =>g.V().HasLabel("person").Values<object>()}},
{"g_addV_propertyXemptyX", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV((string) "person").Property(new Dictionary<object, object> {}), (g,p) =>g.V().HasLabel("person").Values<object>()}},
- {"g_addV_propertyXset_nullX", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV((string) "foo").Property(Cardinality.Set, null), (g,p) =>g.V().HasLabel("foo").Values<object>()}},
+ {"g_addV_propertyXset_nullX", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV((string) "foo").Property(Cardinality.Set, (IDictionary<object, object>) null), (g,p) =>g.V().HasLabel("foo").Values<object>()}},
{"g_addV_propertyXset_emptyX", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV((string) "foo").Property(Cardinality.Set, new Dictionary<object, object> {}), (g,p) =>g.V().HasLabel("person").Values<object>()}},
{"g_addVXpersonX_propertyXname_joshX_propertyXage_nullX", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV((string) "person").Property("name", "josh").Property("age", null), (g,p) =>g.V().Has("person", "age", (object) null)}},
{"g_addVXpersonX_propertyXname_markoX_propertyXfriendWeight_null_acl_nullX", new List<Func<GraphTraversalSource, IDictionary<string, object>, ITraversal>> {(g,p) =>g.AddV((string) "person").Property("name", "marko").Property("friendWeight", null, "acl", null), (g,p) =>g.V().Has("person", "name", "marko").Has("friendWeight", (object) null), (g,p) =>g.V().Has("person", "name", "marko").Properties<object>("friendWeight").Has("acl", (object) null), (g,p) =>g.V().Has("person", [...]
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionTests.cs
index ca499380ba..6a8c2b7856 100644
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ConnectionTests.cs
@@ -282,12 +282,20 @@ namespace Gremlin.Net.UnitTest.Driver
serializer.MimeType.Returns(mimeType);
serializer.SerializeMessageAsync(Arg.Any<RequestMessage>(), Arg.Any<CancellationToken>())
.Returns(Task.FromResult(new byte[] { 0x84 }));
- serializer.DeserializeMessageAsync(Arg.Any<byte[]>(), Arg.Any<CancellationToken>())
- .Returns(Task.FromResult(
- new ResponseMessage<List<object>>(false, results, 200, null, null)));
+ serializer.DeserializeMessageAsync(Arg.Any<Stream>(), Arg.Any<CancellationToken>())
+ .Returns(callInfo => ToAsyncEnumerable(results));
return serializer;
}
+ private static async IAsyncEnumerable<object> ToAsyncEnumerable(List<object> items)
+ {
+ foreach (var item in items)
+ {
+ yield return item;
+ }
+ await Task.CompletedTask;
+ }
+
[Fact]
public async Task ShouldCallInterceptorsInOrder()
{
@@ -864,11 +872,11 @@ namespace Gremlin.Net.UnitTest.Driver
await connection.SubmitAsync<object>(CreateTestRequest());
// Verify the response serializer was called for deserialization
- await responseSerializer.Received(1)
- .DeserializeMessageAsync(Arg.Any<byte[]>(), Arg.Any<CancellationToken>());
+ responseSerializer.Received(1)
+ .DeserializeMessageAsync(Arg.Any<Stream>(), Arg.Any<CancellationToken>());
// Verify the request serializer was NOT called by Connection (interceptor called it directly)
- await requestSerializer.DidNotReceive()
- .DeserializeMessageAsync(Arg.Any<byte[]>(), Arg.Any<CancellationToken>());
+ requestSerializer.DidNotReceive()
+ .DeserializeMessageAsync(Arg.Any<Stream>(), Arg.Any<CancellationToken>());
}
[Fact]
@@ -951,58 +959,60 @@ namespace Gremlin.Net.UnitTest.Driver
}
[Fact]
- public async Task ShouldBoxNonTraverserResultsIntoTraversers()
+ public async Task ShouldReturnStreamingResultSet()
{
var (httpClient, _) = CreateMockHttpClient();
var serializer = CreateMockSerializer(new List<object> { "hello", 42 });
var settings = new ConnectionSettings();
using var connection = new Connection(TestUri, serializer, serializer, settings, httpClient);
- var result = await connection.SubmitAsync<Traverser>(CreateTestRequest());
+ var result = await connection.SubmitAsync<object>(CreateTestRequest());
- var items = result.ToList();
+ var items = await result.ToListAsync();
Assert.Equal(2, items.Count);
- Assert.Equal("hello", items[0].Object);
- Assert.Equal(1, items[0].Bulk);
- Assert.Equal(42, items[1].Object);
- Assert.Equal(1, items[1].Bulk);
+ Assert.Equal("hello", items[0]);
+ Assert.Equal(42, items[1]);
}
[Fact]
- public async Task ShouldNotDoubleBoxTraverserResults()
+ public async Task ShouldReturnEmptyResultSetWhenNoResults()
{
var (httpClient, _) = CreateMockHttpClient();
- var serializer = CreateMockSerializer(new List<object> { new Traverser("existing", 5) });
+ var serializer = CreateMockSerializer(new List<object>());
var settings = new ConnectionSettings();
using var connection = new Connection(TestUri, serializer, serializer, settings, httpClient);
- var result = await connection.SubmitAsync<Traverser>(CreateTestRequest());
+ var result = await connection.SubmitAsync<object>(CreateTestRequest());
- var item = result.Single();
- Assert.Equal("existing", item.Object);
- Assert.Equal(5, item.Bulk);
+ var items = await result.ToListAsync();
+ Assert.Empty(items);
}
[Fact]
- public async Task ShouldBoxMixedTraverserAndNonTraverserResults()
+ public async Task ShouldStreamResponseWithoutFullBuffering()
{
- var (httpClient, _) = CreateMockHttpClient();
- var serializer = CreateMockSerializer(new List<object> { new Traverser("already", 3), "raw" });
+ // Verify that Connection reads the response as a stream (consistent with
+ // HttpCompletionOption.ResponseHeadersRead) rather than buffering the entire
+ // body. We do this by using a SlowStream that tracks whether ReadAsync was
+ // called incrementally (streaming) vs the content being fully buffered upfront.
+ var responseBytes = BuildMinimalResponseBytes();
+ var slowStream = new TrackingStream(new MemoryStream(responseBytes));
+ var handler = new StreamMockHandler(slowStream);
+ var httpClient = new HttpClient(handler);
+ var serializer = CreateMockSerializer();
var settings = new ConnectionSettings();
using var connection = new Connection(TestUri, serializer, serializer, settings, httpClient);
- var result = await connection.SubmitAsync<Traverser>(CreateTestRequest());
+ var result = await connection.SubmitAsync<object>(CreateTestRequest());
- var items = result.ToList();
- Assert.Equal(2, items.Count);
- Assert.Equal("already", items[0].Object);
- Assert.Equal(3, items[0].Bulk);
- Assert.Equal("raw", items[1].Object);
- Assert.Equal(1, items[1].Bulk);
+ Assert.NotNull(result);
+ Assert.True(handler.WasCalled, "SendAsync should have been called");
}
/// <summary>
/// A test HttpMessageHandler that captures the request and returns a canned response.
+ /// The response uses ByteArrayContent but does NOT dispose the content stream
+ /// immediately, allowing streaming reads after SendAsync returns.
/// </summary>
private class MockHandler : HttpMessageHandler
{
@@ -1060,5 +1070,75 @@ namespace Gremlin.Net.UnitTest.Driver
return clone;
}
}
+
+ /// <summary>
+ /// A mock handler that returns a StreamContent wrapping a provided stream,
+ /// used to verify streaming behavior with ResponseHeadersRead.
+ /// </summary>
+ private class StreamMockHandler : HttpMessageHandler
+ {
+ private readonly Stream _responseStream;
+
+ public bool WasCalled { get; private set; }
+
+ public StreamMockHandler(Stream responseStream)
+ {
+ _responseStream = responseStream;
+ }
+
+ protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request,
+ CancellationToken cancellationToken)
+ {
+ WasCalled = true;
+
+ var response = new HttpResponseMessage(HttpStatusCode.OK)
+ {
+ Content = new StreamContent(_responseStream)
+ };
+ response.Content.Headers.ContentType =
+ new MediaTypeHeaderValue(SerializationTokens.GraphBinary4MimeType);
+ return Task.FromResult(response);
+ }
+ }
+
+ /// <summary>
+ /// A stream wrapper that tracks read operations.
+ /// </summary>
+ private class TrackingStream : Stream
+ {
+ private readonly Stream _inner;
+
+ public TrackingStream(Stream inner)
+ {
+ _inner = inner;
+ }
+
+ public override bool CanRead => _inner.CanRead;
+ public override bool CanSeek => _inner.CanSeek;
+ public override bool CanWrite => _inner.CanWrite;
+ public override long Length => _inner.Length;
+ public override long Position
+ {
+ get => _inner.Position;
+ set => _inner.Position = value;
+ }
+
+ public override void Flush() => _inner.Flush();
+ public override int Read(byte[] buffer, int offset, int count) =>
+ _inner.Read(buffer, offset, count);
+ public override long Seek(long offset, SeekOrigin origin) =>
+ _inner.Seek(offset, origin);
+ public override void SetLength(long value) => _inner.SetLength(value);
+ public override void Write(byte[] buffer, int offset, int count) =>
+ _inner.Write(buffer, offset, count);
+
+ public override ValueTask<int> ReadAsync(Memory<byte> buffer,
+ CancellationToken cancellationToken = default) =>
+ _inner.ReadAsync(buffer, cancellationToken);
+
+ public override Task<int> ReadAsync(byte[] buffer, int offset, int count,
+ CancellationToken cancellationToken) =>
+ _inner.ReadAsync(buffer, offset, count, cancellationToken);
+ }
}
}
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/DriverRemoteConnectionTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/DriverRemoteConnectionTests.cs
index 7a67ce668f..1e05943638 100644
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/DriverRemoteConnectionTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/DriverRemoteConnectionTests.cs
@@ -208,8 +208,11 @@ namespace Gremlin.Net.UnitTest.Driver
.Returns(callInfo =>
{
capture(callInfo.Arg<RequestMessage>());
+ var channel = System.Threading.Channels.Channel.CreateUnbounded<object>();
+ channel.Writer.Complete();
+ var cts = new CancellationTokenSource();
var emptyResult = new ResultSet<Traverser>(
- new List<Traverser>(), new Dictionary<string, object>());
+ channel.Reader, cts, Task.CompletedTask);
return Task.FromResult(emptyResult);
});
return client;
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ResultSetTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ResultSetTests.cs
new file mode 100644
index 0000000000..84f19fbe15
--- /dev/null
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Driver/ResultSetTests.cs
@@ -0,0 +1,328 @@
+#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.Channels;
+using System.Threading.Tasks;
+using Gremlin.Net.Driver;
+using Xunit;
+
+namespace Gremlin.Net.UnitTest.Driver
+{
+ public class ResultSetTests
+ {
+ /// <summary>
+ /// Creates a ResultSet backed by a channel. The background task writes
+ /// the provided items to the channel and then completes it.
+ /// </summary>
+ private static ResultSet<T> CreateResultSet<T>(IEnumerable<T> items)
+ {
+ var channel = Channel.CreateUnbounded<object>(
+ new UnboundedChannelOptions { SingleWriter = true });
+ var disposeCts = new CancellationTokenSource();
+
+ var backgroundTask = Task.Run(async () =>
+ {
+ foreach (var item in items)
+ {
+ await channel.Writer.WriteAsync(item!).ConfigureAwait(false);
+ }
+ channel.Writer.Complete();
+ });
+
+ return new ResultSet<T>(channel.Reader, disposeCts, backgroundTask);
+ }
+
+ /// <summary>
+ /// Creates a ResultSet whose background task completes the channel with an error.
+ /// </summary>
+ private static ResultSet<T> CreateFaultedResultSet<T>(Exception exception)
+ {
+ var channel = Channel.CreateUnbounded<object>(
+ new UnboundedChannelOptions { SingleWriter = true });
+ var disposeCts = new CancellationTokenSource();
+
+ var backgroundTask = Task.Run(() =>
+ {
+ channel.Writer.Complete(exception);
+ });
+
+ return new ResultSet<T>(channel.Reader, disposeCts, backgroundTask);
+ }
+
+ /// <summary>
+ /// Creates a ResultSet whose background task blocks until the dispose CTS is cancelled.
+ /// </summary>
+ private static (ResultSet<T> resultSet, TaskCompletionSource<bool> started) CreateBlockingResultSet<T>()
+ {
+ var channel = Channel.CreateUnbounded<object>(
+ new UnboundedChannelOptions { SingleWriter = true });
+ var disposeCts = new CancellationTokenSource();
+ var started = new TaskCompletionSource<bool>();
+
+ var backgroundTask = Task.Run(async () =>
+ {
+ started.SetResult(true);
+ try
+ {
+ // Block until cancelled
+ await Task.Delay(Timeout.Infinite, disposeCts.Token).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException)
+ {
+ // Expected on dispose
+ }
+ finally
+ {
+ channel.Writer.Complete();
+ }
+ });
+
+ return (new ResultSet<T>(channel.Reader, disposeCts, backgroundTask), started);
+ }
+
+ [Fact]
+ public async Task AwaitForeachShouldYieldAllResults()
+ {
+ var expected = new List<int> { 1, 2, 3, 4, 5 };
+ await using var resultSet = CreateResultSet(expected);
+
+ var actual = new List<int>();
+ await foreach (var item in resultSet)
+ {
+ actual.Add(item);
+ }
+
+ Assert.Equal(expected, actual);
+ }
+
+ [Fact]
+ public async Task AwaitForeachShouldYieldStringResults()
+ {
+ var expected = new List<string> { "hello", "world", "gremlin" };
+ await using var resultSet = CreateResultSet(expected);
+
+ var actual = new List<string>();
+ await foreach (var item in resultSet)
+ {
+ actual.Add(item);
+ }
+
+ Assert.Equal(expected, actual);
+ }
+
+ [Fact]
+ public async Task AwaitForeachShouldHandleEmptyResultSet()
+ {
+ await using var resultSet = CreateResultSet(new List<int>());
+
+ var actual = new List<int>();
+ await foreach (var item in resultSet)
+ {
+ actual.Add(item);
+ }
+
+ Assert.Empty(actual);
+ }
+
+ [Fact]
+ public async Task ToListAsyncShouldReturnAllResultsInOrder()
+ {
+ var expected = new List<int> { 10, 20, 30, 40, 50 };
+ await using var resultSet = CreateResultSet(expected);
+
+ var actual = await resultSet.ToListAsync();
+
+ Assert.Equal(expected, actual);
+ }
+
+ [Fact]
+ public async Task ToListAsyncShouldReturnEmptyListForEmptyResultSet()
+ {
+ await using var resultSet = CreateResultSet(new List<string>());
+
+ var actual = await resultSet.ToListAsync();
+
+ Assert.Empty(actual);
+ }
+
+ [Fact]
+ public async Task DisposeAsyncShouldCancelBackgroundTask()
+ {
+ var (resultSet, started) = CreateBlockingResultSet<int>();
+ await started.Task;
+
+ var disposeTask = resultSet.DisposeAsync().AsTask();
+ var completed = await Task.WhenAny(disposeTask, Task.Delay(5000));
+
+ Assert.Same(disposeTask, completed);
+ }
+
+ [Fact]
+ public async Task ErrorInChannelShouldPropagateToConsumerDuringIteration()
+ {
+ var expectedException = new InvalidOperationException("deserialization failed");
+ await using var resultSet = CreateFaultedResultSet<int>(expectedException);
+
+ var ex = await Assert.ThrowsAsync<InvalidOperationException>(async () =>
+ {
+ await foreach (var _ in resultSet)
+ {
+ // Should not yield any items
+ }
+ });
+
+ Assert.Same(expectedException, ex);
+ }
+
+ [Fact]
+ public async Task ErrorInChannelShouldPropagateToToListAsync()
+ {
+ var expectedException = new InvalidOperationException("stream error");
+ await using var resultSet = CreateFaultedResultSet<string>(expectedException);
+
+ var ex = await Assert.ThrowsAsync<InvalidOperationException>(
+ () => resultSet.ToListAsync());
+
+ Assert.Same(expectedException, ex);
+ }
+
+ [Fact]
+ public async Task SecondCallToGetAsyncEnumeratorShouldThrowInvalidOperationException()
+ {
+ await using var resultSet = CreateResultSet(new List<int> { 1, 2, 3 });
+
+ // First call should succeed
+ var enumerator = resultSet.GetAsyncEnumerator();
+ // Consume to avoid leaving the enumerator in a bad state
+ while (await enumerator.MoveNextAsync())
+ {
+ }
+
+ // Second call throws immediately — the guard check is eager (not deferred
+ // to MoveNextAsync) so callers get fast feedback.
+ Assert.Throws<InvalidOperationException>(
+ () => resultSet.GetAsyncEnumerator());
+ }
+
+ [Fact]
+ public async Task ToListAsyncAfterGetAsyncEnumeratorShouldThrowInvalidOperationException()
+ {
+ await using var resultSet = CreateResultSet(new List<int> { 1 });
+
+ // Consume via await foreach first
+ await foreach (var _ in resultSet)
+ {
+ }
+
+ // ToListAsync calls GetAsyncEnumerator internally, so it should throw
+ await Assert.ThrowsAsync<InvalidOperationException>(
+ () => resultSet.ToListAsync());
+ }
+
+ [Fact]
+ public void DisposeShouldNotThrowWhenCalledSynchronously()
+ {
+ var resultSet = CreateResultSet(new List<int> { 1, 2, 3 });
+
+ // Synchronous Dispose should not throw
+ resultSet.Dispose();
+ }
+
+ [Fact]
+ public async Task DoubleDisposeAsyncShouldNotThrow()
+ {
+ var resultSet = CreateResultSet(new List<int> { 1, 2, 3 });
+
+ await resultSet.DisposeAsync();
+ await resultSet.DisposeAsync();
+ // If we get here without throwing, double-dispose is safe
+ }
+
+ [Fact]
+ public void DoubleDisposeSyncShouldNotThrow()
+ {
+ var resultSet = CreateResultSet(new List<int> { 1, 2, 3 });
+
+ resultSet.Dispose();
+ resultSet.Dispose();
+ // If we get here without throwing, double sync dispose is safe
+ }
+
+ [Fact]
+ public async Task MixedDisposeAsyncThenSyncShouldNotThrow()
+ {
+ var resultSet = CreateResultSet(new List<int> { 1, 2, 3 });
+
+ await resultSet.DisposeAsync();
+ resultSet.Dispose();
+ // Mixed disposal order should not throw
+ }
+
+ [Fact]
+ public async Task CancellationDuringIterationShouldStopEnumeration()
+ {
+ // Create a channel that won't complete on its own — items trickle in slowly
+ var channel = Channel.CreateUnbounded<object>(
+ new UnboundedChannelOptions { SingleWriter = true });
+ var disposeCts = new CancellationTokenSource();
+ var backgroundTask = Task.Run(async () =>
+ {
+ for (var i = 0; i < 100; i++)
+ {
+ await channel.Writer.WriteAsync(i).ConfigureAwait(false);
+ try
+ {
+ await Task.Delay(50, disposeCts.Token).ConfigureAwait(false);
+ }
+ catch (OperationCanceledException)
+ {
+ break;
+ }
+ }
+ channel.Writer.Complete();
+ });
+
+ var resultSet = new ResultSet<int>(channel.Reader, disposeCts, backgroundTask);
+ using var iterationCts = new CancellationTokenSource();
+
+ var collected = new List<int>();
+ await Assert.ThrowsAnyAsync<OperationCanceledException>(async () =>
+ {
+ await foreach (var item in resultSet.WithCancellation(iterationCts.Token))
+ {
+ collected.Add(item);
+ if (collected.Count >= 2)
+ {
+ iterationCts.Cancel();
+ }
+ }
+ });
+
+ Assert.True(collected.Count >= 2, "Should have collected at least 2 items before cancellation");
+ await resultSet.DisposeAsync();
+ }
+ }
+}
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TestTraversal.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TestTraversal.cs
index 132be07b75..badc176540 100644
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TestTraversal.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TestTraversal.cs
@@ -23,6 +23,8 @@
using System.Collections.Generic;
using System.Linq;
+using System.Runtime.CompilerServices;
+using System.Threading;
using Gremlin.Net.Process.Traversal;
namespace Gremlin.Net.UnitTest.Process.Traversal
@@ -33,14 +35,14 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
{
var traversers = new List<Traverser>(traverserObjs.Count);
traverserObjs.ForEach(o => traversers.Add(new Traverser(o)));
- Traversers = traversers;
+ Traversers = ToAsyncEnumerable(traversers);
}
public TestTraversal(IReadOnlyList<object?> traverserObjs, IReadOnlyList<long> traverserBulks)
{
var traversers = new List<Traverser>(traverserObjs.Count);
traversers.AddRange(traverserObjs.Select((t, i) => new Traverser(t, traverserBulks[i])));
- Traversers = traversers;
+ Traversers = ToAsyncEnumerable(traversers);
}
public TestTraversal(IList<ITraversalStrategy> traversalStrategies)
@@ -49,5 +51,18 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
}
public override GremlinLang GremlinLang { get; } = new();
+
+ #pragma warning disable CS1998 // Async method lacks 'await' — yield return in async iterator is intentional
+ internal static async IAsyncEnumerable<Traverser> ToAsyncEnumerable(
+ IEnumerable<Traverser> traversers,
+ [EnumeratorCancellation] CancellationToken cancellationToken = default)
+ {
+ foreach (var traverser in traversers)
+ {
+ cancellationToken.ThrowIfCancellationRequested();
+ yield return traverser;
+ }
+ }
+ #pragma warning restore CS1998
}
}
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TestTraversalStrategy.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TestTraversalStrategy.cs
index 374aeca364..8542e0b687 100644
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TestTraversalStrategy.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TestTraversalStrategy.cs
@@ -30,11 +30,11 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
{
public class TestTraversalStrategy : ITraversalStrategy
{
- private readonly IEnumerable<Traverser> _traversers;
+ private readonly IAsyncEnumerable<Traverser> _traversers;
public TestTraversalStrategy(IEnumerable<Traverser> traversersToAddOnApplication)
{
- _traversers = traversersToAddOnApplication;
+ _traversers = TestTraversal.ToAsyncEnumerable(traversersToAddOnApplication);
}
public void Apply<TStart, TEnd>(ITraversal<TStart, TEnd> traversal)
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TraversalTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TraversalTests.cs
index a158ed8634..fa438094f9 100644
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TraversalTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Process/Traversal/TraversalTests.cs
@@ -24,6 +24,7 @@
using System;
using System.Collections.Generic;
using System.Linq;
+using System.Threading.Tasks;
using Xunit;
using Gremlin.Net.Process.Traversal;
using Gremlin.Net.Structure;
@@ -37,7 +38,7 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
[InlineData("test")]
public void ShouldReturnAvailableTraverserObjWhenNextIsCalled(object traverserObj)
{
- var traversal = new TestTraversal(new List<object?> {traverserObj});
+ var traversal = new TestTraversal(new List<object?> { traverserObj });
var actualObj = traversal.Next();
@@ -45,23 +46,39 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
Assert.Null(traversal.Next());
}
+
[Theory]
[InlineData(1)]
[InlineData("test")]
public void ShouldCheckHasNext(object traverserObj)
{
- var traversal = new TestTraversal(new List<object?> {traverserObj});
+ var traversal = new TestTraversal(new List<object?> { traverserObj });
Assert.True(traversal.HasNext());
Assert.True(traversal.HasNext());
-
+
var actualObj = traversal.Next();
Assert.Equal(traverserObj, actualObj);
-
+
Assert.False(traversal.HasNext());
Assert.False(traversal.HasNext());
}
+ [Fact]
+ public void ShouldReturnAllResultObjsWhenToListIsCalled()
+ {
+ var objs = new List<object?>(20);
+ for (var i = 0; i < 20; i++)
+ objs.Add(i);
+ var traversal = new TestTraversal(objs);
+
+ var traversedObjs = traversal.ToList();
+
+ Assert.Equal(objs.Count, traversedObjs.Count);
+ for (var i = 0; i < objs.Count; i++)
+ Assert.Equal(objs[i], traversedObjs[i]);
+ }
+
[Theory]
[InlineData(3)]
[InlineData(10)]
@@ -92,7 +109,7 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
[Fact]
public void ShouldDrainAllTraversersWhenIterateIsCalled()
{
- var someObjs = new List<object?> {1, 2, 3};
+ var someObjs = new List<object?> { 1, 2, 3 };
var traversal = new TestTraversal(someObjs);
var drainedTraversal = traversal.Iterate();
@@ -104,7 +121,7 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
public void ShouldReturnNullWhenNextIsCalledAndNoTraverserIsAvailable()
{
var expectedFirstObj = 1;
- var traversal = new TestTraversal(new List<object?> {expectedFirstObj});
+ var traversal = new TestTraversal(new List<object?> { expectedFirstObj });
var actualFirstObj = traversal.Next();
var actualSecondObj = traversal.Next();
@@ -116,27 +133,18 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
[Fact]
public void ShouldReturnTraversalsTraverserWhenNextTraverserIsCalled()
{
- var someObjs = new List<object?> {1, 2, 3};
+ var someObjs = new List<object?> { 1, 2, 3 };
var traversal = new TestTraversal(someObjs);
var traverser = traversal.NextTraverser();
- Assert.Equal(traversal.Traversers!.First(), traverser);
- }
-
- [Fact]
- public void ShouldThrowNotSupportedExceptionWhenResetIsCalled()
- {
- var someObjs = new List<object?> {1, 2, 3};
- var traversal = new TestTraversal(someObjs);
-
- Assert.Throws<NotSupportedException>(() => traversal.Reset());
+ Assert.Equal(someObjs[0], traverser.Object);
}
[Fact]
public void ShouldReturnAllTraverserObjsWhenToListIsCalled()
{
- var expectedObjs = new List<object?> {1, 2, 3};
+ var expectedObjs = new List<object?> { 1, 2, 3 };
var traversal = new TestTraversal(expectedObjs);
var traversedObjs = traversal.ToList();
@@ -147,7 +155,7 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
[Fact]
public void ShouldReturnAllTraverserObjWithoutDuplicatesWhenToSetIsCalled()
{
- var traverserObjs = new List<object?> {1, 1, 2, 3};
+ var traverserObjs = new List<object?> { 1, 1, 2, 3 };
var traversal = new TestTraversal(traverserObjs);
var traversedObjSet = traversal.ToSet();
@@ -160,8 +168,8 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
public void ShouldApplyStrategiesWhenNextIsCalledAndNoTraversersPresent()
{
const int expectedObj = 531;
- var testStrategy = new TestTraversalStrategy(new List<Traverser> {new Traverser(expectedObj)});
- var testTraversal = new TestTraversal(new List<ITraversalStrategy> {testStrategy});
+ var testStrategy = new TestTraversalStrategy(new List<Traverser> { new Traverser(expectedObj) });
+ var testTraversal = new TestTraversal(new List<ITraversalStrategy> { testStrategy });
var actualObj = testTraversal.Next();
@@ -171,8 +179,8 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
[Fact]
public void ShouldBeUnfoldTraverserBulksWhenToListIsCalled()
{
- var objs = new List<object?> {1, 2, 3};
- var bulks = new List<long> {3, 2, 1};
+ var objs = new List<object?> { 1, 2, 3 };
+ var bulks = new List<long> { 3, 2, 1 };
var traversal = new TestTraversal(objs, bulks);
var traversedObjs = traversal.ToList();
@@ -181,6 +189,44 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
Assert.Equal(expectedObjs, traversedObjs);
}
+ [Fact]
+ public async Task ShouldDisposeAsyncCleanUpProperly()
+ {
+ var someObjs = new List<object?> { 1, 2, 3 };
+ var traversal = new TestTraversal(someObjs);
+
+ // Consume one element to initialize the enumerator
+ traversal.Next();
+
+ // DisposeAsync should complete without error
+ await traversal.DisposeAsync();
+ }
+
+ [Fact]
+ public async Task ShouldMoveNextAsyncReturnFalseWhenEmpty()
+ {
+ var traversal = new TestTraversal(new List<object?>());
+
+ var hasMore = await traversal.MoveNextAsync();
+
+ Assert.False(hasMore);
+ }
+
+ [Fact]
+ public async Task ShouldMoveNextAsyncIterateAllElements()
+ {
+ var objs = new List<object?> { 1, 2, 3 };
+ var traversal = new TestTraversal(objs);
+ var results = new List<object?>();
+
+ while (await traversal.MoveNextAsync())
+ {
+ results.Add(traversal.Current);
+ }
+
+ Assert.Equal(objs, results);
+ }
+
[Fact]
public void ShouldExtractIdFromVertex()
{
@@ -223,5 +269,52 @@ namespace Gremlin.Net.UnitTest.Process.Traversal
Assert.Equal(1, mergeMap2[Direction.Out]);
Assert.Equal(2, mergeMap2[Direction.In]);
}
+
+ [Fact]
+ public async Task ShouldUnfoldBulksViaMoveNextAsync()
+ {
+ var objs = new List<object?> { "a", "b" };
+ var bulks = new List<long> { 2, 1 };
+ var traversal = new TestTraversal(objs, bulks);
+
+ var results = new List<object?>();
+ while (await traversal.MoveNextAsync())
+ {
+ results.Add(traversal.Current);
+ }
+
+ // "a" with bulk 2 should yield "a", "a"; "b" with bulk 1 should yield "b"
+ Assert.Equal(new List<object?> { "a", "a", "b" }, results);
+ }
+
+ [Fact]
+ public async Task ShouldReturnCurrentObjectAsUntypedValue()
+ {
+ var traversal = new TestTraversal(new List<object?> { 42 });
+
+ await traversal.MoveNextAsync();
+
+ Assert.Equal(42, traversal.CurrentObject);
+ }
+
+ [Fact]
+ public async Task ShouldDisposeAsyncBeforeAnyIteration()
+ {
+ var traversal = new TestTraversal(new List<object?> { 1, 2, 3 });
+
+ // DisposeAsync before any iteration should not throw
+ await traversal.DisposeAsync();
+ }
+
+ [Fact]
+ public async Task ShouldIterateAsyncDrainAllResults()
+ {
+ var traversal = new TestTraversal(new List<object?> { 1, 2, 3 });
+
+ var result = await traversal.IterateAsync();
+
+ // After IterateAsync, MoveNextAsync should return false
+ Assert.False(await result.MoveNextAsync());
+ }
}
-}
\ No newline at end of file
+}
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/GraphBinary4MessageSerializerTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/GraphBinary4MessageSerializerTests.cs
index a675c15aba..41298a6f7e 100644
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/GraphBinary4MessageSerializerTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/GraphBinary4MessageSerializerTests.cs
@@ -22,6 +22,7 @@
#endregion
using System;
+using System.IO;
using System.Threading;
using System.Threading.Tasks;
using Gremlin.Net.Driver.Messages;
@@ -72,7 +73,12 @@ namespace Gremlin.Net.UnitTest.Structure.IO.GraphBinary4
var serializer = CreateMessageSerializer();
await Assert.ThrowsAsync<TaskCanceledException>(async () =>
- await serializer.DeserializeMessageAsync(Array.Empty<byte>(), new CancellationToken(true)));
+ {
+ await foreach (var item in serializer.DeserializeMessageAsync(
+ new MemoryStream(Array.Empty<byte>()), new CancellationToken(true)))
+ {
+ }
+ });
}
private static GraphBinary4MessageSerializer CreateMessageSerializer()
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/ResponseMessageSerializerTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/ResponseSerializerTests.cs
similarity index 71%
rename from gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/ResponseMessageSerializerTests.cs
rename to gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/ResponseSerializerTests.cs
index 8fc76c7560..a616ab6a73 100644
--- a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/ResponseMessageSerializerTests.cs
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/ResponseSerializerTests.cs
@@ -21,6 +21,7 @@
#endregion
+using System.Collections.Generic;
using System.IO;
using System.Threading.Tasks;
using Gremlin.Net.Driver.Exceptions;
@@ -30,14 +31,13 @@ using Xunit;
namespace Gremlin.Net.UnitTest.Structure.IO.GraphBinary4
{
- public class ResponseMessageSerializerTests
+ public class ResponseSerializerTests
{
[Fact]
public async Task ShouldDeserializeNonBulkedResponse()
{
// Build a response: version(0x84) + bulked(0x00) + int value 42 + marker + status footer
using var stream = new MemoryStream();
- var writer = new GraphBinaryWriter();
// Version byte
await stream.WriteByteAsync(0x84);
@@ -60,16 +60,16 @@ namespace Gremlin.Net.UnitTest.Structure.IO.GraphBinary4
stream.Position = 0;
var reader = new GraphBinaryReader();
- var serializer = new ResponseMessageSerializer();
+ var serializer = new ResponseSerializer();
- var result = await serializer.ReadValueAsync(stream, reader);
+ var results = new List<object>();
+ await foreach (var item in serializer.ReadStreamingAsync(stream, reader))
+ {
+ results.Add(item);
+ }
- 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);
+ Assert.Single(results);
+ Assert.Equal(42, results[0]);
}
[Fact]
@@ -102,13 +102,16 @@ namespace Gremlin.Net.UnitTest.Structure.IO.GraphBinary4
stream.Position = 0;
var reader = new GraphBinaryReader();
- var serializer = new ResponseMessageSerializer();
+ var serializer = new ResponseSerializer();
- var result = await serializer.ReadValueAsync(stream, reader);
+ var results = new List<object>();
+ await foreach (var item in serializer.ReadStreamingAsync(stream, reader))
+ {
+ results.Add(item);
+ }
- Assert.True(result.Bulked);
- Assert.Single(result.Result);
- var traverser = Assert.IsType<Traverser>(result.Result[0]);
+ Assert.Single(results);
+ var traverser = Assert.IsType<Traverser>(results[0]);
Assert.Equal("hello", (string)traverser.Object);
Assert.Equal(3L, traverser.Bulk);
}
@@ -141,10 +144,15 @@ namespace Gremlin.Net.UnitTest.Structure.IO.GraphBinary4
stream.Position = 0;
var reader = new GraphBinaryReader();
- var serializer = new ResponseMessageSerializer();
+ var serializer = new ResponseSerializer();
- var ex = await Assert.ThrowsAsync<ResponseException>(
- () => serializer.ReadValueAsync(stream, reader));
+ var ex = await Assert.ThrowsAsync<ResponseException>(async () =>
+ {
+ await foreach (var item in serializer.ReadStreamingAsync(stream, reader))
+ {
+ // Should not yield any items before throwing
+ }
+ });
Assert.Equal(500, ex.StatusCode);
Assert.Equal("Server error", ex.Message);
@@ -160,10 +168,15 @@ namespace Gremlin.Net.UnitTest.Structure.IO.GraphBinary4
stream.Position = 0;
var reader = new GraphBinaryReader();
- var serializer = new ResponseMessageSerializer();
+ var serializer = new ResponseSerializer();
- await Assert.ThrowsAsync<IOException>(
- () => serializer.ReadValueAsync(stream, reader));
+ await Assert.ThrowsAsync<IOException>(async () =>
+ {
+ await foreach (var item in serializer.ReadStreamingAsync(stream, reader))
+ {
+ // Should throw before yielding any items
+ }
+ });
}
[Fact]
@@ -197,14 +210,49 @@ namespace Gremlin.Net.UnitTest.Structure.IO.GraphBinary4
stream.Position = 0;
var reader = new GraphBinaryReader();
- var serializer = new ResponseMessageSerializer();
+ var serializer = new ResponseSerializer();
- var result = await serializer.ReadValueAsync(stream, reader);
+ var results = new List<object>();
+ await foreach (var item in serializer.ReadStreamingAsync(stream, reader))
+ {
+ results.Add(item);
+ }
- Assert.Equal(3, result.Result.Count);
- Assert.Equal(1, result.Result[0]);
- Assert.Equal(2, result.Result[1]);
- Assert.Equal(3, result.Result[2]);
+ Assert.Equal(3, results.Count);
+ Assert.Equal(1, results[0]);
+ Assert.Equal(2, results[1]);
+ Assert.Equal(3, results[2]);
+ }
+
+ [Fact]
+ public async Task ShouldDeserializeEmptyResultSet()
+ {
+ using var stream = new MemoryStream();
+
+ // Version byte
+ await stream.WriteByteAsync(0x84);
+ // Bulked = false
+ await stream.WriteByteAsync(0x00);
+ // Marker immediately after bulked flag (no results)
+ 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 ResponseSerializer();
+
+ var results = new List<object>();
+ await foreach (var item in serializer.ReadStreamingAsync(stream, reader))
+ {
+ results.Add(item);
+ }
+
+ Assert.Empty(results);
}
}
}
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/StreamExtensionsTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/StreamExtensionsTests.cs
new file mode 100644
index 0000000000..c926718798
--- /dev/null
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/StreamExtensionsTests.cs
@@ -0,0 +1,382 @@
+#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.IO;
+using System.Threading.Tasks;
+using Gremlin.Net.Structure.IO.GraphBinary4;
+using Xunit;
+
+namespace Gremlin.Net.UnitTest.Structure.IO.GraphBinary4
+{
+ public class StreamExtensionsTests
+ {
+ // --- EOF handling tests ---
+
+ [Fact]
+ public async Task ReadByteAsyncShouldThrowOnEmptyStream()
+ {
+ using var stream = new MemoryStream(Array.Empty<byte>());
+
+ await Assert.ThrowsAsync<IOException>(
+ async () => await stream.ReadByteAsync());
+ }
+
+ [Fact]
+ public async Task ReadIntAsyncShouldThrowOnStreamWithFewerThan4Bytes()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x01, 0x02, 0x03 });
+
+ await Assert.ThrowsAsync<EndOfStreamException>(
+ async () => await stream.ReadIntAsync());
+ }
+
+ [Fact]
+ public async Task ReadLongAsyncShouldThrowOnStreamWithFewerThan8Bytes()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07 });
+
+ await Assert.ThrowsAsync<EndOfStreamException>(
+ async () => await stream.ReadLongAsync());
+ }
+
+ [Fact]
+ public async Task ReadShortAsyncShouldThrowOnStreamWithFewerThan2Bytes()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x01 });
+
+ await Assert.ThrowsAsync<EndOfStreamException>(
+ async () => await stream.ReadShortAsync());
+ }
+
+ [Fact]
+ public async Task ReadFloatAsyncShouldThrowOnStreamWithFewerThan4Bytes()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x01, 0x02, 0x03 });
+
+ await Assert.ThrowsAsync<EndOfStreamException>(
+ async () => await stream.ReadFloatAsync());
+ }
+
+ [Fact]
+ public async Task ReadDoubleAsyncShouldThrowOnStreamWithFewerThan8Bytes()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07 });
+
+ await Assert.ThrowsAsync<EndOfStreamException>(
+ async () => await stream.ReadDoubleAsync());
+ }
+
+ [Fact]
+ public async Task ReadAsyncShouldThrowOnStreamWithFewerThanCountBytes()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x01, 0x02 });
+
+ await Assert.ThrowsAsync<EndOfStreamException>(
+ async () => await stream.ReadAsync(5));
+ }
+
+ // --- Valid big-endian read tests ---
+
+ [Fact]
+ public async Task ReadByteAsyncShouldReturnCorrectValue()
+ {
+ using var stream = new MemoryStream(new byte[] { 0xAB });
+
+ var result = await stream.ReadByteAsync();
+
+ Assert.Equal(0xAB, result);
+ }
+
+ [Fact]
+ public async Task ReadIntAsyncShouldReturnCorrectBigEndianValue()
+ {
+ // Big-endian representation of 0x01020304 = 16909060
+ using var stream = new MemoryStream(new byte[] { 0x01, 0x02, 0x03, 0x04 });
+
+ var result = await stream.ReadIntAsync();
+
+ Assert.Equal(0x01020304, result);
+ }
+
+ [Fact]
+ public async Task ReadIntAsyncShouldHandleNegativeValue()
+ {
+ // Big-endian representation of -1 = 0xFFFFFFFF
+ using var stream = new MemoryStream(new byte[] { 0xFF, 0xFF, 0xFF, 0xFF });
+
+ var result = await stream.ReadIntAsync();
+
+ Assert.Equal(-1, result);
+ }
+
+ [Fact]
+ public async Task ReadLongAsyncShouldReturnCorrectBigEndianValue()
+ {
+ // Big-endian representation of 0x0102030405060708
+ using var stream = new MemoryStream(
+ new byte[] { 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08 });
+
+ var result = await stream.ReadLongAsync();
+
+ Assert.Equal(0x0102030405060708L, result);
+ }
+
+ [Fact]
+ public async Task ReadLongAsyncShouldHandleNegativeValue()
+ {
+ // Big-endian representation of -1 = 0xFFFFFFFFFFFFFFFF
+ using var stream = new MemoryStream(
+ new byte[] { 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF });
+
+ var result = await stream.ReadLongAsync();
+
+ Assert.Equal(-1L, result);
+ }
+
+ [Fact]
+ public async Task ReadShortAsyncShouldReturnCorrectBigEndianValue()
+ {
+ // Big-endian representation of 0x0102 = 258
+ using var stream = new MemoryStream(new byte[] { 0x01, 0x02 });
+
+ var result = await stream.ReadShortAsync();
+
+ Assert.Equal((short)0x0102, result);
+ }
+
+ [Fact]
+ public async Task ReadShortAsyncShouldHandleNegativeValue()
+ {
+ // Big-endian representation of -1 = 0xFFFF
+ using var stream = new MemoryStream(new byte[] { 0xFF, 0xFF });
+
+ var result = await stream.ReadShortAsync();
+
+ Assert.Equal((short)-1, result);
+ }
+
+ [Fact]
+ public async Task ReadFloatAsyncShouldReturnCorrectBigEndianValue()
+ {
+ // Big-endian representation of 1.0f = 0x3F800000
+ using var stream = new MemoryStream(new byte[] { 0x3F, 0x80, 0x00, 0x00 });
+
+ var result = await stream.ReadFloatAsync();
+
+ Assert.Equal(1.0f, result);
+ }
+
+ [Fact]
+ public async Task ReadDoubleAsyncShouldReturnCorrectBigEndianValue()
+ {
+ // Big-endian representation of 1.0d = 0x3FF0000000000000
+ using var stream = new MemoryStream(
+ new byte[] { 0x3F, 0xF0, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00 });
+
+ var result = await stream.ReadDoubleAsync();
+
+ Assert.Equal(1.0d, result);
+ }
+
+ [Fact]
+ public async Task ReadAsyncShouldReturnCorrectBytes()
+ {
+ var expected = new byte[] { 0xDE, 0xAD, 0xBE, 0xEF };
+ using var stream = new MemoryStream(expected);
+
+ var result = await stream.ReadAsync(4);
+
+ Assert.Equal(expected, result);
+ }
+
+ // --- Round-trip tests (write then read) ---
+
+ [Fact]
+ public async Task IntRoundTripShouldPreserveValue()
+ {
+ using var stream = new MemoryStream();
+ await stream.WriteIntAsync(42);
+ stream.Position = 0;
+
+ var result = await stream.ReadIntAsync();
+
+ Assert.Equal(42, result);
+ }
+
+ [Fact]
+ public async Task LongRoundTripShouldPreserveValue()
+ {
+ using var stream = new MemoryStream();
+ await stream.WriteLongAsync(123456789012345L);
+ stream.Position = 0;
+
+ var result = await stream.ReadLongAsync();
+
+ Assert.Equal(123456789012345L, result);
+ }
+
+ [Fact]
+ public async Task ShortRoundTripShouldPreserveValue()
+ {
+ using var stream = new MemoryStream();
+ await stream.WriteShortAsync(12345);
+ stream.Position = 0;
+
+ var result = await stream.ReadShortAsync();
+
+ Assert.Equal((short)12345, result);
+ }
+
+ [Fact]
+ public async Task FloatRoundTripShouldPreserveValue()
+ {
+ using var stream = new MemoryStream();
+ await stream.WriteFloatAsync(3.14f);
+ stream.Position = 0;
+
+ var result = await stream.ReadFloatAsync();
+
+ Assert.Equal(3.14f, result);
+ }
+
+ [Fact]
+ public async Task DoubleRoundTripShouldPreserveValue()
+ {
+ using var stream = new MemoryStream();
+ await stream.WriteDoubleAsync(3.14159265358979);
+ stream.Position = 0;
+
+ var result = await stream.ReadDoubleAsync();
+
+ Assert.Equal(3.14159265358979, result);
+ }
+
+ // --- Bool read/write tests ---
+
+ [Fact]
+ public async Task ReadBoolAsyncShouldReturnTrueForByte1()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x01 });
+
+ var result = await stream.ReadBoolAsync();
+
+ Assert.True(result);
+ }
+
+ [Fact]
+ public async Task ReadBoolAsyncShouldReturnFalseForByte0()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x00 });
+
+ var result = await stream.ReadBoolAsync();
+
+ Assert.False(result);
+ }
+
+ [Fact]
+ public async Task ReadBoolAsyncShouldThrowOnInvalidByte()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x02 });
+
+ await Assert.ThrowsAsync<IOException>(
+ async () => await stream.ReadBoolAsync());
+ }
+
+ [Fact]
+ public async Task ReadBoolAsyncShouldThrowOnEmptyStream()
+ {
+ using var stream = new MemoryStream(Array.Empty<byte>());
+
+ await Assert.ThrowsAsync<IOException>(
+ async () => await stream.ReadBoolAsync());
+ }
+
+ [Fact]
+ public async Task BoolRoundTripShouldPreserveTrueValue()
+ {
+ using var stream = new MemoryStream();
+ await stream.WriteBoolAsync(true);
+ stream.Position = 0;
+
+ var result = await stream.ReadBoolAsync();
+
+ Assert.True(result);
+ }
+
+ [Fact]
+ public async Task BoolRoundTripShouldPreserveFalseValue()
+ {
+ using var stream = new MemoryStream();
+ await stream.WriteBoolAsync(false);
+ stream.Position = 0;
+
+ var result = await stream.ReadBoolAsync();
+
+ Assert.False(result);
+ }
+
+ // --- SByte read/write tests ---
+
+ [Fact]
+ public async Task ReadSByteAsyncShouldReturnCorrectPositiveValue()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x7F }); // sbyte max = 127
+
+ var result = await stream.ReadSByteAsync();
+
+ Assert.Equal((sbyte)127, result);
+ }
+
+ [Fact]
+ public async Task ReadSByteAsyncShouldReturnCorrectNegativeValue()
+ {
+ using var stream = new MemoryStream(new byte[] { 0x80 }); // sbyte min = -128
+
+ var result = await stream.ReadSByteAsync();
+
+ Assert.Equal((sbyte)-128, result);
+ }
+
+ [Fact]
+ public async Task ReadSByteAsyncShouldThrowOnEmptyStream()
+ {
+ using var stream = new MemoryStream(Array.Empty<byte>());
+
+ await Assert.ThrowsAsync<IOException>(
+ async () => await stream.ReadSByteAsync());
+ }
+
+ [Fact]
+ public async Task SByteRoundTripShouldPreserveValue()
+ {
+ using var stream = new MemoryStream();
+ await stream.WriteSByteAsync(-42);
+ stream.Position = 0;
+
+ var result = await stream.ReadSByteAsync();
+
+ Assert.Equal((sbyte)-42, result);
+ }
+ }
+}
diff --git a/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/StreamingDeserializerRoundTripTests.cs b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/StreamingDeserializerRoundTripTests.cs
new file mode 100644
index 0000000000..1d4cf70e51
--- /dev/null
+++ b/gremlin-dotnet/test/Gremlin.Net.UnitTest/Structure/IO/GraphBinary4/StreamingDeserializerRoundTripTests.cs
@@ -0,0 +1,669 @@
+#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.Linq;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+using Gremlin.Net.Driver;
+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
+{
+ /// <summary>
+ /// Round-trip tests for streaming deserialization with randomized inputs.
+ /// </summary>
+ public class StreamingDeserializerRoundTripTests
+ {
+ /// <summary>
+ /// Streaming deserialization round-trip (non-bulked mode).
+ ///
+ /// For any random list of primitive values, serializing them into a valid
+ /// GraphBinary 4.0 non-bulked response stream and deserializing via
+ /// ReadStreamingAsync yields the same values in the same order.
+ /// </summary>
+ [Theory]
+ [InlineData(0)]
+ [InlineData(1)]
+ [InlineData(42)]
+ [InlineData(-1)]
+ [InlineData(12345)]
+ [InlineData(-99999)]
+ [InlineData(int.MaxValue)]
+ [InlineData(int.MinValue)]
+ [InlineData(7777)]
+ [InlineData(314159)]
+ public async Task NonBulkedRoundTrip(int seed)
+ {
+ Assert.True(await NonBulkedRoundTripAsync(seed));
+ }
+
+ private static async Task<bool> NonBulkedRoundTripAsync(int seed)
+ {
+ var rng = new System.Random(seed);
+ var count = rng.Next(0, 20);
+ var values = GenerateRandomPrimitives(rng, count);
+
+ using var stream = new MemoryStream();
+ var writer = new GraphBinaryWriter();
+
+ await WriteNonBulkedResponse(stream, writer, values);
+ stream.Position = 0;
+
+ var reader = new GraphBinaryReader();
+ var serializer = new ResponseSerializer();
+
+ var results = new List<object>();
+ await foreach (var item in serializer.ReadStreamingAsync(stream, reader))
+ {
+ results.Add(item);
+ }
+
+ if (values.Count != results.Count) return false;
+
+ for (var i = 0; i < values.Count; i++)
+ {
+ if (!values[i].Value.Equals(results[i])) return false;
+ }
+
+ return true;
+ }
+
+ /// <summary>
+ /// Streaming deserialization round-trip (bulked mode).
+ ///
+ /// For any random list of primitive values with random bulk counts,
+ /// serializing them into a valid GraphBinary 4.0 bulked response stream
+ /// and deserializing via ReadStreamingAsync yields Traverser objects
+ /// wrapping the original values with the correct bulk counts.
+ /// </summary>
+ [Theory]
+ [InlineData(0)]
+ [InlineData(1)]
+ [InlineData(42)]
+ [InlineData(-1)]
+ [InlineData(12345)]
+ [InlineData(-99999)]
+ [InlineData(int.MaxValue)]
+ [InlineData(int.MinValue)]
+ [InlineData(7777)]
+ [InlineData(314159)]
+ public async Task BulkedRoundTrip(int seed)
+ {
+ Assert.True(await BulkedRoundTripAsync(seed));
+ }
+
+ private static async Task<bool> BulkedRoundTripAsync(int seed)
+ {
+ var rng = new System.Random(seed);
+ var count = rng.Next(0, 20);
+ var values = GenerateRandomPrimitives(rng, count);
+ var bulks = Enumerable.Range(0, count)
+ .Select(_ => (long)(rng.Next(1, 1000)))
+ .ToList();
+
+ var entries = values.Zip(bulks, (v, b) => (v, b)).ToList();
+
+ using var stream = new MemoryStream();
+ var writer = new GraphBinaryWriter();
+
+ await WriteBulkedResponse(stream, writer, entries);
+ stream.Position = 0;
+
+ var reader = new GraphBinaryReader();
+ var serializer = new ResponseSerializer();
+
+ var results = new List<object>();
+ await foreach (var item in serializer.ReadStreamingAsync(stream, reader))
+ {
+ results.Add(item);
+ }
+
+ if (entries.Count != results.Count) return false;
+
+ for (var i = 0; i < entries.Count; i++)
+ {
+ if (results[i] is not Traverser traverser) return false;
+ if (!entries[i].v.Value.Equals((object)traverser.Object)) return false;
+ if (entries[i].b != traverser.Bulk) return false;
+ }
+
+ return true;
+ }
+
+ /// <summary>
+ /// Version byte validation.
+ ///
+ /// For any byte value in the range 0x00–0x7F (MSB not set),
+ /// constructing a stream starting with that byte and attempting to
+ /// iterate ReadStreamingAsync throws IOException.
+ /// </summary>
+ [Theory]
+ [InlineData(0x00)]
+ [InlineData(0x01)]
+ [InlineData(0x7F)]
+ [InlineData(0x40)]
+ [InlineData(0x3F)]
+ [InlineData(0x10)]
+ [InlineData(0x55)]
+ [InlineData(0x6A)]
+ [InlineData(0x20)]
+ [InlineData(0x0F)]
+ public async Task InvalidVersionByteThrowsIOException(byte invalidByte)
+ {
+ Assert.True(await InvalidVersionByteThrowsIOExceptionAsync(invalidByte));
+ }
+
+ private static async Task<bool> InvalidVersionByteThrowsIOExceptionAsync(byte versionByte)
+ {
+ using var stream = new MemoryStream();
+
+ // Write the invalid version byte
+ await stream.WriteByteAsync(versionByte);
+ // Write enough trailing data to avoid EOF errors before the version check
+ await stream.WriteByteAsync(0x00); // bulked flag
+ // Marker
+ await stream.WriteByteAsync(0xFD);
+ await stream.WriteByteAsync(0x00);
+ await stream.WriteByteAsync(0x00);
+ // Status footer
+ await stream.WriteIntAsync(200);
+ await stream.WriteByteAsync(0x01);
+ await stream.WriteByteAsync(0x01);
+
+ stream.Position = 0;
+ var reader = new GraphBinaryReader();
+ var serializer = new ResponseSerializer();
+
+ try
+ {
+ await foreach (var _ in serializer.ReadStreamingAsync(stream, reader))
+ {
+ // Should not yield any items
+ }
+
+ return false; // Should have thrown
+ }
+ catch (IOException)
+ {
+ return true; // Expected
+ }
+ }
+
+ /// <summary>
+ /// Verifies that version byte 0x81 (MSB set) is accepted as valid.
+ /// </summary>
+ [Fact]
+ public async Task ValidVersionByte0x81Succeeds()
+ {
+ using var stream = new MemoryStream();
+
+ // Version byte 0x81 (MSB set, valid)
+ await stream.WriteByteAsync(0x81);
+ // Bulked = false
+ await stream.WriteByteAsync(0x00);
+ // Marker immediately (no results)
+ await stream.WriteByteAsync(0xFD);
+ await stream.WriteByteAsync(0x00);
+ await stream.WriteByteAsync(0x00);
+ // Status footer: 200, null message, null exception
+ await stream.WriteIntAsync(200);
+ await stream.WriteByteAsync(0x01);
+ await stream.WriteByteAsync(0x01);
+
+ stream.Position = 0;
+ var reader = new GraphBinaryReader();
+ var serializer = new ResponseSerializer();
+
+ var results = new List<object>();
+ await foreach (var item in serializer.ReadStreamingAsync(stream, reader))
+ {
+ results.Add(item);
+ }
+
+ // Should succeed with no results (empty result set with valid version byte)
+ Assert.Empty(results);
+ }
+
+ /// <summary>
+ /// Error status code propagation.
+ ///
+ /// For any random non-200 status code (1–199, 201–599), random nullable
+ /// status message, and random nullable exception string, writing a valid
+ /// response stream with an error footer and iterating ReadStreamingAsync
+ /// throws a ResponseException with matching StatusCode, Message, and
+ /// ServerException fields.
+ /// </summary>
+ [Theory]
+ [InlineData(0)]
+ [InlineData(1)]
+ [InlineData(42)]
+ [InlineData(-1)]
+ [InlineData(12345)]
+ [InlineData(-99999)]
+ [InlineData(int.MaxValue)]
+ [InlineData(int.MinValue)]
+ [InlineData(7777)]
+ [InlineData(314159)]
+ public async Task ErrorStatusCodePropagation(int seed)
+ {
+ Assert.True(await ErrorStatusCodePropagationAsync(seed));
+ }
+
+ private static async Task<bool> ErrorStatusCodePropagationAsync(int seed)
+ {
+ var rng = new System.Random(seed);
+
+ // Generate a non-200 status code in range 1–599
+ int statusCode;
+ do
+ {
+ statusCode = rng.Next(1, 600);
+ } while (statusCode == 200);
+
+ // Generate nullable status message
+ var hasMessage = rng.Next(2) == 1;
+ var statusMessage = hasMessage ? GenerateRandomString(rng) : null;
+
+ // Generate nullable exception string
+ var hasException = rng.Next(2) == 1;
+ var exceptionString = hasException ? GenerateRandomString(rng) : null;
+
+ using var stream = new MemoryStream();
+
+ // Write a valid response with error footer (no result data)
+ await WriteErrorResponse(stream, statusCode, statusMessage, exceptionString);
+ stream.Position = 0;
+
+ var reader = new GraphBinaryReader();
+ var serializer = new ResponseSerializer();
+
+ try
+ {
+ await foreach (var _ in serializer.ReadStreamingAsync(stream, reader))
+ {
+ // Should not yield any items before throwing
+ }
+
+ return false; // Should have thrown ResponseException
+ }
+ catch (ResponseException ex)
+ {
+ // Verify StatusCode matches
+ if (ex.StatusCode != statusCode) return false;
+
+ // Verify Message matches: statusMessage if provided, otherwise default
+ var expectedMessage = statusMessage ?? $"Server error: {statusCode}";
+ if (ex.Message != expectedMessage) return false;
+
+ // Verify ServerException matches
+ if (ex.ServerException != exceptionString) return false;
+
+ return true;
+ }
+ }
+
+ /// <summary>
+ /// ToListAsync collects all results in order.
+ ///
+ /// For any random list of int values, writing them to a channel,
+ /// constructing a ResultSet backed by that channel, and calling
+ /// ToListAsync returns a list matching the input in order and length.
+ /// </summary>
+ [Theory]
+ [InlineData(0)]
+ [InlineData(1)]
+ [InlineData(42)]
+ [InlineData(-1)]
+ [InlineData(12345)]
+ [InlineData(-99999)]
+ [InlineData(int.MaxValue)]
+ [InlineData(int.MinValue)]
+ [InlineData(7777)]
+ [InlineData(314159)]
+ public async Task ToListAsyncCollectsAllResultsInOrder(int seed)
+ {
+ Assert.True(await ToListAsyncCollectsAllResultsInOrderAsync(seed));
+ }
+
+ private static async Task<bool> ToListAsyncCollectsAllResultsInOrderAsync(int seed)
+ {
+ var rng = new System.Random(seed);
+ var count = rng.Next(0, 50);
+ var inputValues = new List<int>(count);
+ for (var i = 0; i < count; i++)
+ {
+ inputValues.Add(rng.Next(int.MinValue, int.MaxValue));
+ }
+
+ // Create a channel and write all values, then complete it
+ var channel = Channel.CreateUnbounded<object>(
+ new UnboundedChannelOptions { SingleWriter = true });
+ foreach (var value in inputValues)
+ {
+ await channel.Writer.WriteAsync(value);
+ }
+ channel.Writer.Complete();
+
+ // Create a ResultSet<int> backed by the channel
+ using var disposeCts = new CancellationTokenSource();
+ var backgroundTask = Task.CompletedTask;
+ var resultSet = new ResultSet<int>(channel.Reader, disposeCts, backgroundTask);
+
+ // Call ToListAsync and verify
+ var result = await resultSet.ToListAsync();
+
+ if (result.Count != inputValues.Count) return false;
+
+ for (var i = 0; i < inputValues.Count; i++)
+ {
+ if (result[i] != inputValues[i]) return false;
+ }
+
+ return true;
+ }
+
+ /// <summary>
+ /// EOF detection in stream reads.
+ ///
+ /// For any random byte array of length 0 to 7, attempting to read
+ /// int/long/short values from a stream with insufficient data throws
+ /// IOException or EndOfStreamException.
+ /// </summary>
+ [Theory]
+ [InlineData(0)]
+ [InlineData(1)]
+ [InlineData(42)]
+ [InlineData(-1)]
+ [InlineData(12345)]
+ [InlineData(-99999)]
+ [InlineData(int.MaxValue)]
+ [InlineData(int.MinValue)]
+ [InlineData(7777)]
+ [InlineData(314159)]
+ public async Task EofDetectionInStreamReads(int seed)
+ {
+ Assert.True(await EofDetectionInStreamReadsAsync(seed));
+ }
+
+ private static async Task<bool> EofDetectionInStreamReadsAsync(int seed)
+ {
+ var rng = new System.Random(seed);
+ var length = rng.Next(0, 8); // 0 to 7 bytes
+ var data = new byte[length];
+ rng.NextBytes(data);
+
+ // ReadShortAsync needs 2 bytes — test with arrays of length 0-1
+ if (length < 2)
+ {
+ if (!await ThrowsIOExceptionAsync(() =>
+ {
+ var stream = new MemoryStream(data);
+ return stream.ReadShortAsync().AsTask();
+ }))
+ return false;
+ }
+
+ // ReadIntAsync needs 4 bytes — test with arrays of length 0-3
+ if (length < 4)
+ {
+ if (!await ThrowsIOExceptionAsync(() =>
+ {
+ var stream = new MemoryStream(data);
+ return stream.ReadIntAsync().AsTask();
+ }))
+ return false;
+ }
+
+ // ReadLongAsync needs 8 bytes — test with arrays of length 0-7
+ if (length < 8)
+ {
+ if (!await ThrowsIOExceptionAsync(() =>
+ {
+ var stream = new MemoryStream(data);
+ return stream.ReadLongAsync().AsTask();
+ }))
+ return false;
+ }
+
+ // ReadByteAsync needs 1 byte — test with empty array
+ if (length == 0)
+ {
+ if (!await ThrowsIOExceptionAsync(() =>
+ {
+ var stream = new MemoryStream(data);
+ return stream.ReadByteAsync().AsTask();
+ }))
+ return false;
+ }
+
+ return true;
+ }
+
+ /// <summary>
+ /// Helper that returns true if the given async action throws IOException
+ /// or EndOfStreamException (which is a subclass of IOException).
+ /// </summary>
+ private static async Task<bool> ThrowsIOExceptionAsync(Func<Task> action)
+ {
+ try
+ {
+ await action();
+ return false; // Should have thrown
+ }
+ catch (IOException)
+ {
+ return true; // IOException or EndOfStreamException (subclass)
+ }
+ }
+
+ /// <summary>
+ /// Writes a complete GraphBinary 4.0 response stream with an error status footer.
+ /// Contains no result data (marker immediately after bulked flag).
+ /// </summary>
+ private static async Task WriteErrorResponse(Stream stream, int statusCode,
+ string? statusMessage, string? exceptionString)
+ {
+ // Version byte (0x84)
+ await stream.WriteByteAsync(GraphBinaryWriter.VersionByte);
+ // Bulked = false
+ await stream.WriteByteAsync(0x00);
+
+ // Marker immediately (no results): type_code(0xFD) + value_flag(0x00) + value(0x00)
+ await stream.WriteByteAsync(0xFD);
+ await stream.WriteByteAsync(0x00);
+ await stream.WriteByteAsync(0x00);
+
+ // Status footer: non-nullable int status code
+ await stream.WriteIntAsync(statusCode);
+
+ // Nullable status message
+ await WriteNullableStringAsync(stream, statusMessage);
+
+ // Nullable exception string
+ await WriteNullableStringAsync(stream, exceptionString);
+ }
+
+ /// <summary>
+ /// Writes a nullable string in GraphBinary format:
+ /// value_flag(0x01) for null, or value_flag(0x00) + int32(length) + UTF-8 bytes.
+ /// </summary>
+ private static async Task WriteNullableStringAsync(Stream stream, string? value)
+ {
+ if (value == null)
+ {
+ await stream.WriteByteAsync(0x01); // null flag
+ }
+ else
+ {
+ await stream.WriteByteAsync(0x00); // not null flag
+ var bytes = System.Text.Encoding.UTF8.GetBytes(value);
+ await stream.WriteIntAsync(bytes.Length);
+ if (bytes.Length > 0)
+ {
+ await stream.WriteAsync(bytes);
+ }
+ }
+ }
+
+ /// <summary>
+ /// Writes a complete GraphBinary 4.0 response stream for non-bulked mode.
+ /// </summary>
+ private static async Task WriteNonBulkedResponse(Stream stream, GraphBinaryWriter writer,
+ IReadOnlyList<PrimitiveValueHolder> values)
+ {
+ // Version byte (0x84)
+ await stream.WriteByteAsync(GraphBinaryWriter.VersionByte);
+ // Bulked = false
+ await stream.WriteByteAsync(0x00);
+
+ // Write each value as fully-qualified typed value
+ foreach (var pv in values)
+ {
+ await writer.WriteAsync(pv.Value, stream);
+ }
+
+ // Marker: type_code(0xFD) + value_flag(0x00) + value(0x00)
+ await stream.WriteByteAsync(0xFD);
+ await stream.WriteByteAsync(0x00);
+ await stream.WriteByteAsync(0x00);
+
+ // Status footer: status code 200, null message, null exception
+ await stream.WriteIntAsync(200);
+ await stream.WriteByteAsync(0x01); // null message
+ await stream.WriteByteAsync(0x01); // null exception
+ }
+
+ /// <summary>
+ /// Writes a complete GraphBinary 4.0 response stream for bulked mode.
+ /// </summary>
+ private static async Task WriteBulkedResponse(Stream stream, GraphBinaryWriter writer,
+ IReadOnlyList<(PrimitiveValueHolder Value, long Bulk)> entries)
+ {
+ // Version byte (0x84)
+ await stream.WriteByteAsync(GraphBinaryWriter.VersionByte);
+ // Bulked = true
+ await stream.WriteByteAsync(0x01);
+
+ // Write each value followed by its bulk count as fully-qualified typed values
+ foreach (var (pv, bulk) in entries)
+ {
+ await writer.WriteAsync(pv.Value, stream);
+ await writer.WriteAsync(bulk, stream);
+ }
+
+ // Marker: type_code(0xFD) + value_flag(0x00) + value(0x00)
+ await stream.WriteByteAsync(0xFD);
+ await stream.WriteByteAsync(0x00);
+ await stream.WriteByteAsync(0x00);
+
+ // Status footer: status code 200, null message, null exception
+ await stream.WriteIntAsync(200);
+ await stream.WriteByteAsync(0x01); // null message
+ await stream.WriteByteAsync(0x01); // null exception
+ }
+
+ /// <summary>
+ /// A wrapper for a single primitive value used in test generation.
+ /// </summary>
+ private class PrimitiveValueHolder
+ {
+ public object Value { get; }
+
+ public PrimitiveValueHolder(object value)
+ {
+ Value = value;
+ }
+
+ public override string ToString() => $"{Value?.GetType().Name}({Value})";
+ }
+
+ /// <summary>
+ /// Generates a list of random primitive values using the given RNG.
+ /// </summary>
+ private static List<PrimitiveValueHolder> GenerateRandomPrimitives(System.Random rng, int count)
+ {
+ var result = new List<PrimitiveValueHolder>(count);
+ for (var i = 0; i < count; i++)
+ {
+ result.Add(GenerateOnePrimitive(rng));
+ }
+ return result;
+ }
+
+ /// <summary>
+ /// Generates a single random primitive value from the set:
+ /// int, long, string (non-null), double (finite), float (finite), bool.
+ /// </summary>
+ private static PrimitiveValueHolder GenerateOnePrimitive(System.Random rng)
+ {
+ var typeChoice = rng.Next(6);
+ return typeChoice switch
+ {
+ 0 => new PrimitiveValueHolder(rng.Next(int.MinValue, int.MaxValue)),
+ 1 => new PrimitiveValueHolder((long)rng.Next(int.MinValue, int.MaxValue) * rng.Next(1, 100)),
+ 2 => new PrimitiveValueHolder(GenerateRandomString(rng)),
+ 3 => new PrimitiveValueHolder(GenerateFiniteDouble(rng)),
+ 4 => new PrimitiveValueHolder(GenerateFiniteFloat(rng)),
+ 5 => new PrimitiveValueHolder(rng.Next(2) == 0),
+ _ => throw new InvalidOperationException()
+ };
+ }
+
+ /// <summary>
+ /// Generates a random non-null string of printable ASCII characters.
+ /// </summary>
+ private static string GenerateRandomString(System.Random rng)
+ {
+ var length = rng.Next(0, 50);
+ var chars = new char[length];
+ for (var i = 0; i < length; i++)
+ {
+ chars[i] = (char)rng.Next(32, 127); // printable ASCII
+ }
+ return new string(chars);
+ }
+
+ /// <summary>
+ /// Generates a finite double value (no NaN or Infinity).
+ /// </summary>
+ private static double GenerateFiniteDouble(System.Random rng)
+ {
+ return (rng.NextDouble() - 0.5) * 2.0 * rng.Next(1, 10000);
+ }
+
+ /// <summary>
+ /// Generates a finite float value (no NaN or Infinity).
+ /// </summary>
+ private static float GenerateFiniteFloat(System.Random rng)
+ {
+ return (float)((rng.NextDouble() - 0.5) * 2.0 * rng.Next(1, 10000));
+ }
+ }
+}