# Service-to-Service API Gateway (/learn/rpc/scenarios/api-gateway)



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

<Mermaid
  chart="graph LR
  C[&#x22;HTTP Client&#x22;]
  GW[&#x22;API Gateway<br/>(sender)&#x22;]
  CMD{{&#x22;Commands channel<br/>orders.process&#x22;}}
  QRY{{&#x22;Queries channel<br/>orders.lookup&#x22;}}
  OS[&#x22;Order Service&#x22;]
  IS[&#x22;Inventory Service&#x22;]
  US[&#x22;User Service&#x22;]

  C -- &#x22;HTTP request&#x22; --> GW
  GW -- &#x22;write ops&#x22; --> CMD
  GW -- &#x22;read ops&#x22; --> QRY
  CMD -- &#x22;command&#x22; --> OS
  QRY -- &#x22;query&#x22; --> OS
  CMD -- &#x22;command&#x22; --> IS
  QRY -- &#x22;query&#x22; --> US

  class C external
  class GW,OS,IS,US client
  class CMD command
  class QRY query"
/>

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

## Implementation [#implementation]

### API Gateway (Sender) [#api-gateway-sender]

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

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

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

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

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

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

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

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

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

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

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

### Order Service (Responder) [#order-service-responder]

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="order_service.go"
    client.SubscribeToCommands(ctx, "orders.process", "order-workers", ...)
    client.SubscribeToQueries(ctx, "orders.lookup", "order-workers", ...)
    ```
  </Tab>

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

  <Tab value="Node.js">
    ```javascript title="order_service.js"
    client.subscribeToCommands({ channel: "orders.process", group: "order-workers", ... });
    client.subscribeToQueries({ channel: "orders.lookup", group: "order-workers", ... });
    ```
  </Tab>

  <Tab value="Java">
    ```java title="OrderService.java"
    client.subscribeToCommands(CommandsSubscription.builder()
        .channel("orders.process").group("order-workers")...build());
    client.subscribeToQueries(QueriesSubscription.builder()
        .channel("orders.lookup").group("order-workers")...build());
    ```
  </Tab>

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

  <Tab value="Kotlin">
    ```kotlin title="OrderService.kt"
    client.subscribeToCommands(channel = "orders.process", group = "order-workers", ...)
    client.subscribeToQueries(channel = "orders.lookup", group = "order-workers", ...)
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="order_service.cpp"
    client.subscribeToCommands("orders.process", "order-workers", ...);
    client.subscribeToQueries("orders.lookup", "order-workers", ...);
    ```
  </Tab>

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

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

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

## Production Considerations [#production-considerations]

<Accordions>
  <Accordion title="Timeout Propagation">
    Set the HTTP timeout longer than the KubeMQ timeout to avoid the gateway timing out before the RPC call completes. A good rule: HTTP timeout = KubeMQ timeout + 2–3 seconds for overhead.
  </Accordion>

  <Accordion title="Circuit Breaker per Service">
    Each backend service should have its own [circuit breaker](/learn/rpc/how-to/circuit-breaker). If the order service is down, the gateway can still serve inventory and user queries.
  </Accordion>

  <Accordion title="Request Tracing">
    Pass a correlation ID through KubeMQ tags to trace requests across services. Use the `Tags` field to propagate OpenTelemetry trace context.
  </Accordion>
</Accordions>