# CQRS Implementation (/learn/rpc/scenarios/cqrs-implementation)



## Scenario [#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 [#architecture]

<Mermaid
  chart="flowchart LR
    Client[&#x22;Order Service&#x22;]
    K[&#x22;KubeMQ&#x22;]
    WH[&#x22;Write Handler&#x22;]
    RH[&#x22;Read Handler&#x22;]
    WDB[(&#x22;Write DB&#x22;)]
    RDB[(&#x22;Read Projection&#x22;)]

    Client -->|Commands| K
    Client -->|Queries| K
    K -->|Command| WH
    K -->|Query| RH
    WH --> WDB
    WDB -.->|Sync| RDB
    RH --> RDB

    class Client client
    class K broker
    class WH command
    class RH query
    class WDB,RDB data"
/>

*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 [#implementation]

### Command Handler (Writes) [#command-handler-writes]

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

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="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) }),
    )
    ```
  </Tab>

  <Tab value="Python">
    ```python title="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)
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="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),
    });
    ```
  </Tab>

  <Tab value="Java">
    ```java title="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());
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="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 });
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="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}") }
    )
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="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; }
    );
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="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?;
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="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
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="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
      )
    ```
  </Tab>
</Tabs>

### Query Handler (Reads) [#query-handler-reads]

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

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="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) }),
    )
    ```
  </Tab>

  <Tab value="Python">
    ```python title="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)
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="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),
    });
    ```
  </Tab>

  <Tab value="Java">
    ```java title="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());
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="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}"
            });
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="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}") }
    )
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="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; }
    );
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="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?;
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="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
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="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
      )
    ```
  </Tab>
</Tabs>

### Client (Sends Both) [#client-sends-both]

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="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)
    ```
  </Tab>

  <Tab value="Python">
    ```python title="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')}")
    ```
  </Tab>

  <Tab value="Node.js">
    ```javascript title="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));
    ```
  </Tab>

  <Tab value="Java">
    ```java title="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()));
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="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)}");
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="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)}")
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="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;
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="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));
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="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}"
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="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}")
    ```
  </Tab>
</Tabs>

## Production Considerations [#production-considerations]

<Accordions>
  <Accordion title="Eventual Consistency">
    The read projection may lag behind the write database. After a command succeeds, a query may not immediately reflect the change. Design your UI to handle this — show a "processing" state or use optimistic updates on the client side.
  </Accordion>

  <Accordion title="Query Caching">
    Use KubeMQ's built-in [query caching](/learn/rpc/tutorials/query-caching) on the read side with `CacheKey` and `CacheTTL`. This reduces load on the read database for frequently accessed data.
  </Accordion>
</Accordions>