Send Query
Send a KubeMQ Query and receive a data response in request-reply style using the Go SDK.
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 calls client.SendQuery with a channel and a timeout, then blocks until a reply arrives. client.SubscribeToQueries registers a handler via kubemq.WithOnQueryReceive; the handler builds a *kubemq.QueryReply carrying SetRequestId/SetResponseTo (copied from the incoming query) plus SetBody, and KubeMQ routes that reply back to the exact caller waiting on it.
Gotchas: the timeout must cover however long the handler takes to run — a slow handler times out the caller even though the handler eventually succeeds. No handler subscribed yet also times out rather than erroring immediately, so startup order matters. Body is a raw byte slice — encoding it is your application's job.
Prerequisites
- KubeMQ server running on
localhost:50000 - Go SDK installed (
go get github.com/kubemq-io/kubemq-go/v2)
Code
// Example: queries/send-query
//
// Demonstrates sending a query and receiving a response with data.
// Unlike commands, queries return a body payload in the response.
//
// Channel: go-queries.send-query
// Client ID: go-queries-send-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-send-query-client"),
)
if err != nil {
log.Fatal(err)
}
defer client.Close()
channel := "go-queries.send-query"
done := make(chan struct{})
// Subscribe to handle queries and return data.
sub, err := client.SubscribeToQueries(ctx, channel, "",
kubemq.WithOnQueryReceive(func(q *kubemq.QueryReceive) {
fmt.Printf("Query received: channel=%s body=%s\n", q.Channel, q.Body)
resp := kubemq.NewQueryReply().
SetRequestId(q.Id).
SetResponseTo(q.ResponseTo).
SetBody([]byte(`{"result":"data","status":"ok"}`)).
SetMetadata("ok").
SetExecutedAt(time.Now())
_ = client.SendQueryResponse(ctx, resp)
close(done)
}),
kubemq.WithOnError(func(err error) {
log.Println("Query subscription error:", err)
}),
)
if err != nil {
log.Fatal(err)
}
defer sub.Unsubscribe()
time.Sleep(300 * time.Millisecond)
// Send a query and receive the response.
qResp, err := client.SendQuery(ctx, kubemq.NewQuery().
SetChannel(channel).
SetBody([]byte("fetch-data")).
SetTimeout(10*time.Second))
if err != nil {
log.Fatal(err)
}
fmt.Printf("Query response: executed=%v body=%s\n", qResp.Executed, qResp.Body)
<-done
}
// Expected output:
// Query received: channel=go-queries.send-query body=fetch-data
// Query response: executed=true body={"result":"data","status":"ok"}
How It Works
client.SubscribeToQueries(ctx, channel, "", opts...)registers a query handler viakubemq.WithOnQueryReceive; the handler returns a*kubemq.QueryReplycontaining the response body.kubemq.NewQueryReply().SetRequestId(q.Id).SetResponseTo(q.ResponseTo).SetBody(...)constructs the response; unlike commands, the query reply carries aBodypayload that the caller reads.client.SendQuery(ctx, kubemq.NewQuery().SetTimeout(10*time.Second))blocks until a handler responds; the result inqResp.Bodyis the payload the handler set inSetBody.qResp.Executedistruewhen a handler responded successfully;qResp.Bodycontains the raw byte payload — JSON, Protobuf, or any other encoding is the application's responsibility.
Related
Was this page helpful?