# Custom GraphStage

**URL:** <https://discuss.akka.io/t/custom-graphstage/10971>\
**Category:** Akka Libraries\
**Tags:** streams\
**Created:** [October 2, 2024, 2:30pm UTC](https://discuss.akka.io/t/custom-graphstage/10971 "2024-10-02T14:30:15Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![bcoromina](https://avatars.discourse-cdn.com/v4/letter/b/ed655f/32.png) [@bcoromina](https://discuss.akka.io/u/bcoromina)\
**Post date:** [October 2, 2024, 2:30pm UTC](https://discuss.akka.io/t/custom-graphstage/10971/1 "2024-10-02T14:30:15Z")

</div>

Hi,

I’m thinking about using a custom GraphStage like the following:

case class MapAsyncErrorUnordered[In, Out](parallelism: Int, f: In =\> Future[Out])  
extends GraphStage[FanOutShape2[In, Out, Throwable]]

So if Future is success an element will be generated in output 1 and, if Future ends with failure, a Throwable wil be generated in output 2. This would allow me to apply different pipeline for errors and successful cases.

If I go this way like I sketched up here [custom-shapes/src/main/scala/shapes/MapAsyncErrorUnordered.scala at main · bcoromina/custom-shapes · GitHub](https://github.com/bcoromina/custom-shapes/blob/main/src/main/scala/shapes/MapAsyncErrorUnordered.scala), looks like I break encapsulation of graph component so I should test by myself that I’m not breaking things like correct backpreasure handling in my custom shape.

If you do think this is the way to go I would appreciate some advice for testing the new shape, otherwise, maybe you could suggest a better aproach.

Thanks in advance,

Bernat

---

<div class="post-metadata">

**Author:** ![johanandren](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/johanandren/32/85_2.png) [@johanandren](https://discuss.akka.io/u/johanandren)\
**Post date:** [October 3, 2024, 7:25am UTC](https://discuss.akka.io/t/custom-graphstage/10971/2 "2024-10-03T07:25:44Z")

</div>

Note that it is relatively straight forward to achieve this without a custom graph stage.

Combine existing stream operators. Start `mapAsyncUnordered` but wrap the `In => Future[Out]` lambda and returned future with a function that “lifts” future errors to Scala `In => Future[Try[T]]`, a `partition` that directs error and success down different downstream and finally a `map` in each branch to unwrap the `Try[T]` into `T` in the success branch and `Throwable` in the error branch.

A custom graph stage could be an interesting exercise but I would recommend that unless a composed solution is tried out and turns out to be a bottle neck or have too much overhead.

---

<div class="post-metadata">

**Author:** ![bcoromina](https://avatars.discourse-cdn.com/v4/letter/b/ed655f/32.png) [@bcoromina](https://discuss.akka.io/u/bcoromina)\
**Post date:** [October 3, 2024, 2:00pm UTC](https://discuss.akka.io/t/custom-graphstage/10971/3 "2024-10-03T14:00:12Z")

</div>

Thanks a lot for your answer!

I would like to build something similar than this:

 ![image](https://us1.discourse-cdn.com/flex020/uploads/discuss_lightbend/original/2X/a/a30ccc55d2dd20367cdb4227ab18b7188ba99e1e.png)

where CustomFlow1 can produce errors. You suggest to model CustomFlow1with one Input and one Output, like In =\> Future[Try[T]]. The problem I see with that is that in case of CustomFlow1 producing an Error I would like not to call CustomFlow2. With In =\> Future[Try[T]], if I’ve understood well, I will be forced to check is income is error in each stage of the pipeline forceing a message to be sent accross multiple async boundaries. If I could implement a graph like it appears in the picture I would have a separate success path at graph level.  
And separate error handling logic from the success path.

Does any of this makes sense?

---

<div class="post-metadata">

**Author:** ![johanandren](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/johanandren/32/85_2.png) [@johanandren](https://discuss.akka.io/u/johanandren)\
**Post date:** [October 4, 2024, 1:42pm UTC](https://discuss.akka.io/t/custom-graphstage/10971/4 "2024-10-04T13:42:08Z")

</div>

With the suggested composed operator, and probably also if you go with a custom stage, each of those “custom flow” boxes would be one instance of it, so that it funnels successes on, but routes the errors down to the error handling subflow.

A composed operator could be something like this (compiles but I didn’t test using it):

```scala
def partitionErrors[In, Out](f: In => Future[Out]): Graph[FanOutShape2[In, Out, Throwable], NotUsed] = 
  GraphDSL.create() { implicit builder =>
    import GraphDSL.Implicits._
    val liftedFlow =
      builder.add(Flow[In].mapAsync(1)(in => f(in).transform(tryIn => Success(tryIn))(ExecutionContext.parasitic)))
    val partition = builder.add(Partition[Try[Out]](2, {
      case Success(_) => 0
      case Failure(_) => 1
    }))

    liftedFlow.out ~> partition.in
    val successOut = partition.out(0).map(_.get)
    val throwableOut = partition.out(1).map(_.failed.get)

    new FanOutShape2(liftedFlow.in, successOut.outlet, throwableOut.outlet)
  }

```

---

<div class="post-metadata">

**Author:** ![bcoromina](https://avatars.discourse-cdn.com/v4/letter/b/ed655f/32.png) [@bcoromina](https://discuss.akka.io/u/bcoromina)\
**Post date:** [October 4, 2024, 2:12pm UTC](https://discuss.akka.io/t/custom-graphstage/10971/5 "2024-10-04T14:12:30Z")

</div>

Now I get what ‘partition’ from your fisrt answer means! Thanks!