Re: [akka-user] Stream source fromPublisher does not honor the SupervisionStrategy ?

[antonin perrot-audet]

As promised, I tested it and come back with some results :

This code works as expected :
 - when an error is transmitted from the source via the onError channel of 
the stream:
  - The source is recreated 2 times,
and the message "Source error management" is printed.
  - The third time, the stream collapses as expected.
 
implicit val materializer = ActorMaterializer(ActorMaterializerSettings(
system2)


// source declaration
val publiString: Publisher[String] = MongoClients.create(mongoSettings).
listDatabaseNames()
val source: Source[String, NotUsed] = Source.fromPublisher(publiString)


// source supervision
val source2 = source.recoverWithRetries(
  2,
  {
case _ => {
  println("Source error management")
  // declare a new source
  val sourceBis: Source[String, NotUsed] =
Source.fromPublisher(MongoClients.create(mongoSettings).
listDatabaseNames())
  sourceBis
}
  }
)


// define a printing stage
val printFlow: Flow[String, String, NotUsed] = Flow.fromFunction({ s => 
println(s); s })


// wire the source with the printing stage and a sink, run it and 
terminate once done.
source2.via(printFlow).runWith(Sink.ignore).onComplete(_ => system2.
terminate())



to wrap it up :


# To have exhaustive error management:

## One should define a Supervision.Decider

val errorDecider: Supervision.Decider = {
  case e => {
println("Stream error management" + e)
Supervision.resume
  }
}


 and associate it with the ActorMaterializer, (or with individua stream 
processing stages) :
implicit val materializer = ActorMaterializer(ActorMaterializerSettings(
system2)
  .withSupervisionStrategy(errorDecider))


The Supervision.Decider
deals with :
 - errors produced inside explicitely defined akka Flows or junctions,
but does not deal with :
 - errors produced by ZipWith, GraphStage, actorPublisher, ActorSubscriber, 
and Source.fromPublisher() sources.


## One should define some recoverWith partial functions

For stages that do not support Supervision.Decider:

someSupervisedStage = someStage.recoverWithRetries(
  2,
  {
case _ => {
  println("Stage error management")
  [recover logic producing a new source]
}
  }




Is that correct ?



Thank you again for your help and patience Konrad.



Best regards,
Antonin PA

Le mardi 21 février 2017 17:20:15 UTC+1, Konrad Malawski a écrit :
>
> Thanks, that's good feedback - we should include mention of the recover 
> methods in there more prominently :)
>
> -- 
> Konrad `ktoso` Malawski
> Akka  @ Lightbend 
>
> On 22 February 2017 at 01:14:35, antonin perrot-audet (perro...@gmail.com 
> ) wrote:
>
> Thank you Konrad, I think that this is what I am looking for, I'll check 
> them out and let you know. 
>
> I was blindly following the 7.12 "error handling" chapter of the doc, and 
> though that defining a strategy for managing errors in one place this way 
> was quite neat.
>
> Best,
>
> Antonin PA
>
> Le 21 févr. 2017 16:48, "Konrad Malawski"  > a écrit :
>
>> I don't think you need supervision for those things actually - have you 
>> looked at the various `recover*` methods defined on Source/Flow?
>> You can recover from failures using those.
>>
>> -- 
>> Konrad `ktoso` Malawski
>> Akka  @ Lightbend 
>>
>> On 22 February 2017 at 00:47:01, antonin perrot-audet (perro...@gmail.com 
>> ) wrote:
>>
>> Hi Konrad,
>>
>> thanks a lot, your answer does help a lot.
>>
>> - My bad, the onError from MongoDb publisher returns normally but passes 
>> an Exception object, their implementation seems to be correct according to 
>> the reactive stream spec.
>> - Ok, so the supervision does not work for the source created via 
>> fromPublisher(), is there a way to catch this propagated failure somewhere 
>> else in the stream ? Would it be better to wrap the publisher in an actor ?
>>
>> Best,
>> Antonin PA
>>
>>
>> Le mardi 21 février 2017 16:37:37 UTC+1, Konrad Malawski a écrit : 
>>>
>>> Hi Antonin,
>>> Two things here:
>>>
>>> One: "throws an error on the onError callback"? This is not allowed by 
>>> the spec:
>>>
>>> Calling onSubscribe, onNext, onError or onComplete MUST return normally 
>>> except when any provided parameter is null in which case it MUST throw 
>>> a java.lang.NullPointerException to the caller, for all other 
>>> situations the only legal way for a Subscriber to signal failure is by 
>>> cancelling its Subscription. In the case that this rule is violated, 
>>> any associated Subscription to the Subscriber MUST be considered as 
>>> cancelled, and the caller MUST raise this error condition in a fashion that 
>>> is adequate for the runtime environment.
>>> https://github.com/reactive-streams/reactive-streams-jvm#2.13
>>>
>>> Is that mongo implementation surely tested and conforming to the 
>>> react

Re: [akka-user] Stream source fromPublisher does not honor the SupervisionStrategy ?

[Konrad Malawski]

Thanks, that's good feedback - we should include mention of the recover
methods in there more prominently :)

-- 
Konrad `ktoso` Malawski
Akka  @ Lightbend 

On 22 February 2017 at 01:14:35, antonin perrot-audet (perrotau...@gmail.com)
wrote:

Thank you Konrad, I think that this is what I am looking for, I'll check
them out and let you know.

I was blindly following the 7.12 "error handling" chapter of the doc, and
though that defining a strategy for managing errors in one place this way
was quite neat.

Best,

Antonin PA

Le 21 févr. 2017 16:48, "Konrad Malawski"  a
écrit :

> I don't think you need supervision for those things actually - have you
> looked at the various `recover*` methods defined on Source/Flow?
> You can recover from failures using those.
>
> --
> Konrad `ktoso` Malawski
> Akka  @ Lightbend 
>
> On 22 February 2017 at 00:47:01, antonin perrot-audet (
> perrotau...@gmail.com) wrote:
>
> Hi Konrad,
>
> thanks a lot, your answer does help a lot.
>
> - My bad, the onError from MongoDb publisher returns normally but passes
> an Exception object, their implementation seems to be correct according to
> the reactive stream spec.
> - Ok, so the supervision does not work for the source created via
> fromPublisher(), is there a way to catch this propagated failure somewhere
> else in the stream ? Would it be better to wrap the publisher in an actor ?
>
> Best,
> Antonin PA
>
>
> Le mardi 21 février 2017 16:37:37 UTC+1, Konrad Malawski a écrit :
>>
>> Hi Antonin,
>> Two things here:
>>
>> One: "throws an error on the onError callback"? This is not allowed by
>> the spec:
>>
>> Calling onSubscribe, onNext, onError or onComplete MUST return normally
>> except when any provided parameter is null in which case it MUST throw a
>> java.lang.NullPointerException to the caller, for all other situations
>> the only legal way for a Subscriber to signal failure is by cancelling
>> its Subscription. In the case that this rule is violated, any associated
>> Subscription to the Subscriber MUST be considered as cancelled, and the
>> caller MUST raise this error condition in a fashion that is adequate for
>> the runtime environment.
>> https://github.com/reactive-streams/reactive-streams-jvm#2.13
>>
>> Is that mongo implementation surely tested and conforming to the reactive
>> streams spec?
>>
>>
>> Two: Supervision does not work for arbitrary 3rd party publishers - it
>> only works within Akka Streams (specific stages, specifically handle it),
>> it's an additional feature Akka Streams provide over what Reactive Streams
>> do.
>>
>> Hope this helps
>>
>> --
>> Konrad `ktoso` Malawski
>> Akka  @ Lightbend 
>>
>> On 22 February 2017 at 00:04:06, antonin perrot-audet (perro...@gmail.com)
>> wrote:
>>
>> Hello,
>>
>> has anyone succeded at having a source fromPublisher() conform to the
>> SupervisionStrategy defined in the ActorMaterializer ?
>>
>>
>> I have that publisher  :
>> MongoClients.create(mongoSettings).listDatabaseNames()
>> That throws an exception on the onError callback, but the errorDecider
>> never gets called. The stream colapses on a failed Future :
>>
>>
>> implicit val system2 = ActorSystem("Sys2")
>> import system2.dispatcher
>>
>> val errorDecider : Supervision.Decider = {
>>   case _ => {
>> println("errorDecider does something, yeiii !")
>> Supervision.stop}
>> }
>>
>> implicit val materializer = ActorMaterializer(ActorMateria
>> lizerSettings(system2)
>>   .withSupervisionStrategy(errorDecider))
>>
>> val publiString : Publisher[String] = MongoClients.create(mongoSetti
>> ngs).listDatabaseNames()
>> val stream: Source[String, NotUsed] = Source.fromPublisher(publiStri
>> ng)
>>
>> stream.map( {s => println(s);s}).runWith(Sink.ignore).onComplete(_ =>
>> system2.terminate())
>>
>>
>>
>> thanks in advace for your responses.
>>
>> Best,
>>
>> Antonin PA
>> --
>> >> Read the docs: http://akka.io/docs/
>> >> Check the FAQ: http://doc.akka.io/docs/akka/c
>> urrent/additional/faq.html
>> >> Search the archives: https://groups.google.com/group/akka-user
>> ---
>> You received this message because you are subscribed to the Google Groups
>> "Akka User List" group.
>> To unsubscribe from this group and stop receiving emails from it, send an
>> email to akka-user+...@googlegroups.com.
>> To post to this group, send email to akka...@googlegroups.com.
>> Visit this group at https://groups.google.com/group/akka-user.
>> For more options, visit https://groups.google.com/d/optout.
>>
>> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ: http://doc.akka.io/docs/akka/
> current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscrib

Re: [akka-user] Stream source fromPublisher does not honor the SupervisionStrategy ?

[antonin perrot-audet]

Thank you Konrad, I think that this is what I am looking for, I'll check
them out and let you know.

I was blindly following the 7.12 "error handling" chapter of the doc, and
though that defining a strategy for managing errors in one place this way
was quite neat.

Best,

Antonin PA

Le 21 févr. 2017 16:48, "Konrad Malawski"  a
écrit :

> I don't think you need supervision for those things actually - have you
> looked at the various `recover*` methods defined on Source/Flow?
> You can recover from failures using those.
>
> --
> Konrad `ktoso` Malawski
> Akka  @ Lightbend 
>
> On 22 February 2017 at 00:47:01, antonin perrot-audet (
> perrotau...@gmail.com) wrote:
>
> Hi Konrad,
>
> thanks a lot, your answer does help a lot.
>
> - My bad, the onError from MongoDb publisher returns normally but passes
> an Exception object, their implementation seems to be correct according to
> the reactive stream spec.
> - Ok, so the supervision does not work for the source created via
> fromPublisher(), is there a way to catch this propagated failure somewhere
> else in the stream ? Would it be better to wrap the publisher in an actor ?
>
> Best,
> Antonin PA
>
>
> Le mardi 21 février 2017 16:37:37 UTC+1, Konrad Malawski a écrit :
>>
>> Hi Antonin,
>> Two things here:
>>
>> One: "throws an error on the onError callback"? This is not allowed by
>> the spec:
>>
>> Calling onSubscribe, onNext, onError or onComplete MUST return normally
>> except when any provided parameter is null in which case it MUST throw a
>> java.lang.NullPointerException to the caller, for all other situations
>> the only legal way for a Subscriber to signal failure is by cancelling
>> its Subscription. In the case that this rule is violated, any associated
>> Subscription to the Subscriber MUST be considered as cancelled, and the
>> caller MUST raise this error condition in a fashion that is adequate for
>> the runtime environment.
>> https://github.com/reactive-streams/reactive-streams-jvm#2.13
>>
>> Is that mongo implementation surely tested and conforming to the reactive
>> streams spec?
>>
>>
>> Two: Supervision does not work for arbitrary 3rd party publishers - it
>> only works within Akka Streams (specific stages, specifically handle it),
>> it's an additional feature Akka Streams provide over what Reactive Streams
>> do.
>>
>> Hope this helps
>>
>> --
>> Konrad `ktoso` Malawski
>> Akka  @ Lightbend 
>>
>> On 22 February 2017 at 00:04:06, antonin perrot-audet (perro...@gmail.com)
>> wrote:
>>
>> Hello,
>>
>> has anyone succeded at having a source fromPublisher() conform to the
>> SupervisionStrategy defined in the ActorMaterializer ?
>>
>>
>> I have that publisher  :
>> MongoClients.create(mongoSettings).listDatabaseNames()
>> That throws an exception on the onError callback, but the errorDecider
>> never gets called. The stream colapses on a failed Future :
>>
>>
>> implicit val system2 = ActorSystem("Sys2")
>> import system2.dispatcher
>>
>> val errorDecider : Supervision.Decider = {
>>   case _ => {
>> println("errorDecider does something, yeiii !")
>> Supervision.stop}
>> }
>>
>> implicit val materializer = ActorMaterializer(ActorMateria
>> lizerSettings(system2)
>>   .withSupervisionStrategy(errorDecider))
>>
>> val publiString : Publisher[String] = MongoClients.create(mongoSetti
>> ngs).listDatabaseNames()
>> val stream: Source[String, NotUsed] = Source.fromPublisher(publiStri
>> ng)
>>
>> stream.map( {s => println(s);s}).runWith(Sink.ignore).onComplete(_ =>
>> system2.terminate())
>>
>>
>>
>> thanks in advace for your responses.
>>
>> Best,
>>
>> Antonin PA
>> --
>> >> Read the docs: http://akka.io/docs/
>> >> Check the FAQ: http://doc.akka.io/docs/akka/c
>> urrent/additional/faq.html
>> >> Search the archives: https://groups.google.com/group/akka-user
>> ---
>> You received this message because you are subscribed to the Google Groups
>> "Akka User List" group.
>> To unsubscribe from this group and stop receiving emails from it, send an
>> email to akka-user+...@googlegroups.com.
>> To post to this group, send email to akka...@googlegroups.com.
>> Visit this group at https://groups.google.com/group/akka-user.
>> For more options, visit https://groups.google.com/d/optout.
>>
>> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ: http://doc.akka.io/docs/akka/
> current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://grou

Re: [akka-user] Stream source fromPublisher does not honor the SupervisionStrategy ?

[Konrad Malawski]

I don't think you need supervision for those things actually - have you
looked at the various `recover*` methods defined on Source/Flow?
You can recover from failures using those.

-- 
Konrad `ktoso` Malawski
Akka  @ Lightbend 

On 22 February 2017 at 00:47:01, antonin perrot-audet (perrotau...@gmail.com)
wrote:

Hi Konrad,

thanks a lot, your answer does help a lot.

- My bad, the onError from MongoDb publisher returns normally but passes an
Exception object, their implementation seems to be correct according to the
reactive stream spec.
- Ok, so the supervision does not work for the source created via
fromPublisher(), is there a way to catch this propagated failure somewhere
else in the stream ? Would it be better to wrap the publisher in an actor ?

Best,
Antonin PA


Le mardi 21 février 2017 16:37:37 UTC+1, Konrad Malawski a écrit :
>
> Hi Antonin,
> Two things here:
>
> One: "throws an error on the onError callback"? This is not allowed by the
> spec:
>
> Calling onSubscribe, onNext, onError or onComplete MUST return normally
> except when any provided parameter is null in which case it MUST throw a
> java.lang.NullPointerException to the caller, for all other situations
> the only legal way for a Subscriber to signal failure is by cancelling
> its Subscription. In the case that this rule is violated, any associated
> Subscription to the Subscriber MUST be considered as cancelled, and the
> caller MUST raise this error condition in a fashion that is adequate for
> the runtime environment.
> https://github.com/reactive-streams/reactive-streams-jvm#2.13
>
> Is that mongo implementation surely tested and conforming to the reactive
> streams spec?
>
>
> Two: Supervision does not work for arbitrary 3rd party publishers - it
> only works within Akka Streams (specific stages, specifically handle it),
> it's an additional feature Akka Streams provide over what Reactive Streams
> do.
>
> Hope this helps
>
> --
> Konrad `ktoso` Malawski
> Akka  @ Lightbend 
>
> On 22 February 2017 at 00:04:06, antonin perrot-audet (perro...@gmail.com)
> wrote:
>
> Hello,
>
> has anyone succeded at having a source fromPublisher() conform to the
> SupervisionStrategy defined in the ActorMaterializer ?
>
>
> I have that publisher  :
> MongoClients.create(mongoSettings).listDatabaseNames()
> That throws an exception on the onError callback, but the errorDecider
> never gets called. The stream colapses on a failed Future :
>
>
> implicit val system2 = ActorSystem("Sys2")
> import system2.dispatcher
>
> val errorDecider : Supervision.Decider = {
>   case _ => {
> println("errorDecider does something, yeiii !")
> Supervision.stop}
> }
>
> implicit val materializer = ActorMaterializer(ActorMateria
> lizerSettings(system2)
>   .withSupervisionStrategy(errorDecider))
>
> val publiString : Publisher[String] = MongoClients.create(mongoSetti
> ngs).listDatabaseNames()
> val stream: Source[String, NotUsed] = Source.fromPublisher(publiString
> )
>
> stream.map( {s => println(s);s}).runWith(Sink.ignore).onComplete(_ =>
> system2.terminate())
>
>
>
> thanks in advace for your responses.
>
> Best,
>
> Antonin PA
> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ: http://doc.akka.io/docs/akka/
> current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+...@googlegroups.com.
> To post to this group, send email to akka...@googlegroups.com.
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
> --
>> Read the docs: http://akka.io/docs/
>> Check the FAQ:
http://doc.akka.io/docs/akka/current/additional/faq.html
>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups
"Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send an
email to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to t

Re: [akka-user] Stream source fromPublisher does not honor the SupervisionStrategy ?

[antonin perrot-audet]

Hi Konrad,

thanks a lot, your answer does help a lot.

- My bad, the onError from MongoDb publisher returns normally but passes an 
Exception object, their implementation seems to be correct according to the 
reactive stream spec.
- Ok, so the supervision does not work for the source created via 
fromPublisher(), is there a way to catch this propagated failure somewhere 
else in the stream ? Would it be better to wrap the publisher in an actor ?

Best,
Antonin PA


Le mardi 21 février 2017 16:37:37 UTC+1, Konrad Malawski a écrit :
>
> Hi Antonin,
> Two things here:
>
> One: "throws an error on the onError callback"? This is not allowed by the 
> spec:
>
> Calling onSubscribe, onNext, onError or onComplete MUST return normally 
> except when any provided parameter is null in which case it MUST throw a 
> java.lang.NullPointerException to the caller, for all other situations 
> the only legal way for a Subscriber to signal failure is by cancelling 
> its Subscription. In the case that this rule is violated, any associated 
> Subscription to the Subscriber MUST be considered as cancelled, and the 
> caller MUST raise this error condition in a fashion that is adequate for 
> the runtime environment.
> https://github.com/reactive-streams/reactive-streams-jvm#2.13
>
> Is that mongo implementation surely tested and conforming to the reactive 
> streams spec?
>
>
> Two: Supervision does not work for arbitrary 3rd party publishers - it 
> only works within Akka Streams (specific stages, specifically handle it), 
> it's an additional feature Akka Streams provide over what Reactive Streams 
> do.
>
> Hope this helps
>
> -- 
> Konrad `ktoso` Malawski
> Akka  @ Lightbend 
>
> On 22 February 2017 at 00:04:06, antonin perrot-audet (perro...@gmail.com 
> ) wrote:
>
> Hello,
>
> has anyone succeded at having a source fromPublisher() conform to the 
> SupervisionStrategy defined in the ActorMaterializer ?
>
>
> I have that publisher  :
> MongoClients.create(mongoSettings).listDatabaseNames()
> That throws an exception on the onError callback, but the errorDecider 
> never gets called. The stream colapses on a failed Future :
>
>
> implicit val system2 = ActorSystem("Sys2")
> import system2.dispatcher
>
> val errorDecider : Supervision.Decider = {
>   case _ => {
> println("errorDecider does something, yeiii !")
> Supervision.stop}
> }
>
> implicit val materializer = ActorMaterializer(
> ActorMaterializerSettings(system2)
>   .withSupervisionStrategy(errorDecider))
>
> val publiString : Publisher[String] = MongoClients.create(
> mongoSettings).listDatabaseNames()
> val stream: Source[String, NotUsed] = Source.fromPublisher(publiString
> )
>
> stream.map( {s => println(s);s}).runWith(Sink.ignore).onComplete(_ => 
> system2.terminate())
>
>
>
> thanks in advace for your responses.
>
> Best,
>
> Antonin PA
> --
> >> Read the docs: http://akka.io/docs/
> >> Check the FAQ: 
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups 
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an 
> email to akka-user+...@googlegroups.com .
> To post to this group, send email to akka...@googlegroups.com 
> .
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

Re: [akka-user] Stream source fromPublisher does not honor the SupervisionStrategy ?

[Konrad Malawski]

Hi Antonin,
Two things here:

One: "throws an error on the onError callback"? This is not allowed by the
spec:

Calling onSubscribe, onNext, onError or onComplete MUST return normally
except when any provided parameter is null in which case it MUST throw a
java.lang.NullPointerException to the caller, for all other situations the
only legal way for a Subscriber to signal failure is by cancelling its
Subscription. In the case that this rule is violated, any associated
Subscription to the Subscriber MUST be considered as cancelled, and the
caller MUST raise this error condition in a fashion that is adequate for
the runtime environment.
https://github.com/reactive-streams/reactive-streams-jvm#2.13

Is that mongo implementation surely tested and conforming to the reactive
streams spec?


Two: Supervision does not work for arbitrary 3rd party publishers - it only
works within Akka Streams (specific stages, specifically handle it), it's
an additional feature Akka Streams provide over what Reactive Streams do.

Hope this helps

-- 
Konrad `ktoso` Malawski
Akka  @ Lightbend 

On 22 February 2017 at 00:04:06, antonin perrot-audet (perrotau...@gmail.com)
wrote:

Hello,

has anyone succeded at having a source fromPublisher() conform to the
SupervisionStrategy defined in the ActorMaterializer ?


I have that publisher  :
MongoClients.create(mongoSettings).listDatabaseNames()
That throws an exception on the onError callback, but the errorDecider
never gets called. The stream colapses on a failed Future :


implicit val system2 = ActorSystem("Sys2")
import system2.dispatcher

val errorDecider : Supervision.Decider = {
  case _ => {
println("errorDecider does something, yeiii !")
Supervision.stop}
}

implicit val materializer = ActorMaterializer(ActorMaterializerSettings(
system2)
  .withSupervisionStrategy(errorDecider))

val publiString : Publisher[String] = MongoClients.create(mongoSettings
).listDatabaseNames()
val stream: Source[String, NotUsed] = Source.fromPublisher(publiString)

stream.map( {s => println(s);s}).runWith(Sink.ignore).onComplete(_ =>
system2.terminate())



thanks in advace for your responses.

Best,

Antonin PA
--
>> Read the docs: http://akka.io/docs/
>> Check the FAQ:
http://doc.akka.io/docs/akka/current/additional/faq.html
>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups
"Akka User List" group.
To unsubscribe from this group and stop receiving emails from it, send an
email to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>  Search the archives: https://groups.google.com/group/akka-user
--- 
You received this message because you are subscribed to the Google Groups "Akka 
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email 
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.