akka-stream retry flow

akka-stream-retry.md

#Thoughts on retry in streams ##This document is a WIP Too often, when writing akka-stream code, I find my self in need of retrying some logic.

a very common scenario, is when I use connection pools, e.g: mySource.via(Http().superPool()) in this case, what we get from superPool is a flow of type: Flow[(HttpRequest,T),(Try[HttpResponse,T),NotUsed]. we can't do anything with this flow, let alone retry based on some smart decision.

So why not have a generic retry combinator?

since akka-http connection pools yield flows that look like Flow[(A,S),(Try[B],S),_], I decided to base the API on this kind of flows.

assume the existence of this FanOutShape2, which would be easy to implement:

class PartitionWith[In,Out0,Out1](p: In => Either[Out0,Out1]) extends GraphStage[FanOutShape2[In,Out0,Out1]]

and would work very similar to a method Iv'e written in my personal util, to operate on collections.

with this stage and 2 other trivial stages, we can come up with a fairly reasonable API:

def retry[I,O,S,M](flow: Flow[(I,S),(Try[O],S),M])(retryWith: S => Option[(I,S)]): Flow[(I,S),Try[O],M]

which could be implemented like:

the custom increment and decrement stages is needed for graceful stream completion. the increment stage will intercept upstream completion, and will only complete itself if there is no retry leftovers cycling in the graph. if there are such leftovers, it will wait until enough decrement counter messages is received.

PartitionWith stage is created using retryWith as follows:

{
  case (t@Success(_),_) => Right(t)
  case (t,s) => retryWith(s).fold(Right(t))(Left.apply)
}

the generic type S should be some state object that the user can extract another (I,S) - input and state tuple, similar to how you might use unfold.

##Bonus also, we can consider a "family" of retries, such as:

def retryConcat[I,O,S,M](flow: Flow[(I,S),(Try[O],S),M])(retryWith: S => Option[Seq[(I,S)]]): Flow[(I,S),Try[O],M]

//or (empty sequence means to not retry again):
def retryConcat[I,O,S,M](flow: Flow[(I,S),(Try[O],S),M])(retryWith: S => Seq[(I,S)]): Flow[(I,S),Try[O],M]

which might be handy if the input elements I can be broke down to smaller pieces, that may succeed where the uber object failed. e.g: an HttpRequest for some web service bulk API, with heavy body, may be broken down to multiple requests using the regular API.

def retryAsync[I,O,S,M](flow: Flow[(I,S),(Try[O],S),M])(retryWith: S => Future[Option[(I,S)]]): Flow[(I,S),Try[O],M]

which is pretty obvious. if I had the base version analogous to unfold, this may be the unfoldAsync equivalent.

feedback, and c&c are very welcome.

PartitionWith.scala

package hochgi.util

package object collections {
  /**
   * `partition` and `map` combined.
   * for a given collection, and a function from the collection elements to `Either[A,B]`,
   * generates a tuple of 2 collections of types `A` and `B`
   *
   * @param xs the collection of elements
   * @param f a function that convert an element to an `Either[A,B]`
   * @tparam A left collection elements' type
   * @tparam B right collection elements' type
   * @tparam T original collection ekements' type
   * @tparam Coll collection's type
   */
  def partitionWith[A, B, T, Coll[_]]
  (xs: Coll[T])
  (f: T => Either[A,B])
  (implicit ev: Coll[T] <:< TraversableLike[T,Coll[T]],
   cbf1: CanBuildFrom[Coll[T], A, Coll[A]],
   cbf2: CanBuildFrom[Coll[T], B, Coll[B]]): (Coll[A],Coll[B]) = {

    val b1 = cbf1(xs)
    val b2 = cbf2(xs)

    for(x <- xs) f(x) match {
      case Left(a) => b1 += a
      case Right(b) => b2 += b
    }

    b1.result() -> b2.result()
  }
}

retry-concat-stream-diagram.png

retry-concat-stream-diagram.svg

<?xml version="1.0"?>
<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN" "http://www.w3.org/Graphics/SVG/1.1/DTD/svg11.dtd">
<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="1007px" height="167px" version="1.1" content="&lt;mxfile userAgent=&quot;Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:47.0) Gecko/20100101 Firefox/47.0&quot; version=&quot;5.5.2.3&quot; editor=&quot;www.draw.io&quot; type=&quot;device&quot;&gt;&lt;diagram&gt;5V1bc6M4E/01rpp9CIUQ18exJ7vfPGx9qfJWze4jYxObGmyymEyS/fXTGDUXiZsxYJH4wQaBkTnndEtqtfCCrg6vf0Tu0/7PcOsFC03dvi7ol4WmEYvo8JGUvKUlumOlBbvI37KT8oK1/5/HClVW+uxvvVPpxDgMg9h/KhduwuPR28SlMjeKwpfyaY9hUK71yd1hjXnBeuMGYuk3fxvv01JbM/Py/3n+bo81E9NJj3x3Nz92Ufh8ZPUtNPp4fqWHDy5ei93oae9uw5dCEb0HXKMwhCsnW4fXlRck2CJs6fd+rzma/e7IO7Lf1vwFjabf+OkGz+ze2Q+L3xCMl70fe+snd5PsvwDhC7rcx4cA9ghsnuIo/JHBBHewFH8E+10/vSj2UCTZ3YKKvPDgxdEbnMKOWqqtGBq1VSd9Z8AzOWm6odgGqExL3xn4LzlVBAHeF2gyWZnL1LHLqs0Rgg0GUg1gTNmyAWZrjYAZrYBV4IVlV+FltOPlHbefE6OFvWN4hMIiWgBJ9PY37Ki480+yo8CFYf/Vj5Njd6qimhYrSI4TRaWU7T94kQ+/24vYRWrRPoXP0ZmzAs/etuQpRPwLGBoVGGJZ5AVu7P8s+5cqYFkND6EPvy6nlzTSa1XSi5dP74tdsegM+EpouRJGHVZiqk2VxG6082KhkrM+Mri6SYbdXH/JMFWcFcMEgdvvQgzUqmIiE4OpEEfVLZO9MxqvVQMnucnUwHoOM3O49s0crj1Th8t4noXDpeooDpfTkDOVibErS+RwJRNDo8Ol6iQOdyo1UIaZbA7XdJq7QJXwTOBwKd7JzBwu8iyJjZn2FervamItw8rmbvSAJsYUIo/DlUwMltrYw3XIIA5XFjVIGoNBLvMe/3RhFippmEXnMLEqMBmrnZlpJAWpTA2mLHlJ3A16v7yx6edQUPl4HbSOEXyGdDESpFkSSnVHocUXzkgU+hOl4/0I512kxbVEAxIuaRiEc4hUndAhzjTSgVTK6fwo77R62gKESEazBeniFZJR2uL8gOIRnN+IhKOTkc35Eez+5VHfybyfPtOwA3IpiakQrs8mqLirNRC7pRM5oDlIFz6QjVSLd4BC9H4A/zch46akIQLNNJU8QgLvt4sYmJJGDDSjEaIJWwxzpgEEZFYS56LpHKEMxWsbkBZbGtG5SBdLkI1xYKYQbYb3MZqT2/EvaWiB2mVA2HD/Fp5zppEGZFaWSANkJtUT2ttxUqdRJyMajnRxCNkIB2bGd5w34x9H/9LFZB1dMQsvi4t9Q+peIckTkxun6KRbHYIWN8ktIRxiQm5JC2AjNTzYt5xbw4M8y5JbojbSm8wFifRePPmnlSsRswkaKhnQKXUIFUzbKMkmBsS+QH+x+UB1zJV/SeMgzS4W5g1v5WJnGhVBnmfhYs8zTmO72HN3chITky5gIpsYOBeb0D++i52Of1lzMezmfj+sSbmRj51p/ASJlmVa32rkt0X/XY2sbfTY3Fse0Mqki67IJgdH5H8ILyuLAOzLBr+bwD2d/E21BnA7dRtnr1HjUpJjF8XainmtsikE2kvS5DBoJdGXC4bPoWi+7oASuWywWy2RBiHUy6dWIqU05+q0iKrh8c26aphMl68e7ruYQlNsGE3ZpqaatoOTzYUHEBgGsSBtiVoOJePlitkdxr/XSKKWeGnWKgqLZXouR3RaLjQgZ10GVADvmu2evN0B4L/Pi4Clvg3BRU8AKFm3XIMuovLLGPhs9M4z0TWLXFqIB/zd5HbxtKfkhNMFSmUzN7WC5LMtLz2f4SOc3/U4wzPXeXrHvVXfYRg5muqv7v5go16yCMmWbePDZ65NbedT5Pkx8UAGYdbkotaGQ/jzGW51Am49Xq7veoGLw/7V8ykOD1D29biJvETOC80MgPbl9wi2dsmWYAQQ/kjKhfDJKgzCZADHhn6PfhBwRW7g746JcUBFyVhvmQRTfHjO12d24OBvt2dLqorXhHD2Y3C2rj2c58EXxlhXiWmkbROwvHr7RGJscVCdUfLF+7CUwEpgboq8IrloLE6wX1Xg5MEFTGI/PH7zofqPZyHoYBtMBMecg7MhDinPd/jhSKAqn6rMLlEgAfP+BydBnJT79HWhrda/vXfUifhck+lQFzuka+/fhbFE7I20j/quRc8lJepsvwA/9rpK8LOu+1Xwi92lT3/BJYzl/8/IfwD167zjr1B/JfxDqF/sGhXRf9/Am7icMBtbMzc0gdshqtgB+jjI2+hh+DDbJMiLUyzZKGAFTzFOYHrvHh/T0xB/7PaUuv6sUzg8AR0inz1TCQQIKoDq/JybCRdbECRAumyBxmmcjMruIFNHgSfhqKoDw0+LWhgu4RPXOwbL+j/xoe/CCz6QMV6mPcFu2YTJAEOzrYPgMqpxhVtmX7riJI9FwjNEcxuG+rYnQwyzImNKYVyWJdJzuk/QQnfa74ijOIWXjZNdo5v5HTct3/uxVnxjMCKdROwKzmfgf1WLe8fPVk048CdkiDyKK+aa2u2r+nkTxcmn1KU2BFb4Lo3T8sDYYaywY70Xz9GO5mNhN///kfT0/E9e6P0v&lt;/diagram&gt;&lt;/mxfile&gt;">
  <defs/>
  <g transform="translate(0.5,0.5)">
    <rect x="806.52" y="39.86" width="120" height="60" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <rect x="926.52" y="49.86" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 925.67 60.33 L 936.52 54.86" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 926.5 49.5 L 936.52 54.86" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <rect x="926.52" y="79.86" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 925.67 90.33 L 936.52 84.86" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 926.5 79.5 L 936.52 84.86" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <rect x="796.52" y="64.86" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 795.67 75.33 L 806.52 69.86" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 796.5 64.5 L 806.52 69.86" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <rect x="148" y="54" width="120" height="60" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <rect x="138" y="64" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 137.33 74.5 L 148.17 68.67" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 138.17 63.67 L 148 69" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <rect x="138" y="94" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 137.33 104.5 L 148 99" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 138.17 93.67 L 148 99" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <rect x="268" y="79" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 267.33 89.5 L 278 84" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 268.17 78.67 L 278 84" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <rect x="364.71" y="54" width="120" height="60" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <rect x="354.71" y="79" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 354 89.5 L 364.71 84" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 354.83 78.67 L 364.71 84" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <rect x="484.71" y="79" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 484 89.5 L 494.71 84" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 484.83 78.67 L 494.71 84" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <rect x="592.67" y="54.38" width="120" height="60" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <rect x="712.67" y="64.38" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 712.33 74.5 L 722.67 69.38" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 712.33 64.5 L 722.67 69.38" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <rect x="712.67" y="94.38" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 712.33 104.5 L 722.67 99.38" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 712.33 94.5 L 722.67 99.38" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <rect x="582.67" y="79.38" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 582.33 89.5 L 592.67 84.38" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 582.33 79.5 L 592.67 84.38" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 494.83 83.67 L 575.97 84.44" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 581.22 84.49 L 574.18 87.92 L 575.97 84.44 L 574.25 80.92 Z" fill="#000000" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 722.33 69.5 L 790.13 69.5" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 795.38 69.5 L 788.38 73 L 790.13 69.5 L 788.38 66 Z" fill="#000000" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 937 85 L 990.63 85" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 995.88 85 L 988.88 88.5 L 990.63 85 L 988.88 81.5 Z" fill="#000000" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 936.5 54.5 L 936.5 53.25 Q 936.5 52 946.5 52 L 958.17 52 Q 968.17 52 968.17 42 L 968.17 32.83 Q 968.17 23.67 958.17 23.67 L 108.17 23.67 Q 98.17 23.67 98.17 33.67 L 98.17 58.67 Q 98.17 68.67 108.17 68.67 L 131.63 68.67" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 136.88 68.67 L 129.88 72.17 L 131.63 68.67 L 129.88 65.17 Z" fill="#000000" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 722.33 99.5 L 748.17 99.5 Q 758.17 99.5 758.17 109.5 L 758.17 133.67 Q 758.17 143.67 748.17 143.67 L 108.17 143.67 Q 98.17 143.67 98.17 133.67 L 98.17 108.67 Q 98.17 98.67 108.17 98.67 L 131.8 98.67" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 137.05 98.67 L 130.05 102.17 L 131.8 98.67 L 130.05 95.17 Z" fill="#000000" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <g transform="translate(157.5,77.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="99" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 16px; max-width: 116px; width: 100px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">Custom Increment</div>
          </div>
        </foreignObject>
        <text x="50" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">Custom Increment</text>
      </switch>
    </g>
    <g transform="translate(813.5,62.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="105" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 16px; max-width: 116px; width: 106px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">Custom Decrement</div>
          </div>
        </foreignObject>
        <text x="53" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">Custom Decrement</text>
      </switch>
    </g>
    <g transform="translate(618.5,77.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="69" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 16px; max-width: 86px; width: 70px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">PartitionWith</div>
          </div>
        </foreignObject>
        <text x="35" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">PartitionWith</text>
      </switch>
    </g>
    <g transform="translate(413.5,78.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="22" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 16px; max-width: 36px; width: 23px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">flow</div>
          </div>
        </foreignObject>
        <text x="11" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">flow</text>
      </switch>
    </g>
    <g transform="translate(302.5,67.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="22" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 16px; max-width: 36px; width: 23px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">(I,S)</div>
          </div>
        </foreignObject>
        <text x="11" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">(I,S)</text>
      </switch>
    </g>
    <g transform="translate(410.5,146.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="50" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 18px; max-width: 56px; width: 51px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">Seq[(I,S)]</div>
          </div>
        </foreignObject>
        <text x="25" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">Seq[(I,S)]</text>
      </switch>
    </g>
    <g transform="translate(512.5,67.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="50" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 16px; max-width: 56px; width: 51px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">(Try[O],S)</div>
          </div>
        </foreignObject>
        <text x="25" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">(Try[O],S)</text>
      </switch>
    </g>
    <g transform="translate(743.5,53.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="31" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 16px; max-width: 36px; width: 32px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">Try[O]</div>
          </div>
        </foreignObject>
        <text x="16" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">Try[O]</text>
      </switch>
    </g>
    <g transform="translate(950.5,88.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="31" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 16px; max-width: 36px; width: 32px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">Try[O]</div>
          </div>
        </foreignObject>
        <text x="16" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">Try[O]</text>
      </switch>
    </g>
    <g transform="translate(478.5,4.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="104" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 16px; max-width: 119px; width: 105px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">DecrementCounter</div>
          </div>
        </foreignObject>
        <text x="52" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">DecrementCounter</text>
      </switch>
    </g>
    <rect x="138" y="79" width="10" height="10" fill="#ffffff" stroke="#000000" stroke-width="2" pointer-events="none"/>
    <path d="M 137.33 89.5 L 148 84" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 138.17 78.67 L 148 84" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 8 84 L 131.63 84" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 136.88 84 L 129.88 87.5 L 131.63 84 L 129.88 80.5 Z" fill="#000000" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <g transform="translate(36.5,67.5)">
      <switch>
        <foreignObject style="overflow:visible;" pointer-events="all" width="22" height="12" requiredFeatures="http://www.w3.org/TR/SVG11/feature#Extensibility">
          <div xmlns="http://www.w3.org/1999/xhtml" style="display: inline-block; font-size: 12px; font-family: Helvetica; color: rgb(0, 0, 0); line-height: 1.2; vertical-align: top; overflow: hidden; max-height: 16px; max-width: 36px; width: 23px; white-space: normal; word-wrap: normal; text-align: center;">
            <div xmlns="http://www.w3.org/1999/xhtml" style="display:inline-block;text-align:inherit;text-decoration:inherit;">(I,S)</div>
          </div>
        </foreignObject>
        <text x="11" y="12" fill="#000000" text-anchor="middle" font-size="12px" font-family="Helvetica">(I,S)</text>
      </switch>
    </g>
    <path d="M 278.17 83.67 L 348.47 83.67" fill="none" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
    <path d="M 353.72 83.67 L 346.72 87.17 L 348.47 83.67 L 346.72 80.17 Z" fill="#000000" stroke="#000000" stroke-miterlimit="10" pointer-events="none"/>
  </g>
</svg>

retry-stream-diagram.svg

retry-stream.png

That's interesting. Do you have an implementation for "Custom Increment" and "Custom Decrement" stages?

@oridag it was implemented in a much simpler way using BidiFlow. Check implementation at:
https://github.com/akka/akka-stream-contrib

There was no need for the custom increment & decrement.