AkkaStreamsSample.linq

<Query Kind="Program">
  <NuGetReference>Akka.Streams</NuGetReference>
  <Namespace>Akka.Actor</Namespace>
  <Namespace>Akka.Streams.Dsl</Namespace>
  <Namespace>Akka.Streams</Namespace>
  <Namespace>System.Collections.Immutable</Namespace>
  <Namespace>System.Threading.Tasks</Namespace>
</Query>

async Task Main()
{

	// make an ActorSystem as well as the Materializer
	// Both can be treated as singletons.
	var sys = ActorSystem.Create("test-sys");
	
	
	//We use a smaller input buffer here to make the example output smaller,
	//Default size is 16, you can also specify per-stage buffer sizes via Attr
	var materializer = sys.Materializer(ActorMaterializerSettings.Create(sys).WithInputBuffer(2,2));

    //Sample Data:
	var data = Enumerable.Range(1, 10).Select(e => e.ToString()).ToList();


	//Note: The Async() puts an explicit boundary between stages
	//      Some stages have an -implicit- boundary (i.e. always have a boundary)
	//      Boundaries incur some passing overhead,
	//      however memory cost is fixed
	var result = await Source.From<string>(data)
	.Select(s => 
	{
		Console.WriteLine($"Stream - First stage - {s}");
		return s;
	}
	)
	.Async()
	.Select(s =>
	{
		Console.WriteLine($"Stream - Second Stage - {s}");
		return s;
	})
	.RunWith(Sink.Seq<string>(),materializer);

	//vs Enumerable

	data
	.Select(s =>
	{
		Console.WriteLine($"Enumerable First Stage - {s}");
		return s;
	})
	.Select(s =>
	{
		Console.WriteLine($"Enumerable Second Stage - {s}");
		return s;
	} )
	.ToImmutableList();
	
	//Terminate allows, if graceful shutdown is enabled, a 'cleaner' shutdown of any running streams
	// or other work in an app.
	// Additional hooks can be added if needed in composition root
	// via 
	// sys.RegisterOnTermination(()=>{/*stuff*/})
	await sys.Terminate();
}