notebooks/lesson3-graph-lifecycles.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Akka.Streams Graph Lifecycles\n",
    "Just like how Akka.NET actors have discrete lifecycles, so do Akka.Streams graphs!\n",
    "\n",
    "* **Running** – the stream is currently processing elements and has not run out of elements to process.\n",
    "* **Completed** – the stream has terminated because:\n",
    "    * The `Source<T>` ran out of elements or\n",
    "    * One or more downstream stages signaled completion.\n",
    "* **Failed** – the stream terminated because an error was thrown in one of the stages.\n",
    "\n",
    "Let's consider some examples.\n",
    "\n",
    "## Completed Streams\n",
    "A stream completes usually when it's completed processing all possible inputs - or if one of the graph stages signals that it wants to terminate, i.e. a [`KillSwitch`](https://getakka.net/api/Akka.Streams.IKillSwitch.html)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "dotnet-interactive.csharp"
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div><div></div><div></div><div><strong>Installed Packages</strong><ul><li><span>Akka.Streams, 1.4.47</span></li></ul></div></div>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1-->a\r\n"
     ]
    }
   ],
   "source": [
    "#r \"nuget: Akka.Streams, 1.4.47\"\n",
    "\n",
    "using System.Linq;\n",
    "using System.Collections.Immutable;\n",
    "using System.Threading;\n",
    "using Akka;\n",
    "using Akka.Actor;\n",
    "using Akka.Streams;\n",
    "using Akka.Streams.Dsl;\n",
    "\n",
    "ActorSystem actorSystem = ActorSystem.Create(\"StreamsExample\");\n",
    "\n",
    "IMaterializer materializer = actorSystem.Materializer();\n",
    "\n",
    "// a source representing a range of integers\n",
    "Source<int, NotUsed> source1 = Source.From(Enumerable.Range(1, 10));\n",
    "\n",
    "// a source representing a single string value\n",
    "Source<string, NotUsed> source2 = Source.Single(\"a\");\n",
    "\n",
    "// let's combine these two sources such that we create 10 int / string tuples\n",
    "IAsyncEnumerable<(int i, string s)> merged1 = source1.Zip(source2).RunAsAsyncEnumerable(materializer);\n",
    "\n",
    "await foreach(var (i, s) in merged1){\n",
    "    Console.WriteLine($\"{i}-->{s}\");\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This stream terminated with a single output, despite the fact that we had `source1` defined as \n",
    "\n",
    "```csharp\n",
    "Source<int, NotUsed> source1 = Source.From(Enumerable.Range(1, 10));\n",
    "```\n",
    "\n",
    "Which is at least 10 elements! What gives?\n",
    "\n",
    "Well in this case, it's because we combined `source1` with `source2`, which is defined as\n",
    "\n",
    "```csharp\n",
    "Source<string, NotUsed> source2 = Source.Single(\"a\");\n",
    "```\n",
    "\n",
    "A single element! So due to the semantics of how the `Zip<T>` stage works - it will \"complete\" once at least one of its upstream `Source<T>`s completes, the entire stream terminates with a single output.\n",
    "\n",
    "So how could we fix this in order to ensure that we had all 10 outputs from `source1` get rendered?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "dotnet-interactive.csharp"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1-->a\r\n",
      "2-->a\r\n",
      "3-->a\r\n",
      "4-->a\r\n",
      "5-->a\r\n",
      "6-->a\r\n",
      "7-->a\r\n",
      "8-->a\r\n",
      "9-->a\r\n",
      "10-->a\r\n"
     ]
    }
   ],
   "source": [
    "// a source representing a single string value\n",
    "// BUT this source will be repeated each time a downstream requests it.\n",
    "Source<string, NotUsed> source3 = Source.Repeat(\"a\");\n",
    "\n",
    "// let's combine these two sources such that we create 10 int / string tuples\n",
    "IAsyncEnumerable<(int i, string s)> merged1 = source1.Zip(source3).RunAsAsyncEnumerable(materializer);\n",
    "\n",
    "await foreach(var (i, s) in merged1){\n",
    "    Console.WriteLine($\"{i}-->{s}\");\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now we've had all 10 values from `source1` written out, all because we changed to a new `Source<T>` that had different semantics.\n",
    "\n",
    "> **N.B.** This type of use case is exactly why it's a good idea to read \"[Overview of Built-In Akka.Streams Stages and Their Semantics](https://getakka.net/articles/streams/builtinstages.html)\" as it spells out exactly how these streams stages work with regard to lifetime, completion order, and so on."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Infinite Streams\n",
    "Many of the examples we've used so far have been for \"finite\" streams - where the number of possible inputs and outputs can be determined ahead of actually running the code.\n",
    "\n",
    "I.e. when we create a `Source<int>` from a `List<int>`, we know how many items in the `List<int>` will need to be processed before we reach the end. That's an example of a finite stream.\n",
    "\n",
    "An infinite stream, on the other hand, is unbounded - it has no discrete end, so it runs until it's told to terminate via some external or internal signal such as an `IKillSwitch`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "dotnet-interactive.csharp"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hit1\r\n",
      "hit2\r\n",
      "hit3\r\n"
     ]
    }
   ],
   "source": [
    "// create a source that will be materialized into an IActorRef\n",
    "Source<string, IActorRef> actorSource = Source.ActorRef<string>(1000, OverflowStrategy.DropHead);\n",
    "var (preMaterializedRef, standAloneSrc) = actorSource.PreMaterialize(materializer);\n",
    "\n",
    "// materialize the rest of the stream into an IAsyncEnumerable\n",
    "IAsyncEnumerable<string> strResponses = standAloneSrc.Via(Flow.Create<string>().Select(x => x.ToLowerInvariant())).RunAsAsyncEnumerable(materializer);\n",
    "\n",
    "// send some messages to our head actor to drive the stream\n",
    "preMaterializedRef.Tell(\"HIT1\");\n",
    "preMaterializedRef.Tell(\"HIT2\");\n",
    "preMaterializedRef.Tell(\"HIT3\");\n",
    "//preMaterializedRef.GracefulStop(TimeSpan.FromSeconds(1));\n",
    "\n",
    "// need to timeout our IAsyncEnumerable otherwise it will run forever (by design)\n",
    "var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(500));\n",
    "\n",
    "await foreach(var str in strResponses.WithCancellation(cts.Token)){\n",
    "    Console.WriteLine(str);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In order to get this code sample to terminate, we had to pass in a `CancellationToken` and invoke it behind the scenes - otherwise we would wait on our `IAsyncEnumerable` indefinitely!\n",
    "\n",
    "### Using KillSwitches to Terminate Streams\n",
    "\n",
    "However, another approach we could have used is an `IKillSwitch` to programmatically terminate the stream when we're done with it - even though we could, in theory, continue to process more events."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "dotnet-interactive.csharp"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hit1\r\n",
      "hit2\r\n",
      "hit3\r\n",
      "Stream completed!\r\n"
     ]
    }
   ],
   "source": [
    "// create another source that will be materialized into an IActorRef\n",
    "Source<string, IActorRef> actorSource2 = Source.ActorRef<string>(1000, OverflowStrategy.DropTail);\n",
    "var (preMaterializedRef2, standAloneSrc2) = actorSource2.PreMaterialize(materializer);\n",
    "\n",
    "// going to use this as part of our KillSwitch\n",
    "var cts = new CancellationTokenSource();\n",
    "\n",
    "// materialize the rest of the stream into an IAsyncEnumerable\n",
    "IAsyncEnumerable<string> strResponses2 = standAloneSrc2.Via(Flow.Create<string>()\n",
    "    .Select(x => x.ToLowerInvariant()))\n",
    "    .Via(cts.Token.AsFlow<string>(true))\n",
    "    //.Via(KillSwitches.AsFlow<string>(cts.Token, cancelGracefully:true))\n",
    "    .RunAsAsyncEnumerable(materializer);\n",
    "\n",
    "// send some messages to our head actor to drive the stream\n",
    "preMaterializedRef2.Tell(\"HIT1\");\n",
    "preMaterializedRef2.Tell(\"HIT2\");\n",
    "preMaterializedRef2.Tell(\"HIT3\");\n",
    "\n",
    "var count = 0;\n",
    "await foreach(var str in strResponses2){\n",
    "    Console.WriteLine(str);\n",
    "    if(++count == 3){\n",
    "        cts.Cancel(); // shut down the stream\n",
    "    }\n",
    "}\n",
    "Console.WriteLine(\"Stream completed!\");"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "There are other ways to instrument a kill switch - including \"shared\" kill switches that can be used to terminate multiple streams at once, but this is the most idiomatic way of doing it in .NET."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Other Ways of Completing Streams\n",
    "Some stream stages support their own custom methods of completion - the `Source.ActorRef<T>` being one of them. We can complete this stage by stopping the `IActorRef` we receive from the materializer."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "dotnet-interactive.csharp"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "hit1\r\n",
      "hit2\r\n",
      "hit3\r\n",
      "Stream completed!\r\n"
     ]
    }
   ],
   "source": [
    "// create a source that will be materialized into an IActorRef\n",
    "Source<string, IActorRef> actorSource3 = Source.ActorRef<string>(1000, OverflowStrategy.DropTail);\n",
    "var (preMaterializedRef3, standAloneSrc3) = actorSource3.PreMaterialize(materializer);\n",
    "\n",
    "// materialize the rest of the stream into an IAsyncEnumerable\n",
    "IAsyncEnumerable<string> strResponses3 = standAloneSrc3.Via(Flow.Create<string>().Select(x => x.ToLowerInvariant())).RunAsAsyncEnumerable(materializer);\n",
    "\n",
    "// send some messages to our head actor to drive the stream\n",
    "preMaterializedRef3.Tell(\"HIT1\");\n",
    "preMaterializedRef3.Tell(\"HIT2\");\n",
    "preMaterializedRef3.Tell(\"HIT3\");\n",
    "preMaterializedRef3.Tell(PoisonPill.Instance);\n",
    "\n",
    "await foreach(var str in strResponses3){\n",
    "    Console.WriteLine(str);\n",
    "}\n",
    "Console.WriteLine(\"Stream completed!\");"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "And there you have it - stream completed via sending the `IActorRef` a `PoisonPill` message."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Dealing with Failures\n",
    "Failures are a part of every programming exercise - and Akka.Streams is no different."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "dotnet-interactive.csharp"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1/9 is 0 w/ integer division\r\n",
      "1/8 is 0 w/ integer division\r\n",
      "1/7 is 0 w/ integer division\r\n",
      "1/6 is 0 w/ integer division\r\n"
     ]
    },
    {
     "ename": "Error",
     "evalue": "System.DivideByZeroException: Attempted to divide by zero.\r\n   at Submission#10.<>c.<<Initialize>>b__0_0(Int32 x)\r\n   at Akka.Streams.Implementation.Fusing.Select`2.Logic.OnPush()\r\n--- End of stack trace from previous location ---\r\n   at Akka.Streams.Dsl.SinkQueueAsyncEnumerator`1.MoveNextAsync()\r\n   at Submission#10.<<Initialize>>d__0.MoveNext()\r\n--- End of stack trace from previous location ---\r\n   at Submission#10.<<Initialize>>d__0.MoveNext()\r\n--- End of stack trace from previous location ---\r\n   at Microsoft.CodeAnalysis.Scripting.ScriptExecutionState.RunSubmissionsAsync[TResult](ImmutableArray`1 precedingExecutors, Func`2 currentExecutor, StrongBox`1 exceptionHolderOpt, Func`2 catchExceptionOpt, CancellationToken cancellationToken)",
     "output_type": "error",
     "traceback": [
      "System.DivideByZeroException: Attempted to divide by zero.\r\n",
      "   at Submission#10.<>c.<<Initialize>>b__0_0(Int32 x)\r\n",
      "   at Akka.Streams.Implementation.Fusing.Select`2.Logic.OnPush()\r\n",
      "--- End of stack trace from previous location ---\r\n",
      "   at Akka.Streams.Dsl.SinkQueueAsyncEnumerator`1.MoveNextAsync()\r\n",
      "   at Submission#10.<<Initialize>>d__0.MoveNext()\r\n",
      "--- End of stack trace from previous location ---\r\n",
      "   at Submission#10.<<Initialize>>d__0.MoveNext()\r\n",
      "--- End of stack trace from previous location ---\r\n",
      "   at Microsoft.CodeAnalysis.Scripting.ScriptExecutionState.RunSubmissionsAsync[TResult](ImmutableArray`1 precedingExecutors, Func`2 currentExecutor, StrongBox`1 exceptionHolderOpt, Func`2 catchExceptionOpt, CancellationToken cancellationToken)"
     ]
    }
   ],
   "source": [
    "// create a source of integers - including one bad apple (zero)\n",
    "Source<int, NotUsed> numbers = Source.From(new []{ 9,8,7,6,0,5,4,3,2,1 });\n",
    "\n",
    "IAsyncEnumerable<string> integerDivision = numbers.Via(Flow.Create<int>()\n",
    "    .Select(x => $\"1/{x} is {1/x} w/ integer division\"))\n",
    "    .RunAsAsyncEnumerable(materializer);\n",
    "\n",
    "await foreach(var d in integerDivision){\n",
    "    Console.WriteLine(d);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "When the stream attempts to divide by zero we get a `System.DivideByZeroException` and the stream stops processing any additional items, even though there might still be outstanding work to perform inside the stream.\n",
    "\n",
    "In this example it would be pretty easy to fix the source of the bug by just checking for zero or converting to a floating-point number first, but let's take a look at how we can recover from failures using Akka.Streams."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Handling Failed Akka.Streams Stages\n",
    "One of the simplest methods we can use to [handle errors in Akka.Streams stages is the `Recover` method](https://getakka.net/articles/streams/error-handling.html#recover)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "dotnet-interactive.csharp"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1/9 is 0 w/ integer division\r\n",
      "1/8 is 0 w/ integer division\r\n",
      "1/7 is 0 w/ integer division\r\n",
      "1/6 is 0 w/ integer division\r\n",
      "Whoops - attempted to divide by zero!\r\n"
     ]
    }
   ],
   "source": [
    "using Akka.Util; // need this for Option<T>, which Recover depends upon\n",
    "\n",
    "IAsyncEnumerable<string> integerDivision2 = numbers.Via(Flow.Create<int>()\n",
    "    .Select(x => $\"1/{x} is {1/x} w/ integer division\"))\n",
    "    .Recover(ex => {\n",
    "        if(ex is DivideByZeroException){\n",
    "            return new Option<string>(\"Whoops - attempted to divide by zero!\");\n",
    "        }\n",
    "\n",
    "        // otherwise just return nothing - a gap will appear in the output\n",
    "        return Option<string>.None;\n",
    "    })\n",
    "    .RunAsAsyncEnumerable(materializer);\n",
    "\n",
    "await foreach(var d in integerDivision2){\n",
    "    Console.WriteLine(d);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this instance, the stream still terminated without processing all of its inputs - but rather than throwing an exception we had a chance to return a final output that could be processed by our `Sink<T>` instead.\n",
    "\n",
    "What if we want something more robust? What if we want the opportunity to retry a failed stream stage and keep processing the rest of the upstream events? This is [where `RecoverWithRetries<T>` can be very helpful](https://getakka.net/articles/streams/error-handling.html#recover-with-retries)!"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "dotnet-interactive.csharp"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1/9 is 0 w/ integer division\r\n",
      "1/8 is 0 w/ integer division\r\n",
      "1/7 is 0 w/ integer division\r\n",
      "1/6 is 0 w/ integer division\r\n",
      "1/5 is 0 w/ integer division\r\n",
      "1/4 is 0 w/ integer division\r\n",
      "1/3 is 0 w/ integer division\r\n",
      "1/2 is 0 w/ integer division\r\n",
      "1/1 is 1 w/ integer division\r\n"
     ]
    }
   ],
   "source": [
    "IAsyncEnumerable<string> integerDivision3 = numbers.Via(Flow.Create<int>()\n",
    "    .Select(x => $\"1/{x} is {1/x} w/ integer division\")\n",
    "    .RecoverWithRetries(ex => {\n",
    "        if(ex is DivideByZeroException){\n",
    "            // have to return a new Source<int> here\n",
    "            return numbers.Via(Flow.Create<int>()\n",
    "                .SkipWhile(x => x != 0)\n",
    "                .Where(x => x != 0) // so we can skip the 0 element itself too\n",
    "                .Select(x => $\"1/{x} is {1/x} w/ integer division\"));\n",
    "        }\n",
    "\n",
    "        // otherwise just bail and don't attempt to recover\n",
    "        return null;\n",
    "    }, attempts: 3)) // allow up to three restart attempts\n",
    "    .RunAsAsyncEnumerable(materializer);\n",
    "\n",
    "await foreach(var d in integerDivision3){\n",
    "    Console.WriteLine(d);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Alright! We got all of the non-zero output this time - but that `RecoverWithRetries` stage was really cumbersome to write for this particular case. I wonder if there's an even better way to do this?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "dotnet_interactive": {
     "language": "csharp"
    },
    "vscode": {
     "languageId": "dotnet-interactive.csharp"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1/9 is 0 w/ integer division\r\n",
      "1/8 is 0 w/ integer division\r\n",
      "1/7 is 0 w/ integer division\r\n",
      "1/6 is 0 w/ integer division\r\n",
      "1/5 is 0 w/ integer division\r\n",
      "1/4 is 0 w/ integer division\r\n",
      "1/3 is 0 w/ integer division\r\n",
      "1/2 is 0 w/ integer division\r\n",
      "1/1 is 1 w/ integer division\r\n"
     ]
    }
   ],
   "source": [
    "using Akka.Streams.Supervision; // needed for stream-stage deciders\n",
    "\n",
    "// create a custom Decider with a \"Restart\" directive in the event of DivideByZeroException\n",
    "Akka.Streams.Supervision.Decider decider = cause => cause is DivideByZeroException\n",
    "    ? Akka.Streams.Supervision.Directive.Restart\n",
    "    : Akka.Streams.Supervision.Directive.Stop;\n",
    "\n",
    "IAsyncEnumerable<string> integerDivision = numbers.Via(Flow.Create<int>()\n",
    "    .Select(x => $\"1/{x} is {1/x} w/ integer division\"))\n",
    "    .WithAttributes(ActorAttributes.CreateSupervisionStrategy(decider))\n",
    "    .RunAsAsyncEnumerable(materializer);\n",
    "\n",
    "await foreach(var d in integerDivision){\n",
    "    Console.WriteLine(d);\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Similar to [how actors are programmed to handle their children's failures, primarily by restarting them](https://petabridge.com/blog/akkadotnet-actors-restart/), so too can Akka.Streams stages be handled using some metadata we provide during stream construction.\n",
    "\n",
    "This is exactly what we've done by [defining a `Decider`](https://getakka.net/api/Akka.Streams.Supervision.Decider.html) - the algorithm that gets used to evaluate a stream stage's `Exception`s when they're thrown:\n",
    "\n",
    "```csharp\n",
    "Akka.Streams.Supervision.Decider decider = cause => cause is DivideByZeroException\n",
    "    ? Akka.Streams.Supervision.Directive.Restart\n",
    "    : Akka.Streams.Supervision.Directive.Stop;\n",
    "```\n",
    "\n",
    "By default a `Directive.Stop` will be returned - hence why we had to go through so much trouble using the `RecoverWithRetries` stage to rebuild the stream - we literally had to recreate the entire upstream part of the stream in order to resume processing.\n",
    "\n",
    "In this case, however, we can simplify our error handling by configuring the underlying actors that are created by the `IMaterializer` to execute the graph to simply restart in the event of a `DivideByZeroException`. This is exactly what we did when we passed in these `ActorAttributes`:\n",
    "\n",
    "```csharp\n",
    ".WithAttributes(ActorAttributes.CreateSupervisionStrategy(decider))\n",
    "```\n",
    "\n",
    "There are a number of other attributes you can configure in your streams, such as:\n",
    "\n",
    "* The name of the stream stage actor;\n",
    "* The dispatcher it runs on;\n",
    "* And a variety of other settings, which need to be more well-documented.\n",
    "\n",
    "But for now you should have the tools to properly manage the lifecycles of your streams."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".NET (C#)",
   "language": "C#",
   "name": ".net-csharp"
  },
  "language_info": {
   "name": "C#"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}