Merge pull request #682 from LennartKleymann/681-add-update-job-timeo…

…ut-command

Implementation of UpdateJobTimeoutCommand

Client.UnitTests/GatewayTestService.cs

@@ -53,6 +53,7 @@ public GatewayTestService()
             typedRequestHandler.Add(typeof(CompleteJobRequest), request => new CompleteJobResponse());
             typedRequestHandler.Add(typeof(FailJobRequest), request => new FailJobResponse());
             typedRequestHandler.Add(typeof(UpdateJobRetriesRequest), request => new UpdateJobRetriesResponse());
+            typedRequestHandler.Add(typeof(UpdateJobTimeoutRequest), request => new UpdateJobTimeoutResponse());
             typedRequestHandler.Add(typeof(ThrowErrorRequest), request => new ThrowErrorResponse());
 
             typedRequestHandler.Add(typeof(DeployResourceRequest), request => new DeployResourceResponse());
@@ -106,6 +107,11 @@ public override Task<UpdateJobRetriesResponse> UpdateJobRetries(UpdateJobRetries
             return Task.FromResult((UpdateJobRetriesResponse)HandleRequest(request, context));
         }
 
+        public override Task<UpdateJobTimeoutResponse> UpdateJobTimeout(UpdateJobTimeoutRequest request, ServerCallContext context)
+        {
+            return Task.FromResult((UpdateJobTimeoutResponse)HandleRequest(request, context));
+        }
+
         public override Task<ThrowErrorResponse> ThrowError(ThrowErrorRequest request, ServerCallContext context)
         {
             return Task.FromResult((ThrowErrorResponse)HandleRequest(request, context));

Client.UnitTests/TestDataProvider.cs

@@ -143,6 +143,15 @@ public static IEnumerable<TestCaseData> Provider()
                 new EvaluateDecisionResponse(),
                 (RequestCreator<IEvaluateDecisionResponse>)
                 (zeebeClient => zeebeClient.NewEvaluateDecisionCommand().DecisionId("decision")));
+            yield return new TestCaseData(
+                new UpdateJobTimeoutRequest()
+                {
+                    JobKey = 12113L,
+                    Timeout = 20000
+                }, new UpdateJobTimeoutResponse(),
+                (RequestCreator<IUpdateJobTimeoutResponse>)
+                (zeebeClient => zeebeClient.NewUpdateJobTimeoutCommand(12113L)
+                    .Timeout(new TimeSpan(0, 0, 0, 20))));
         }
     }
 }

Client.UnitTests/UpdateJobTimeoutTest.cs

@@ -0,0 +1,67 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using GatewayProtocol;
+using Grpc.Core;
+using NUnit.Framework;
+
+namespace Zeebe.Client;
+
+[TestFixture]
+public class UpdateJobTimeoutTest : BaseZeebeTest
+{
+    [Test]
+    public async Task ShouldSendRequestAsExpected()
+    {
+        // given
+        var expectedRequest = new UpdateJobTimeoutRequest()
+        {
+            JobKey = 1024,
+            Timeout = 2000
+        };
+
+        // when
+        await ZeebeClient
+            .NewUpdateJobTimeoutCommand(1024)
+            .Timeout(new TimeSpan(0, 0, 2))
+            .Send();
+
+        // then
+        var request = TestService.Requests[typeof(UpdateJobTimeoutRequest)][0];
+        Assert.AreEqual(expectedRequest, request);
+    }
+
+    [Test]
+    public void ShouldTimeoutRequest()
+    {
+        // given
+
+        // when
+        var task = ZeebeClient
+            .NewUpdateJobTimeoutCommand(1024)
+            .Timeout(new TimeSpan(0, 0, 2))
+            .Send(TimeSpan.Zero);
+        var aggregateException = Assert.Throws<AggregateException>(() => task.Wait());
+        var rpcException = (RpcException)aggregateException.InnerExceptions[0];
+
+        // then
+        Assert.AreEqual(StatusCode.DeadlineExceeded, rpcException.Status.StatusCode);
+    }
+
+    [Test]
+    public void ShouldCancelRequest()
+    {
+        // given
+
+        // when
+        var task = ZeebeClient
+            .NewUpdateJobTimeoutCommand(1024)
+            .Timeout(new TimeSpan(0, 0, 2))
+            .Send(new CancellationTokenSource(TimeSpan.Zero).Token);
+        var aggregateException = Assert.Throws<AggregateException>(() => task.Wait());
+        var rpcException = (RpcException)aggregateException.InnerExceptions[0];
+
+        // then
+        Assert.AreEqual(StatusCode.Cancelled, rpcException.Status.StatusCode);
+    }
+}

Client/Api/Commands/IUpdateJobTimeoutCommandSteps.cs

@@ -0,0 +1,25 @@
+using System;
+using Zeebe.Client.Api.Responses;
+
+namespace Zeebe.Client.Api.Commands;
+
+public interface IUpdateJobTimeoutCommandStep1
+{
+    /// <summary>
+    ///   Update the timeout for this job.
+    /// </summary>
+    ///
+    /// <para>
+    ///   If the job's timeout is set to zero, the job will be directly retried.
+    /// </para>
+    /// <param name="timeout">The duration of the new timeout as a TimeSpan, starting from the current moment.</param>
+    /// <returns>
+    ///   The builder for this command. Call <see cref="IFinalCommandStep{T}.Send"/> to complete the command and send it to the broker.
+    /// </returns>
+    IUpdateJobTimeoutCommandStep2 Timeout(TimeSpan timeout);
+}
+
+public interface IUpdateJobTimeoutCommandStep2 : IFinalCommandWithRetryStep<IUpdateJobTimeoutResponse>
+{
+    // the place for new optional parameters
+}

Client/Api/Responses/IUpdateJobTimeoutResponse.cs

@@ -0,0 +1,9 @@
+namespace Zeebe.Client.Api.Responses
+{
+    /// <summary>
+    /// Response for an update job timeout request.
+    /// </summary>
+    public interface IUpdateJobTimeoutResponse
+    {
+    }
+}

Client/IZeebeClient.cs

@@ -118,6 +118,31 @@ public interface IZeebeClient : IJobClient, IDisposable
         /// </returns>
         IUpdateRetriesCommandStep1 NewUpdateRetriesCommand(long jobKey);
 
+        /// <summary>
+        /// Command to update the timeout of a job.
+        /// </summary>
+        /// <example>
+        /// <code>
+        /// long jobKey = ..;
+        ///
+        /// zeebeClient
+        ///  .NewUpdateJobTimeoutCommand(jobKey)
+        ///  .Timeout(new TimeSpan(0, 0, 0, 10))
+        ///  .Send();
+        /// </code>
+        /// </example>
+        ///
+        /// <para>
+        ///   If the job's timeout is zero, the job will be directly retried.
+        /// </para>
+        /// <param name="jobKey">
+        ///     the key of the job to update
+        /// </param>
+        /// <returns>
+        ///     a builder for the command
+        /// </returns>
+        IUpdateJobTimeoutCommandStep1 NewUpdateJobTimeoutCommand(long jobKey);
+
         /// <summary>
         /// Command to deploy new resources, i.e. BPMN process models and DMN decision models.
         /// </summary>

Client/Impl/Commands/UpdateJobTimeoutCommand.cs

@@ -0,0 +1,50 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using GatewayProtocol;
+using Zeebe.Client.Api.Commands;
+using Zeebe.Client.Api.Misc;
+using Zeebe.Client.Api.Responses;
+using UpdateJobTimeoutResponse = Zeebe.Client.Impl.Responses.UpdateJobTimeoutResponse;
+
+namespace Zeebe.Client.Impl.Commands;
+
+public class UpdateJobTimeoutCommand : IUpdateJobTimeoutCommandStep1, IUpdateJobTimeoutCommandStep2
+{
+    private readonly UpdateJobTimeoutRequest request;
+    private readonly Gateway.GatewayClient client;
+    private readonly IAsyncRetryStrategy asyncRetryStrategy;
+
+    public UpdateJobTimeoutCommand(Gateway.GatewayClient client, IAsyncRetryStrategy asyncRetryStrategy, long jobKey)
+    {
+        request = new UpdateJobTimeoutRequest()
+        {
+            JobKey = jobKey
+        };
+        this.client = client;
+        this.asyncRetryStrategy = asyncRetryStrategy;
+    }
+
+    public IUpdateJobTimeoutCommandStep2 Timeout(TimeSpan timeout)
+    {
+        request.Timeout = (long)timeout.TotalMilliseconds;
+        return this;
+    }
+
+    public async Task<IUpdateJobTimeoutResponse> Send(TimeSpan? timeout = null, CancellationToken token = default)
+    {
+        var asyncReply = client.UpdateJobTimeoutAsync(request, deadline: timeout?.FromUtcNow(), cancellationToken: token);
+        await asyncReply.ResponseAsync;
+        return new UpdateJobTimeoutResponse();
+    }
+
+    public async Task<IUpdateJobTimeoutResponse> Send(CancellationToken cancellationToken)
+    {
+        return await Send(token: cancellationToken);
+    }
+
+    public async Task<IUpdateJobTimeoutResponse> SendWithRetry(TimeSpan? timeout = null, CancellationToken token = default)
+    {
+        return await asyncRetryStrategy.DoWithRetry(() => Send(timeout, token));
+    }
+}

Client/Impl/Responses/UpdateJobTimeoutResponse.cs

@@ -0,0 +1,8 @@
+using Zeebe.Client.Api.Responses;
+
+namespace Zeebe.Client.Impl.Responses;
+
+public class UpdateJobTimeoutResponse : IUpdateJobTimeoutResponse
+{
+
+}

Client/ZeebeClient.cs

@@ -188,6 +188,11 @@ public IUpdateRetriesCommandStep1 NewUpdateRetriesCommand(long jobKey)
             return new UpdateRetriesCommand(gatewayClient, asyncRetryStrategy, jobKey);
         }
 
+        public IUpdateJobTimeoutCommandStep1 NewUpdateJobTimeoutCommand(long jobKey)
+        {
+            return new UpdateJobTimeoutCommand(gatewayClient, asyncRetryStrategy, jobKey);
+        }
+
         public IThrowErrorCommandStep1 NewThrowErrorCommand(long jobKey)
         {
             return new ThrowErrorCommand(gatewayClient, asyncRetryStrategy, jobKey);