ActorSrcGen is a C# Source Generator that converts simple C# classes into TPL Dataflow-compatible pipelines. It simplifies working with TPL Dataflow by generating boilerplate code to handle errors without interrupting the pipeline, ideal for long-lived processes with ingesters that continually pump messages into the pipeline.
Install the package:
dotnet add package ActorSrcGen
Declare the pipeline class:
[Actor]publicpartialclassMyPipeline{}
The class must be
partialto allow the source generator to add boilerplate code.If you are using Visual Studio, you can see the generated part of the code under the ActorSrcGen analyzer:
Create ingester functions:
[Ingest(1)][NextStep(nameof(DoSomethingWithRequest))]publicasyncTask<string>ReceivePollRequest(CancellationTokencancellationToken){returnawaitGetTheNextRequest();}
Ingesters define a
Priorityand are visited in priority order. If no messages are available, the pipeline sleeps for a second before retrying.Implement pipeline steps:
[FirstStep("decode incoming poll request")][NextStep(nameof(ActOnTheRequest))]publicPollRequestDecodeRequest(stringjson){Console.WriteLine(nameof(DecodeRequest));varpollRequest=JsonSerializer.Deserialize<PollRequest>(json);returnpollRequest;}
The first step controls the pipeline's interface. Implement additional steps as needed, ensuring input and output types match.
Now implement other steps are needed in the pipeline. The outputs and input types of successive steps need to match.
[Step][NextStep(nameof(DeliverResults))]publicPollResultsActOnTheRequest(PollRequestreq){Console.WriteLine(nameof(ActOnTheRequest));varresult=SomeApiClient.GetTheResults(req.Id);returnresult;}
Define the last step:
[LastStep]publicboolDeliverResults(PollResultsres){returnmyQueue.TryPush(res);}
Generated code example:
usingSystem.Threading.Tasks.Dataflow;usingGridsum.DataflowEx;publicpartialclassMyActor:Dataflow<string,bool>,IActor<string>{publicMyActor(DataflowOptionsdataflowOptions=null):base(DataflowOptions.Default){_DeliverResults=newTransformBlock<PollResults,bool>((PollResultsx)=>{try{returnDeliverResults(x);}catch(Exceptione){LogMessage(LogLevel.Error,$"Error in DeliverResults: {e.Message}\nStack Trace: {e.StackTrace}");returndefault;}},newExecutionDataflowBlockOptions(){BoundedCapacity=1,MaxDegreeOfParallelism=1});RegisterChild(_DeliverResults);_ActOnTheRequest=newTransformBlock<PollRequest,PollResults>((PollRequestx)=>{try{returnActOnTheRequest(x);}catch(Exceptione){LogMessage(LogLevel.Error,$"Error in ActOnTheRequest: {e.Message}\nStack Trace: {e.StackTrace}");returndefault;}},newExecutionDataflowBlockOptions(){BoundedCapacity=1,MaxDegreeOfParallelism=1});RegisterChild(_ActOnTheRequest);_DecodeRequest=newTransformBlock<string,PollRequest>((stringx)=>{try{returnDecodeRequest(x);}catch(Exceptione){LogMessage(LogLevel.Error,$"Error in DecodeRequest: {e.Message}\nStack Trace: {e.StackTrace}");returndefault;}},newExecutionDataflowBlockOptions(){BoundedCapacity=1,MaxDegreeOfParallelism=1});RegisterChild(_DecodeRequest);_ActOnTheRequest.LinkTo(_DeliverResults,newDataflowLinkOptions{PropagateCompletion=true});_DecodeRequest.LinkTo(_ActOnTheRequest,newDataflowLinkOptions{PropagateCompletion=true});}TransformBlock<PollResults,bool>_DeliverResults;TransformBlock<PollRequest,PollResults>_ActOnTheRequest;TransformBlock<string,PollRequest>_DecodeRequest;publicoverrideITargetBlock<string>InputBlock{get=>_DecodeRequest;}publicoverrideISourceBlock<bool>OutputBlock{get=>_DeliverResults;}publicboolCall(stringinput)=>InputBlock.Post(input);publicasyncTask<bool>Cast(stringinput)=>awaitInputBlock.SendAsync(input);publicasyncTask<bool>AcceptAsync(CancellationTokencancellationToken){try{varresult=await_DeliverResults.ReceiveAsync(cancellationToken);returnresult;}catch(OperationCanceledExceptionoperationCanceledException){returnawaitTask.FromCanceled<bool>(cancellationToken);}}publicasyncTaskIngest(CancellationTokenct){// start the message pumpwhile(!ct.IsCancellationRequested){varfoundSomething=false;try{// cycle through ingesters IN PRIORITY ORDER.{varmsg=awaitReceivePollRequest(ct);if(msg!=null){Call(msg);foundSomething=true;// then jump back to the start of the pumpcontinue;}}if(!foundSomething)awaitTask.Delay(1000,ct);}catch(TaskCanceledException){// if nothing was found on any of the receivers, then sleep for a while.continue;}catch(Exceptione){LogMessage(LogLevel.Error,$"Exception in Ingest loop: {e.Message}\nStack Trace: {e.StackTrace}");}}}}
Using the pipeline:
varactor=newMyActor();// this is your pipelinetry{// call into the pipeline synchronouslyif(actor.Call(""" { "something": "here" } """))Console.WriteLine("Called Synchronously");// stop the pipeline after 10 secsvarcts=newCancellationTokenSource(TimeSpan.FromSeconds(10));// kick off an endless process to keep ingesting input into the pipelinevart=Task.Run(async()=>awaitactor.Ingest(cts.Token),cts.Token);// consume results from the last step via the AcceptAsync methodwhile(!cts.Token.IsCancellationRequested){varresult=awaitactor.AcceptAsync(cts.Token);Console.WriteLine($"Result: {result}");}awaitt;// cancel the message pump taskawaitactor.SignalAndWaitForCompletionAsync();// wait for all pipeline tasks to complete}catch(OperationCanceledException_){Console.WriteLine("All Done!");}
- Simplifies TPL Dataflow usage: Automatically generates boilerplate code.
- Concurrency: Efficient use of multiple CPU cores.
- Fault tolerance: Errors in pipeline steps are trapped and handled.
- Encapsulation: Easier to reason about and test code.
- Run the suite:
dotnet test - With coverage (85% threshold, critical paths 100%):
dotnet test /p:CollectCoverage=true /p:CoverletOutputFormat=cobertura - See quickstart for full workflow: specs/001-generator-reliability-hardening/quickstart.md
- ASG0001 Non-disjoint input types: ensure entry steps have distinct input signatures
- ASG0002 Missing input types: add at least one [FirstStep] or [Step] method
- ASG0003 Generation error: inspect the diagnostic message for the underlying exception
- Full reference: doc/DIAGNOSTICS.md
Built on DataflowEx and Bnaya.SourceGenerator.Template.