# Consuming elements from a Source.queue and connecting to cachedHostConnectionPool

**URL:** <https://discuss.akka.io/t/consuming-elements-from-a-source-queue-and-connecting-to-cachedhostconnectionpool/9147>\
**Category:** Akka HTTP\
**Tags:** akka, akka-http, java, streams\
**Created:** [November 22, 2021, 6:58pm UTC](https://discuss.akka.io/t/consuming-elements-from-a-source-queue-and-connecting-to-cachedhostconnectionpool/9147 "2021-11-22T18:58:45Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![AND2797](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/and2797/32/2616_2.png) [@AND2797](https://discuss.akka.io/u/AND2797)\
**Post date:** [November 22, 2021, 6:58pm UTC](https://discuss.akka.io/t/consuming-elements-from-a-source-queue-and-connecting-to-cachedhostconnectionpool/9147/1 "2021-11-22T18:58:45Z")

</div>

I am trying to process HTTP requests in a way that will not result in

```auto
Exceeded configured max-open-requests value of [32]

```

My idea for this is to use a `cachedHostConnectionPool` that will open a limited amount of connections, and while the connections are in use, the other queries will be buffered. I will pass requests to the flow via a `Source.queue`

My code currently looks like this -

```auto
    private Source<Pair<HttpRequest, Object>, SourceQueueWithComplete<Pair<HttpRequest, Object>>> sourceQueue = Source.queue(4, OverflowStrategy.backpressure(), 4);

private final Flow<Pair<HttpRequest, Object>, Pair<Try<HttpResponse>, Object>, HostConnectionPool> flow 
 = Http.get(system).cachedHostConnectionPool(host);

private final Materializer materializer = SystemMaterializer.get(system).materializer();

 private SourceQueueWithComplete<Pair<HttpRequest, Object>> queue = sourceQueue.via(flow).toMat(Sink.foreach(p -> {
            // processing ? 

        }), Keep.left()).run(materializer);

```

My question is how should I process the requests and get a response out of it? Particularly, what should I be doing here -

```auto
(Sink.foreach(p -> {
            // processing ? 

        })

```

If I have understood correctly, the `Sink.foreach` construct is responsible for consuming the elements that are queued, so the processing, i.e getting the HttpResponse should be completed here? If someone can provide an example based off this - that would be very helpful.  
Here, `p.first()` gives a `Try<HttpResponse>`, So is this the result of the `HttpRequest` that is offered to the queue? I’m unsure of this as well.

---

<div class="post-metadata">

**Author:** ![pbernet](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/pbernet/32/370_2.png) [@pbernet](https://discuss.akka.io/u/pbernet)\
**Post date:** [November 23, 2021, 7:21pm UTC](https://discuss.akka.io/t/consuming-elements-from-a-source-queue-and-connecting-to-cachedhostconnectionpool/9147/2 "2021-11-23T19:21:20Z")

</div>

Maybe this Scala example answers some of your questions, this is the interesting part:

> <https://github.com/pbernet/akka_streams_tutorial/blob/dc727e1552525ea227a535063c3e00c60244ab22/src/main/scala/akkahttp/HttpFileEchoStream.scala#L150>

Although it is a HttpRequest to download a file here, it shows the handling of the `HttpResponse` in the `SourceQueue` / `cachedHostConnectionPool` scenario.  
Note that this example has a homegrown retry meccano, which may be improved.

Hope this helps  
Paul

---

<div class="post-metadata">

**Author:** ![AND2797](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/and2797/32/2616_2.png) [@AND2797](https://discuss.akka.io/u/AND2797)\
**Post date:** [November 30, 2021, 4:42pm UTC](https://discuss.akka.io/t/consuming-elements-from-a-source-queue-and-connecting-to-cachedhostconnectionpool/9147/3 "2021-11-30T16:42:54Z")

</div>

Thanks for your reply. This is what I ended up following.