Synchronous request-reply patterns. Commands carry no response body (success/failure only). Queries return data in the response and support server-side caching.

Commands

SendCommand

[[nodiscard]] StatusOr<CommandResponse> SendCommand(const Command& command);

Send a command and block until a handler responds or the timeout expires.

SubscribeToCommands

[[nodiscard]] StatusOr<std::unique_ptr<Subscription>> SubscribeToCommands(
    const std::string& channel, const std::string& group,
    std::function<void(const CommandReceive&)> on_command,
    std::function<void(const Status&)> on_error);

Subscribe to incoming commands. Use client->SendCommandResponse(reply) to respond.

SendCommandResponse

[[nodiscard]] Status SendCommandResponse(const CommandReply& reply);

Send a response to a received command.

Command Type

Built via Command::Builder.

Method	Type	Required	Description
`SetChannel(s)`	`string`	Yes	Target channel
`SetBody(s)`	`string`	No	Command body
`SetMetadata(s)`	`string`	No	String metadata
`SetTimeout(d)`	`chrono::milliseconds`	Yes	Max wait time
`SetTags(m)`	`map<string,string>`	No	Key-value tags

CommandReceive

Field	Type	Description
`id`	`string`	Command identifier
`channel`	`string`	Channel name
`body`	`string`	Command body
`metadata`	`string`	String metadata
`response_to`	`string`	Reply-to address
`tags`	`map<string,string>`	Key-value tags

CommandReply

Built via CommandReply::Builder.

Method	Type	Required	Description
`SetRequestId(s)`	`string`	Yes	Original command ID
`SetResponseTo(s)`	`string`	Yes	Reply-to address
`SetExecuted(b)`	`bool`	Yes	Success or failure
`SetExecutedAt(n)`	`int64_t`	No	Execution timestamp
`SetBody(s)`	`string`	No	Optional response body

Queries

SendQuery

[[nodiscard]] StatusOr<QueryResponse> SendQuery(const Query& query);

Send a query and block until a handler responds or the timeout expires.

SubscribeToQueries

[[nodiscard]] StatusOr<std::unique_ptr<Subscription>> SubscribeToQueries(
    const std::string& channel, const std::string& group,
    std::function<void(const QueryReceive&)> on_query,
    std::function<void(const Status&)> on_error);

Subscribe to incoming queries. Use client->SendQueryResponse(reply) to respond.

SendQueryResponse

[[nodiscard]] Status SendQueryResponse(const QueryReply& reply);

Send a response to a received query.

Query Type

Built via Query::Builder.

Method	Type	Required	Description
`SetChannel(s)`	`string`	Yes	Target channel
`SetBody(s)`	`string`	No	Query body
`SetMetadata(s)`	`string`	No	String metadata
`SetTimeout(d)`	`chrono::milliseconds`	Yes	Max wait time
`SetCacheKey(s)`	`string`	No	Cache key for server-side caching
`SetCacheTTL(d)`	`chrono::milliseconds`	No	Cache duration
`SetTags(m)`	`map<string,string>`	No	Key-value tags

QueryResponse

Field	Type	Description
`executed`	`bool`	Whether query was handled
`body`	`string`	Response body (data)
`metadata`	`string`	Response metadata
`cache_hit`	`bool`	Whether result came from cache
`tags`	`map<string,string>`	Response tags

Quick Usage

rpc.cc

// Command
auto cmd_or = kubemq::Command::Builder()
    .SetChannel("commands")
    .SetBody("deploy")
    .SetTimeout(std::chrono::seconds(10))
    .Build();
auto resp = client->SendCommand(*cmd_or);

// Query with caching
auto q_or = kubemq::Query::Builder()
    .SetChannel("queries")
    .SetBody("get-users")
    .SetTimeout(std::chrono::seconds(10))
    .SetCacheKey("users-list")
    .SetCacheTTL(std::chrono::seconds(60))
    .Build();
auto result = client->SendQuery(*q_or);

RPC