feat: add support for FluxRecord streaming

Sylvain2703 · Sylvain2703 · commit e50c11fbd049 · 2025-03-18T14:51:37.000+01:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,6 +4,9 @@
 
 ⚠️ Important Notice: Starting from this release, we won’t be listing every dependency change in our changelog. This helps us maintain the project faster and focus on important features for our InfluxDB client.
 
+### Features:
+1. [#705](https://github.com/influxdata/influxdb-client-csharp/pull/705): `QueryAsyncEnumerable()` now supports `FluxRecord` streaming.
+
 ### CI
 1. [#681](https://github.com/influxdata/influxdb-client-csharp/pull/681): Add build for `dotnet8`
 
diff --git a/Client.Core/Internal/AbstractQueryClient.cs b/Client.Core/Internal/AbstractQueryClient.cs
@@ -203,10 +203,11 @@ protected async IAsyncEnumerable<T> QueryEnumerable<T>(
                 )
             };
 
-            var stream = await RestClient.DownloadStreamAsync(query, 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(new StreamReader(stream), cancellationToken)
-                               .ConfigureAwait(false))
+                               .ParseFluxResponseAsync(sr, cancellationToken).ConfigureAwait(false))
                 if (!(record is null))
                 {
                     yield return convert.Invoke(record);
diff --git a/Client.Test/QueryApiTest.cs b/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]
diff --git a/Client/InvokableScriptsApi.cs b/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)
diff --git a/Client/QueryApi.cs b/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>
@@ -801,9 +838,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);
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -203,10 +203,11 @@ protected async IAsyncEnumerable<T> QueryEnumerable<T>(`
`203`	`203`	`)`
`204`	`204`	`};`
`205`	`205`
`206`		`- var stream = await RestClient.DownloadStreamAsync(query, cancellationToken).ConfigureAwait(false);`
	`206`	`+ using var stream = await RestClient.DownloadStreamAsync(query, cancellationToken).ConfigureAwait(false);`
	`207`	`+ using var sr = new StreamReader(stream);`
	`208`	`+`
`207`	`209`	`await foreach (var (_, record) in _csvParser`
`208`		`- .ParseFluxResponseAsync(new StreamReader(stream), cancellationToken)`
`209`		`- .ConfigureAwait(false))`
	`210`	`+ .ParseFluxResponseAsync(sr, cancellationToken).ConfigureAwait(false))`
`210`	`211`	`if (!(record is null))`
`211`	`212`	`{`
`212`	`213`	`yield return convert.Invoke(record);`