Hi Evgeniy
> Of course, but it will produce a lot of boilerplate (instead .map(_.param)
> you will write .map(_.map(_.param)) - for all transformers).
> It is same as saying: "Library has no such behavior, but you can write it
> by himself".
>
Streams is a basic building block and its purpose is exactly that: build
more complex behavior around it.
> Absent behavior with help of which you can remove boilerplate is issue.
>
The problem is that without the wrapper it is impossible to know wether a
Producer[T] is producing a faithful sequence of Ts or drops them. Also, the
behavior of recovery is completely application dependent.
What would a filter emit in case of a failure? Will it drop an element?
Will it emit an error marker? Also, what would filter do, if it encounters
an error marker? Will it drop that? Will it emit it downstream?
And above is just one of the simpler scenarios. What would groupBy do?
Which group does an error marker belong to? If it encounters an error,
where should it emit that? On the master stream (the stream of groups)?
All of the groups? One of the groups, then which one? Or should it open an
error group?
What would splitWhen do? Or toFuture, especially if the stream emits only
one element which fails? What would mapConcat do if it gets an error
marker? What would buffer do if it is full? Drop error markers?
Or should we just drop elements silently without any error marker?
All of the questions above have more than one possible answer, and all of
them belong to a valid use-case. Yours is one, but not the only one.
>
> 2) In case of finite stream you can get result value and operate with it.
>>> In case of infinite you cant doing anything with data without custom
>>> decorator.
>>>
>>
>> Could you elaborate?
>>
> Without decorator you can not take last element of infinite stream for
> example. With decorator it would be like:
>
> class FlowDecorator[T](flow: Flow[T], initial: Option[T] = None)(implicit
> materializer: FlowMaterializer) extends Flow[T] with Builder[T] {
> private val last = new AtomicReference(initial)
> private val producer = flow.map { e => last.set(e); e
> }.toProducer(materializer)
>
> def apply() = last.get()
>
> override def toFuture(..) =
> last.get().map(Future.successful).getOrElse(Flow(producer).toFuture(materializer))
>
> override def map[U](f: T => U) = FlowDecorator(Flow(producer).map(f),
> last.get().map(f))
> ... a lot of boilerplate for all transformers
> }
>
Why would you extend Flow? You can just build an implicit wrapper that
calls into the appropriate Flow methods. The above code touches internals
and it might break if we change things.
-Endre
>
> Thank you for answer.
>
>> --
> >>>>>>>>>> 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 [email protected].
> To post to this group, send email to [email protected].
> Visit this group at http://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 [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
