EventBus and chunked HTTP responses (streaming support) - a.k.a multiple responses to an eventbus request?

[Stwissel]

I have a kind of "proxy" application where POSTing to an endpoint triggers an EventBus message which in return triggers an http client to fetch some data and return in via msg.reply. Roughly like this:

sequenceDiagram
        autonumber
        actor client
        participant h as httpListener
        participant e as EventBus
        participant v as Verticle
        participant c as HttpClient
        participant d as Target
        v->>e: listen for processurl
        client->>h: POST {url: ..}
        h->>e: request processurl
        e->>v: request processurl
        v->>c: request processurl
        c->>d: GET url
        d->>c: return result
        c->>v: return result
        v->>e: package for eventbus
        e->>h: return result
        h->>client: return result

Loading

I created a sample where I stripped out all the pre- and post- processing to focus on the flow (there is plenty in the production app). So far the approach worked - until it doesn't :-(

I need to add support for URLs that return chunked responses. I don't want to load all the chunks eventual blowing my memory. In principal I know how to implement that with the http client. My stumbling block now is: how to reply to one eventBus.request(...) multiple times, once for each chunk. Or what would be a good approach for such a task?

[jponge]

@tsegismont @vietj might give more details (it's a Vert.x question). The main problem is indeed the lack of back-pressure for streaming use-cases at the event-bus level.

[vietj]

I think best option would be to pause the HttpClientResponse and pass it over to the httpListener that would unpause it and stream it in all cases

[Stwissel]

How roughly would that look like code wise? Still using an EventBus response?

[vietj]

I agree, event bus is not designed for this :-)

Best option would be to use indeed a response but you do need a codec for this. Otherwise you can put the response in a static concurrent map with an identifier (e.g. a sequence of long) and pass the identifier in the response