[akka-user] Re: [akka-streams] Generic streams and abstract types
My understanding from the documentation is that using Http.singleRequest will fail requests when in a backpressure situation, while using a MergeHub with the flow will in fact backpressure. Is that a correct understanding? Jeff On Thursday, July 13, 2017 at 4:46:39 AM UTC-7, johannes...@lightbend.com wrote: > > On Wednesday, July 12, 2017 at 9:08:52 PM UTC+2, Jeff wrote: >> >> As for the issue of complexity, it's actually not as complex as it >> sounds. I'm using Http().superPool() to make api requests and I wanted to >> avoid having to create a separate stream for every single iteration of api >> request when the only thing that changed was the Unmarshaller. Instead of >> materializing multiple streams where the only thing that changed was the >> Sink, I just created one stream where the Sink.foreach(...) take the >> Unmarshaller function and resolves the Promise. >> > > You could just use Http.singleRequest in that case because it implements > almost the same kind of logic and also uses the same pool as superPool. > > Johannes > -- >> 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.
[akka-user] Re: [akka-streams] Generic streams and abstract types
On Wednesday, July 12, 2017 at 9:08:52 PM UTC+2, Jeff wrote: > > As for the issue of complexity, it's actually not as complex as it sounds. > I'm using Http().superPool() to make api requests and I wanted to avoid > having to create a separate stream for every single iteration of api > request when the only thing that changed was the Unmarshaller. Instead of > materializing multiple streams where the only thing that changed was the > Sink, I just created one stream where the Sink.foreach(...) take the > Unmarshaller function and resolves the Promise. > You could just use Http.singleRequest in that case because it implements almost the same kind of logic and also uses the same pool as superPool. Johannes -- >> 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.
[akka-user] Re: [akka-streams] Generic streams and abstract types
Thanks for the great suggestions - I eventually came to the "custom tuple"
solution myself and it seems to work well.
As for the issue of complexity, it's actually not as complex as it sounds.
I'm using Http().superPool() to make api requests and I wanted to avoid
having to create a separate stream for every single iteration of api
request when the only thing that changed was the Unmarshaller. Instead of
materializing multiple streams where the only thing that changed was the
Sink, I just created one stream where the Sink.foreach(...) take the
Unmarshaller function and resolves the Promise.
On Wednesday, July 12, 2017 at 5:24:07 AM UTC-7, johannes...@lightbend.com
wrote:
>
> Hi Jeff,
>
> your API seems quite complex. I don't know the purpose of it so I cannot
> suggest anything but I'd try to simplify. :)
>
> That said, your problem seems to be that you cannot write a concrete type
> that would express the dependency between the two components of the tuple
> `(RequestBuilder, Promise[???])`. There are two ways to solve it:
>
> 1) make `Out` a type parameter and convert `sink` to a `def sink[Out]`,
> then you can use the tuple `(RequestBuilder[Out], Promise[Out])`
> 2) create your custom tuple type that allows to express the dependency:
>
> case class BuilderWithPromise[O](builder: RequestBuilder { type Out = O },
> promise: Promise[O])
>
> and then use it as MergeHub.source[BuilderWithPromise[_]]
>
> But I can only repeat that getting rid of the dependent types if possible
> is usually the best solution ;)
>
> Johannes
>
> On Thursday, July 6, 2017 at 11:23:50 PM UTC+2, Jeff wrote:
>>
>> Here is a strawman program which illustrates the issue I am having
>>
>> trait RequestBuilder {
>> type Out
>>
>> def complete(p: Promise[Out]): Unit
>> }
>>
>> def makeRequest(in: RequestBuilder): Source[(RequestBuilder,
>> Promise[in.Out]), Future[in.Out]] = {
>> val p = Promise[in.Out]
>>
>> Source.single(in -> p).mapMaterializedValue(_ => p.future)
>> }
>>
>> val sink = MergeHub.source[(RequestBuilder, Promise[???])].to(Sink.foreach {
>> case (r, p) => r.complete(p)
>> }).run()
>>
>> sink.runWith(makeRequest(new RequestBuilder {
>> type Out = Int
>>
>> def complete(p: Promise[Out]): Unit = p.success(1)
>> }))
>>
>>
>> The issue is, how do I type the Promise[???] in the sink? I have been
>> able to work around this by making the Promise a part of the RequestBuilder
>> trait itself, but this seems like a code smell to me
>>
>
--
>> 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.
[akka-user] Re: [akka-streams] Generic streams and abstract types
Hi Jeff,
your API seems quite complex. I don't know the purpose of it so I cannot
suggest anything but I'd try to simplify. :)
That said, your problem seems to be that you cannot write a concrete type
that would express the dependency between the two components of the tuple
`(RequestBuilder, Promise[???])`. There are two ways to solve it:
1) make `Out` a type parameter and convert `sink` to a `def sink[Out]`,
then you can use the tuple `(RequestBuilder[Out], Promise[Out])`
2) create your custom tuple type that allows to express the dependency:
case class BuilderWithPromise[O](builder: RequestBuilder { type Out = O },
promise: Promise[O])
and then use it as MergeHub.source[BuilderWithPromise[_]]
But I can only repeat that getting rid of the dependent types if possible
is usually the best solution ;)
Johannes
On Thursday, July 6, 2017 at 11:23:50 PM UTC+2, Jeff wrote:
>
> Here is a strawman program which illustrates the issue I am having
>
> trait RequestBuilder {
> type Out
>
> def complete(p: Promise[Out]): Unit
> }
>
> def makeRequest(in: RequestBuilder): Source[(RequestBuilder,
> Promise[in.Out]), Future[in.Out]] = {
> val p = Promise[in.Out]
>
> Source.single(in -> p).mapMaterializedValue(_ => p.future)
> }
>
> val sink = MergeHub.source[(RequestBuilder, Promise[???])].to(Sink.foreach {
> case (r, p) => r.complete(p)
> }).run()
>
> sink.runWith(makeRequest(new RequestBuilder {
> type Out = Int
>
> def complete(p: Promise[Out]): Unit = p.success(1)
> }))
>
>
> The issue is, how do I type the Promise[???] in the sink? I have been
> able to work around this by making the Promise a part of the RequestBuilder
> trait itself, but this seems like a code smell to me
>
--
>> 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.
[akka-user] Re: [akka-streams] Generic streams and abstract types
Any thoughts?
On Thursday, July 6, 2017 at 2:23:50 PM UTC-7, Jeff wrote:
>
> Here is a strawman program which illustrates the issue I am having
>
> trait RequestBuilder {
> type Out
>
> def complete(p: Promise[Out]): Unit
> }
>
> def makeRequest(in: RequestBuilder): Source[(RequestBuilder,
> Promise[in.Out]), Future[in.Out]] = {
> val p = Promise[in.Out]
>
> Source.single(in -> p).mapMaterializedValue(_ => p.future)
> }
>
> val sink = MergeHub.source[(RequestBuilder, Promise[???])].to(Sink.foreach {
> case (r, p) => r.complete(p)
> }).run()
>
> sink.runWith(makeRequest(new RequestBuilder {
> type Out = Int
>
> def complete(p: Promise[Out]): Unit = p.success(1)
> }))
>
>
> The issue is, how do I type the Promise[???] in the sink? I have been
> able to work around this by making the Promise a part of the RequestBuilder
> trait itself, but this seems like a code smell to me
>
--
>> 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.
