CODEWITHSUNDEEP
rx-javavert.xrx-java2
paul
paul

Reputation: 13471

Back pressure with Vertx HttpClient and RxJava

I´m having problems to create a "back pressure system". I´m using Vertx HttpClient and RxJava. I need to do 6000 request to an external service, and in order to avoid a full in the waitingForQueue, Since this external service cannot process so fast as I send, I put a delay between request/response.

Since this journey is working as batch process no worries if it´s takes a minute.

Here my code

return from(subGroups)
        .flatMap(subGroup -> getProductIdsForSubGroup(subGroup))
        .delay(50, TimeUnit.MILLISECONDS)

This method it´s invoked from a Observable interval that run every 24H passing this list of subgroups(6000)

But after check my logs I cannot see a delay between my request of 50ms

Here 3 of my logs

{"@timestamp":"2016-11-30T10:32:48.973+00:00","event":"started","requestHost":"localhost","requestMethod":"GET","requestUri":"/v3/comercial?category=T15EB&clientId=ERROR_NOT_SUPPLIED","requestHash":189630582,"level":"INFO","thread_name":"vert.x-eventloop-thread-5"}
{"@timestamp":"2016-11-30T10:32:48.978+00:00","event":"started","requestHost":"localhost","requestMethod":"GET","requestUri":"/v3/commercial?category=T15EE&clientId=ERROR_NOT_SUPPLIED","requestHash":1296199359,"level":"INFO","thread_name":"vert.x-eventloop-thread-5"}
{"@timestamp":"2016-11-30T10:32:48.981+00:00","event":"started","requestHost":"localhost","requestMethod":"GET","requestUri":"/v3/commercial?category=T15EG&clientId=ERROR_NOT_SUPPLIED","requestHash":228306365,"level":"INFO","thread_name":"vert.x-eventloop-thread-5"}

Any idea what I need to do in order to achieve this?.

Regards.

SOLUTION

I end up using concatMap Please if you have a better solution, let me know

return from(subGroups)
        .concatMap(subGroup -> Observable.just(subGroup).delay(50, TimeUnit.MILLISECONDS))
        .flatMap(subGroup -> getProductIdsForSubGroup(subGroup))

Upvotes: 2

Views: 874

Answers (1)

Tassos Bassoukos
Tassos Bassoukos

Reputation: 16142

Be aware that delay just delays the emission, so it's essentially wasted time.

If you can query the remote system with let's say up to 10 simultaneous requests / connections, you can use the 2-parameter flatMap:

return from(subGroups)
    .flatMap(subGroup -> getProductIdsForSubGroup(subGroup), 10);

Upvotes: 1

Related Questions