#moonrpc

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

#rpc.message
pub struct HelloRequest {
  name : String
}

#rpc.message
pub struct HelloReply {
  message : String
}

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

#rpc.method
fn Greeter::say_hello(self : Greeter, req : HelloRequest) -> HelloReply {
  { message: self.greeting + ", " + req.name }
}

test {
  let greeter = Greeter::{ greeting: "hello" }
  assert_eq(greeter.say_hello({ name: "Leo" }).message, "hello, Leo")
}

#Annotations

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

#rpc.message
pub struct Profile {
  name : String               // 1
  #rpc.field(number="4")
  tags : Array[String]        // 4
  nickname : String?          // 5, proto3 optional
  mood : Mood                 // 6
  extras : Map[String, Int64] // 7, map<string, int64>
}

#rpc.message(name="HelloReply.Badge")   // nested: .greet.HelloReply.Badge
pub struct Badge { label : String }

#rpc.message
pub struct Upto { n : Int }

#rpc.message
pub struct Number { value : Int }

#rpc.method
fn Greeter::count_to(_self : Greeter, req : Upto) -> Array[Number] {
  let out = []
  for i = 1; i <= req.n; i = i + 1 {
    out.push({ value: i })
  }
  out
}

#rpc.method
fn Greeter::echo(_self : Greeter, req : Array[Number]) -> Array[Number] {
  req
}

// From the declarations above the generator writes a `Message` impl per message, and
// `Greeter::register(server)` plus `Greeter::descriptor()` per service. Written out by
// hand, registration is one line per method:
fn wire_greeter(server : @net.GrpcServer) -> Unit {
  server.register_unary("/greet.Greeter/SayHello", (_ctx, req) => req)
}

// The descriptor is what reflection serves, and it can be printed back as a .proto.
fn print_descriptor() -> Unit raise {
  let file = @moonrpc.FileDescriptor::new("greet.proto", "greet")
  println(file.source())
}

// A generated client takes a channel and returns typed replies:
//   let client = GreeterClient::new(@net.Channel::connect("127.0.0.1", @net.port))
//   let reply = client.say_hello({ name: "Leo" })

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

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

#The four call kinds by hand

fn register_four_kinds(server : @moonrpc.H2Server) -> Unit {
  // unary: one request, one reply.
  server.register("/greet.Greeter/SayHello", req => b"hello:" + req)

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

  // client-streaming: every request is collected, then one reply.
  server.register_client_streaming("/sum.S/Add", (_ctx, msgs) => msgs[0])

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

#Packages

#The Channel client

async fn channel_calls(chan : @net.Channel) -> Unit raise {
  // unary: the reply messages, in order, and the status that ended the call.
  let reply = chan.unary("/greet.Greeter/SayHello", b"Leo")
  assert_eq(reply.grpc_status, 0)
  assert_eq(reply.messages.length(), 1)

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

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

  // 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", b"x", timeout_millis=200)
  assert_eq(bounded.grpc_status is 4 || bounded.grpc_status is 0, true)
}

#Health

fn health_service() -> @moonrpc.HealthService {
  let health = @moonrpc.HealthService::new()
  health.set_status("greet.Greeter", @service.ServingStatus::Serving)
  // Check and Watch answer on /grpc.health.v1.Health/*; a status change is pushed to
  // every live Watch.
  health.set_status("greet.Greeter", @service.ServingStatus::NotServing)
  health
}

#Interceptors

fn logging_interceptor() -> @server.Interceptor {
  (info, _ctx, requests, next) => {
    // Answer here without calling `next` to serve the call yourself.
    let replies = next(requests)
    println("{info.path} answered {replies.length()}")
    replies
  }
}

fn authorize() -> @client.Interceptor {
  // A client interceptor sees the invocation and may change what goes out.
  async fn(invocation, next) {
    let metadata = invocation.metadata
    metadata.push(@header.Header::of("authorization", "Bearer ..."))
    next({ ..invocation, metadata })
  }
}

#TLS

async fn serve_over_tls(cert_text : String, key_text : String) -> Unit raise {
  let server = @net.GrpcServer::new()
  server.register("/echo.Echo/Say", req => req)
  // leaf certificate first, then the chain; ALPN settles on `h2`.
  server.serve(
    port=12000,
    tls=@net.server_tls(@x509.pems(cert_text), @x509.PrivateKey::pem(key_text)),
  )
}

async fn call_over_tls(roots : @x509.Roots) -> Unit raise {
  let chan = @net.Channel::connect(
    "api.example.com",
    12000,
    tls=@net.client_tls(roots),
  )
  chan.close()
}

#Compression

async fn compressed_traffic(server : @net.GrpcServer, host : String) -> Unit raise {
  // The server compresses a reply when the caller says it accepts the encoding.
  server.set_encoding(@moonrpc.Encoding::Gzip)
  let chan = @net.Channel::connect(
    host,
    12000,
    encoding=@moonrpc.Encoding::Gzip,
  )
  chan.close()
}

#Interoperability

#Conformance

#Stats and channelz

fn stats_handler() -> @moonrpc.StatsHandler {
  @moonrpc.StatsHandler::new(rpc=event => {
    match event {
      @wire.RpcStats::Begin(path~, name~) => println("begin {path} {name}")
      @wire.RpcStats::End(status~, message~) => println("end {status} {message}")
      _ => ()
    }
  })
}

fn channelz_registry() -> @moonrpc.Registry {
  let registry = @moonrpc.Registry::new()
  // The handler a channel reports its calls through.
  let id = registry.add_channel("127.0.0.1:12000")
  registry.channel_handler(id) |> ignore
  registry
}

fn install_channelz(server : @net.GrpcServer) -> Unit {
  let registry = @moonrpc.Registry::new()
  let server_id = registry.add_server()
  ignore(registry.add_server_socket(server_id, "127.0.0.1:12000"))
  let channelz = @moonrpc.ChannelzService::new(registry)
  channelz.install(server)
  let reflection = @moonrpc.ReflectionService::new()
  channelz.add_to_reflection(reflection)
  server.register_reflection(reflection)
}

#Service configuration and WaitForReady

async fn configured_call(host : String) -> Unit raise {
  let text =
    #|{
    #|  "methodConfig": [{
    #|    "name": [{"service": "greet.Greeter", "method": "SayHello"}],
    #|    "timeout": "1.5s",
    #|    "waitForReady": true,
    #|    "retryPolicy": {
    #|      "maxAttempts": 4, "initialBackoff": "0.1s", "maxBackoff": "1s",
    #|      "backoffMultiplier": 2, "retryableStatusCodes": ["UNAVAILABLE"]
    #|    }
    #|  }]
    #|}
  let config = @moonrpc.ServiceConfig::parse(text) catch {
    _ => @moonrpc.ServiceConfig::empty()
  }
  let chan = @net.ManagedChannel::connect(
    [("127.0.0.1", 12000)],
    config=config,
  )
  chan.close()
}

#Thrift

async fn serve_and_call_thrift() -> Unit raise {
  let processor = @thrift.Processor::new()
  processor.add("ping", proto => {
    ignore(proto.read_field_begin())                 // field 1: the argument
    let text = proto.read_string()
    ignore(proto.read_field_begin())                 // STOP
    proto.write_field_begin(@thrift.type_string, 0)  // field 0: the result
    proto.write_string("pong:" + text)
  })
  let listener = @net.bind_thrift(host="127.0.0.1", port=0)
  let port = listener.addr().port()
  // Serve on a socket that is already bound, so the port is known before the loop starts.
  let serving = @async.with_task_group(group => {
    group.spawn_bg(() => {
      @net.serve_thrift_on(listener, processor, () => @thrift.Binary::new(b""))
    })
    let proto : &@thrift.Protocol = @thrift.Compact::new(b"")
    let client = @net.ThriftClient::connect("127.0.0.1", port, proto)
    let pong = @thrift.call(client, proto, "ping", 1, proto => {
      proto.write_field_begin(@thrift.type_string, 1)
      proto.write_string("moonrpc")
    }, proto => proto.read_string())
    client.close()
    assert_eq(pong, "pong:moonrpc")
  })
  ignore(serving)
}

fn thrift_failures() -> @thrift.Processor {
  let processor = @thrift.Processor::new()
  processor.add("find", proto => {
    ignore(proto.read_field_begin())
    ignore(proto.read_string())
    raise @thrift.Application("no such key")   // the caller reads this as a Thrift error
  })
  // A method the IDL declares `oneway`: it runs and no REPLY goes back. The flag belongs to
  // the method — Apache's Go client sends a oneway call as a plain CALL and never reads an
  // answer, so a server that replies to one desynchronises the connection.
  processor.add_oneway("notify", proto => {
    ignore(proto.read_field_begin())
    ignore(proto.read_string())
  })
  processor
}

fn header_endpoint() -> @thrift.Header {
  let header = @thrift.Header::new()
  header.set_protocol(@thrift.header_protocol_compact)   // or header_protocol_binary
  header.add_header("tenant", "acme")                    // beside the call, not in it
  header.add_transform(@thrift.header_transform_zlib)    // the payload travels compressed
  header
}

async fn fire_and_forget(client : @net.ThriftClient) -> Unit raise {
  let proto : &@thrift.Protocol = @thrift.Binary::new(b"")
  @thrift.oneway_call(client, proto, "notify", 1, proto => {
    proto.write_field_begin(@thrift.type_string, 1)
    proto.write_string("done")
  })
}

moon run src/cmd/thriftc -- greet.thrift -o greet.g.mbt

union Contact {
  1: string email
  2: i64 phone
}

pub(all) enum Contact { Email(String); Phone(Int64) }

const i32 MAX_AGE = 130
const Person ADA = { "name": "Ada", "age": 37 }

// `const i32 MAX_AGE = 130` in the IDL, with the name MoonBit wants for a value.
pub let max_age : Int32 = 130

test {
  assert_eq(max_age, 130)
}

exception Missing {
  1: string message
  2: i32 code
}

service Greeter {
  string find(1: string key) throws (1: Missing missing)
}

// `find(1: string key) throws (1: Missing m)` becomes a `Result`, not an exception:
// the error enum names the declared exception, the handler returns
// `Result[String, GreeterFindError]`, and the client gets the same type back.
pub(all) enum GreeterFindError {
  Missing(Missing)
}

#derive(Debug)
pub struct Missing {
  message : String
}

fn found_or_missing(found : Result[String, GreeterFindError]) -> String {
  match found {
    Ok(value) => value
    Err(Missing(missing)) => missing.message
  }
}

test {
  assert_eq(found_or_missing(Ok("found ada")), "found ada")
  assert_eq(found_or_missing(Err(Missing({ message: "no ada" }))), "no ada")
}

#Connect

async fn serve_echo(server : @net.ConnectServer) -> Unit raise {
  server.add("/connect.Echo/Say", {
    input: ".connect.EchoRequest",   // the message a JSON body is read as
    output: ".connect.EchoReply",
    handler: @stream.Handler::Unary((_ctx, req) => req),
  })
  // One listener for both versions: a connection that opens with the HTTP/2 preface is
  // served over HTTP/2, any other over HTTP/1.1.
  server.serve(port=12000)
}

async fn call_echo(transport : @net.Http1Transport) -> Unit raise {
  let client : @connect.Client[@net.Http1Transport] = @connect.Client::new(transport)
  let reply = client.unary("/connect.Echo/Say", b"hi")
  assert_eq(reply.messages[0], b"hi")
}

#Concurrency on a connection

fn slow_method(engine : @moonrpc.H2Server) -> Unit {
  engine.register_handler("/slow.S/Wait", @stream.Handler::Pushed((_ctx, requests, out) => {
    @async.sleep(400)               // a slow handler, on its own task
    (out.send)(requests)
    (out.finish)()
  }))
}

#Protobuf and Server Reflection

pub fn message_as_json(schema : @moonrpc.Schema) -> Json raise {
  // `Profile` is a message whose descriptor the schema holds; `bytes` is its wire form.
  let bytes = b""
  let value = schema.to_json(".test.Profile", bytes)
  // The same value back on the wire, so the two directions are checked against each other.
  ignore(schema.to_bytes(".test.Profile", value))
  value
}

fn load_schema() -> @moonrpc.Schema raise {
  @moonrpc.Schema::compile([
    (
      "greet.proto",
      #|syntax = "proto3";
      #|package greet.v1;
      #|message Hello { string name = 1; }
      #|service Greeter { rpc Say(Hello) returns (Hello); }
    ),
  ])
}

fn protobuf_bytes() -> Unit raise {
  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)
  assert_eq(field, 1)
  assert_eq(r.read_string(), "testing")
}

fn describe_service() -> @moonrpc.FileDescriptor {
  let greeter : @moonrpc.ServiceDescriptor = {
    name: "Greeter",
    methods: [
      {
        name: "SayHello",
        input_type: ".greet.HelloRequest",
        output_type: ".greet.HelloReply",
        client_streaming: false,
        server_streaming: false,
      },
    ],
  }
  @moonrpc.FileDescriptor::new("greet.proto", "greet", services=[greeter])
}

#The engines

fn engine_reply(engine : @moonrpc.H2Server, frame : @http2.Frame) -> Unit raise {
  engine.register("/echo.Echo/Say", req => req)
  // [HEADERS(:status 200), DATA(reply), HEADERS(grpc-status:0)]
  let out = engine.feed(frame)
  assert_eq(out.length() > 0, true)
}

#Framing and status

test {
  let frame = @moonrpc.encode_message(b"hello grpc")
  // [0x00][0x00 0x00 0x00 0x0A]["hello grpc"]
  assert_eq(frame.length(), 5 + 10)

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

  assert_eq(@wire.Status::code(@wire.Status::NotFound), 5)
  assert_eq(@wire.Status::name(@wire.Status::Unauthenticated), "UNAUTHENTICATED")

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

#Retries and hedging

async fn throttled_call(host : String) -> Unit raise {
  let chan = @net.Channel::connect(
    host,
    12000,
    throttling=@moonrpc.Throttling::new(tokens=10, ratio=0.1),
  )
  let reply = chan.unary(
    "/greet.Greeter/SayHello",
    b"Ada",
    timeout_millis=2000,
    retry=@moonrpc.retry,
  )
  ignore(reply)
  chan.close()
}

#Not yet

• Descriptor options other than map_entry; oneof and imports from #rpc; extensions anywhere in the descriptor model — the reflection queries are answered, but there is nothing to report.
• HPACK never-indexed literals: the encoder indexes what it is given, and sends the dynamic table size update the peer asks for.
• -bin metadata goes out base64-padded. Receivers accept both forms (and metadata_value_from_wire reads unpadded), and the suite pins the padded form.
• Thrift: the JSON protocol, and an unframed (TBufferedTransport) peer — this reader takes a body that arrives whole, so an unframed message, which has no length in front of it, has nothing to read to. Everything else is served and driven by a real Apache peer: the framed and header transports, the binary and compact protocols inside either, the zlib transform, ONEWAY methods, and both directions of Apache's own cross test.
• Connect: not declared to the conformance suite, and therefore not tested by it — Connect over GET (an idempotent call with its message in query parameters), a message receive limit, TLS, and compressions other than gzip. scripts/conformance/connect.yaml is the statement of scope the runner reads.

#Licence