From a message to a stream

1 · Install

Add the package with the MoonBit toolchain. The root package re-exports the pure engines and every typed helper, so one import is enough.

$ moon add moonbitstack/moonrpc

  import "moonbitstack/moonrpc"          // @moonrpc
  // sockets, when you need them:
  import "moonbitstack/moonrpc/net"      // @net  (native only)

@net is flagged supported_targets="native"; everything under @moonrpc compiles to JS, native and Wasm.

2 · Two pure engines

A gRPC call is two state machines trading HTTP/2 frames. H2Client and H2Server are pure: you hand each one a frame, it returns the frames to send back. No socket, no I/O — just data in, data out. That is why the same code runs in a server and in the browser showcase.

let server = @moonrpc.H2Server::new()
let client = @moonrpc.H2Client::new()

// client.unary opens a stream and returns the first frames to send
let (id, frames) = client.unary("/greet.Greeter/SayHello", request_bytes)

// shuttle frames between the engines until the call finishes
while !client.is_done(id) {
  for f in frames { for r in server.feed(f) { /* frames to client */ } }
  // ... and back: client.feed(r) -> frames to server
}

See the whole loop in src/examples/07-unary-call. The homepage shows the nine frames a single unary call produces.

3 · Define a message

A message is just bytes on the wire, packed by the protobuf codec. You can do it by hand, or let #rpc.message derive the codec from the struct.

With #rpc.message

The annotation derives @moonrpc.Message.

#rpc.message
pub struct HelloRequest {
  #rpc.field(number="1")
  name : String
}

#rpc.message
pub struct HelloReply {
  #rpc.field(number="1")
  name : String
}

#rpc.enum
pub enum Mood { Calm = 0; Glad = 1 }   // MOOD_CALM = 0, MOOD_GLAD = 1

#rpc.message
pub struct Profile {
  mood : Mood                    // an enum field
  scores : Map[String, Int64]    // map<string, int64>
  nickname : String?             // proto3 optional
}

By hand

fn encode_request(name : String) -> Bytes {
  let w = @moonrpc.PbWriter::new()
  w.string_(1, name)        // field 1, wire type LengthDelimited
  w.to_bytes()
}

fn decode_name(msg : Bytes) -> String {
  let r = @moonrpc.PbReader::new(msg)
  let _ = r.read_tag()      // consume the tag for field 1
  r.read_string()
}

#rpc.field(number="…") pins the protobuf tag. Omitted fields get consecutive numbers; #rpc.skip leaves a field out of the wire form. An enum's values take MoonBit's own tags, or count up from 0.

4 · Handle a unary call

Register a path and a function from request bytes to reply bytes. For typed messages, @moonrpc.unary decodes the input and encodes the output for you.

let server = @moonrpc.H2Server::new()

// hand-rolled: request bytes -> reply bytes
server.register("/greet.Greeter/SayHello", req => {
  let name = decode_name(req)
  let w = @moonrpc.PbWriter::new()
  w.string_(1, "hello, " + name)
  w.to_bytes()
})

// typed: works once HelloRequest / HelloReply are #rpc.message
server.register_handler(
  "/greet.Greeter/SayHello",
  @moonrpc.unary(fn (_ctx : @moonrpc.RpcContext, req : HelloRequest) -> HelloReply {
    { name : "hello, " + req.name }
  }),
)

Or describe the whole service once and let #rpc wire it up:

#rpc.service(name="greet.Greeter")
pub struct Greeter {}

#rpc.method
pub fn Greeter::say_hello(
  self : Greeter, _ctx : @moonrpc.RpcContext, req : HelloRequest,
) -> HelloReply {
  { name : "hello, " + req.name }
}

// later:
let greeter = Greeter::{}
greeter.register(server)   // generated by #rpc.service

The method's type decides the call kind: a plain reply is unary, an Array[…] reply is server-streaming, an Array[…] request is client-streaming, and both is bidi.

5 · Run it in memory

Because the engines are pure, a full call needs no socket — only the frame shuttle. This is the exact shape the showcase runs in your browser.

let server = @moonrpc.H2Server::new()
server.register("/greet.Greeter/SayHello", req => /* … */)

let client = @moonrpc.H2Client::new()
let (id, to_server) = client.unary("/greet.Greeter/SayHello", encode_request("heke"))

let mut to_client : Array[@http2.Frame] = []
let mut steps = 0
while !client.is_done(id) && steps < 100000 {
  steps = steps + 1
  for f in to_server { for r in server.feed(f) { to_client.push(r) } }
  to_server = []
  for r in to_client { for f in client.feed(r) { to_server.push(f) } }
  to_client = []
}

let reply = client.reply(id)
println("grpc-status \{reply.grpc_status}")
println("reply: \{decode_name(reply.messages[0])}")

6 · Serve it for real

To actually accept TCP connections you add the native socket layer. @net.GrpcServer wraps an engine and listens on a port — native only, because MoonBit's async runtime has no JS/Wasm sockets.

let greeter = Greeter::{}
let server = @net.GrpcServer::new()
greeter.register(server)     // the #rpc-generated registrar
server.serve()              // listens on :12000 (native target)

// elsewhere, the client #rpc.service generates alongside:
let client = GreeterClient::new(@net.Channel::connect("127.0.0.1", @net.port))
let reply = client.say_hello({ name: "heke" })   // a HelloReply, or a StatusError

Build with moon build --target native. The engines you tested in memory are the same ones serving here.

7 · The four call kinds

All four are the same frame machinery; only the number of messages each side sends differs. Each has a typed wrapper under @moonrpc.

// server streaming: one request, a sequence back
server.register_handler(
  "/demo.Counter/Up",
  @moonrpc.server_streaming(fn (_ctx, _req : UpRequest) -> Array[UpReply] {
    [{ n : 1 }, { n : 2 }, { n : 3 }]
  }),
)

Run all four live in the Showcase — each shows the frames it produces.

8 · Status and errors

Raise a StatusError and the engine answers the call with that grpc-status and message in the trailers. Any other error becomes UNKNOWN.

server.register_handler(
  "/greet.Greeter/SayHello",
  @moonrpc.unary(fn (_ctx, req : HelloRequest) -> HelloReply raise {
    if req.name == "" {
      raise @wire.StatusError(@wire.Status::InvalidArgument, "name required")
    }
    { name : "hello, " + req.name }
  }),
)

// on the client:
let reply = client.reply(id)
// reply.grpc_status  -> Int32  (0 = OK)
// reply.grpc_message -> String

9 · TLS

HTTP/2 over TLS is chosen by ALPN: both ends name h2 in the handshake and check what it settled on before a frame is exchanged. @net.server_tls and @net.client_tls build the two configurations with h2 alone, so the agreement cannot be forgotten.

// the server: a certificate chain, leaf first, and its key
let server = @net.GrpcServer::new()
greeter.register(server)
server.serve(port=12000, tls=@net.server_tls(@x509.pems(cert_text), @x509.PrivateKey::pem(key_text)))

// the client: the CA that signed the server, and nothing else to configure
let chan = @net.Channel::connect("api.example.com", 12000, tls=@net.client_tls(ca_roots))
let reply = chan.unary("/greet.Greeter/SayHello", request)

MuxChannel::connect, ManagedChannel::connect and dns_connect take the same tls. A peer that offers no h2 is closed rather than spoken to in a version it did not agree to, and a handshake failure on serve is a refused client, not a broken server.

10 · Interceptors

One interceptor shape covers all four call kinds, on both ends: a call is the request messages it carries and the replies they produce.

// server: (info, ctx, requests, next) -> replies
server_engine.add_interceptor((info, ctx, requests, next) => {
  let started = @async.now()
  let replies = next(requests)          // or answer here and never call it
  println("\{info.path} took \{@async.now() - started} ms")
  replies
})

// client: (invocation, next) -> reply
chan.add_interceptor((invocation, next) => {
  let metadata = invocation.metadata.copy()
  metadata.push({ name: b"authorization", value: b"Bearer ..." })
  next({ ..invocation, metadata })
})

An interceptor sees CallInfo — the path and which sides stream — so passing unary calls through untouched and handling only those is one if. The client chain wraps a call once, outside the retry loop: a retry is one invocation to an interceptor.

11 · Compression

gRPC names three encodings — identity, gzip and deflate — and negotiates them with grpc-encoding and grpc-accept-encoding. Set one and messages go compressed; a peer that did not ask for it still gets identity.

// the server answers in gzip to a caller that accepts it
let server = @net.GrpcServer::new()
server.set_encoding(@moonrpc.Encoding::Gzip)
greeter.register(server)

// the client compresses its own requests, and decodes whatever comes back
let chan = @net.Channel::connect("127.0.0.1", 12000, encoding=@moonrpc.Encoding::Gzip)
let reply = chan.unary("/greet.Greeter/SayHello", request)

A message whose compression flag is set under identity — compressed and not compressed at once — is refused rather than read as plaintext, and a body that will not inflate fails the call with INTERNAL.

Where to next

The API reference lists every public item by package. The Showcase runs the real engine in your browser. The repository's src/examples/* cover HTTP/2 frames, HPACK, flow control, health, reflection, deadlines, retries and more.

Back to home