I misread it as wanting the last value to pass through the stream before
completion, but I guess it was, read the last value that passed by in a
stream.

There's no stage that does that out of the box, so providing your own safe
concurrent field of some kind, or firing of the elements to an actor that
keeps the latest value would probably be easiest, creating a stage that
does the same into some materialized value would also be possible (if you
think about the case where a stream is materialized multiple times you
wouldn't want all streams updating the same last-seen-value).

--
Johan
Akka Team

On Mon, Aug 7, 2017 at 8:51 PM, Alexey Shuksto <seig...@gmail.com> wrote:

> 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> 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/c
>>> urrent/additional/faq.html
>>> >>>>>>>>>> Search the archives: https://groups.google.com/grou
>>> p/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 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.

Reply via email to