feat: add support for FluxRecord streaming

Author: Sylvain2703

Client.Core/Internal/AbstractQueryClient.cs

@@ -203,14 +203,14 @@ protected async IAsyncEnumerable<T> QueryEnumerable<T>(
                 )
             };
 
-            var stream = await RestClient.DownloadStreamAsync(query, cancellationToken).ConfigureAwait(false);
-            await foreach (var (_, record) in _csvParser
-                               .ParseFluxResponseAsync(new StreamReader(stream), cancellationToken)
-                               .ConfigureAwait(false))
+            using var stream = await RestClient.DownloadStreamAsync(query, cancellationToken).ConfigureAwait(false);
+            using var sr = new StreamReader(stream);
+
+            await foreach (var (_, record) in _csvParser.ParseFluxResponseAsync(sr, cancellationToken).ConfigureAwait(false))
+            {
                 if (!(record is null))
-                {
                     yield return convert.Invoke(record);
-                }
+            }
         }
 
         protected abstract void BeforeIntercept(RestRequest query);

Client.Test/QueryApiTest.cs

@@ -98,6 +98,31 @@ public async Task QueryAsyncEnumerable()
             await foreach (var item in measurements.ConfigureAwait(false)) list.Add(item);
 
             Assert.AreEqual(2, list.Count);
+            Assert.AreEqual("mem", list[0].Measurement);
+            Assert.AreEqual("mem", list[1].Measurement);
+            Assert.AreEqual(12.25, list[0].Value);
+            Assert.AreEqual(13.00, list[1].Value);
+        }
+
+        [Test]
+        public async Task QueryAsyncEnumerableOfFluxRecords()
+        {
+            MockServer
+                .Given(Request.Create().WithPath("/api/v2/query").UsingPost())
+                .RespondWith(CreateResponse(Data));
+
+            var measurements = _queryApi.QueryAsyncEnumerable(
+                new Query(null, "from(...)"),
+                "my-org");
+
+            var list = new List<FluxRecord>();
+            await foreach (var item in measurements.ConfigureAwait(false)) list.Add(item);
+
+            Assert.AreEqual(2, list.Count);
+            Assert.AreEqual("mem", list[0].GetMeasurement());
+            Assert.AreEqual("mem", list[1].GetMeasurement());
+            Assert.AreEqual(12.25, list[0].GetValue());
+            Assert.AreEqual(13.00, list[1].GetValue());
         }
 
         [Test]

Client/InvokableScriptsApi.cs

@@ -1,7 +1,5 @@
 using System;
 using System.Collections.Generic;
-using System.Net.Http;
-using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using InfluxDB.Client.Api.Domain;
@@ -248,15 +246,13 @@ await QueryRaw(CreateRequest(scriptId, bindParams), Consumer, ErrorConsumer, Emp
         /// <param name="bindParams">Represent key/value pairs parameters to be injected into script</param>
         /// <param name="cancellationToken">Cancellation token</param>
         /// <returns>stream of FluxRecord</returns>
-        public async IAsyncEnumerable<FluxRecord> InvokeScriptEnumerableAsync(string scriptId,
+        public IAsyncEnumerable<FluxRecord> InvokeScriptEnumerableAsync(string scriptId,
             Dictionary<string, object> bindParams = default,
-            [EnumeratorCancellation] CancellationToken cancellationToken = default)
+            CancellationToken cancellationToken = default)
         {
             var requestMessage = CreateRequest(scriptId, bindParams);
 
-            await foreach (var record in QueryEnumerable(requestMessage, it => it, cancellationToken)
-                               .ConfigureAwait(false))
-                yield return record;
+            return QueryEnumerable(requestMessage, r => r, cancellationToken);
         }
 
         /// <summary>
@@ -266,15 +262,13 @@ public async IAsyncEnumerable<FluxRecord> InvokeScriptEnumerableAsync(string scr
         /// <param name="bindParams">Represent key/value pairs parameters to be injected into script</param>
         /// <param name="cancellationToken">Cancellation token</param>
         /// <returns>stream of Measurement</returns>
-        public async IAsyncEnumerable<T> InvokeScriptMeasurementsEnumerableAsync<T>(string scriptId,
+        public IAsyncEnumerable<T> InvokeScriptMeasurementsEnumerableAsync<T>(string scriptId,
             Dictionary<string, object> bindParams = default,
-            [EnumeratorCancellation] CancellationToken cancellationToken = default)
+            CancellationToken cancellationToken = default)
         {
             var requestMessage = CreateRequest(scriptId, bindParams);
 
-            await foreach (var record in QueryEnumerable(requestMessage, it => Mapper.ConvertToEntity<T>(it),
-                               cancellationToken).ConfigureAwait(false))
-                yield return record;
+            return QueryEnumerable(requestMessage, r => Mapper.ConvertToEntity<T>(r), cancellationToken);
         }
 
         protected override void BeforeIntercept(RestRequest request)

Client/QueryApi.cs

@@ -1,7 +1,6 @@
 using System;
 using System.Collections.Generic;
 using System.Net.Http;
-using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
 using InfluxDB.Client.Api.Domain;
@@ -236,6 +235,28 @@ IAsyncEnumerable<T> QueryAsyncEnumerable<T>(string query, string org = null,
         IAsyncEnumerable<T> QueryAsyncEnumerable<T>(Query query, string org = null,
             CancellationToken cancellationToken = default);
 
+        /// <summary>
+        /// Executes the Flux query against the InfluxDB 2.x and returns
+        /// an asynchronous enumerable of <see cref="FluxRecord"/>.
+        /// </summary>
+        /// <param name="query">the flux query to execute</param>
+        /// <param name="org">specifies the source organization. If the org is not specified then is used config from <see cref="InfluxDBClientOptions.Org" />.</param>
+        /// <param name="cancellationToken">cancellation token</param>
+        /// <returns>An asynchronous enumerable of <see cref="FluxRecord"/>.</returns>
+        IAsyncEnumerable<FluxRecord> QueryAsyncEnumerable(string query, string org = null,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Executes the Flux query against the InfluxDB 2.x and returns
+        /// an asynchronous enumerable of <see cref="FluxRecord"/>.
+        /// </summary>
+        /// <param name="query">the flux query to execute</param>
+        /// <param name="org">specifies the source organization. If the org is not specified then is used config from <see cref="InfluxDBClientOptions.Org" />.</param>
+        /// <param name="cancellationToken">cancellation token</param>
+        /// <returns>An asynchronous enumerable of <see cref="FluxRecord"/>.</returns>
+        IAsyncEnumerable<FluxRecord> QueryAsyncEnumerable(Query query, string org = null,
+            CancellationToken cancellationToken = default);
+
         /// <summary>
         /// Executes the Flux query against the InfluxDB and synchronously map whole response to <see cref="string"/> result.
         /// 
@@ -442,16 +463,10 @@ await QueryAsync(query, consumer, ErrorConsumer, EmptyAction, org, cancellationT
         /// <param name="cancellationToken">cancellation token</param>
         /// <typeparam name="T">the type of measurement</typeparam>
         /// <returns>Measurements which are matched the query</returns>
-        public async IAsyncEnumerable<T> QueryAsyncEnumerable<T>(string query, string org = null,
-            [EnumeratorCancellation] CancellationToken cancellationToken = default)
+        public IAsyncEnumerable<T> QueryAsyncEnumerable<T>(string query, string org = null,
+            CancellationToken cancellationToken = default)
         {
-            Arguments.CheckNonEmptyString(query, nameof(query));
-
-            var requestMessage = CreateRequest(CreateQuery(query), org);
-
-            await foreach (var record in QueryEnumerable(requestMessage, it => Mapper.ConvertToEntity<T>(it),
-                               cancellationToken).ConfigureAwait(false))
-                yield return record;
+            return QueryAsyncEnumerable<T>(CreateQuery(query), org, cancellationToken);
         }
 
         /// <summary>
@@ -463,16 +478,40 @@ public async IAsyncEnumerable<T> QueryAsyncEnumerable<T>(string query, string or
         /// <param name="cancellationToken">cancellation token</param>
         /// <typeparam name="T">the type of measurement</typeparam>
         /// <returns>Measurements which are matched the query</returns>
-        public async IAsyncEnumerable<T> QueryAsyncEnumerable<T>(Query query, string org = null,
-            [EnumeratorCancellation] CancellationToken cancellationToken = default)
+        public IAsyncEnumerable<T> QueryAsyncEnumerable<T>(Query query, string org = null,
+            CancellationToken cancellationToken = default)
         {
-            Arguments.CheckNotNull(query, nameof(query));
+            var request = CreateRequest(query, org);
+            return QueryEnumerable(request, r => Mapper.ConvertToEntity<T>(r), cancellationToken);
+        }
 
-            var requestMessage = CreateRequest(query, org);
+        /// <summary>
+        /// Executes the Flux query against the InfluxDB 2.x and returns
+        /// an asynchronous enumerable of <see cref="FluxRecord"/>.
+        /// </summary>
+        /// <param name="query">the flux query to execute</param>
+        /// <param name="org">specifies the source organization. If the org is not specified then is used config from <see cref="InfluxDBClientOptions.Org" />.</param>
+        /// <param name="cancellationToken">cancellation token</param>
+        /// <returns>An asynchronous enumerable of <see cref="FluxRecord"/>.</returns>
+        public IAsyncEnumerable<FluxRecord> QueryAsyncEnumerable(string query, string org = null,
+            CancellationToken cancellationToken = default)
+        {
+            return QueryAsyncEnumerable(CreateQuery(query), org, cancellationToken);
+        }
 
-            await foreach (var record in QueryEnumerable(requestMessage, it => Mapper.ConvertToEntity<T>(it),
-                               cancellationToken).ConfigureAwait(false))
-                yield return record;
+        /// <summary>
+        /// Executes the Flux query against the InfluxDB 2.x and returns
+        /// an asynchronous enumerable of <see cref="FluxRecord"/>.
+        /// </summary>
+        /// <param name="query">the flux query to execute</param>
+        /// <param name="org">specifies the source organization. If the org is not specified then is used config from <see cref="InfluxDBClientOptions.Org" />.</param>
+        /// <param name="cancellationToken">cancellation token</param>
+        /// <returns>An asynchronous enumerable of <see cref="FluxRecord"/>.</returns>
+        public IAsyncEnumerable<FluxRecord> QueryAsyncEnumerable(Query query, string org = null,
+            CancellationToken cancellationToken = default)
+        {
+            var request = CreateRequest(query, org);
+            return QueryEnumerable(request, r => r, cancellationToken);
         }
 
         /// <summary>
@@ -494,8 +533,7 @@ public Task QueryAsync(string query, Action<FluxRecord> onNext, Action<Exception
 
             var consumer = new FluxResponseConsumerRecord(onNext);
 
-            return QueryAsync(CreateQuery(query, Dialect), consumer, onError, onComplete, org,
-                cancellationToken);
+            return QueryAsync(CreateQuery(query, Dialect), consumer, onError, onComplete, org, cancellationToken);
         }
 
         /// <summary>
@@ -540,8 +578,7 @@ public Task QueryAsync<T>(string query, Action<T> onNext, Action<Exception> onEr
 
             var consumer = new FluxResponseConsumerPoco<T>(onNext, Mapper);
 
-            return QueryAsync(CreateQuery(query, Dialect), consumer, onError, onComplete, org,
-                cancellationToken);
+            return QueryAsync(CreateQuery(query, Dialect), consumer, onError, onComplete, org, cancellationToken);
         }
 
         /// <summary>
@@ -779,8 +816,7 @@ private Task QueryAsync(Query query, FluxCsvParser.IFluxResponseConsumer consume
 
             var requestMessage = CreateRequest(query, org);
 
-            return Query(requestMessage, consumer, onError ?? ErrorConsumer, onComplete ?? EmptyAction,
-                cancellationToken);
+            return Query(requestMessage, consumer, onError ?? ErrorConsumer, onComplete ?? EmptyAction, cancellationToken);
         }
 
         private RestRequest CreateRequest(Query query, string org = null)
@@ -801,9 +837,7 @@ internal static Query CreateQuery(string query, Dialect dialect = null)
         {
             Arguments.CheckNonEmptyString(query, nameof(query));
 
-            var created = new Query(query: query, dialect: dialect ?? Dialect);
-
-            return created;
+            return new Query(query: query, dialect: dialect ?? Dialect);
         }
     }
 }