diff --git a/src/main/scala/com/xeppaka/app/App.scala b/src/main/scala/com/xeppaka/app/App.scala index e588fda..b3c3761 100644 --- a/src/main/scala/com/xeppaka/app/App.scala +++ b/src/main/scala/com/xeppaka/app/App.scala @@ -12,6 +12,7 @@ import akka.stream.scaladsl.Source import akka.util.Timeout import com.lightbend.cinnamon.akka.stream.CinnamonAttributes import com.lightbend.cinnamon.akka.stream.CinnamonAttributes.GraphWithInstrumented +import com.lightbend.cinnamon.akka.stream.CinnamonAttributes.FlowWithInstrumented import scala.concurrent.ExecutionContext import scala.concurrent.duration._ @@ -33,14 +34,9 @@ object App { .via(RestartFlow.onFailuresWithBackoff(restartSettings) { () => Flow[Int] .map(identity) + .instrumentedPartial() .throttle(1, 1.second) .via(processor1) - // .map { i => - // if (i % 100000 == 0) { - // throw new Exception("BOOM!") - // } - // i - // } }) .to(RestartSink.withBackoff(restartSettings) { () => Sink.foreach(println)