[ADR] RPC streaming design doc

doc/reference/error-model.md

@@ -151,6 +151,7 @@
     ExecutionException,
     MqttError,
     UnsupportedVersion,
+    Canceled,
 }
 ```
 
@@ -263,6 +264,7 @@
     EXECUTION_EXCEPTION,
     MQTT_ERROR,
     UNSUPPORTED_VERSION,
+    CANCELED,
 }
 ```
 
@@ -327,6 +329,7 @@
     ExecutionException,
     MqttError,
     UnsupportedVersion,
+    Canceled,
 }
 ```
 
@@ -399,6 +402,7 @@
     ExecutionError
     MqttError
     UnsupportedVersion
+    Canceled
 }
 ```
 
@@ -423,7 +427,7 @@
 }
 ```
 
 The `AkriMqttError` struct must provide an `Error()` method, but the detais are omitted from this document:
 
 ```go
 func (err AkriMqttError) Error() string {
@@ -454,6 +458,7 @@
     EXECUTION_EXCEPTION = 10
     MQTT_ERROR = 11
     UNSUPPORTED_VERSION = 12
+    CANCELED = 13
 ```
 
 The Akri.Mqtt error type is defined as follows:
@@ -569,6 +574,7 @@
 | 400 | Bad Request | false | no | | invalid payload |
 | 408 | Request Timeout | false | yes | yes | timeout |
 | 415 | Unsupported Media Type | false | yes | yes | invalid header |
+| 452 | Request Cancelled | false | no | no | canceled |
 | 500 | Internal Server Error | false | no | | unknown error |
 | 500 | Internal Server Error | false | yes | | internal logic error |
 | 500 | Internal Server Error | true | maybe | | execution error |

dotnet/src/Azure.Iot.Operations.Protocol/AkriSystemProperties.cs

@@ -1,4 +1,4 @@
-// Copyright (c) Microsoft Corporation.
+// Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
 using System;

dotnet/src/Azure.Iot.Operations.Protocol/RPC/ExtendedResponse.cs

@@ -1,4 +1,4 @@
-// Copyright (c) Microsoft Corporation.
+// Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
 using System;

dotnet/src/Azure.Iot.Operations.Protocol/Streaming/IStreamContext.cs

@@ -0,0 +1,76 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Azure.Iot.Operations.Protocol.Streaming
+{
+    /// <summary>
+    /// A stream of requests or responses that can be gracefully ended or canceled (with confirmation) at any time.
+    /// </summary>
+    /// <typeparam name="T">The type of the payload of the request/response stream</typeparam>
+    public interface IStreamContext<T>
+        where T : class
+    {
+        /// <summary>
+        /// The asynchronously readable entries in the stream
+        /// </summary>
+        IAsyncEnumerable<T> Entries { get; set; }
+
+        /// <summary>
+        /// Cancel this RPC streaming exchange.
+        /// </summary>
+        /// <param name="userData">
+        /// The optional user properties to include in this cancellation request. the receiving side of this cancellation request
+        /// will be given these properties alongside the notification that the streaming exchange has been canceled.
+        /// </param>
+        /// <param name="cancellationToken">Cancellation token to wait for confirmation from the receiving side that the cancellation succeeded.</param>
+        /// <remarks>
+        /// When called by the invoker, the executor will be notified about this cancellation and the executor will attempt
+        /// to stop any user-defined handling of the streaming request. When called by the executor, the invoker will be notified
+        /// and will cease sending requests and will throw an <see cref="AkriMqttException"/> with <see cref="AkriMqttException.Kind"/>
+        /// of <see cref="AkriMqttErrorKind.Cancellation"/>.
+        /// 
+        /// This method may be called by the streaming invoker or executor at any time. For instance, if the request stream
+        /// stalls unexpectedly, the executor can call this method to notify the invoker to stop sending requests.
+        /// Additionally, the invoker can call this method if its response stream has stalled unexpectedly.
+        /// </remarks>
+        Task CancelAsync(Dictionary<string, string>? userData = null, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// The token that tracks if the streaming exchange has been cancelled by the other party and/or timed out.
+        /// </summary>
+        /// <remarks>
+        /// For instance, if the invoker side cancels the streaming exchange, the executor side callback's <see cref="IStreamContext{T}.CancellationToken"/>
+        /// will be triggered. If the executor side cancels the streaming exchange, the invoker side's returned <see cref="IStreamContext{T}.CancellationToken"/>
+        /// will be triggered.
+        ///
+        /// To see if this was triggered because the stream exchange was cancelled, see <see cref="IsCanceled"/>. To see if it was triggered because
+        /// the stream exchange timed out, see <see cref="HasTimedOut"/>.
+        /// </remarks>
+        CancellationToken CancellationToken { get; }
+
+        /// <summary>
+        /// Get the user properties associated with a cancellation request started with <see cref="CancelAsync(Dictionary{string, string}?, CancellationToken)"/>.
+        /// </summary>
+        /// <returns>The user properties associated with a cancellation request</returns>
+        /// <remarks>
+        /// If the stream has not been cancelled, this will return null. If the stream has been cancelled, but no user properties were
+        /// provided in that cancellation request, this will return null.
+        /// </remarks>
+        Dictionary<string, string>? GetCancellationRequestUserProperties();
+
+        /// <summary>
+        /// True if this stream exchange has timed out. If a stream has timed out, <see cref="CancellationToken"/> will trigger as well.
+        /// </summary>
+        bool HasTimedOut { get; internal set; }
+
+        /// <summary>
+        /// True if this stream exchange has been canceled by the other party. If a stream has been cancelled, <see cref="CancellationToken"/> will trigger as well.
+        /// </summary>
+        bool IsCanceled { get; internal set; }
+    }
+}

dotnet/src/Azure.Iot.Operations.Protocol/Streaming/ReceivedStreamingExtendedRequest.cs

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using System.Threading.Tasks;
+
+namespace Azure.Iot.Operations.Protocol.Streaming
+{
+    /// <summary>
+    /// The payload and metadata associated with a single request in a request stream.
+    /// </summary>
+    /// <typeparam name="TReq">The type of the payload of the request</typeparam>
+    public class ReceivedStreamingExtendedRequest<TReq> : StreamingExtendedRequest<TReq>
+        where TReq : class
+    {
+        private readonly Task _acknowledgementFunc;
+
+        internal ReceivedStreamingExtendedRequest(TReq request, StreamMessageMetadata metadata, Task acknowledgementFunc)
+            : base(request, metadata)
+        {
+            _acknowledgementFunc = acknowledgementFunc;
+        }
+
+        public async Task AcknowledgeAsync()
+        {
+            await _acknowledgementFunc;
+        }
+    }
+}

dotnet/src/Azure.Iot.Operations.Protocol/Streaming/ReceivedStreamingExtendedResponse.cs

@@ -0,0 +1,28 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System.Threading.Tasks;
+
+namespace Azure.Iot.Operations.Protocol.Streaming
+{
+    /// <summary>
+    /// The payload and metadata associated with a single response in a response stream.
+    /// </summary>
+    /// <typeparam name="TResp">The type of the payload of the response</typeparam>
+    public class ReceivedStreamingExtendedResponse<TResp> : StreamingExtendedResponse<TResp>
+        where TResp : class
+    {
+        private readonly Task _acknowledgementFunc;
+
+        internal ReceivedStreamingExtendedResponse(TResp response, StreamMessageMetadata metadata, Task acknowledgementFunc)
+            : base(response, metadata)
+        {
+            _acknowledgementFunc = acknowledgementFunc;
+        }
+
+        public async Task AcknowledgeAsync()
+        {
+            await _acknowledgementFunc;
+        }
+    }
+}

dotnet/src/Azure.Iot.Operations.Protocol/Streaming/RequestStreamMetadata.cs

@@ -0,0 +1,45 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using System.Collections.Generic;
+using Azure.Iot.Operations.Protocol.Models;
+
+namespace Azure.Iot.Operations.Protocol.Streaming
+{
+    /// <summary>
+    /// Metadata for a request stream as a whole.
+    /// </summary>
+    public class RequestStreamMetadata
+    {
+        /// <summary>
+        /// The correlationId for tracking this streaming request
+        /// </summary>
+        public Guid CorrelationId { get; set; }
+
+        /// <summary>
+        /// The Id of the client that invoked this streaming request
+        /// </summary>
+        public string? InvokerClientId { get; set; }
+
+        /// <summary>
+        /// The MQTT topic tokens used in this streaming request.
+        /// </summary>
+        public Dictionary<string, string> TopicTokens { get; } = new();
+
+        /// <summary>
+        /// The partition associated with this streaming request.
+        /// </summary>
+        public string? Partition { get; }
+
+        /// <summary>
+        /// The content type of all messages sent in this request stream.
+        /// </summary>
+        public string? ContentType { get; set; }
+
+        /// <summary>
+        /// The payload format indicator for all messages sent in this request stream.
+        /// </summary>
+        public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; }
+    }
+}

dotnet/src/Azure.Iot.Operations.Protocol/Streaming/ResponseStreamMetadata.cs

@@ -0,0 +1,23 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using Azure.Iot.Operations.Protocol.Models;
+
+namespace Azure.Iot.Operations.Protocol.Streaming
+{
+    /// <summary>
+    /// Metadata for a response stream as a whole.
+    /// </summary>
+    public class ResponseStreamMetadata
+    {
+        /// <summary>
+        /// The content type of all messages in this response stream
+        /// </summary>
+        public string? ContentType { get; set; }
+
+        /// <summary>
+        /// The payload format indicator for all messages in this response stream
+        /// </summary>
+        public MqttPayloadFormatIndicator PayloadFormatIndicator { get; set; }
+    }
+}

dotnet/src/Azure.Iot.Operations.Protocol/Streaming/StreamMessageMetadata.cs

@@ -0,0 +1,29 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System.Collections.Generic;
+
+namespace Azure.Iot.Operations.Protocol.Streaming
+{
+    /// <summary>
+    /// Metadata for a specific message within a request stream
+    /// </summary>
+    public class StreamMessageMetadata
+    {
+        /// <summary>
+        /// The timestamp attached to this particular message
+        /// </summary>
+        public HybridLogicalClock? Timestamp { get; internal set; }
+
+        /// <summary>
+        /// User properties associated with this particular message
+        /// </summary>
+        public Dictionary<string, string> UserData { get; init; } = new();
+
+        /// <summary>
+        /// The index of this message within the stream as a whole
+        /// </summary>
+        /// <remarks>This value is automatically assigned when sending messages in a request/response stream and cannot be overriden.</remarks>
+        public int Index { get; internal set; }
+    }
+}

dotnet/src/Azure.Iot.Operations.Protocol/Streaming/StreamingCommandExecutor.cs

@@ -0,0 +1,75 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+#pragma warning disable IDE0060 // Remove unused parameter
+#pragma warning disable CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider adding the 'required' modifier or declaring as nullable.
+
+namespace Azure.Iot.Operations.Protocol.Streaming
+{
+    public abstract class StreamingCommandExecutor<TReq, TResp> : IAsyncDisposable
+        where TReq : class
+        where TResp : class
+    {
+        /// <summary>
+        /// A streaming command was invoked
+        /// </summary>
+        /// <remarks>
+        /// The callback provides the stream of requests and requires the user to return one to many responses.
+        /// </remarks>
+        public required Func<IStreamContext<ReceivedStreamingExtendedRequest<TReq>>, RequestStreamMetadata, IAsyncEnumerable<StreamingExtendedResponse<TResp>>> OnStreamingCommandReceived { get; set; }
+
+        public string RequestTopicPattern { get; init; }
+
+        /// <summary>
+        /// The topic token replacement map that this executor will use by default. Generally, this will include the token values
+        /// for topic tokens such as "executorId" which should be the same for the duration of this command executor's lifetime.
+        /// </summary>
+        /// <remarks>
+        /// Tokens replacement values can also be specified when starting the executor by specifying the additionalTopicToken map in <see cref="StartAsync(int?, Dictionary{string, string}?, CancellationToken)"/>.
+        /// </remarks>
+        public Dictionary<string, string> TopicTokenMap { get; protected set; }
+
+        /// <summary>
+        /// If true, this executor will acknowledge the MQTT message associated with each streaming request as soon as it arrives.
+        /// If false, the user must call <see cref="ReceivedStreamingExtendedRequest{TReq}.AcknowledgeAsync"/> once they are done processing
+        /// each request message.
+        /// </summary>
+        /// <remarks>
+        /// Generally, delaying acknowledgement allows for re-delivery by the broker in cases where the executor crashes or restarts unexpectedly.
+        /// However, MQTT acknowledgements must be delivered in order, so delaying these acknowledgements may affect the flow of acknowledgements
+        /// being sent by other processes using this same MQTT client. Additionally, the MQTT broker has a limit on the number of un-acknowledged messages
+        /// that are allowed to be in-flight at a single moment, so delaying too many acknowledgements may halt all further MQTT traffic on the underlying
+        /// MQTT client.
+        /// </remarks>
+        public bool AutomaticallyAcknowledgeRequests { get; set; } = true;
+
+        public Task StartAsync(CancellationToken cancellationToken = default)
+        {
+            // TODO: derive the expected request topic (like command executor does)
+
+            // TODO: subscribe to the shared subscription prefixed request topic
+
+            throw new NotImplementedException();
+        }
+
+        public Task StopAsync(CancellationToken cancellationToken = default)
+        {
+            // TODO: Unsubscribe from the request topic derived in StartAsync
+
+            throw new NotImplementedException();
+        }
+
+        public async ValueTask DisposeAsync()
+        {
+            await StopAsync();
+
+            GC.SuppressFinalize(this);
+        }
+    }
+}
+#pragma warning restore IDE0060 // Remove unused parameter
+#pragma warning restore CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider adding the 'required' modifier or declaring as nullable.