Handle Query

Overview

A query handler is the answering side of KubeMQ's request/response RPC pattern — the code that does real work and sends back data, unlike a Command handler, which only acknowledges receipt. Reach for it whenever a caller needs an actual answer — a lookup result, a computed value, a status object — not just confirmation that a message arrived.

Registering a handler with SubscribeToQueries and WithOnQueryReceive opens a subscription; the broker delivers every matching query to your callback as it arrives. The callback builds a reply carrying the original query's correlation id (SetRequestId/SetResponseTo) back to the broker, so the answer routes to the specific caller blocked waiting, and sets a body with the real result via SetBody before sending it with SendQueryResponse.

Gotchas: if the handler never sends a response, the caller blocks until its own SetTimeout elapses and fails with a timeout, not a fast error. An exception inside the handler doesn't automatically become a failure reply, so uncaught errors can leave the sender hanging. And because every matching query lands on the same callback, slow handler code delays every other in-flight caller.

Prerequisites

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

Code

// Example: queries/handle-query
//
// Demonstrates subscribing to queries and handling them with business logic.
// The handler processes incoming queries and returns data in the response.
//
// Channel: go-queries.handle-query
// Client ID: go-queries-handle-query-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-queries-handle-query-client"),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer client.Close()

	channel := "go-queries.handle-query"
	done := make(chan struct{})

	// Register a query handler.
	sub, err := client.SubscribeToQueries(ctx, channel, "",
		kubemq.WithOnQueryReceive(func(q *kubemq.QueryReceive) {
			fmt.Printf("Handling query: id=%s body=%s\n", q.Id, q.Body)

			// Process the query and prepare a response.
			resp := kubemq.NewQueryReply().
				SetRequestId(q.Id).
				SetResponseTo(q.ResponseTo).
				SetBody([]byte(`{"users":["alice","bob"]}`)).
				SetMetadata("success").
				SetExecutedAt(time.Now())
			if err := client.SendQueryResponse(ctx, resp); err != nil {
				log.Printf("Failed to send response: %v", err)
			}
			close(done)
		}),
		kubemq.WithOnError(func(err error) {
			log.Println("Handler error:", err)
		}),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer sub.Unsubscribe()
	time.Sleep(300 * time.Millisecond)

	// Send a query.
	qResp, err := client.SendQuery(ctx, kubemq.NewQuery().
		SetChannel(channel).
		SetBody([]byte("list-users")).
		SetTimeout(10*time.Second))
	if err != nil {
		log.Fatal(err)
	}
	fmt.Printf("Response: executed=%v body=%s metadata=%s\n",
		qResp.Executed, qResp.Body, qResp.Metadata)

	<-done
}

How It Works

1. client.SubscribeToQueries(ctx, channel, "", opts...) subscribes to query requests on go-queries.handle-query; the kubemq.WithOnQueryReceive callback receives each *kubemq.QueryReceive.
2. The handler reads q.Id and q.ResponseTo from the incoming request and echoes them back in kubemq.NewQueryReply().SetRequestId(q.Id).SetResponseTo(q.ResponseTo) so the broker knows which caller to reply to.
3. SetBody([]byte(...)) sets the response payload — any byte slice works; the caller reads it back in qResp.Body.
4. client.SendQueryResponse(ctx, resp) sends the reply from the handler goroutine; if this call fails, the caller will block until its SetTimeout fires and returns an error.

Related

• Pattern overview
• Go SDK Reference
• Send Query
• Query Group