Workflow Management - Initial Methods

.github/workflows/itests.yml

@@ -41,15 +41,15 @@ jobs:
           prefix: 'net6'
     env:
       NUPKG_OUTDIR: bin/Release/nugets
-      GOVER: 1.17
+      GOVER: 1.19
       GOOS: linux
       GOARCH: amd64
       GOPROXY: https://proxy.golang.org
-      DAPR_CLI_VER: 1.8.0
-      DAPR_RUNTIME_VER: 1.8.0
+      DAPR_CLI_VER: 1.9.1
+      DAPR_RUNTIME_VER: 1.9.5
       DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/3dacfb672d55f1436c249057aaebbe597e1066f3/install/install.sh
       DAPR_CLI_REF: ''
-      DAPR_REF: ''
+      DAPR_REF: '68159926dea9694f12a132c80935df0da92ae7bf' #TODO, REMOVE ONCE TESTING IS DONE
     steps:
       - name: Set up Dapr CLI
         run: wget -q ${{ env.DAPR_INSTALL_URL }} -O - | /bin/bash -s ${{ env.DAPR_CLI_VER }}
@@ -69,7 +69,7 @@ jobs:
         uses: actions/checkout@v2
         if: env.DAPR_REF != ''
         with:
-          repository: dapr/dapr
+          repository: RyanLettieri/dapr #TODO, REMOVE ONCE TESTING IS DONE
           ref: ${{ env.DAPR_REF }}
           path: dapr
       - name: Build and override dapr cli with referenced commit.

src/Dapr.Client/DaprClient.cs

@@ -941,6 +941,53 @@ public abstract Task<UnlockResponse> Unlock(
             string lockOwner,
             CancellationToken cancellationToken = default);
 
+        /// <summary>
+        /// Attempt to start the given workflow with response indicating success.
+        /// </summary>
+        /// <param name="instanceID">Identifier of the specific run.</param>
+        /// <param name="workflowComponent">The component to interface with.</param>
+        /// <param name="workflowType">Name of the workflow to run.</param>
+        /// <param name="workflowOptions">The list of options that are potentially needed to start a workflow.</param>
+        /// <param name="input">The input input for the given workflow.</param>
+        /// <param name="cancellationToken">A <see cref="CancellationToken" /> that can be used to cancel the operation.</param>
+        /// <returns>A <see cref="Task"/> containing a <see cref="WorkflowReference"/></returns>
+        [Obsolete("This API is currently not stable as it is in the Alpha stage. This attribute will be removed once it is stable.")]
+        public abstract Task<WorkflowReference> StartWorkflow(
+            string instanceID,
+            string workflowComponent,
+            string workflowType,
+            Object input,
+            IReadOnlyDictionary<string, string> workflowOptions = default,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Attempt to get information about the given workflow.
+        /// </summary>
+        /// <param name="instanceID">Identifier of the specific run.</param>
+        /// <param name="workflowComponent">The component to interface with.</param>
+        /// <param name="workflowType">Name of the workflow to run.</param>
+        /// <param name="cancellationToken">A <see cref="CancellationToken" /> that can be used to cancel the operation.</param>
+        /// <returns>A <see cref="Task"/> containing a <see cref="GetWorkflowResponse"/></returns>
+        [Obsolete("This API is currently not stable as it is in the Alpha stage. This attribute will be removed once it is stable.")]
+        public abstract Task<GetWorkflowResponse> GetWorkflow(
+            string instanceID,
+            string workflowComponent,
+            string workflowType,
+            CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Attempt to get terminate the given workflow.
+        /// </summary>
+        /// <param name="instanceID">Identifier of the specific run.</param>
+        /// <param name="workflowComponent">The component to interface with.</param>
+        /// <param name="cancellationToken">A <see cref="CancellationToken" /> that can be used to cancel the operation.</param>
+        /// <returns>A <see cref="Task" /> that will complete when the terminate operation has been scheduled. If the wrapped value is true the operation suceeded.</returns>
+        [Obsolete("This API is currently not stable as it is in the Alpha stage. This attribute will be removed once it is stable.")]
+        public abstract Task TerminateWorkflow(
+            string instanceID,
+            string workflowComponent,
+            CancellationToken cancellationToken = default);
+
         /// <inheritdoc />
         public void Dispose()
         {

src/Dapr.Client/DaprClientGrpc.cs

@@ -1374,6 +1374,121 @@ public async override Task<UnlockResponse> Unlock(
 
         #endregion
 
+
+        #region Workflow API
+        /// <inheritdoc/>
+        [Obsolete]
+        public async override Task<WorkflowReference> StartWorkflow(
+            string instanceId,
+            string workflowComponent,
+            string workflowType,
+            Object input,
+            IReadOnlyDictionary<string, string> workflowOptions = default,
+            CancellationToken cancellationToken = default)
+        {
+            ArgumentVerifier.ThrowIfNullOrEmpty(instanceId, nameof(instanceId));
+            ArgumentVerifier.ThrowIfNullOrEmpty(workflowComponent, nameof(workflowComponent));
+            ArgumentVerifier.ThrowIfNullOrEmpty(workflowType, nameof(workflowType));
+            ArgumentVerifier.ThrowIfNull(input, nameof(input));
+
+            // Serialize json data. Converts input object to json string and then bytestring inside the request.
+            string jsonUtf8Bytes =JsonSerializer.Serialize(input);
+
+            var request = new Autogenerated.StartWorkflowRequest()
+            {
+                InstanceId = instanceId,
+                WorkflowComponent = workflowComponent,
+                WorkflowName = workflowType,
+                Input = ByteString.CopyFromUtf8(jsonUtf8Bytes)
+            };
+
+            if (workflowOptions?.Count > 0)
+                {
+                foreach (var item in workflowOptions)
+                {
+                    request.Options[item.Key] = item.Value;
+                }
+            }
+
+            try
+            {
+                var options = CreateCallOptions(headers: null, cancellationToken);
+                var response = await client.StartWorkflowAlpha1Async(request, options);
+                return new WorkflowReference(response.InstanceId);
+
+            }
+            catch (RpcException ex)
+            {
+                throw new DaprException("Start Workflow operation failed: the Dapr endpoint indicated a failure. See InnerException for details.", ex);
+            }
+        }
+
+        /// <inheritdoc/>
+        [Obsolete]
+        public async override Task<GetWorkflowResponse> GetWorkflow(
+            string instanceId,
+            string workflowComponent,
+            string workflowType,
+            CancellationToken cancellationToken = default)
+        {
+            ArgumentVerifier.ThrowIfNullOrEmpty(instanceId, nameof(instanceId));
+            ArgumentVerifier.ThrowIfNullOrEmpty(workflowComponent, nameof(workflowComponent));
+            ArgumentVerifier.ThrowIfNullOrEmpty(workflowType, nameof(workflowType));
+
+            var request = new Autogenerated.GetWorkflowRequest()
+            {
+                InstanceId = instanceId,
+                WorkflowComponent = workflowComponent,
+                WorkflowType = workflowType
+            };
+
+            try
+            {
+                var options = CreateCallOptions(headers: null, cancellationToken);
+                var response = await client.GetWorkflowAlpha1Async(request, options);
+                var dateTimeValue = new DateTime(response.StartTime);
+                return new GetWorkflowResponse(response.InstanceId, dateTimeValue, response.Metadata);
+            }
+            catch (RpcException ex)
+            {
+                throw new DaprException("Get workflow operation failed: the Dapr endpoint indicated a failure. See InnerException for details.", ex);
+            }
+
+        }
+
+
+        /// <inheritdoc/>
+        [Obsolete]
+        public async override Task TerminateWorkflow(
+            string instanceID,
+            string workflowComponent,
+            CancellationToken cancellationToken = default)
+        {
+            ArgumentVerifier.ThrowIfNullOrEmpty(instanceID, nameof(instanceID));
+            ArgumentVerifier.ThrowIfNullOrEmpty(workflowComponent, nameof(workflowComponent));
+
+            var request = new Autogenerated.TerminateWorkflowRequest()
+            {
+                InstanceId = instanceID,
+                WorkflowComponent = workflowComponent
+            };
+
+            var options = CreateCallOptions(headers: null, cancellationToken);
+
+            try
+            {
+                await client.TerminateWorkflowAlpha1Async(request, options);
+            }
+            catch (RpcException ex)
+            {
+                throw new DaprException("Terminate workflow operation failed: the Dapr endpoint indicated a failure. See InnerException for details.", ex);
+            }
+
+        }
+
+        #endregion
+
+
         #region Dapr Sidecar Methods
 
         /// <inheritdoc/>

src/Dapr.Client/GetWorkflowResponse.cs

@@ -0,0 +1,56 @@
+// ------------------------------------------------------------------------
+// Copyright 2021 The Dapr Authors
+// 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;
+using System.Collections.Generic;
+
+namespace Dapr.Client
+{
+    /// <summary>
+    /// Represents the response from invoking a binding.
+    /// </summary>
+    public sealed record GetWorkflowResponse
+    {
+        /// <summary>
+        /// Initializes a new <see cref="GetWorkflowResponse" />.`
+        /// </summary>
+        /// <param name="instanceId">The instance ID assocated with this response.</param>
+        /// <param name="startTime">The time at which the workflow started executing.</param>
+        /// <param name="metadata">The response metadata.</param>
+        public GetWorkflowResponse(string instanceId, DateTime startTime, IReadOnlyDictionary<string, string> metadata)
+        {
+            ArgumentVerifier.ThrowIfNull(instanceId, nameof(instanceId));
+            ArgumentVerifier.ThrowIfNull(startTime, nameof(startTime));
+            ArgumentVerifier.ThrowIfNull(metadata, nameof(metadata));
+
+            this.InstanceId = instanceId;
+            this.StartTime = startTime;
+            this.Metadata = metadata;
+        }
+
+        /// <summary>
+        /// Gets the workflow instance ID assocated with this response.
+        /// </summary>
+        public string InstanceId { set; get; }
+
+        /// <summary>
+        /// Gets the time that the workflow started.
+        /// </summary>
+        public DateTime StartTime { set; get; }
+
+        /// <summary>
+        /// Gets the response metadata from the associated workflow. This includes information such as start time and status of workflow.
+        /// </summary>
+        public IReadOnlyDictionary<string, string> Metadata { set; get; }
+    }
+}