Client/Impl/Worker/JobWorker.cs
/// <summary>
/// Opens the configured JobWorker to activate jobs in the given poll interval
/// and handle with the given handler.
/// </summary>
Opens the configured JobWorker to activate jobs in the given poll interval
and handle with the given handler.
Opens the configured JobWorker to activate jobs in the given poll interval and handle with the given handler .
internal void Open()
        {
            isRunning = true;
            var cancellationToken = source.Token;
            activateJobsCommand = jobWorkerBuilder.Command;
            maxJobsActive = jobWorkerBuilder.Command.Request.MaxJobsToActivate;
            var bufferOptions = new DataflowBlockOptions
            {
                CancellationToken = cancellationToken,
                EnsureOrdered = false
            };
            var executionOptions = new ExecutionDataflowBlockOptions
            {
                MaxDegreeOfParallelism = jobWorkerBuilder.ThreadCount,
                CancellationToken = cancellationToken,
                EnsureOrdered = false
            };
            var input = new BufferBlock<IJob>(bufferOptions);
            var transformer = new TransformBlock<IJob, IJob>(async activatedJob =>
                {
                    var jobClient = JobClientWrapper.Wrap(jobWorkerBuilder.JobClient);
                    try
                    {
                        await jobHandler(jobClient, activatedJob);
                        await TryToAutoCompleteJob(jobClient, activatedJob);
                    }
                    catch (Exception exception)
                    {
                        await FailActivatedJob(jobClient, activatedJob, cancellationToken, exception);
                    }
                    finally
                    {
                        jobClient.Reset();
                    }
                    return activatedJob;
                },
                executionOptions);
            var output = new ActionBlock<IJob>(activatedJob => { currentJobsActive--; },
                executionOptions);
            input.LinkTo(transformer);
            transformer.LinkTo(output);
            Task.Run(async () =>
                {
                    while (!source.IsCancellationRequested)
                    {
                        if (currentJobsActive >= maxJobsActive)
                        {
                            await Task.Delay(pollInterval);
                            continue;
                        }
                        var jobCount = maxJobsActive - currentJobsActive;
                        activateJobsCommand.MaxJobsToActivate(jobCount);
                        try
                        {
                            var response = await activateJobsCommand.Send(null, cancellationToken);
                            logger?.LogDebug(
                                "Job worker ({worker}) activated {activatedCount} of {requestCount} successfully.",
                                activateJobsCommand.Request.Worker,
                                response.Jobs.Count,
                                jobCount);
                            foreach (var job in response.Jobs)
                            {
                                await input.SendAsync(job);
                                currentJobsActive++;
                            }
                        }
                        catch (RpcException rpcException)
                        {
                            LogRpcException(rpcException);
                        }
                    }
                },
                cancellationToken).ContinueWith(
                t => logger?.LogError(t.Exception, "Job polling failed."),
                TaskContinuationOptions.OnlyOnFaulted);
            var command = jobWorkerBuilder.Command;
            logger?.LogDebug(
                "Job worker ({worker}) for job type {type} has been opened.",
                command.Request.Worker,
                command.Request.Type);
        }
internal void Open ( ) { isRunning = true ; var cancellationToken = source . Token ; activateJobsCommand = jobWorkerBuilder . Command ; maxJobsActive = jobWorkerBuilder . Command . Request . MaxJobsToActivate ; var bufferOptions = new DataflowBlockOptions { CancellationToken = cancellationToken , EnsureOrdered = false } ; var executionOptions = new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = jobWorkerBuilder . ThreadCount , CancellationToken = cancellationToken , EnsureOrdered = false } ; var input = new BufferBlock < IJob > ( bufferOptions ) ; var transformer = new TransformBlock < IJob , IJob > ( async activatedJob => { var jobClient = JobClientWrapper . Wrap ( jobWorkerBuilder . JobClient ) ; try { await jobHandler ( jobClient , activatedJob ) ; await TryToAutoCompleteJob ( jobClient , activatedJob ) ; } catch ( Exception exception ) { await FailActivatedJob ( jobClient , activatedJob , cancellationToken , exception ) ; } finally { jobClient . Reset ( ) ; } return activatedJob ; } , executionOptions ) ; var output = new ActionBlock < IJob > ( activatedJob => { currentJobsActive -- ; } , executionOptions ) ; input . LinkTo ( transformer ) ; transformer . LinkTo ( output ) ; Task . Run ( async ( ) => { while ( ! source . IsCancellationRequested ) { if ( currentJobsActive >= maxJobsActive ) { await Task . Delay ( pollInterval ) ; continue ; } var jobCount = maxJobsActive - currentJobsActive ; activateJobsCommand . MaxJobsToActivate ( jobCount ) ; try { var response = await activateJobsCommand . Send ( null , cancellationToken ) ; logger ? . LogDebug ( " " , activateJobsCommand . Request . Worker , response . Jobs . Count , jobCount ) ; foreach ( var job in response . Jobs ) { await input . SendAsync ( job ) ; currentJobsActive ++ ; } } catch ( RpcException rpcException ) { LogRpcException ( rpcException ) ; } } } , cancellationToken ) . ContinueWith ( t => logger ? . LogError ( t . Exception , " " ) , TaskContinuationOptions . OnlyOnFaulted ) ; var command = jobWorkerBuilder . Command ; logger ? . LogDebug ( " " , command . Request . Worker , command . Request . Type ) ; }
Opens the configured JobWorker to activate jobs in the given poll interval
and handle with the given handler.
Opens the configured JobWorker to activate jobs in the given poll interval and handle with the given handler .
// Create options for blocks // Buffer for polled jobs // Transformblock to process polled jobs // Action block to finalize handled tasks // Link blocks // Start polling