KubeMQ
LearnRPCScenarios

Service-to-Service API Gateway

Build a microservice API gateway using KubeMQ RPC for inter-service communication.

Scenario

An API gateway receives HTTP requests from clients and dispatches them to backend microservices via KubeMQ RPC. Commands handle write operations (create order, update inventory), while queries handle read operations (get order, list products). Each backend service subscribes to its own channel.

Architecture

The gateway maps HTTP writes to Commands and HTTP reads to Queries, each routed through KubeMQ to the owning backend service.

Implementation

API Gateway (Sender)

The gateway maps incoming HTTP requests to KubeMQ commands or queries.

gateway.go
package main

import (
    "context"
    "encoding/json"
    "log"
    "net/http"
    "time"

    "github.com/kubemq-io/kubemq-go/v2"
)

var client *kubemq.Client

func createOrder(w http.ResponseWriter, r *http.Request) {
    var body map[string]interface{}
    json.NewDecoder(r.Body).Decode(&body)
    data, _ := json.Marshal(body)

    ctx, cancel := context.WithTimeout(r.Context(), 8*time.Second)
    defer cancel()

    resp, err := client.SendCommand(ctx, kubemq.NewCommand().
        SetChannel("orders.process").
        SetBody(data).
        SetTimeout(5 * time.Second))
    if err != nil || !resp.Executed {
        http.Error(w, "order creation failed", http.StatusInternalServerError)
        return
    }
    w.WriteHeader(http.StatusCreated)
    w.Write([]byte(`{"status":"created"}`))
}

func getOrder(w http.ResponseWriter, r *http.Request) {
    orderId := r.URL.Query().Get("id")
    ctx, cancel := context.WithTimeout(r.Context(), 8*time.Second)
    defer cancel()

    resp, err := client.SendQuery(ctx, kubemq.NewQuery().
        SetChannel("orders.lookup").
        SetBody([]byte(orderId)).
        SetTimeout(5 * time.Second))
    if err != nil || !resp.Executed {
        http.Error(w, "order not found", http.StatusNotFound)
        return
    }
    w.Header().Set("Content-Type", "application/json")
    w.Write(resp.Body)
}
gateway.py
from flask import Flask, request, jsonify
from kubemq.cq import Client as CQClient, CommandMessage, QueryMessage

app = Flask(__name__)
client = CQClient(address="localhost:50000")

@app.route("/orders", methods=["POST"])
def create_order():
    data = request.get_json()
    response = client.send_command(CommandMessage(
        channel="orders.process",
        body=str(data).encode(),
        timeout_in_seconds=5))
    if response.is_executed:
        return jsonify({"status": "created"}), 201
    return jsonify({"error": response.error}), 500

@app.route("/orders/<order_id>", methods=["GET"])
def get_order(order_id):
    response = client.send_query(QueryMessage(
        channel="orders.lookup",
        body=order_id.encode(),
        timeout_in_seconds=5))
    if response.is_executed:
        return response.body, 200, {"Content-Type": "application/json"}
    return jsonify({"error": response.error}), 404
gateway.js
const express = require("express");
const { KubeMQClient } = require("kubemq-js");

const app = express();
app.use(express.json());
const client = new KubeMQClient({ address: "localhost:50000" });

app.post("/orders", async (req, res) => {
  const response = await client.sendCommand({
    channel: "orders.process",
    body: Buffer.from(JSON.stringify(req.body)),
    timeoutInSeconds: 5,
  });
  if (response.isExecuted) return res.status(201).json({ status: "created" });
  res.status(500).json({ error: response.error });
});

app.get("/orders/:id", async (req, res) => {
  const response = await client.sendQuery({
    channel: "orders.lookup",
    body: Buffer.from(req.params.id),
    timeoutInSeconds: 5,
  });
  if (response.isExecuted)
    return res.json(JSON.parse(Buffer.from(response.body).toString()));
  res.status(404).json({ error: response.error });
});

app.listen(3000);
Gateway.java
@RestController
public class OrderGateway {
    private final CQClient client = CQClient.builder()
        .address("localhost:50000").clientId("api-gateway").build();

    @PostMapping("/orders")
    public ResponseEntity<String> createOrder(@RequestBody String body) {
        var resp = client.sendCommandRequest(CommandMessage.builder()
            .channel("orders.process").body(body.getBytes()).timeout(5000).build());
        if (resp.isExecuted())
            return ResponseEntity.status(201).body("{\"status\":\"created\"}");
        return ResponseEntity.status(500).body("{\"error\":\"" + resp.getError() + "\"}");
    }

    @GetMapping("/orders/{id}")
    public ResponseEntity<String> getOrder(@PathVariable String id) {
        var resp = client.sendQueryRequest(QueryMessage.builder()
            .channel("orders.lookup").body(id.getBytes()).timeout(5000).build());
        if (resp.isExecuted())
            return ResponseEntity.ok(new String(resp.getBody()));
        return ResponseEntity.status(404).body("{\"error\":\"" + resp.getError() + "\"}");
    }
}
Gateway.cs
app.MapPost("/orders", async (HttpContext ctx, KubeMQClient client) =>
{
    var body = await new StreamReader(ctx.Request.Body).ReadToEndAsync();
    var resp = await client.SendCommandAsync(new CommandMessage
    {
        Channel = "orders.process",
        Body = Encoding.UTF8.GetBytes(body),
        Timeout = TimeSpan.FromSeconds(5)
    });
    if (resp.IsExecuted)
        return Results.Created("/orders", new { status = "created" });
    return Results.Problem(resp.Error);
});

app.MapGet("/orders/{id}", async (string id, KubeMQClient client) =>
{
    var resp = await client.SendQueryAsync(new QueryMessage
    {
        Channel = "orders.lookup",
        Body = Encoding.UTF8.GetBytes(id),
        Timeout = TimeSpan.FromSeconds(5)
    });
    if (resp.IsExecuted)
        return Results.Text(Encoding.UTF8.GetString(resp.Body.Span), "application/json");
    return Results.NotFound(new { error = resp.Error });
});
Gateway.kt
@RestController
class OrderGateway {
    private val client = CQClient("localhost:50000")

    @PostMapping("/orders")
    fun createOrder(@RequestBody body: String): ResponseEntity<String> {
        val resp = client.sendCommand(CommandMessage(
            channel = "orders.process", body = body.toByteArray(), timeout = 5000))
        return if (resp.isExecuted)
            ResponseEntity.status(201).body("""{"status":"created"}""")
        else ResponseEntity.status(500).body("""{"error":"${resp.error}"}""")
    }

    @GetMapping("/orders/{id}")
    fun getOrder(@PathVariable id: String): ResponseEntity<String> {
        val resp = client.sendQuery(QueryMessage(
            channel = "orders.lookup", body = id.toByteArray(), timeout = 5000))
        return if (resp.isExecuted)
            ResponseEntity.ok(String(resp.body))
        else ResponseEntity.status(404).body("""{"error":"${resp.error}"}""")
    }
}
gateway.cpp
auto client = kubemq::CQClient("localhost:50000");

void handleCreateOrder(const HttpRequest& req, HttpResponse& res) {
    kubemq::CommandMessage cmd;
    cmd.channel = "orders.process";
    cmd.body = req.body;
    cmd.timeout = 5000;

    auto resp = client.sendCommand(cmd);
    if (resp.isExecuted) {
        res.status = 201;
        res.body = R"({"status":"created"})";
    } else {
        res.status = 500;
        res.body = R"({"error":")" + resp.error + R"("})";
    }
}

void handleGetOrder(const HttpRequest& req, HttpResponse& res) {
    kubemq::QueryMessage query;
    query.channel = "orders.lookup";
    query.body = req.params["id"];
    query.timeout = 5000;

    auto resp = client.sendQuery(query);
    if (resp.isExecuted) {
        res.body = resp.body;
    } else {
        res.status = 404;
    }
}
gateway.rs
use kubemq::prelude::*;
use kubemq::{CommandBuilder, QueryBuilder};
use std::time::Duration;

// Build once at startup, share the client across request handlers.
let client = KubemqClient::builder()
    .host("localhost")
    .port(50000)
    .build()
    .await?;

// POST /orders -> command (write)
async fn create_order(client: &KubemqClient, body: Vec<u8>) -> kubemq::Result<bool> {
    let command = CommandBuilder::new()
        .channel("orders.process")
        .body(body)
        .timeout(Duration::from_secs(5))
        .build();
    let resp = client.send_command(command).await?;
    Ok(resp.executed)
}

// GET /orders/{id} -> query (read)
async fn get_order(client: &KubemqClient, id: &str) -> kubemq::Result<Vec<u8>> {
    let query = QueryBuilder::new()
        .channel("orders.lookup")
        .body(id.as_bytes().to_vec())
        .timeout(Duration::from_secs(5))
        .build();
    let resp = client.send_query(query).await?;
    Ok(resp.body)
}
gateway.rb
require 'kubemq'

# Build once at startup, share across request handlers.
client = KubeMQ::CQClient.new(address: 'localhost:50000', client_id: 'api-gateway')

# POST /orders -> command (write)
def create_order(client, body)
  msg = KubeMQ::CQ::CommandMessage.new(
    channel: 'orders.process',
    timeout: 5,
    body: body
  )
  result = client.send_command(msg)
  result.executed
end

# GET /orders/:id -> query (read)
def get_order(client, id)
  msg = KubeMQ::CQ::QueryMessage.new(
    channel: 'orders.lookup',
    timeout: 5,
    body: id
  )
  result = client.send_query(msg)
  result.executed ? result.body : nil
end
gateway.exs
# Build once at startup, share the client across request handlers.
{:ok, client} =
  KubeMQ.Client.start_link(address: "localhost:50000", client_id: "api-gateway")

# POST /orders -> command (write)
def create_order(client, body) do
  command =
    KubeMQ.Command.new(
      channel: "orders.process",
      body: body,
      timeout: 5_000
    )

  case KubeMQ.Client.send_command(client, command) do
    {:ok, resp} -> resp.executed
    {:error, _} -> false
  end
end

# GET /orders/:id -> query (read)
def get_order(client, id) do
  query =
    KubeMQ.Query.new(
      channel: "orders.lookup",
      body: id,
      timeout: 5_000
    )

  case KubeMQ.Client.send_query(client, query) do
    {:ok, %{executed: true} = resp} -> {:ok, resp.body}
    _ -> {:error, :not_found}
  end
end

Order Service (Responder)

order_service.go
client.SubscribeToCommands(ctx, "orders.process", "order-workers", ...)
client.SubscribeToQueries(ctx, "orders.lookup", "order-workers", ...)
order_service.py
client.subscribe_to_commands(CommandsSubscription(
    channel="orders.process", group="order-workers", ...))
client.subscribe_to_queries(QueriesSubscription(
    channel="orders.lookup", group="order-workers", ...))
order_service.js
client.subscribeToCommands({ channel: "orders.process", group: "order-workers", ... });
client.subscribeToQueries({ channel: "orders.lookup", group: "order-workers", ... });
OrderService.java
client.subscribeToCommands(CommandsSubscription.builder()
    .channel("orders.process").group("order-workers")...build());
client.subscribeToQueries(QueriesSubscription.builder()
    .channel("orders.lookup").group("order-workers")...build());
OrderService.cs
await foreach (var cmd in client.SubscribeToCommandsAsync(
    new CommandsSubscription { Channel = "orders.process", Group = "order-workers" })) { }
await foreach (var q in client.SubscribeToQueriesAsync(
    new QueriesSubscription { Channel = "orders.lookup", Group = "order-workers" })) { }
OrderService.kt
client.subscribeToCommands(channel = "orders.process", group = "order-workers", ...)
client.subscribeToQueries(channel = "orders.lookup", group = "order-workers", ...)
order_service.cpp
client.subscribeToCommands("orders.process", "order-workers", ...);
client.subscribeToQueries("orders.lookup", "order-workers", ...);
order_service.rs
// Each handler joins the "order-workers" group so commands and queries are
// load-balanced across service replicas. Replies are sent back to the gateway.
let _cmd_sub = client
    .subscribe_to_commands("orders.process", "order-workers", on_command, None)
    .await?;
let _qry_sub = client
    .subscribe_to_queries("orders.lookup", "order-workers", on_query, None)
    .await?;
order_service.rb
cmd_sub = KubeMQ::CQ::CommandsSubscription.new(channel: 'orders.process', group: 'order-workers')
client.subscribe_to_commands(cmd_sub, cancellation_token: cancel) { |cmd| handle_command(cmd) }

qry_sub = KubeMQ::CQ::QueriesSubscription.new(channel: 'orders.lookup', group: 'order-workers')
client.subscribe_to_queries(qry_sub, cancellation_token: cancel) { |query| handle_query(query) }
order_service.exs
{:ok, _cmd_sub} =
  KubeMQ.Client.subscribe_to_commands(client, "orders.process",
    group: "order-workers", on_command: &handle_command/1)

{:ok, _qry_sub} =
  KubeMQ.Client.subscribe_to_queries(client, "orders.lookup",
    group: "order-workers", on_query: &handle_query/1)

Production Considerations

Was this page helpful?

Implement Circuit Breaker

Protect services from cascading failures with a circuit breaker pattern for RPC calls.

CQRS Implementation

Implement Command Query Responsibility Segregation with KubeMQ Commands and Queries.

On this page

ScenarioArchitectureImplementationAPI Gateway (Sender)Order Service (Responder)Production Considerations