moonzero — a service framework for MoonBit (← go-zero): config-driven assembly of a moonapi application with middleware, producing a runnable moonasgi AsgiApp. v0.6.2 runs real unary, server/client-streaming, and bidirectional zRPC calls over moonrpc's h2c transport (an in-process RpcChannel drives HPACK HEADERS + length-prefixed DATA + grpc-status trailers through the H2Server engine), graceful shutdown draining in-flight calls, a persisted, watchable etcd v3-shaped registry (watch, events_since catch-up, snapshot/restore) with round-robin/pick-first balancers and a load-balanced client, real file-backed registry I/O over moonbitlang/async's fs with a live filesystem watcher (native discov driver), request metrics (counter + latency histogram), and W3C traceparent propagation.
Dependencies
flowchart LR
conf["ServiceConf"] --> srv["**moonzero** Server"]
api["moonapi App"] --> srv
mw["middleware<br/>(logging, ...)"] --> srv
srv -->|"to_asgi()"| asgi(["moonasgi AsgiApp"])
asgi --> cat["mooncat serves it"]let app = @moonapi.App::new()
let api = @moonzero.Group::new(app, "/api/v1") // prefix a set of routes
api.get("/ping", _ctx => @moonapi.text(200, "pong"))
let conf = @moonzero.ServiceConf::new(
name="greet", host="127.0.0.1", port=8888, timeout_ms=3000, log_level=Info,
)
let server = @moonzero.Server::new(conf, app)
.use_(@moonzero.cors(@moonzero.CorsConf::new())) // Access-Control-* headers
.use_(@moonzero.request_id()) // x-request-id per request
.use_(@moonzero.recovery) // 500 instead of a panic
.use_(@moonzero.logging)
server.describe() // "greet listening on 127.0.0.1:8888"
@mooncat.serve(server.to_asgi(), host=conf.host, port=conf.port) // run it (native)let clock = @moonzero.Clock::new(() => now_ms()) // real time at the async edge
let server = @moonzero.Server::new(conf, app)
.use_(@moonzero.maxbytes(1 << 20)) // 413 over 1 MiB
.use_(@moonzero.rate_limit(@moonzero.TokenBucket::new(100.0), clock))// 429 when empty
.use_(@moonzero.breaker(@moonzero.Breaker::new(), clock)) // 503 while open
.use_(@moonzero.timeout(3000L, clock)) // deadline
.use_(@moonzero.structured_logging(clock)) // one JSON line/req// JWT HS256 — self-built SHA-256/HMAC (verified against NIST/RFC vectors)
let token = @moonzero.jwt_sign(
Map([("sub", Json::string("alice")), ("exp", Json::number(1893456000.0))]),
"topsecret",
)
let server = @moonzero.Server::new(conf, app)
.use_(@moonzero.auth("topsecret", clock)) // 401 unless a valid Bearer JWT
.use_(@moonzero.tracing()) // W3C traceparent in/out
.use_(@moonzero.metrics(m, clock)) // request counter + latency histogram
// YAML config — the etc/*.yaml format go-zero ships, same lenient defaults as JSON
let conf = @moonzero.ServiceConf::from_yaml("name: greet\nport: 9000\nlog_level: error\n")
// A real zRPC call over moonrpc's h2c transport — unary and streaming
let rpc = @moonzero.RpcServer::new(@moonzero.RpcServerConf::new(name="greeter", port=9090))
let g = rpc.group("hello.Greeter")
g.register("SayHello", req => handle(req)) // unary
g.register_server_streaming("Tail", req => chunks(req)) // one in, many out
g.register_client_streaming("Upload", msgs => summarize(msgs)) // many in, one out
g.register_bidi_streaming("Chat", () => @moonzero.BidiStreamHandler::{
on_message: m => [echo(m)], // a reply the instant each message arrives
on_end: () => [b"bye"], // a farewell after the client half-closes
})
let ch = @moonzero.RpcChannel::connect(rpc)
ch.call("/hello.Greeter/SayHello", request) // Ok(reply) | Err(status)
ch.call_server_streaming("/hello.Greeter/Tail", request) // Ok([msg, ...]) | Err(status)
ch.call_client_streaming("/hello.Greeter/Upload", [a, b]) // Ok(reply) | Err(status)
let call = ch.open_bidi("/hello.Greeter/Chat") // stream both ways
call.send(a) // -> the replies produced right then (interleaved)
call.close_send() // -> Ok([on_end replies...]) | Err(status)// Persisted, watchable registry (etcd v3-shaped): register, watch, snapshot
let reg = @moonzero.PersistentRegistry::new()
reg.watch(e => log(e)) // Put/Delete events in revision order
reg.register("greeter", @moonzero.Endpoint::new("10.0.0.1", 9090))
reg.register("greeter", @moonzero.Endpoint::new("10.0.0.2", 9090))
let saved = reg.snapshot() // persist to a file/etcd; restore reloads it
// Real registry I/O over a file (native): a publisher persists, a reader watches
@discov.persist_registry(path, reg) // write the snapshot to a real file
let reader = @discov.FileRegistry::load(path) // load it back
reader.reload() // re-read -> the Put/Delete diff since last load
reader.watch(dir, e => log(e)) // reload on every filesystem change
// Load-balanced client: resolve an instance, then call it over h2c
let ch = @moonzero.LoadBalancedChannel::new(reg.resolver(), cluster, "greeter")
ch.call("/hello.Greeter/SayHello", request) // dials 10.0.0.1, then .2, cycling
// Metrics read out for a /metrics scrape after serving
let m = @moonzero.ServerMetrics::new()
m.requests().value("GET /ping 200") // request count for that label
m.latency().mean() // mean request latency, mstype RpcHandler = (Bytes) -> Bytespub suberror ConfigError {
ConfigError(String)
}pub suberror JwtError {
MalformedToken(String)
UnsupportedAlg(String)
BadSignature
Expired
NotYetValid
}pub struct BidiCall {
channel : RpcChannel
sid : Int
pending : Bytes
status_code : Int
ended : Bool
}pub struct Breaker {
max_failures : Int
open_ms : Int64
half_open_max : Int
state : BreakerState
failures : Int
opened_at : Int64
probes : Int
}pub struct Clock {
now_ms : () -> Int64
}pub(all) struct CorsConf {
allow_origin : String
allow_methods : String
allow_headers : String
allow_credentials : Bool
max_age : Int
}pub struct Deadline {
budget_ms : Int64
started_ms : Int64
}fn InMemoryRegistry::register(self : InMemoryRegistry, service : String, endpoint : Endpoint) -> Stringpub struct LoadBalancedChannel {
resolve : (String) -> Array[Endpoint]
cluster : RpcCluster
balancer : Balancer
service : String
}fn LoadBalancedChannel::call(self : LoadBalancedChannel, path : String, request : Bytes) -> Result[Bytes, Status] raisefn LoadBalancedChannel::call_bidi_streaming(self : LoadBalancedChannel, path : String, requests : Array[Bytes]) -> Result[Array[Bytes], Status] raisefn LoadBalancedChannel::call_server_streaming(self : LoadBalancedChannel, path : String, request : Bytes) -> Result[Array[Bytes], Status] raisefn LoadBalancedChannel::new(resolve : (String) -> Array[Endpoint], cluster : RpcCluster, service : String, balancer? : Balancer) -> LoadBalancedChannelpub struct ManualClock {
ms : Int64
}pub struct PersistentRegistry {
instances : Map[String, Map[String, Endpoint]]
key_rev : Map[String, Int64]
seq : Int
revision : Int64
events : Array[RegistryEvent]
watchers : Array[(RegistryEvent) -> Unit]
}fn PersistentRegistry::deregister(self : PersistentRegistry, service : String, key : String) -> Boolfn PersistentRegistry::events_since(self : PersistentRegistry, revision : Int64) -> Array[RegistryEvent]fn PersistentRegistry::register(self : PersistentRegistry, service : String, endpoint : Endpoint) -> Stringpub(all) struct RequestLog {
http_method : String
path : String
status : Int
duration_ms : Int64
request_id : String
client_ip : String
user_agent : String
}pub struct RoundRobin {
cursor : Int
}pub struct RpcChannel {
engine : H2Server
encoder : HpackEncoder
decoder : HpackDecoder
authority : String
next_stream_id : Int
}fn RpcChannel::call(self : RpcChannel, path : String, request : Bytes) -> Result[Bytes, Status] raisefn RpcChannel::call_bidi_streaming(self : RpcChannel, path : String, requests : Array[Bytes]) -> Result[Array[Bytes], Status] raisefn RpcChannel::call_client_streaming(self : RpcChannel, path : String, requests : Array[Bytes]) -> Result[Bytes, Status] raisefn RpcChannel::call_server_streaming(self : RpcChannel, path : String, request : Bytes) -> Result[Array[Bytes], Status] raisefn RpcGroup::register_bidi_streaming(self : RpcGroup, name : String, factory : () -> BidiStreamHandler) -> Unitpub struct RpcServer {
conf : RpcServerConf
handlers : Map[String, (Bytes) -> Bytes]
server_streaming : Map[String, (Bytes) -> Array[Bytes]]
client_streaming : Map[String, (Array[Bytes]) -> Bytes]
bidi_streaming : Map[String, () -> BidiStreamHandler]
}fn RpcServer::dispatch_graceful(self : RpcServer, coord : ShutdownCoordinator, path : String, request : Bytes) -> Result[Bytes, Status]fn RpcServer::register_bidi_streaming(self : RpcServer, desc : Method, factory : () -> BidiStreamHandler) -> Unitfn RpcServerConf::new(name? : String, host? : String, port? : Int, timeout_ms? : Int) -> RpcServerConfpub struct Server {
conf : ServiceConf
handler : async (Scope, async () -> Event, async (Event) -> Unit) -> Unit
}fn ServiceConf::new(name? : String, host? : String, port? : Int, timeout_ms? : Int, log_level? : LogLevel) -> ServiceConfpub struct ShutdownCoordinator {
shutting_down : Bool
in_flight : Int
}pub struct TokenBucket {
capacity : Double
refill_per_ms : Double
tokens : Double
last_ms : Int64
}pub(all) struct TraceContext {
trace_id : String
span_id : String
flags : Int
}fn base64url_decode(s : String) -> Bytesfn base64url_encode(data : Bytes) -> Stringfn constant_time_eq(a : Bytes, b : Bytes) -> Boolfn generate_span_id(seed : Int64) -> Stringfn generate_trace_id(seed : Int64) -> Stringfn hmac_sha256(key : Bytes, msg : Bytes) -> Bytesfn jwt_authorized(token : String?, secret : String, now_secs : Int64) -> Boolfn sha256(msg : Bytes) -> Bytesmoonzero — a service framework for MoonBit (← go-zero): config-driven assembly of a moonapi application with middleware, producing a runnable moonasgi AsgiApp. v0.6.2 runs real unary, server/client-streaming, and bidirectional zRPC calls over moonrpc's h2c transport (an in-process RpcChannel drives HPACK HEADERS + length-prefixed DATA + grpc-status trailers through the H2Server engine), graceful shutdown draining in-flight calls, a persisted, watchable etcd v3-shaped registry (watch, events_since catch-up, snapshot/restore) with round-robin/pick-first balancers and a load-balanced client, real file-backed registry I/O over moonbitlang/async's fs with a live filesystem watcher (native discov driver), request metrics (counter + latency histogram), and W3C traceparent propagation.
Dependencies