Bidirectional Concurrent Streaming in HTTP/2

[AbanoubGhadban]

Is it possible to achieve bidirectional streaming in an HTTP/2 POST request using HTTPx?

Here is an example scenario I am working on:

# Create a new HTTPX session
http = HTTPX
       .plugin(:stream)
       .with(fallback_protocol: "h2")

# Create an Enumerator that yields JSON objects
data = Enumerator.new do |yielder|
  5.times do |i|
    chunk = "#{{ message: "Hello #{i}", timestamp: Time.now.to_i }.to_json}\n"
    puts "Sending chunk #{i}: #{chunk}"
    yielder << chunk
    sleep 1 # Simulate some delay between chunks
  end
end

# Send the request with streaming body
response = http.post(
  "http://localhost:3000/ndjson",
  body: data,
  headers: {
    "Content-Type" => "application/x-ndjson"
  },
  stream: true
)

response.each_line do |line|
  parsed_response = JSON.parse(line)
  puts "Received response: #{parsed_response}"
end

This code sends chunks of NDJSON to an echo server that returns the processed chunk, formatted as {"processed" => <message>}.

However, the responses are only printed after the entire request body is sent. The output looks like this:

Sending chunk 0: {"message":"Hello 0","timestamp":1737389106}
Sending chunk 1: {"message":"Hello 1","timestamp":1737389107}
Sending chunk 2: {"message":"Hello 2","timestamp":1737389108}
Sending chunk 3: {"message":"Hello 3","timestamp":1737389109}
Sending chunk 4: {"message":"Hello 4","timestamp":1737389110}
Received response: {"processed"=>{"message"=>"Hello 0", "timestamp"=>1737389106}}
Received response: {"processed"=>{"message"=>"Hello 1", "timestamp"=>1737389107}}
Received response: {"processed"=>{"message"=>"Hello 2", "timestamp"=>1737389108}}
Received response: {"processed"=>{"message"=>"Hello 3", "timestamp"=>1737389109}}
Received response: {"processed"=>{"message"=>"Hello 4", "timestamp"=>1737389110}}

What I want to achieve is immediate processing of each chunk, with output like this:

Sending chunk 0: {"message":"Hello 0","timestamp":1737389106}
Received response: {"processed"=>{"message"=>"Hello 0", "timestamp"=>1737389106}}
Sending chunk 1: {"message":"Hello 1","timestamp":1737389107}
Received response: {"processed"=>{"message"=>"Hello 1", "timestamp"=>1737389107}}
Sending chunk 2: {"message":"Hello 2","timestamp":1737389108}
Received response: {"processed"=>{"message"=>"Hello 2", "timestamp"=>1737389108}}
Sending chunk 3: {"message":"Hello 3","timestamp":1737389109}
Received response: {"processed"=>{"message"=>"Hello 3", "timestamp"=>1737389109}}
Sending chunk 4: {"message":"Hello 4","timestamp":1737389110}
Received response: {"processed"=>{"message"=>"Hello 4", "timestamp"=>1737389110}}

Is there a way to achieve this behavior using HTTPx or any of its plugins? If not, are there alternative approaches within HTTPx or recommendations for other libraries to enable such bidirectional streaming in HTTP/2?

FYI, the following nodejs code worked as expected, so I think the problem is not at the server

const http2 = require('http2');

// Form entries to send
const formEntries = [
  { name: 'John Doe' },
  { email: '[email protected]' },
  { message: 'This is a test message' },
];

// Create an HTTP/2 client
const client = http2.connect('http://localhost:3000');

// Create and send request
const req = client.request({
  ':path': '/ndjson',
  ':method': 'POST',
  'content-type': 'application/x-ndjson',
});

let responseData = '';
req.on('data', (chunk) => {
  console.log('chunk', chunk.toString())
  responseData += chunk.toString();
});

req.on('end', () => {
  client.close();
});

req.on('error', (err) => {
  console.error('Request error:', err);
  client.close();
});

// Send each form entry with a delay to simulate streaming
let index = 0;
const sendNextEntry = () => {
  if (index < formEntries.length) {
    const entry = JSON.stringify(formEntries[index]) + '\n'; // NDJSON format
    req.write(entry);
    console.log(`Sent form entry: ${entry.trim()}`);
    index++;
    setTimeout(sendNextEntry, 1000); // Simulate delay between entries
  } else {
    req.end(); // End the request after all entries are sent
  }
};

sendNextEntry();

Its output is

Sent form entry: {"name":"John Doe"}
chunk {"processed":{"name":"John Doe"}}

Sent form entry: {"email":"[email protected]"}
chunk {"processed":{"email":"[email protected]"}}

Sent form entry: {"message":"This is a test message"}
chunk {"processed":{"message":"This is a test message"}}

[HoneyryderChuck]

Hi @AbanoubGhadban ,

I believe it's possible, at the very least with the stream plugin. The :grpc plugin piggybacks on it to support its bidirectional streaming.

this can probably work without plugins, as long as you provide the corresponding custom functionality in a plugin. FWIW this is an example how you could send the body in chunks (it may require adjustments):

req_body = form_entries.lazy.map do |entry|
   entry = JSON.dump(entry) + '\n'
   sleep(1)
   puts "Sent form entry: #{entry.strip}"
   entry
end

stream_session = HTTPX.plugin(:stream)
response = .post("https://localhost:3000/ndjson", headers: { "content-type" => "application/nd+json" }, body: req_body, stream: true)

response.each do |chunk|
  puts "chunk: #{chunk}"
end

[HoneyryderChuck]

perhaps something like this?

last_chunk = nil

req_body = Enumerator.new do |y|
  loop do
    entry = form_entries.shift

    raise StopIteration if entry.nil?

    if last_chunk.nil?
      # first body chunk
      y << JSON.dump(entry) + '\n'
    end

    entry = dyo_process(entry, last_chunk)

    # send processed entry
    y << JSON.dump(entry) + '\n'

    # allows processing next response chunk

    y << nil
    # or, if the above may not work, perhaps this workaround does:
    # throw(:buffer_full)
  end
end
  
stream_session = HTTPX.plugin(:stream)
response = stream_session.post("https://localhost:3000/ndjson", headers: { "content-type" => "application/nd+json" }, body: req_body, stream: true)

response.each do |chunk|
  puts "chunk: #{chunk}"
  last_chunk = chunk
end

again, not sure if this works fully (hard for me without having the peer available to test something like this with), so you'll have to iterate on the idea above, which boils down to delaying the processing of the next request body chunk until the response yields the next response body chunk.

However, it'd be great to have a peer server to test this with, it'd help pinpoint whether httpx already sorta/kinda supports the pattern, and if it doesn't, work on it, perhaps with a better API.

[AbanoubGhadban]

@HoneyryderChuck, the issue is that the each block is only executed after the body enumerator has fully completed execution.. You can't use a feedback variable like last_chunk betweem each block and the enumerator block.

If you'd like to test your code, you can use this dummy Node.js server I’ve shared:
https://gist.github.com/AbanoubGhadban/d849ca61a72c28d56211ce04f9c2a39d

[HoneyryderChuck]

@AbanoubGhadban thx for the node example. I see I may need to install this "fastify" with npm, shouldn't it have a package.json? I'm not very familiar with node, would be possible to provide it as a repository, with instruction on how to set up the server and run the node client-side script? I can see if I can provide a ruby example that works for your use case, or worst case scenario implement the required feature.

[AbanoubGhadban]

Apologies for not including instructions earlier! This is just a minimal demo server designed to work without any additional npm packages or dependencies. You can find the code here:
https://gist.github.com/AbanoubGhadban/6974c035e0939011e138c2aa712b1809

[HoneyryderChuck]

Thx for that! I've started a new plugin in the gh-disc-71 branch. You can follow the specs here to see how you can use it. I'd kindly ask you to test this branch in order to find loose ends.