Expose IASyncEnumerable in callback to allow async linq to be used (#676)

Author: RichardIrons-neo4j

Neo4j.Driver/Neo4j.Driver.Tests.TestBackend/Protocol/DriverQuery/ExecuteQuery.cs

@@ -11,7 +11,7 @@ internal class ExecuteQuery : IProtocolObject
 {
     public ExecuteQueryDto data { get; set; }
     [JsonIgnore]
-    public EagerResult Result { get; set; }
+    public EagerResult<IReadOnlyList<IRecord>> Result { get; set; }
 
     public class ExecuteQueryDto
     {
@@ -36,11 +36,13 @@ public override async Task Process()
         var driver = ObjManager.GetObject<NewDriver>(data.driverId).Driver;
         var queryConfig = BuildConfig();
 
-        Result = await driver
+        var queryResult = await driver
             .ExecutableQuery(data.cypher)
             .WithParameters(data.parameters)
             .WithConfig(queryConfig)
             .ExecuteAsync();
+
+        Result = queryResult;
     }
 
     private QueryConfig BuildConfig()
@@ -73,7 +75,7 @@ private QueryConfig BuildConfig()
 
     public override string Respond()
     {
-        var mappedList = Result.Records
+        var mappedList = Result.Result
             .Select(x => new
             {
                 values = x.Values

Neo4j.Driver/Neo4j.Driver/Experimental/EagerResult.cs

@@ -1,64 +1,70 @@
-// Copyright (c) 2002-2022 "Neo4j,"
+// Copyright (c) "Neo4j"
 // Neo4j Sweden AB [http://neo4j.com]
-// 
+//
 // This file is part of Neo4j.
-// 
-// Licensed under the Apache License, Version 2.0 (the "License");
+//
+// Licensed 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.
 
-using System.Collections;
-using System.Collections.Generic;
-
 namespace Neo4j.Driver.Experimental;
 
 /// <summary>
 /// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.
 /// Complete result from a cypher query.
 /// </summary>
-public sealed class EagerResult : IReadOnlyList<IRecord>
+public sealed class EagerResult<T>
 {
+    internal EagerResult(T result, IResultSummary summary, string[] keys)
+    {
+        Result = result;
+        Summary = summary;
+        Keys = keys;
+    }
+
     /// <summary>
     /// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.
-    /// Least common set of fields in <see cref="Records"/>.
+    /// Least common set of fields in <see cref="Result" />.
     /// </summary>
     public string[] Keys { get; init; }
 
     /// <summary>
     /// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.
     /// All Records from query.
     /// </summary>
-    public IRecord[] Records { get; init; }
+    public T Result { get; init; }
 
     /// <summary>
     /// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.
     /// Query summary.
     /// </summary>
     public IResultSummary Summary { get; init; }
 
-    /// <inheritdoc />
-    public IEnumerator<IRecord> GetEnumerator()
+    /// <summary>Deconstructs the result into its constituent parts.</summary>
+    /// <param name="result">The result returned from the query.</param>
+    /// <param name="summary">The summary of the result.</param>
+    /// <param name="keys">The keys present in the result.</param>
+    public void Deconstruct(out T result, out IResultSummary summary, out string[] keys)
     {
-        return ((IEnumerable<IRecord>)Records).GetEnumerator();
+        keys = Keys;
+        result = Result;
+        summary = Summary;
     }
 
-    /// <inheritdoc />
-    IEnumerator IEnumerable.GetEnumerator()
+    /// <summary>Deconstructs the result into its constituent parts.</summary>
+    /// <param name="result">The result returned from the query.</param>
+    /// <param name="summary">The summary of the result.</param>
+    public void Deconstruct(out T result, out IResultSummary summary)
     {
-        return Records.GetEnumerator();
+        result = Result;
+        summary = Summary;
     }
-
-    /// <inheritdoc />
-    public int Count => Records.Length;
-
-    /// <inheritdoc />
-    public IRecord this[int index] => Records[index];
 }

Neo4j.Driver/Neo4j.Driver/Experimental/ExperimentalExtensions.cs

@@ -1,9 +1,9 @@
-// Copyright (c) 2002-2022 "Neo4j,"
+// Copyright (c) "Neo4j"
 // Neo4j Sweden AB [http://neo4j.com]
 // 
 // This file is part of Neo4j.
 // 
-// Licensed under the Apache License, Version 2.0 (the "License");
+// Licensed 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
 // 
@@ -15,66 +15,82 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+using System.Collections.Generic;
 using Neo4j.Driver.Experimental.FluentQueries;
 using Neo4j.Driver.Internal;
 
 namespace Neo4j.Driver.Experimental;
 
 /// <summary>
-/// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.<br/>
-/// This class provides access to experimental APIs on existing non-static classes.
+/// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.
+/// <br /> This class provides access to experimental APIs on existing non-static classes.
 /// </summary>
 public static class ExperimentalExtensions
 {
     /// <summary>
-    /// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.<br/>
-    /// Sets the <see cref="IBookmarkManager"/> for maintaining bookmarks for the lifetime of the session.
+    /// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.
+    /// <br /> Sets the <see cref="IBookmarkManager" /> for maintaining bookmarks for the lifetime of the session.
     /// </summary>
-    /// <param name="builder">This <see cref="SessionConfigBuilder"/> instance.</param>
-    /// <param name="bookmarkManager">An instance of <see cref="IBookmarkManager"/> to use in the session.</param>
-    /// <returns>this <see cref="SessionConfigBuilder"/> instance.</returns>
-    public static SessionConfigBuilder WithBookmarkManager(this SessionConfigBuilder builder, IBookmarkManager bookmarkManager)
+    /// <param name="builder">This <see cref="SessionConfigBuilder" /> instance.</param>
+    /// <param name="bookmarkManager">An instance of <see cref="IBookmarkManager" /> to use in the session.</param>
+    /// <returns>this <see cref="SessionConfigBuilder" /> instance.</returns>
+    public static SessionConfigBuilder WithBookmarkManager(
+        this SessionConfigBuilder builder,
+        IBookmarkManager bookmarkManager)
     {
         return builder.WithBookmarkManager(bookmarkManager);
     }
 
     /// <summary>
     /// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.
-    /// Gets an <see cref="IExecutableQuery"/> that can be used to configure and execute a query using
-    /// fluent method chaining.
+    /// Gets an <see cref="IExecutableQuery&lt;IRecord&gt;" /> that can be used to configure and execute a query using fluent
+    /// method chaining.
     /// </summary>
     /// <example>
     /// The following example configures and executes a simple query, then iterates over the results.
     /// <code language="cs">
-    /// var queryResult = await driver
-    ///     .ExecutableQuery("MATCH (m:Movie) WHERE m.released > $releaseYear RETURN m.title AS title")
-    ///     .WithParameters(new { releaseYear = 2005 })
-    ///     .ExecuteAsync();
-    ///
-    /// foreach(var record in queryResult)
-    /// {
-    ///     Console.WriteLine(record["title"].As&lt;string&gt;());
-    /// }
-    /// </code>
+    ///  var eagerResult = await driver
+    ///      .ExecutableQuery("MATCH (m:Movie) WHERE m.released > $releaseYear RETURN m.title AS title")
+    ///      .WithParameters(new { releaseYear = 2005 })
+    ///      .ExecuteAsync();
+    ///  <para></para>
+    ///  foreach(var record in eagerResult.Result)
+    ///  {
+    ///      Console.WriteLine(record["title"].As&lt;string&gt;());
+    ///  }
+    ///  </code>
+    /// <para></para>
+    /// The following example gets a single scalar value from a query.
+    /// <code>
+    ///  var born = await driver
+    ///      .ExecutableQuery("MATCH (p:Person WHERE p.name = $name) RETURN p.born AS born")
+    ///      .WithStreamProcessor(async stream => (await stream.Where(_ => true).FirstAsync())["born"].As&lt;int&gt;())
+    ///      .WithParameters(new Dictionary&lt;string, object&gt; { ["name"] = "Tom Hanks" })
+    ///      .ExecuteAsync();
+    ///  <para></para>
+    ///  Console.WriteLine($"Tom Hanks born {born.Result}");
+    ///  </code>
     /// </example>
     /// <param name="driver">The driver.</param>
     /// <param name="cypher">The cypher of the query.</param>
-    /// <returns>An <see cref="IExecutableQuery"/> that can be used to configure and execute a query using
-    /// fluent method chaining.</returns>
-    public static IExecutableQuery ExecutableQuery(this IDriver driver, string cypher)
+    /// <returns>
+    /// An <see cref="IExecutableQuery&lt;IRecord&gt;" /> that can be used to configure and execute a query using
+    /// fluent method chaining.
+    /// </returns>
+    public static IExecutableQuery<IReadOnlyList<IRecord>> ExecutableQuery(this IDriver driver, string cypher)
     {
-        return new ExecutableQuery((IInternalDriver)driver, cypher);
+        return ExecutableQuery<IReadOnlyList<IRecord>>.GetDefault((IInternalDriver)driver, cypher);
     }
-    
+
     /// <summary>
-    /// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.<br/>
-    /// Experimental: This method will be removed and replaced with a readonly property "BookmarkManager" on the
-    /// <see cref="SessionConfig"/> class.<br/>
-    /// Gets the configured experimental bookmark manager from this <see cref="SessionConfig"/> instance.
+    /// There is no guarantee that anything in Neo4j.Driver.Experimental namespace will be in a next minor version.
+    /// <br /> Experimental: This method will be removed and replaced with a readonly property "BookmarkManager" on the
+    /// <see cref="SessionConfig" /> class.<br /> Gets the configured experimental bookmark manager from this
+    /// <see cref="SessionConfig" /> instance.
     /// </summary>
-    /// <seealso cref="WithBookmarkManager"/>
-    /// <param name="config">This <see cref="SessionConfig"/> instance.</param>
-    /// <returns>This <see cref="SessionConfig"/>'s configured <see cref="IBookmarkManager"/> instance.</returns>
+    /// <seealso cref="WithBookmarkManager" />
+    /// <param name="config">This <see cref="SessionConfig" /> instance.</param>
+    /// <returns>This <see cref="SessionConfig" />'s configured <see cref="IBookmarkManager" /> instance.</returns>
     public static IBookmarkManager GetBookmarkManager(this SessionConfig config)
     {
         return config.BookmarkManager;

Neo4j.Driver/Neo4j.Driver/Experimental/FluentQueries/ExecutableQuery.cs

@@ -23,55 +23,77 @@
 
 namespace Neo4j.Driver.Experimental.FluentQueries;
 
-internal class ExecutableQuery : IExecutableQuery
+internal class ExecutableQuery<T> : IExecutableQuery<T>
 {
     private IInternalDriver _driver;
+    private Func<IAsyncEnumerable<IRecord>, ValueTask<T>> _streamProcessor;
     private Query _query;
     private QueryConfig _queryConfig;
 
-    public ExecutableQuery(IInternalDriver driver, string cypher)
+    public static ExecutableQuery<IReadOnlyList<IRecord>> GetDefault(IInternalDriver driver, string cypher)
     {
+        return new ExecutableQuery<IReadOnlyList<IRecord>>(new Query(cypher), driver, null, ToListAsync);
+    }
+
+    private ExecutableQuery(
+        Query query,
+        IInternalDriver driver,
+        QueryConfig queryConfig,
+        Func<IAsyncEnumerable<IRecord>, ValueTask<T>> streamProcessor)
+    {
+        _query = query;
         _driver = driver;
-        _query = new Query(cypher, new {});
+        _queryConfig = queryConfig;
+        _streamProcessor = streamProcessor;
+    }
+
+    private static async ValueTask<IReadOnlyList<T>> ToListAsync<T>(IAsyncEnumerable<T> enumerable)
+    {
+        var result = new List<T>();
+        await foreach (var item in enumerable)
+        {
+            result.Add(item);
+        }
+
+        return result;
     }
 
-    public IExecutableQuery WithConfig(QueryConfig config)
+    public IExecutableQuery<T> WithConfig(QueryConfig config)
     {
         _queryConfig = config;
         return this;
     }
 
-    public IExecutableQuery WithParameters(object parameters)
+    public IExecutableQuery<T> WithParameters(object parameters)
     {
         _query = new Query(_query.Text, parameters);
         return this;
     }
 
-    public IExecutableQuery WithParameters(Dictionary<string, object> parameters)
+    public IExecutableQuery<T> WithParameters(Dictionary<string, object> parameters)
     {
         _query = new Query(_query.Text, parameters);
         return this;
     }
 
+    public IExecutableQuery<TResult> WithStreamProcessor<TResult>(
+        Func<IAsyncEnumerable<IRecord>, ValueTask<TResult>> streamProcessor)
+    {
+        return new ExecutableQuery<TResult>(_query, _driver, _queryConfig, streamProcessor);
+    }
+
     // removing since behaviour is different to WithParameters, pending discussion
     // public IExecutableQuery WithParameter(string name, object value)
     // {
     //     _query.Parameters[name] = value;
     //     return this;
     // }
 
-    public Task<EagerResult> ExecuteAsync(CancellationToken cancellationToken = default)
-    {
-        return _driver.ExecuteQueryAsync(_query, _queryConfig, cancellationToken);
-    }
-
-    private Task<IReadOnlyList<T>> TransformRecordsAsync<T>(
-        Func<IRecord, T> transform,
-        CancellationToken cancellationToken = default)
+    public Task<EagerResult<T>> ExecuteAsync(CancellationToken cancellationToken = default)
     {
         return _driver.ExecuteQueryAsync(
             _query,
-            MapTransformer<T>.GetFactoryMethod(transform),
+            _streamProcessor,
             _queryConfig,
             cancellationToken);
     }