Send Query

Overview

This tutorial builds the RPC half of KubeMQ's request/reply patterns: a query, where the caller blocks for a handler's data payload instead of just a completion status. Reach for it whenever a caller needs an answer — fetching a record, running a lookup, or asking another service to compute a value on demand. You'll run a handler and a sender in the same process to see the full round trip.

The sender builds a KubeMQ::CQ::QueryMessage and calls client.send_query(msg), which blocks until a reply arrives. client.subscribe_to_queries(sub, ...) registers a handler block; the handler builds a KubeMQ::CQ::QueryResponseMessage with request_id: query.id, reply_channel: query.reply_channel — copied from the incoming query — plus executed: true and a body, sent with client.send_response(response). KubeMQ routes that reply back to the caller waiting on it.

Gotchas: the timeout must cover however long the handler takes to run — a slow handler leaves result.executed false even though the handler eventually succeeds. No handler subscribed yet also times out rather than erroring immediately. cache_hit only applies with server-side caching configured — for a plain query it's always false.

Prerequisites

• KubeMQ server running on localhost:50000
• Ruby SDK installed (gem install kubemq)

Code

require 'kubemq'

address = ENV.fetch('KUBEMQ_ADDRESS', 'localhost:50000')
channel = 'ruby-rpc.query-send'

begin
  client = KubeMQ::CQClient.new(address: address, client_id: 'query-sender')
  puts "Connected to #{address}"

  cancel = KubeMQ::CancellationToken.new
  sub = KubeMQ::CQ::QueriesSubscription.new(channel: channel)
  client.subscribe_to_queries(sub, cancellation_token: cancel, on_error: lambda { |e|
    puts "Error: #{e.message}"
  }) do |query|
    puts "Handler received query: #{query.metadata}"
    response = KubeMQ::CQ::QueryResponseMessage.new(
      request_id: query.id,
      reply_channel: query.reply_channel,
      executed: true,
      body: "answer-to-#{query.metadata}",
      metadata: 'result'
    )
    client.send_response(response)
  end
  sleep 1

  msg = KubeMQ::CQ::QueryMessage.new(
    channel: channel,
    timeout: 10_000,
    metadata: 'get-user',
    body: 'user-id-42'
  )
  result = client.send_query(msg)
  puts "Query result: executed=#{result.executed}, body=#{result.body}, cache_hit=#{result.cache_hit}"
rescue KubeMQ::Error => e
  puts "KubeMQ error: #{e.message}"
ensure
  cancel&.cancel
  client&.close
  puts 'Done'
end

How It Works

• Queries return data from the handler — unlike commands which only return execution status.
• The handler populates body and metadata in the QueryResponseMessage.
• cache_hit indicates whether the response was served from the server-side cache.
• Review timeouts, channel names, and client IDs before running against shared environments.

Related

• Pattern overview
• Ruby SDK Reference
• Handle Query
• Cached Query