Implement changes to Cursor API to include the stream transaction id …

arangodb-net-standard.Test/CollectionApi/CollectionApiClientTest.cs

@@ -1,13 +1,13 @@
-using ArangoDBNetStandard;
+using System.Collections.Generic;
+using System.Linq;
+using System.Net;
+using System.Threading.Tasks;
+using ArangoDBNetStandard;
 using ArangoDBNetStandard.CollectionApi;
 using ArangoDBNetStandard.CollectionApi.Models;
 using ArangoDBNetStandard.DocumentApi.Models;
 using ArangoDBNetStandard.Transport;
 using Moq;
-using System.Collections.Generic;
-using System.Linq;
-using System.Net;
-using System.Threading.Tasks;
 using Xunit;
 
 namespace ArangoDBNetStandardTest.CollectionApi
@@ -160,8 +160,9 @@ public async Task PostCollectionAsync_ShouldUseQueryParameter()
 
             mockTransport.Setup(x => x.PostAsync(
                 It.IsAny<string>(),
-                It.IsAny<byte[]>()))
-                .Returns((string uri, byte[] content) =>
+                It.IsAny<byte[]>(),
+                It.IsAny<WebHeaderCollection>()))
+                .Returns((string uri, byte[] content, WebHeaderCollection webHeaderCollection) =>
                 {
                     requestUri = uri;
                     return Task.FromResult(mockResponse.Object);

arangodb-net-standard.Test/CursorApi/CursorApiClientTest.cs

@@ -1,12 +1,12 @@
-using ArangoDBNetStandard;
-using ArangoDBNetStandard.CursorApi;
-using ArangoDBNetStandard.CursorApi.Models;
-using System.Collections.Generic;
+using System.Collections.Generic;
 using System.IO;
 using System.Linq;
 using System.Net;
 using System.Text;
 using System.Threading.Tasks;
+using ArangoDBNetStandard;
+using ArangoDBNetStandard.CursorApi;
+using ArangoDBNetStandard.CursorApi.Models;
 using ArangoDBNetStandard.Serialization;
 using ArangoDBNetStandard.Transport;
 using Moq;
@@ -187,7 +187,8 @@ public async Task PostCursorAsync_ShouldThrow_WhenErrorDeserializationFailed()
 
             mockTransport.Setup(x => x.PostAsync(
                     It.IsAny<string>(),
-                    It.IsAny<byte[]>()))
+                    It.IsAny<byte[]>(),
+                    It.IsAny<WebHeaderCollection>()))
                 .Returns(Task.FromResult(mockResponse.Object));
 
             var cursorApi = new CursorApiClient(mockTransport.Object);
@@ -215,6 +216,58 @@ public async Task PostCursorAsync_ShouldThrowException_WhenResponseDeserializati
             Assert.NotNull(ex.InnerException);
         }
 
+        [Fact]
+        public async Task PostCursorAsync_ShouldUseHeaderProperties()
+        {
+            // Mock the IApiClientTransport.
+            var mockTransport = new Mock<IApiClientTransport>();
+
+            // Mock the IApiClientResponse.
+            var mockResponse = new Mock<IApiClientResponse>();
+
+            // Mock the IApiClientResponseContent.
+            var mockResponseContent = new Mock<IApiClientResponseContent>();
+
+            // Setup the mocked api client response.
+            mockResponse.Setup(x => x.Content)
+                .Returns(mockResponseContent.Object);
+            mockResponse.Setup(x => x.IsSuccessStatusCode)
+                .Returns(true);
+
+            // Setup the mocked api client transport.
+            WebHeaderCollection requestHeader = null;
+            mockTransport.Setup(x => x.PostAsync(
+                It.IsAny<string>(),
+                It.IsAny<byte[]>(),
+                It.IsAny<WebHeaderCollection>()))
+                .Returns((string uri, byte[] content, WebHeaderCollection webHeaderCollection) =>
+                {
+                    requestHeader = webHeaderCollection;
+                    return Task.FromResult(mockResponse.Object);
+                });
+
+            string transactionHeaderKey = "x-arango-trx-id";
+            string dummyTransactionId = "dummy transaction Id";
+
+            // Call the method to create the cursor.
+            var apiClient = new CursorApiClient(mockTransport.Object);
+            await apiClient.PostCursorAsync<MyModel>(
+                 new PostCursorBody
+                 {
+                     Query = "FOR doc IN [{ myProperty: CONCAT('This is a ', @testString) }] LIMIT 1 RETURN doc",
+                     BindVars = new Dictionary<string, object> { ["testString"] = "robbery" }
+                 },
+                 new CursorHeaderProperties
+                 {
+                     TransactionId = dummyTransactionId
+                 });
+
+            // Check that the header and values are there.
+            Assert.NotNull(requestHeader);
+            Assert.Contains(transactionHeaderKey, requestHeader.AllKeys);
+            Assert.Equal(dummyTransactionId, requestHeader.Get(transactionHeaderKey));
+        }
+
         [Fact]
         public async Task PutCursorAsync_ShouldSucceed()
         {

arangodb-net-standard/CursorApi/CursorApiClient.cs

@@ -1,9 +1,9 @@
-using ArangoDBNetStandard.CursorApi.Models;
-using ArangoDBNetStandard.Serialization;
-using ArangoDBNetStandard.Transport;
-using System.Collections.Generic;
+using System.Collections.Generic;
 using System.Net;
 using System.Threading.Tasks;
+using ArangoDBNetStandard.CursorApi.Models;
+using ArangoDBNetStandard.Serialization;
+using ArangoDBNetStandard.Transport;
 
 namespace ArangoDBNetStandard.CursorApi
 {
@@ -45,6 +45,25 @@ public CursorApiClient(IApiClientTransport client, IApiClientSerialization seria
             _client = client;
         }
 
+        /// <summary>
+        /// Method to get the header collection.
+        /// </summary>
+        /// <param name="headerProperties">The <see cref="CursorHeaderProperties"/> values.</param>
+        /// <returns><see cref="WebHeaderCollection"/> values.</returns>
+        protected virtual WebHeaderCollection GetHeaderCollection(CursorHeaderProperties headerProperties)
+        {
+            var headerCollection = new WebHeaderCollection();
+            if (headerProperties != null)
+            {
+                if (!string.IsNullOrWhiteSpace(headerProperties.TransactionId))
+                {
+                    headerCollection.Add("x-arango-trx-id", headerProperties.TransactionId);
+                }
+            }
+
+            return headerCollection;
+        }
+
         /// <summary>
         /// Execute an AQL query, creating a cursor which can be used to page query results.
         /// </summary>
@@ -57,6 +76,7 @@ public CursorApiClient(IApiClientTransport client, IApiClientSerialization seria
         /// <param name="cache"></param>
         /// <param name="memoryLimit"></param>
         /// <param name="ttl"></param>
+        /// <param name="transactionId">Optional. The stream transaction Id.</param>
         /// <returns></returns>
         public virtual async Task<CursorResponse<T>> PostCursorAsync<T>(
                 string query,
@@ -66,37 +86,48 @@ public virtual async Task<CursorResponse<T>> PostCursorAsync<T>(
                 long? batchSize = null,
                 bool? cache = null,
                 long? memoryLimit = null,
-                int? ttl = null)
+                int? ttl = null,
+                string transactionId = null)
         {
-            return await PostCursorAsync<T>(new PostCursorBody
+            var headerProperties = new CursorHeaderProperties();
+            if (!string.IsNullOrWhiteSpace(transactionId))
             {
-                Query = query,
-                BindVars = bindVars,
-                Options = options,
-                Count = count,
-                BatchSize = batchSize,
-                Cache = cache,
-                MemoryLimit = memoryLimit,
-                Ttl = ttl
-            }).ConfigureAwait(false);
+                headerProperties.TransactionId = transactionId;
+            }
+
+            return await PostCursorAsync<T>(
+                new PostCursorBody
+                {
+                    Query = query,
+                    BindVars = bindVars,
+                    Options = options,
+                    Count = count,
+                    BatchSize = batchSize,
+                    Cache = cache,
+                    MemoryLimit = memoryLimit,
+                    Ttl = ttl
+                }, headerProperties).ConfigureAwait(false);
         }
 
         /// <summary>
         /// Execute an AQL query, creating a cursor which can be used to page query results.
         /// </summary>
         /// <param name="postCursorBody">Object encapsulating options and parameters of the query.</param>
+        /// <param name="headerProperties">Optional. Additional Header properties.</param>
         /// <returns></returns>
         public virtual async Task<CursorResponse<T>> PostCursorAsync<T>(
-            PostCursorBody postCursorBody)
+            PostCursorBody postCursorBody, CursorHeaderProperties headerProperties = null)
         {
             var content = GetContent(postCursorBody, new ApiClientSerializationOptions(true, true));
-            using (var response = await _client.PostAsync(_cursorApiPath, content).ConfigureAwait(false))
+            var headerCollection = GetHeaderCollection(headerProperties);
+            using (var response = await _client.PostAsync(_cursorApiPath, content, headerCollection).ConfigureAwait(false))
             {
                 if (response.IsSuccessStatusCode)
                 {
                     var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
                     return DeserializeJsonFromStream<CursorResponse<T>>(stream);
                 }
+
                 throw await GetApiErrorException(response).ConfigureAwait(false);
             }
         }
@@ -116,6 +147,7 @@ public virtual async Task<DeleteCursorResponse> DeleteCursorAsync(string cursorI
                     var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
                     return DeserializeJsonFromStream<DeleteCursorResponse>(stream);
                 }
+
                 throw await GetApiErrorException(response).ConfigureAwait(false);
             }
         }
@@ -136,6 +168,7 @@ public virtual async Task<PutCursorResponse<T>> PutCursorAsync<T>(string cursorI
                     var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false);
                     return DeserializeJsonFromStream<PutCursorResponse<T>>(stream);
                 }
+
                 throw await GetApiErrorException(response).ConfigureAwait(false);
             }
         }

arangodb-net-standard/CursorApi/ICursorApiClient.cs

@@ -1,7 +1,6 @@
-using ArangoDBNetStandard.CursorApi.Models;
-using ArangoDBNetStandard.Serialization;
-using System.Collections.Generic;
+using System.Collections.Generic;
 using System.Threading.Tasks;
+using ArangoDBNetStandard.CursorApi.Models;
 
 namespace ArangoDBNetStandard.CursorApi
 {
@@ -22,6 +21,7 @@ public interface ICursorApiClient
         /// <param name="cache"></param>
         /// <param name="memoryLimit"></param>
         /// <param name="ttl"></param>
+        /// <param name="transactionId">Optional. The stream transaction Id.</param>
         /// <returns></returns>
         Task<CursorResponse<T>> PostCursorAsync<T>(
                 string query,
@@ -31,15 +31,17 @@ Task<CursorResponse<T>> PostCursorAsync<T>(
                 long? batchSize = null,
                 bool? cache = null,
                 long? memoryLimit = null,
-                int? ttl = null);
+                int? ttl = null,
+                string transactionId = null);
 
         /// <summary>
         /// Execute an AQL query, creating a cursor which can be used to page query results.
         /// </summary>
         /// <param name="postCursorBody">Object encapsulating options and parameters of the query.</param>
+        /// <param name="headerProperties">Optional. Additional Header properties.</param>
         /// <returns></returns>
         Task<CursorResponse<T>> PostCursorAsync<T>(
-            PostCursorBody postCursorBody);
+            PostCursorBody postCursorBody, CursorHeaderProperties headerProperties);
 
         /// <summary>
         /// Deletes an existing cursor and frees the resources associated with it.

arangodb-net-standard/CursorApi/Models/CursorHeaderProperties.cs

@@ -0,0 +1,13 @@
+namespace ArangoDBNetStandard.CursorApi.Models
+{
+    /// <summary>
+    /// Class representing the additional header properties for Cursor Api.
+    /// </summary>
+    public class CursorHeaderProperties
+    {
+        /// <summary>
+        /// Gets or sets the stream transaction Id.
+        /// </summary>
+        public string TransactionId { get; set; }
+    }
+}

arangodb-net-standard/Transport/Http/HttpApiTransport.cs

@@ -40,6 +40,22 @@ public HttpApiTransport(HttpClient client, HttpContentType contentType)
             _contentType = contentType;
         }
 
+        /// <summary>
+        /// Method to apply the additional headers.
+        /// </summary>
+        /// <param name="webHeaderCollection">Object containing a dictionary of Header keys and values.</param>
+        /// <param name="headers">The header to update.</param>
+        private static void ApplyHeaders(WebHeaderCollection webHeaderCollection, HttpHeaders headers)
+        {
+            if (webHeaderCollection != null)
+            {
+                foreach (var key in webHeaderCollection.AllKeys)
+                {
+                    headers.Add(key, webHeaderCollection[key]);
+                }
+            }
+        }
+
         /// <summary>
         /// Get an instance of <see cref="HttpApiTransport"/> that uses no authentication.
         /// </summary>
@@ -197,11 +213,14 @@ public async Task<IApiClientResponse> DeleteAsync(string requestUri, byte[] cont
         /// </summary>
         /// <param name="requestUri"></param>
         /// <param name="content">The content of the request, must not be null.</param>
+        /// <param name="webHeaderCollection">Object containing a dictionary of Header keys and values.</param>
         /// <returns></returns>
-        public async Task<IApiClientResponse> PostAsync(string requestUri, byte[] content)
+        public async Task<IApiClientResponse> PostAsync(
+            string requestUri, byte[] content, WebHeaderCollection webHeaderCollection = null)
         {
             var httpContent = new ByteArrayContent(content);
             httpContent.Headers.ContentType = new MediaTypeHeaderValue(_contentTypeMap[_contentType]);
+            ApplyHeaders(webHeaderCollection, httpContent.Headers);
             var response = await _client.PostAsync(requestUri, httpContent).ConfigureAwait(false);
             return new HttpApiClientResponse(response);
         }
@@ -212,7 +231,8 @@ public async Task<IApiClientResponse> PostAsync(string requestUri, byte[] conten
         /// <param name="requestUri"></param>
         /// <param name="content">The content of the request, must not be null.</param>
         /// <returns></returns>
-        public async Task<IApiClientResponse> PutAsync(string requestUri, byte[] content)
+        public async Task<IApiClientResponse> PutAsync(
+            string requestUri, byte[] content)
         {
             var httpContent = new ByteArrayContent(content);
             httpContent.Headers.ContentType = new MediaTypeHeaderValue(_contentTypeMap[_contentType]);

arangodb-net-standard/Transport/IApiClientTransport.cs

@@ -14,8 +14,10 @@ public interface IApiClientTransport : IDisposable
         /// </summary>
         /// <param name="requestUri"></param>
         /// <param name="content"></param>
+        /// <param name="webHeaderCollection">Object containing a dictionary of Header keys and values.</param>
         /// <returns></returns>
-        Task<IApiClientResponse> PostAsync(string requestUri, byte[] content);
+        Task<IApiClientResponse> PostAsync(
+            string requestUri, byte[] content, WebHeaderCollection webHeaderCollection = null);
 
         /// <summary>
         /// Send a DELETE request.
@@ -38,7 +40,8 @@ public interface IApiClientTransport : IDisposable
         /// <param name="requestUri"></param>
         /// <param name="content"></param>
         /// <returns></returns>
-        Task<IApiClientResponse> PutAsync(string requestUri, byte[] content);
+        Task<IApiClientResponse> PutAsync(
+            string requestUri, byte[] content);
 
         /// <summary>
         /// Send a GET request.