Guide

From a first method to two-way calls and a transport of your own. The code below runs as written; the output is what it prints.

Install

moon add moonbitstack/moonjrpc
moon add moonbitlang/async

Import the root package for messages and the method table. Import peer only if you call both ways.

import {
  "moonbitstack/moonjrpc",
  "moonbitstack/moonjrpc/peer",
  "moonbitlang/async",
}

A first method

typed decodes the parameters and encodes the result through FromJson and ToJson. body takes the text of a request and returns the text of the answer.

let methods = @moonjrpc.Methods::new()
methods.typed("sum", (numbers : Array[Int32]) => {
  numbers.fold(init=0, (a, b) => a + b)
})
let request =
  #|{"jsonrpc":"2.0","id":1,"method":"sum","params":[1,2,3]}
let answer = methods.body(request)

{"jsonrpc":"2.0","id":1,"result":6}

This is all an HTTP endpoint needs: read the POST body, call body, write what it returns. When it returns None there is nothing to send.

A handler that waits for something is written async fn(params) { ... }. One that does not is an arrow, as above.

Errors

fail answers with an error object of your choosing. The rest come from the library.

methods.call("no", _ => @moonjrpc.fail(7, "said no", data=Json::string("why")))
let refused =
  #|{"jsonrpc":"2.0","id":1,"method":"no"}
let unknown =
  #|{"jsonrpc":"2.0","id":2,"method":"nobody"}

{"jsonrpc":"2.0","id":1,"error":{"code":7,"message":"said no","data":"why"}}
{"jsonrpc":"2.0","id":2,"error":{"code":-32601,"message":"Method not found"}}
{"jsonrpc":"2.0","id":null,"error":{"code":-32700,"message":"Parse error"}}

What a handler trips over is not the caller's to read. Pass on_error to Methods::new to hear it yourself; it also hears failures in notices, which have nobody else to go to.

Notices and batches

A notice gets no answer, so body returns None. An array is a batch, answered in one array that leaves out the notices.

methods.notice("log", params => logged.push(params))
let notice =
  #|{"jsonrpc":"2.0","method":"log","params":["hello"]}
assert_eq(methods.body(notice), None)
methods.call("ping", _ => Json::string("pong"))
let batch =
  #|[{"jsonrpc":"2.0","id":1,"method":"ping"},{"jsonrpc":"2.0","method":"log"}]

[{"jsonrpc":"2.0","id":1,"result":"pong"}]

Some protocols forbid batches. methods.body(batch, batch=false) refuses an array with -32600.

Messages as values

A Message is a plain enum with ToJson and FromJson, for when you need to look at what is going across. The field the specification calls method is name here, because method is reserved in MoonBit.

let call : @moonjrpc.Message = Call(
  id=@moonjrpc.Id::int(1),
  name="ping",
  params=None,
)

{"jsonrpc":"2.0","id":1,"method":"ping"}

Calling both ways

@peer.Peer is for transports that stay open. Each side has a method table and can call the other, even in the middle of answering a call.

let (agent_reads, editor_writes) = @io.pipe()
let (editor_reads, agent_writes) = @io.pipe()

let editor_methods = @moonjrpc.Methods::new()
editor_methods.call("read_file", _ => Json::string("hello"))
let editor = @peer.Peer::new(
  @peer.lines(editor_reads, editor_writes),
  methods=editor_methods,
)

let agent_methods = @moonjrpc.Methods::new()
let agent : Ref[@peer.Peer?] = Ref(None)
agent_methods.call("prompt", async fn(_) {
  let text = agent.val.unwrap().call("read_file")
  Json::array([Json::string("the file says"), text])
})
agent.val = Some(
  @peer.Peer::new(
    @peer.lines(agent_reads, agent_writes),
    methods=agent_methods,
  ),
)

@async.with_task_group(async fn(group) {
  group.spawn_bg(no_wait=true, () => editor.run())
  group.spawn_bg(no_wait=true, () => agent.val.unwrap().run())
  let answer = editor.call("prompt")
})

["the file says","hello"]

Peer::run reads until the other side closes. It waits for calls still running, so a client that writes its requests and closes its end still gets its answers. Cancel the task to stop sooner. See this conversation drawn in the playground.

Order

Messages are taken in the order they arrive:

• A notice finishes before the next message is looked at.
• A call starts in its turn and runs on in a task of its own.
• A reply wakes its caller in its turn.

So the updates a side sends before answering are handled before its caller sees the answer, and a change sent before a call is in place when the call starts.

When a call fails

Peer::call raises one of three errors. The default limit is 60 seconds; timeout_ms=None waits as long as it takes.

let said = try editor.call("slow", timeout_ms=Some(10)) catch {
  @moonjrpc.Failed(failure) => "refused: \{failure.message}"
  @moonjrpc.TimedOut(name~, limit~) => "\{name} took over \{limit} ms"
  @moonjrpc.Closed(name~) => "\{name}: the other side went away"
  _ => "something else"
} noraise {
  _ => "answered"
}

slow took over 10 ms

Giving up does not tell the other side. Protocols have their own notice for that, so send it yourself. A late reply is dropped.

Framing

@peer.lines puts one message on a line. @peer.content_length writes a header with the length in bytes:

let frames = @peer.content_length(nothing, writes)

Content-Length: 9

{"猫":1}

Both raise TooLong past a limit, 64 MiB unless you pass limit=. Bytes that are not UTF-8 are answered like text that is not JSON.

Your own transport

A transport is a read and a write of one message's text. Here is one over two queues; a WebSocket looks the same.

struct Memory {
  incoming : @aqueue.Queue[String]
  outgoing : @aqueue.Queue[String]
}

impl @peer.Frames for Memory with fn read(self) {
  Some(self.incoming.get())
}

impl @peer.Frames for Memory with fn write(self, text) {
  self.outgoing.put(text)
}

Building a protocol

A protocol on top of this usually needs three things of its own:

• Its types, as records with FromJson and ToJson, registered through typed.
• Its HTTP shape, if it has one. Whether an answer goes back on the POST or on a separate stream differs from protocol to protocol, so it is not here.
• Its cancellation notice.

Pass batch=false to Peer::new if the protocol forbids batches, and timeout_ms=None if a call can legitimately run for a long time.