Request-Reply

Overview

Request-reply gives you synchronous RPC on top of KubeMQ's messaging fabric: a caller sends a query and blocks until the handler actually processing the request sends back a real answer — not just an acknowledgment. Reach for it whenever the caller needs a return value to proceed — a lookup, a computed result, a status check — the same shape as an HTTP call, but routed by KubeMQ instead of a service mesh or DNS.

A handler subscribes with SubscribeToQueries and, inside its WithOnQueryReceive callback, builds a QueryReply that copies the request's q.Id and q.ResponseTo back from the incoming QueryReceive — that's what lets KubeMQ route the response to the one caller waiting, not broadcast it. The caller's client.SendQuery blocks until that reply lands or its SetTimeout elapses, then returns a response carrying Executed, Body, and Metadata.

Gotchas: if no subscriber is listening — or the handler crashes before replying — SendQuery simply times out; there's no way to distinguish "no handler" from "handler is slow" from the timeout alone. Every reply must echo back the same q.Id and q.ResponseTo unchanged, or the response is silently dropped or misrouted. If you don't actually need a return value, use commands instead — they only need an ack, so they don't tie up a caller waiting on a round trip.

Prerequisites

• KubeMQ server running on localhost:50000
• Go SDK installed (go get github.com/kubemq-io/kubemq-go/v2)

Code

// Example: patterns/request-reply
//
// Demonstrates the request-reply pattern using queries.
// A client sends a query and waits for a response from a handler.
// This is the fundamental RPC pattern in KubeMQ.
//
// Channel: go-patterns.request-reply
// Client ID: go-patterns-request-reply-client
//
// Run with a KubeMQ server on localhost:50000
// (see https://docs.kubemq.io/deploy).
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/kubemq-io/kubemq-go/v2"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	client, err := kubemq.NewClient(ctx,
		kubemq.WithAddress("localhost", 50000),
		kubemq.WithClientId("go-patterns-request-reply-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-patterns.request-reply"
	done := make(chan struct{})

	// Service: subscribe to handle requests.
	sub, err := client.SubscribeToQueries(ctx, channel, "",
		kubemq.WithOnQueryReceive(func(q *kubemq.QueryReceive) {
			fmt.Printf("Service received request: body=%s\n", q.Body)

			// Process the request and return a reply.
			resp := kubemq.NewQueryReply().
				SetRequestId(q.Id).
				SetResponseTo(q.ResponseTo).
				SetBody([]byte(`{"order_id":"ORD-123","status":"confirmed"}`)).
				SetMetadata("success").
				SetExecutedAt(time.Now())
			_ = client.SendQueryResponse(ctx, resp)
			close(done)
		}),
		kubemq.WithOnError(func(err error) {
			log.Println("Service error:", err)
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Unsubscribe()
	time.Sleep(300 * time.Millisecond)

	// Client: send a request and wait for the reply.
	reply, err := client.SendQuery(ctx, kubemq.NewQuery().
		SetChannel(channel).
		SetBody([]byte(`{"action":"create_order","item":"widget"}`)).
		SetTimeout(10*time.Second))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Client received reply: executed=%v body=%s\n",
		reply.Executed, reply.Body)

	<-done
}

How It Works

1. A service subscribes to go-patterns.request-reply using SubscribeToQueries, registering a WithOnQueryReceive callback.
2. The callback constructs a QueryReply by reflecting back q.Id as the request ID and q.ResponseTo as the routing address, then calls client.SendQueryResponse to deliver the reply.
3. The requester calls client.SendQuery which blocks until the response arrives or the 10-second timeout expires; it returns a *QueryResponse with Executed, Body, and Metadata fields.
4. A done channel synchronises the service goroutine — close(done) in the callback signals the main goroutine that the handler has run.

Related

• Pattern overview
• Go SDK Reference
• Fan-Out
• Work Queue