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: 'john.doe@example.com' },
  { 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":"john.doe@example.com"}
chunk {"processed":{"email":"john.doe@example.com"}}

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

[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.

[HoneyryderChuck]

this has been merged.

[AbanoubGhadban]

Thanks @HoneyryderChuck – your solution works; however, it appears that sending request chunks from a different thread isn’t fully supported. Here’s a clearer explanation with examples:

Running the following code:

require 'httpx'
require 'json'

uri = "http://localhost:3000/ndjson"

http = HTTPX
      .plugin(:retries, max_retries: 1, retry_change_requests: true)
      .plugin(:stream_bidi)
      .with(fallback_protocol: "h2")

start_msg = { message: "Hello, world!" }.to_json + "\n"
ping_msg = { message: "Ping!" }.to_json + "\n"

request = http.build_request(
  "POST",
  uri,
  headers: { "content-type" => "application/x-ndjson" },
  body: [start_msg]
)

response = http.request(request, stream: true)

t = Thread.new do
  4.times do |i|
    puts "sending ping #{i}"
    request << ping_msg
    sleep 1
  end
  request.close
end

puts "waiting for response"
response.each.each_with_index do |chunk, idx|
  puts "chunk: #{chunk}"
end

puts "waiting for thread to finish"
t.join

produces the output:

waiting for response
sending ping 0
chunk: {"processed":{"message":"Hello, world!"}}
chunk: {"processed":{"message":"Ping!"}}
sending ping 1
sending ping 2
sending ping 3

Despite the additional pings being queued, the program hangs, and the server only receives the initial messages ({ message: "Hello, world!" } and the first { message: "Ping!" }). This indicates that subsequent request chunks sent from another thread are not reaching the server.

FYI, the following code using the http gem successfully demonstrates bidirectional streaming with multithreading:

require 'http/2'
require 'socket'
require 'json'

# Step 1: Create a TCP connection to the server
socket = TCPSocket.new('localhost', 3000)

# Step 2: Initialize the HTTP/2 client
client = HTTP2::Client.new

# Step 3: Send the HTTP/2 request
client.on(:frame) do |bytes|
  socket.print bytes
  socket.flush
end

# Step 4: Create a new stream for the conversation
stream = client.new_stream

# Step 5: Define the HTTP/2 headers
headers = [
  [':method', 'POST'],
  [':scheme', 'http'],
  [':authority', 'localhost'],
  [':path', '/ndjson'],
  ['content-type', 'application/x-ndjson'],
  ['user-agent', 'http-2-ruby-client'],
]

# Step 6: Send the headers
stream.headers(headers, end_stream: false)

# Step 7: Send the first message
stream.data(JSON.generate({ name: 'testName' }) + "\n", end_stream: false)

# Step 8: Handle the server's responses
response_complete = false
response_body = ''

stream.on(:headers) do |h|
  puts "Received headers: #{h}"
end

stream.on(:data) do |d|
  response_body += d
  puts "Received response: #{d}"
end

stream.on(:close) do
  response_complete = true
  puts "Stream closed"
end

Thread.new do
  sleep 1
  stream.data(JSON.generate({ name: 'testName2' }) + "\n", end_stream: false)
  sleep 1
  stream.data(JSON.generate({ name: 'testName3' }) + "\n", end_stream: true)
end

# Step 10: Read responses from the server
loop do
  begin
    data = socket.read_nonblock(1024)
    client << data
  rescue IO::WaitReadable
    if response_complete
      puts "Response complete, closing connection."
      socket.close
      break
    else
      IO.select([socket])
      retry
    end
  rescue EOFError
    puts "Connection closed by server."
    socket.close
    break
  end
end

[HoneyryderChuck]

Indeed, sending messages to the request from a separate thread is not (yet) supported. It's not impossible to support it, but I'd need to add some additional plumbing that is not yet available, such as the ability to interrupt the loop in the main thread when more data is available. Moreover, I don't think that multithread request body buffering is thread-safe, so that'll also have to be done.

FYI, the following code using the http gem successfully demonstrates bidirectional streaming with multithreading

You meant http-2 right? I don't think that's stable enough to use in production, as http-2 connections aren't thread-safe. YMMV.

Just for me to access the need, how important is that for your use-case? I'm asking because you presented the nodeJS example, and nodeJS can't do multithreading anyway.

[AbanoubGhadban]

@HoneyryderChuck I don’t actually need multithreading in my use case—it's just that the blocking nature of response.each forces me into it.

To illustrate, imagine a scenario where the server can either send data to the client or request data from the client:

response.each.each_with_index do |chunk, idx| # rubocop:disable Style/RedundantEach
  json = JSON.parse(chunk)
  if json["type"] == "query"
    file_path = json["file_path"]
    # file might be large
    file_content = File.read(file_path)
    response << { type: "response", content: file_content }.to_json + "\n"
  elsif json["type"] == "response"
    process_response(json["content"])
  end
end

In this case, if the server sends a "query" asking for a large file, the client will block while reading the file and won’t be able to receive any further chunks from the server during that time.

What I’d ideally like is to separate handling server messages from sending responses—so receiving and sending can happen concurrently without blocking each other.

Thanks again for the insight—really appreciate the work you’re doing on this!