#moonrpc

Moved on mooncakes from Lfan-ke/moonrpc to moonbitstack/moonrpc.

let server = @net.GrpcServer::new()

// unary: one request, one reply.
server.register("/greet.Greeter/SayHello", req => handle(req))

// server-streaming: one request, an ordered run of replies.
server.register_server_streaming("/count.C/Up", (ctx, req) => [
  first(req), second(req), third(req),
])

// client-streaming: many requests collected, one reply at half-close.
server.register_client_streaming("/sum.S/Add", (ctx, msgs) => fold(msgs))

// bidi: each request answered as it arrives, a farewell at half-close.
server.register_bidi("/chat.C/Echo", ctx => @moonrpc.BidiHandler::{
  on_message: m => [reply_to(m)],
  on_end: () => [b"bye"],
})

server.serve()             // listens on 12000

#Packages

#The Channel client

let chan = @net.Channel::connect("127.0.0.1", @net.port)

// unary.
let reply = chan.unary("/greet.Greeter/SayHello", request)
reply.messages[0]   // the reply message; reply.grpc_status is 0 on success

// server-streaming: one request, every framed reply reassembled in order.
let out = chan.server_streaming("/count.C/Up", request)   // out.messages

// client-streaming: many requests, one reply.
let sum = chan.client_streaming("/sum.S/Add", [a, b, c])

// a deadline: sent as grpc-timeout and enforced locally — if it elapses the
// stream is reset and grpc_status comes back DEADLINE_EXCEEDED (4).
let bounded = chan.unary("/slow.S/Wait", request, timeout_millis=200)

#Health and interceptors

let health = @moonrpc.HealthService::new()
health.set_status("greet.Greeter", Serving)
server.register_health(health)      // Check + Watch on /grpc.health.v1.Health/*

engine.add_unary_interceptor((ctx, req, next) => {
  // inspect, then proceed — or return without calling next to short-circuit.
  next(ctx, req)
})

#Protobuf and Server Reflection

let w = @moonrpc.PbWriter::new()
w.string_(1, "testing")          // field 1, a length-delimited string
w.int32(2, -1)                   // field 2, a varint (negative -> ten octets)
let r = @moonrpc.PbReader::new(w.to_bytes())
let (field, wire) = r.read_tag() // (1, LengthDelim)
r.read_string()                  // "testing"

let greeter : @moonrpc.ServiceDescriptor = {
  name: "Greeter",
  methods: [
    { name: "SayHello", input_type: ".greet.HelloRequest",
      output_type: ".greet.HelloReply",
      client_streaming: false, server_streaming: false },
  ],
}
let refl = @moonrpc.ReflectionService::new()
refl.add_file(@moonrpc.FileDescriptor::new("greet.proto", "greet", services=[greeter]))
server.register_reflection(refl)

#The engines

let engine = @moonrpc.H2Server::new()
engine.register("/echo.Echo/Say", req => req)
let out = engine.feed(frame)   // -> [HEADERS(:status 200), DATA(reply), HEADERS(grpc-status:0)]

#Framing and status

let frame = @moonrpc.encode_message(b"hello grpc")
// [0x00][0x00 0x00 0x00 0x0A]["hello grpc"]  — 1-byte flag + 4-byte big-endian length + payload

let (compressed, payload) = @moonrpc.decode_message(frame).unwrap()
// (false, b"hello grpc")

@moonrpc.Status::code(NotFound)          // 5
@moonrpc.Status::name(Unauthenticated)   // "UNAUTHENTICATED", one of the 17 codes

let m : @moonrpc.Method = { service: "greet.Greeter", name: "SayHello" }
m.path()                                 // "/greet.Greeter/SayHello"

#Retries and hedging

let chan = @net.Channel::connect(
  "127.0.0.1",
  @net.port,
  throttling=@moonrpc.Throttling::new(tokens=10, ratio=0.1),
)
let reply = chan.unary(
  "/greet.Greeter/SayHello",
  b"Ada",
  timeout_millis=2000,
  retry=@moonrpc.retry,
)

#Annotations

#rpc.message
pub struct HelloRequest {
  name : String               // 1
  #rpc.field(number="4")
  tags : Array[String]        // 4
  nickname : String?          // 5, proto3 optional
}

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

#rpc.method
fn Greeter::say_hello(self : Greeter, req : HelloRequest) -> HelloReply raise { ... }

#rpc.method
fn Greeter::count_to(self : Greeter, req : Upto) -> Array[Number] { ... }

Greeter::{ greeting: "hello" }.register(server)   // an H2Server or a GrpcServer
reflection.add_file(descriptor())

rule(name: "rpc", command: "moonx moonbitstack/moonrpc/cmd/rpc@0.21.1 -C \"$mod_dir\" \"$input\" -o \"$output\"")

dev_build(rule: "rpc", input: "service.mbt", output: "service.g.mbt")
formatter(ignore: [ "service.g.mbt" ])

#Not yet

• TLS. moonbitlang/async's TLS exposes no ALPN, so HTTP/2 runs as h2c until moontls carries it over TCP.
• Response and request compression; incoming gzip is decompressed, outgoing messages are not compressed.
• Concurrent dispatch on one connection: a slow handler holds up the other streams, which is also why health Watch sends once and server-side deadlines are not preemptive.
• Connection backoff, client interceptors, channelz and service config.
• Descriptor fidelity for enums, nested types, imports and options, and transitive dependencies in reflection; #rpc waits on the same for enums and maps.
• Bidirectional streaming and client stubs from #rpc.

#Licence