Hi, Johan,
Thanks for your reply, but would not `Sink.last` complete it's Future only
after Source fully completed? I need to check for last passed value
in-between:
```
val bh = Source.unfold(0)(v => Some((v+1,
v))).delay(1.minute).runWith(BroadcastHub.sink)
bh.runWith(Sink.ignore)
```
So, if `bh` would be materialized after 2.5 minutes, it would not only
observer value `3` after another 30 seconds, but also value `2` just after
materialization.
понедельник, 7 августа 2017 г., 15:49:06 UTC+3 пользователь Akka Team
написал:
>
> You can achieve this by using Sink.last which will materialize into a
> Future[T] that completes with the last element emitted.
>
> It does however not complete the promise with the last element on
> failures, this can be remedied by prepending it with a
> .recoverWithRetries(0, { case _ => Source.empty }) making sure any
> upstream failure leads to a completion of the stream.
>
> --
> Johan
> Akka Team
>
> On Sat, Jul 29, 2017 at 8:17 AM, Alexey Shuksto <sei...@gmail.com
> <javascript:>> wrote:
>
>> Hello hAkkers,
>>
>> In our project we use some number of flows that follow same building
>> logic:
>> ```
>> val input: Source[T, NotUsed] = ???
>> val output = input.runWith(BroadcastHub.sink)
>>
>> output.runWith(Sink.ignore)
>> ```
>>
>> Sink.ignore here used to prevent BroadcastHub inner buffer overflow and,
>> also, to dump out old stream elements as they are usually do not matter.
>>
>> But in some rare cases we need to provide to clients of our source 'the
>> last seen element' -- the last element which passed our flow before client
>> materialization of BroadcastHub' Source.
>>
>> Currently it is done via some private volatile var:
>>
>> ```
>> @volatile private var last: Option[T] = None
>> _output.runForeach(e => last = Some(e))
>>
>> val output = _output.prepend(Source.lazily(() => last).collect { case
>> Some(e} => e})
>> ```
>>
>> I wonder maybe there is better way to do so, without mutable state and
>> stuff?
>>
>> --
>> >>>>>>>>>> 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 <javascript:>.
>> To post to this group, send email to akka...@googlegroups.com
>> <javascript:>.
>> 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.
