Scenario

An e-commerce platform separates write and read operations: Commands create, update, and cancel orders in the write database, while Queries read order data from an optimized read projection. KubeMQ's built-in distinction between commands and queries maps directly to the CQRS pattern.

Architecture

Commands flow to the write side (orange) and mutate the write database; queries flow to the read side (amber) and serve from a synced projection.

Implementation

Command Handler (Writes)

The command handler subscribes to write operations, persists changes to the write database, and returns Executed: true.

write_handler.go

_, err := client.SubscribeToCommands(ctx, "orders.write", "write-handlers",
    kubemq.WithOnCommandReceive(func(cmd *kubemq.CommandReceive) {
        var order map[string]interface{}
        json.Unmarshal(cmd.Body, &order)

        err := db.SaveOrder(order)

        resp := kubemq.NewCommandReply().
            SetRequestId(cmd.Id).
            SetResponseTo(cmd.ResponseTo)
        if err != nil {
            resp.SetError(err.Error())
        } else {
            resp.SetExecutedAt(time.Now())
        }
        client.SendCommandResponse(ctx, resp)
    }),
    kubemq.WithOnError(func(err error) { log.Println("Error:", err) }),
)

write_handler.py

def on_command(request):
    order = json.loads(request.body)
    try:
        db.save_order(order)
        client.send_response_message(
            CommandResponse(command_received=request, is_executed=True))
    except Exception as e:
        client.send_response_message(
            CommandResponse(command_received=request, is_executed=False,
                error=str(e)))

client.subscribe_to_commands(CommandsSubscription(
    channel="orders.write", group="write-handlers",
    on_receive_command_callback=on_command,
    on_error_callback=lambda e: print(f"Error: {e}")),
    cancel=cancel)

write_handler.js

client.subscribeToCommands({
  channel: "orders.write",
  group: "write-handlers",
  onCommand: async (cmd) => {
    const order = JSON.parse(new TextDecoder().decode(cmd.body));
    try {
      await db.saveOrder(order);
      await client.sendCommandResponse({
        id: cmd.id, replyChannel: cmd.replyChannel, executed: true,
      });
    } catch (err) {
      await client.sendCommandResponse({
        id: cmd.id, replyChannel: cmd.replyChannel,
        executed: false, error: err.message,
      });
    }
  },
  onError: (err) => console.error("Error:", err.message),
});

WriteHandler.java

client.subscribeToCommands(CommandsSubscription.builder()
    .channel("orders.write").group("write-handlers")
    .onReceiveCommandCallback(cmd -> {
        try {
            var order = new Gson().fromJson(new String(cmd.getBody()), Order.class);
            db.saveOrder(order);
            return CommandResponseMessage.builder()
                .requestId(cmd.getId()).isExecuted(true).build();
        } catch (Exception e) {
            return CommandResponseMessage.builder()
                .requestId(cmd.getId()).isExecuted(false)
                .error(e.getMessage()).build();
        }
    })
    .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
    .build());

WriteHandler.cs

await foreach (var cmd in client.SubscribeToCommandsAsync(
    new CommandsSubscription { Channel = "orders.write", Group = "write-handlers" }))
{
    try
    {
        var order = JsonSerializer.Deserialize<Order>(cmd.Body.Span);
        await db.SaveOrderAsync(order!);
        await client.SendCommandResponseAsync(new CommandResponse
            { RequestId = cmd.Id, IsExecuted = true });
    }
    catch (Exception ex)
    {
        await client.SendCommandResponseAsync(new CommandResponse
            { RequestId = cmd.Id, IsExecuted = false, Error = ex.Message });
    }
}

WriteHandler.kt

client.subscribeToCommands(
    channel = "orders.write", group = "write-handlers",
    onCommand = { cmd ->
        try {
            val order = Json.decodeFromString<Order>(String(cmd.body))
            db.saveOrder(order)
            client.sendCommandResponse(requestId = cmd.id, isExecuted = true)
        } catch (e: Exception) {
            client.sendCommandResponse(
                requestId = cmd.id, isExecuted = false, error = e.message ?: "")
        }
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)

write_handler.cpp

client.subscribeToCommands("orders.write", "write-handlers",
    [&](const kubemq::CommandReceive& cmd) {
        try {
            auto order = nlohmann::json::parse(cmd.body);
            db.saveOrder(order);
            client.sendCommandResponse(cmd.id, true);
        } catch (const std::exception& e) {
            client.sendCommandResponse(cmd.id, false, e.what());
        }
    },
    [](const std::string& err) { std::cerr << "Error: " << err << std::endl; }
);

write_handler.rs

let rc = client.clone();
let sub = client
    .subscribe_to_commands("orders.write", "write-handlers", move |cmd| {
        let c = rc.clone();
        Box::pin(async move {
            let order: serde_json::Value =
                serde_json::from_slice(&cmd.body).unwrap_or_default();
            let reply = match db.save_order(&order) {
                Ok(_) => CommandReplyBuilder::new()
                    .request_id(&cmd.id)
                    .response_to(&cmd.response_to)
                    .executed(true)
                    .build(),
                Err(e) => CommandReplyBuilder::new()
                    .request_id(&cmd.id)
                    .response_to(&cmd.response_to)
                    .executed(false)
                    .error(&e.to_string())
                    .build(),
            };
            tokio::spawn(async move {
                let _ = c.send_command_response(reply).await;
            });
        })
    }, None)
    .await?;

write_handler.rb

sub = KubeMQ::CQ::CommandsSubscription.new(
  channel: "orders.write", group: "write-handlers")
client.subscribe_to_commands(sub, cancellation_token: cancel,
  on_error: ->(e) { puts "Error: #{e.message}" }) do |cmd|
  order = JSON.parse(cmd.body)
  begin
    db.save_order(order)
    response = KubeMQ::CQ::CommandResponseMessage.new(
      request_id: cmd.id, reply_channel: cmd.reply_channel, executed: true)
  rescue => e
    response = KubeMQ::CQ::CommandResponseMessage.new(
      request_id: cmd.id, reply_channel: cmd.reply_channel,
      executed: false, error: e.message)
  end
  client.send_response(response)
end

write_handler.exs

{:ok, sub} =
  KubeMQ.Client.subscribe_to_commands(client, "orders.write",
    group: "write-handlers",
    on_command: fn cmd ->
      order = Jason.decode!(cmd.body)

      case Orders.save(order) do
        :ok ->
          KubeMQ.CommandReply.new(
            request_id: cmd.id, response_to: cmd.reply_channel, executed: true)

        {:error, reason} ->
          KubeMQ.CommandReply.new(
            request_id: cmd.id, response_to: cmd.reply_channel,
            executed: false, error: reason)
      end
    end,
    on_error: fn err -> IO.puts("Error: #{err.message}") end
  )

Query Handler (Reads)

The query handler reads from the optimized read projection and returns full data in the response body.

read_handler.go

_, err := client.SubscribeToQueries(ctx, "orders.read", "read-handlers",
    kubemq.WithOnQueryReceive(func(query *kubemq.QueryReceive) {
        orderId := string(query.Body)
        order, err := readDB.GetOrder(orderId)

        resp := kubemq.NewQueryReply().
            SetRequestId(query.Id).
            SetResponseTo(query.ResponseTo)
        if err != nil {
            resp.SetError("order not found: " + orderId)
        } else {
            data, _ := json.Marshal(order)
            resp.SetBody(data).SetExecutedAt(time.Now())
        }
        client.SendQueryResponse(ctx, resp)
    }),
    kubemq.WithOnError(func(err error) { log.Println("Error:", err) }),
)

read_handler.py

def on_query(request):
    order_id = request.body.decode("utf-8")
    order = read_db.get_order(order_id)
    if order:
        client.send_response_message(QueryResponse(
            query_received=request, is_executed=True,
            body=json.dumps(order).encode()))
    else:
        client.send_response_message(QueryResponse(
            query_received=request, is_executed=False,
            error=f"order not found: {order_id}"))

client.subscribe_to_queries(QueriesSubscription(
    channel="orders.read", group="read-handlers",
    on_receive_query_callback=on_query,
    on_error_callback=lambda e: print(f"Error: {e}")),
    cancel=cancel)

read_handler.js

client.subscribeToQueries({
  channel: "orders.read",
  group: "read-handlers",
  onQuery: async (query) => {
    const orderId = new TextDecoder().decode(query.body);
    const order = await readDB.getOrder(orderId);
    if (order) {
      await client.sendQueryResponse({
        id: query.id, replyChannel: query.replyChannel, executed: true,
        body: new TextEncoder().encode(JSON.stringify(order)),
      });
    } else {
      await client.sendQueryResponse({
        id: query.id, replyChannel: query.replyChannel,
        executed: false, error: `order not found: ${orderId}`,
      });
    }
  },
  onError: (err) => console.error("Error:", err.message),
});

ReadHandler.java

client.subscribeToQueries(QueriesSubscription.builder()
    .channel("orders.read").group("read-handlers")
    .onReceiveQueryCallback(query -> {
        String orderId = new String(query.getBody());
        var order = readDB.getOrder(orderId);
        if (order != null) {
            return QueryResponseMessage.builder()
                .requestId(query.getId()).isExecuted(true)
                .body(new Gson().toJson(order).getBytes()).build();
        }
        return QueryResponseMessage.builder()
            .requestId(query.getId()).isExecuted(false)
            .error("order not found: " + orderId).build();
    })
    .onErrorCallback(err -> System.err.println("Error: " + err.getMessage()))
    .build());

ReadHandler.cs

await foreach (var query in client.SubscribeToQueriesAsync(
    new QueriesSubscription { Channel = "orders.read", Group = "read-handlers" }))
{
    var orderId = Encoding.UTF8.GetString(query.Body.Span);
    var order = await readDB.GetOrderAsync(orderId);
    if (order is not null)
    {
        await client.SendQueryResponseAsync(new QueryResponse
        {
            RequestId = query.Id, IsExecuted = true,
            Body = JsonSerializer.SerializeToUtf8Bytes(order)
        });
    }
    else
    {
        await client.SendQueryResponseAsync(new QueryResponse
        {
            RequestId = query.Id, IsExecuted = false,
            Error = $"order not found: {orderId}"
        });
    }
}

ReadHandler.kt

client.subscribeToQueries(
    channel = "orders.read", group = "read-handlers",
    onQuery = { query ->
        val orderId = String(query.body)
        val order = readDB.getOrder(orderId)
        if (order != null) {
            client.sendQueryResponse(requestId = query.id, isExecuted = true,
                body = Json.encodeToString(order).toByteArray())
        } else {
            client.sendQueryResponse(requestId = query.id, isExecuted = false,
                error = "order not found: $orderId")
        }
    },
    onError = { err -> System.err.println("Error: ${err.message}") }
)

read_handler.cpp

client.subscribeToQueries("orders.read", "read-handlers",
    [&](const kubemq::QueryReceive& query) {
        auto order = readDB.getOrder(query.body);
        if (!order.empty()) {
            client.sendQueryResponse(query.id, true, order);
        } else {
            client.sendQueryResponse(query.id, false, "",
                "order not found: " + query.body);
        }
    },
    [](const std::string& err) { std::cerr << "Error: " << err << std::endl; }
);

read_handler.rs

let rc = client.clone();
let sub = client
    .subscribe_to_queries("orders.read", "read-handlers", move |query| {
        let c = rc.clone();
        Box::pin(async move {
            let order_id = String::from_utf8_lossy(&query.body).to_string();
            let reply = match read_db.get_order(&order_id) {
                Some(order) => QueryReplyBuilder::new()
                    .request_id(&query.id)
                    .response_to(&query.response_to)
                    .body(serde_json::to_vec(&order).unwrap())
                    .build(),
                None => QueryReplyBuilder::new()
                    .request_id(&query.id)
                    .response_to(&query.response_to)
                    .executed(false)
                    .error(&format!("order not found: {}", order_id))
                    .build(),
            };
            tokio::spawn(async move {
                let _ = c.send_query_response(reply).await;
            });
        })
    }, None)
    .await?;

read_handler.rb

sub = KubeMQ::CQ::QueriesSubscription.new(
  channel: "orders.read", group: "read-handlers")
client.subscribe_to_queries(sub, cancellation_token: cancel,
  on_error: ->(e) { puts "Error: #{e.message}" }) do |query|
  order_id = query.body
  order = read_db.get_order(order_id)
  response =
    if order
      KubeMQ::CQ::QueryResponseMessage.new(
        request_id: query.id, reply_channel: query.reply_channel,
        executed: true, body: order.to_json, metadata: "application/json")
    else
      KubeMQ::CQ::QueryResponseMessage.new(
        request_id: query.id, reply_channel: query.reply_channel,
        executed: false, error: "order not found: #{order_id}")
    end
  client.send_response(response)
end

read_handler.exs

{:ok, sub} =
  KubeMQ.Client.subscribe_to_queries(client, "orders.read",
    group: "read-handlers",
    on_query: fn query ->
      order_id = query.body

      case ReadStore.get_order(order_id) do
        nil ->
          KubeMQ.QueryReply.new(
            request_id: query.id, response_to: query.reply_channel,
            executed: false, error: "order not found: #{order_id}")

        order ->
          KubeMQ.QueryReply.new(
            request_id: query.id, response_to: query.reply_channel,
            executed: true, body: Jason.encode!(order),
            metadata: "application/json")
      end
    end,
    on_error: fn err -> IO.puts("Error: #{err.message}") end
  )

Client (Sends Both)

client.go

cmdResp, _ := client.SendCommand(ctx, kubemq.NewCommand().
    SetChannel("orders.write").
    SetBody([]byte(`{"orderId":"ORD-100","item":"Widget","qty":3}`)).
    SetTimeout(10 * time.Second))
log.Printf("Create order — Executed: %v", cmdResp.Executed)

qryResp, _ := client.SendQuery(ctx, kubemq.NewQuery().
    SetChannel("orders.read").
    SetBody([]byte("ORD-100")).
    SetTimeout(10 * time.Second))
log.Printf("Get order — Data: %s", qryResp.Body)

client.py

cmd_resp = client.send_command(CommandMessage(
    channel="orders.write",
    body=b'{"orderId":"ORD-100","item":"Widget","qty":3}',
    timeout_in_seconds=10))
print(f"Create order — Executed: {cmd_resp.is_executed}")

qry_resp = client.send_query(QueryMessage(
    channel="orders.read", body=b"ORD-100", timeout_in_seconds=10))
print(f"Get order — Data: {qry_resp.body.decode('utf-8')}")

client.js

const cmdResp = await client.sendCommand(createCommand({
  channel: "orders.write",
  body: JSON.stringify({ orderId: "ORD-100", item: "Widget", qty: 3 }),
  timeoutInSeconds: 10,
}));
console.log("Create order — Executed:", cmdResp.executed);

const qryResp = await client.sendQuery(createQuery({
  channel: "orders.read",
  body: "ORD-100",
  timeoutInSeconds: 10,
}));
console.log("Get order — Data:", new TextDecoder().decode(qryResp.body));

Client.java

var cmdResp = client.sendCommandRequest(CommandMessage.builder()
    .channel("orders.write")
    .body("{\"orderId\":\"ORD-100\",\"item\":\"Widget\",\"qty\":3}".getBytes())
    .timeout(10000).build());
System.out.println("Create order — Executed: " + cmdResp.isExecuted());

var qryResp = client.sendQueryRequest(QueryMessage.builder()
    .channel("orders.read").body("ORD-100".getBytes()).timeout(10000).build());
System.out.println("Get order — Data: " + new String(qryResp.getBody()));

Client.cs

var cmdResp = await client.SendCommandAsync(new CommandMessage
{
    Channel = "orders.write",
    Body = Encoding.UTF8.GetBytes("{\"orderId\":\"ORD-100\",\"item\":\"Widget\",\"qty\":3}"),
    Timeout = TimeSpan.FromSeconds(10)
});
Console.WriteLine($"Create order — Executed: {cmdResp.IsExecuted}");

var qryResp = await client.SendQueryAsync(new QueryMessage
{
    Channel = "orders.read", Body = Encoding.UTF8.GetBytes("ORD-100"),
    Timeout = TimeSpan.FromSeconds(10)
});
Console.WriteLine($"Get order — Data: {Encoding.UTF8.GetString(qryResp.Body.Span)}");

Client.kt

val cmdResp = client.sendCommand(CommandMessage(
    channel = "orders.write",
    body = """{"orderId":"ORD-100","item":"Widget","qty":3}""".toByteArray(),
    timeout = 10000))
println("Create order — Executed: ${cmdResp.isExecuted}")

val qryResp = client.sendQuery(QueryMessage(
    channel = "orders.read", body = "ORD-100".toByteArray(), timeout = 10000))
println("Get order — Data: ${String(qryResp.body)}")

client.cpp

kubemq::CommandMessage cmd;
cmd.channel = "orders.write";
cmd.body = R"({"orderId":"ORD-100","item":"Widget","qty":3})";
cmd.timeout = 10000;
auto cmdResp = client.sendCommand(cmd);
std::cout << "Create order — Executed: " << cmdResp.isExecuted << std::endl;

kubemq::QueryMessage query;
query.channel = "orders.read";
query.body = "ORD-100";
query.timeout = 10000;
auto qryResp = client.sendQuery(query);
std::cout << "Get order — Data: " << qryResp.body << std::endl;

client.rs

let command = CommandBuilder::new()
    .channel("orders.write")
    .body(br#"{"orderId":"ORD-100","item":"Widget","qty":3}"#.to_vec())
    .timeout(Duration::from_secs(10))
    .build();
let cmd_resp = client.send_command(command).await?;
println!("Create order — Executed: {}", cmd_resp.executed);

let query = QueryBuilder::new()
    .channel("orders.read")
    .body(b"ORD-100".to_vec())
    .timeout(Duration::from_secs(10))
    .build();
let qry_resp = client.send_query(query).await?;
println!("Get order — Data: {}", String::from_utf8_lossy(&qry_resp.body));

client.rb

cmd = KubeMQ::CQ::CommandMessage.new(
  channel: "orders.write", timeout: 10_000,
  body: '{"orderId":"ORD-100","item":"Widget","qty":3}')
cmd_resp = client.send_command(cmd)
puts "Create order — Executed: #{cmd_resp.executed}"

query = KubeMQ::CQ::QueryMessage.new(
  channel: "orders.read", timeout: 10_000, body: "ORD-100")
qry_resp = client.send_query(query)
puts "Get order — Data: #{qry_resp.body}"

client.exs

command =
  KubeMQ.Command.new(
    channel: "orders.write",
    body: ~s({"orderId":"ORD-100","item":"Widget","qty":3}),
    timeout: 10_000)

{:ok, cmd_resp} = KubeMQ.Client.send_command(client, command)
IO.puts("Create order — Executed: #{cmd_resp.executed}")

query = KubeMQ.Query.new(channel: "orders.read", body: "ORD-100", timeout: 10_000)
{:ok, qry_resp} = KubeMQ.Client.send_query(client, query)
IO.puts("Get order — Data: #{qry_resp.body}")

CQRS Implementation

Scenario

Architecture

Implementation

Command Handler (Writes)

Query Handler (Reads)

Client (Sends Both)

Production Considerations

On this page

CQRS Implementation

Scenario

Architecture

Implementation

Command Handler (Writes)

Query Handler (Reads)

Client (Sends Both)

Production Considerations

Eventual Consistency

Query Caching

On this page