🥮 mooncakes.io

    moonamqp1

    AMQP 1.0 typed wire codec and bounded sans-I/O messaging state machines

    amqp1
    messaging
    sans-io
    Download zip
    Version
    0.1.0
    License
    Apache-2.0
    Repository
    Last updated
    4 days ago
    Downloads
    14

    #MoonAMQP1

    MoonBit 原生 AMQP 1.0 类型、消息、增量分帧与有界协议状态核心。同一实现用于 native、JavaScript、Wasm、Wasm-GC。核心不绑定 Rhea、C 消息库或网络运行时；原生示例宿主使用 async 处理文件和标准输入输出。

    模块：binhongcha/moonamqp1@0.1.0；GitHub · Mooncakes。

    #生态定位与使用场景

    AMQP 1.0 的类型、described value、会话窗口、单向 link、credit 与 settlement 构成跨系统消息交换能力。RabbitMQ 官方明确它与 AMQP 0-9-1 使用不同线协议和客户端。已有 MoonBit 0-9-1 库仍适用于其自身协议；本项目补充独立的 1.0 协议核心，不提供 0-9-1 exchange/queue 方法或透明转码。

    • 后台任务与服务作业：发送 5KB 作业，只有一个 credit，消息分为多个 transfer；工作端明确 accepted 或 released，上层自行决定是否重试。适用于文件处理、构建任务、通知作业、订单异步步骤。运行 examples/tasks。
    • 设备和应用遥测：慢速历史链路没有 credit 时，实时样本和告警链路仍能使用共享 session；精确传递 uint64 ID、timestamp 和单位。适用于采集网关、监控、应用事件和边缘上报。运行 examples/telemetry。
    • 跨系统消息桥与 Web/Wasm 宿主：两个独立连接保留带类型的属性、注解、多个 data section 与 footer，下游接受后再确认上游。适用于消息代理适配、迁移工具、离线诊断和可嵌入集成测试。运行 examples/bridge；该场景可直接编译为 JS，不附带浏览器 UI。

    本地已有项目分别处理文件格式、HTTP 缓存、更新信任、编码恢复、事件日志挖掘等，本项目处理的是 AMQP 1.0 线上消息与链路状态。检索属于有日期、有关键词范围的排重证据，不能证明整个生态绝无相近工作。

    #安装与运行

    安装官方 MoonBit 工具链。本机验证 moonc 0.10.16，符合赛事最低要求；CI 固定 compiler/core 0.10.16+b1a0e846e；依赖版本见 moon.mod。

    moon version --all
moon check --target all --deny-warn
moon build --target all --deny-warn
moon test --target all --deny-warn
moon run examples/tasks --target native --deny-warn
moon run examples/telemetry --target native --deny-warn
moon run examples/bridge --target js --deny-warn
moon run examples/scale --target native --deny-warn
moon fmt --check
moon info
git diff --exit-code

    从项目目录运行即可复现，示例已配置导入。Moon20261016 的 Windows OCaml 子工具可能拒绝 \\?\ 路径，可使用进程参数适配器；首次构建需要 MSVC Build Tools，不修改全局环境或编译器：

    .\tools\windows\moon.ps1 check --target all --deny-warn
.\tools\windows\moon.ps1 test --target all --deny-warn
.\tools\windows\moon.ps1 run examples/tasks --target native --deny-warn

    #最小使用样例

    在已有 MoonBit 工程中安装：

    moon add binhongcha/moonamqp1@0.1.0

    使用方 moon.pkg 配置导入：

    import { "binhongcha/moonamqp1" @amqp, }
    pkgtype(kind: "executable")

    fn main raise {
      let message = @amqp.Message::data(b"payload", properties=[
        ("sequence", ULong(18446744073709551615UL)),
        ("kind", Text("job")),
      ])
      let bytes = @amqp.encode_message(message)
      let restored = @amqp.decode_message(bytes)
      assert_eq(restored.data_bytes(), b"payload")
    }

    消息编解码可以独立使用。Message::new 接受已知 section 的有序列表；sections() 返回深层副本，调用方修改数组不会改变消息。无 body、amqp-value(null) 和空 data section 保持区别。data_bytes() 连接 data sections，遇到 sequence/value body 会报错。

    #兼容范围

    层次支持当前边界
    值全部原始域、十进制原始字节、精确整数/时间戳、list/map、typed array、数字或符号 descriptor十进制不提供算术；数组公共 constructor 最多一层 descriptor；编码采用全宽 canonical 形式
    文本严格 UTF-8、Unicode scalar、ASCII symbol，保留 BOM 和 ASCII NUL拒绝非法 UTF-8 与未配对 UTF-16 代理项，不静默替换
    帧AMQP1.0 header、9 类 transport performative、扩展头、空 heartbeat、任意 feed 分块SASL frame/type、TLS header、未知额外 performative 字段明确拒绝
    消息header、注解、properties、application-properties、data/sequence/value body、footer检查顺序、唯一性、互斥及字段域；未知 section 不静默跳过
    状态一个 connection、一个 session、多 link、本地/远端 channel 与 handle 独立映射不支持 crossed begin、多 session、session/link 恢复、动态/持久 terminus、过滤及能力协商
    收发link credit、session window、自动发送分片、接收组装/aborted、三种 sender settle mode、receiver first不支持 receiver second、drain、resume、事务、非零 message-format、分片中改变 settlement
    确认accepted、rejected（保留无 error 的拒绝）、released、modified 的两个布尔标志、范围 dispositionreceived/事务状态、modified 附加注解明确不支持；确认不等同业务 exactly-once
    宿主bytes/commands → bytes/events；原生 CLI 文件及 stdio 示例无内置 broker、TCP 客户端、SASL、TLS、持久 outbox、自动重连、定时器或自动重试

    首版终结点仅接受 idle-time-out 缺省或 0；定时器应由未来宿主集成实现。source/target 的非空高级字段会被明确拒绝，而不是假装完成过滤或 durability。typed value/message 层可独立处理状态核心尚不协商的内容。

    Value::ArrayValue(code, descriptor, bodies) 的 code 使用全宽基础类型，如 uint 0x70、symbol 0xb3、list 0xd0；bodies 不重复包含公共 descriptor。读取接受 array8/32 及 compact/zero-width constructor，再规范化。通用 map 保留顺序，键按类型与规范化值检查重复；注解、fields、application-properties 另有各自键域限制。

    #驱动连接与处理流控

    1. 创建 Endpoint；调用 start()，将 drain() 字节交给宿主的可靠有序传输。
    2. 每次输入调用 receive(bytes)，处理 events()，再发送 drain()。被动端收到协议 header 会自动响应 open。
    3. 主动一方在 RemoteOpen 后调用 begin()，另一方自动接受单 session；收到 SessionReady 后 attach sender/receiver。
    4. 接收 link ready 后调用 grant(handle, credit)。这设置以当前 delivery-count 为基准的新额度，不是累加旧额度。
    5. sender 使用本地 handle 调用 send(message, tag)；收到完整 Received 后按业务结果 acknowledge(id, outcome)。tag 最多 32 字节，同一 link 的在途未确认标签必须唯一；标签不提供跨重连幂等性。
    6. 显式 refresh_windows(incoming, outgoing) 重新通告宿主可承担的会话窗口；对端 flow 会更新远端容量。它与 grant link credit 是两个独立操作。
    7. 正常结束用 end_session() 或 close() 完成相应握手；传输 EOF 调用 finish_input()，无正常 close 会被报告。

    每个 transfer 帧消耗一次会话窗口，一条多帧 delivery 只消耗一次 link credit。send 在编码全部分片并通过队列/窗口预检后才更新状态；失败可以在补足窗口或清空队列后重试。首版不暂存“发送一半等待窗口”的命令，因此整条 delivery 的帧数必须能装入当前两个发送窗口。其 frame count 可从前后 session_positions() 差值观察。

    收到非法 wire 数据后，endpoint 永久失败，释放在途状态并清空待发送回复；不得拿同一实例继续解析。正常 detach 会产生 UnsettledLost，向宿主说明未确认结果不能再等待。连接/session 关闭使其所有未确认结果未知，宿主必须自行保留业务任务状态。核心对象不支持多个执行者并发修改，需串行驱动。

    #资源与复杂度

    Limits 默认限制：单值/输入 feed/逻辑聚合接收载荷/输出队列 16MiB、单帧 1MiB、单消息 16MiB、总值节点 65536、嵌套32层、link256、事件/各方向未确认256、每消息4096片。可调整，深度上限128；本地 session window 和 link credit 限于不产生 serial 歧义的半范围。接受合法 UINT_MAX 远端 session 容量；不会据其大小直接分配内存。

    增量分帧按字节追加，输入总长 N 时 O(N)，只保留一个未完成帧；消息分片按字节组装 O(M)。嵌套容器编码临时缓冲与 map key 规范化的成本至多 O(N·D)，D 为受限深度。handle 查找 O(L)，范围确认只扫描 O(P) 个已知 pending ID，排序最坏 O(P log P)，不会按对端声明的巨大 ID 区间逐个遍历。

    预算限制的是逻辑字节和节点数量，不是精确进程 RSS：编码/解码副本、JS 对象、Message 树与宿主已取走的事件仍有开销。核心保留输出字节；drain() 不代表网络已可靠发送。examples/scale 验证2048条、6144帧、64条有界批次的持续正确性，不是跨语言速度基准。

    #独立互操作与 CLI

    moon build --target native --deny-warn
npm ci --ignore-scripts --no-audit --no-fund
npm run interop
moon run cmd/main --target native -- value INPUT.bin NEW_OUTPUT.bin
moon run cmd/main --target native -- message INPUT.bin NEW_OUTPUT.bin
moon run cmd/main --target native -- frames INPUT.bin NEW_OUTPUT.bin

    CLI 拒绝覆盖已有输出。peer 使用二进制 stdin/stdout 承载 AMQP1.0：为接收 link 授予32个 credit 并 accepted；为获额度的发送 link 发送一次5KB原创测试消息。它是互操作示例宿主，不是生产服务。

    固定 Rhea3.0.5 为开发依赖：27组合法类型（含准确 UINT64、数组/描述类型）、消息 section 和帧流双向对照；真实 Rhea 客户端通过 stdio 与 MoonBit 双向收发5KB消息，协商512字节帧并验证分片、accepted、关闭。没有连接 RabbitMQ、Azure 或外部 broker，不能据此宣称这些服务已验证。Rhea 的宽松解码不作为恶意输入合法性的依据，截断/错类型/错状态/预算路径另行测试。

    Linux/Windows CI 覆盖检查、构建、四后端测试、格式、接口、五个场景、独立 Rhea 互操作和包内容；工程验证不等于正式赛事验收。申报书及审查/AI/汇总材料可在项目本地文件夹保存，由忽略及历史/包检查排除，不进入 Git 或发行包。许可证为 Apache-2.0，来源与 AI 辅助范围见 THIRD_PARTY.md。规范参考：Types、Transport、Messaging。

    #十一月新增功能

    完整交付发送预检（Endpoint::plan_send）；显式消息到期判断（Message::expiry_status）。详见功能、用途和边界，运行 moon run examples/november --target native --deny-warn。功能包含于0.1.0，安装方式见上文。

    AmqpError

    source
    pub suberror AmqpError {
      Invalid(String, Int)
      Unsupported(String)
      Limit(String)
      State(String)
    } derive(
    Debug    )

    Delivery

    source
    pub(all) struct Delivery {
      handle : UInt
      id : UInt
      tag : Bytes
      settled : Bool
      message : Message
    }

    Endpoint

    source
    pub struct Endpoint {
      // private fields
    }

    Endpoint::acknowledge

    source
    fn Endpoint::acknowledge(self : Endpoint, id : UInt, outcome : Outcome) -> Unit raise AmqpError

    Endpoint::attach

    source
    fn Endpoint::attach(self : Endpoint, name : String, sender : Bool, address : String, handle? : UInt) -> UInt raise AmqpError

    Endpoint::begin

    source
    fn Endpoint::begin(self : Endpoint) -> Unit raise AmqpError

    Endpoint::close

    source
    fn Endpoint::close(self : Endpoint) -> Unit raise AmqpError

    Endpoint::detach

    source
    fn Endpoint::detach(self : Endpoint, handle : UInt) -> Unit raise AmqpError

    Endpoint::drain

    source
    fn Endpoint::drain(self : Endpoint) -> Bytes

    Endpoint::end_session

    source
    fn Endpoint::end_session(self : Endpoint) -> Unit raise AmqpError

    Endpoint::events

    source
    fn Endpoint::events(self : Endpoint) -> Array[Event]

    Endpoint::finish_input

    source
    fn Endpoint::finish_input(self : Endpoint) -> Unit raise AmqpError

    Endpoint::grant

    source
    fn Endpoint::grant(self : Endpoint, handle : UInt, credit : UInt) -> Unit raise AmqpError

    Endpoint::heartbeat

    source
    fn Endpoint::heartbeat(self : Endpoint) -> Unit raise AmqpError

    Endpoint::is_closed

    source
    fn Endpoint::is_closed(self : Endpoint) -> Bool

    Endpoint::is_failed

    source
    fn Endpoint::is_failed(self : Endpoint) -> Bool

    fn Endpoint::link_credit(self : Endpoint, handle : UInt) -> UInt raise AmqpError

    Endpoint::new

    source
    fn Endpoint::new(container : String, channel? : Int, window? : UInt, limits? : Limits) -> Endpoint raise AmqpError

    Endpoint::pending

    source
    fn Endpoint::pending(self : Endpoint) -> (Int, Int)

    Endpoint::plan_send

    source
    fn Endpoint::plan_send(self : Endpoint, handle : UInt, message : Message, tag : Bytes, settled? : Bool) -> SendPlan raise AmqpError

    Encode and validate a delivery without consuming credit or modifying queues.

    Endpoint::receive

    source
    fn Endpoint::receive(self : Endpoint, bytes : Bytes) -> Unit raise AmqpError

    Endpoint::refresh_windows

    source
    fn Endpoint::refresh_windows(self : Endpoint, incoming : UInt, outgoing : UInt) -> Unit raise AmqpError

    Endpoint::send

    source
    fn Endpoint::send(self : Endpoint, handle : UInt, message : Message, tag : Bytes, settled? : Bool) -> UInt raise AmqpError

    Endpoint::session_positions

    source
    fn Endpoint::session_positions(self : Endpoint) -> (UInt, UInt)

    Endpoint::start

    source
    fn Endpoint::start(self : Endpoint) -> Unit raise AmqpError

    Start both the protocol header and local open. No network I/O occurs here.

    Event

    source
    pub enum Event {
      RemoteOpen(String)
      SessionReady
      LinkReady(UInt, String, Bool)
      Credit(UInt, UInt)
      Received(Delivery)
      Settled(UInt, Outcome)
      PeerSettled(UInt)
      UnsettledLost(UInt, Bool)
      Aborted(UInt, UInt)
      Detached(UInt)
      SessionEnded(Value)
      RemoteClosed(Value)
    }

    ExpiryStatus

    source
    pub(all) struct ExpiryStatus {
      expired : Bool
      remaining_milliseconds : UInt64?
    } derive(
    Debug    )

    Frame

    source
    pub(all) struct Frame {
      channel : Int
      performative : Performative?
      payload : Bytes
      extended_header : Bytes
    }

    FrameStream

    source
    pub struct FrameStream {
      // private fields
    }

    FrameStream::buffered_bytes

    source
    fn FrameStream::buffered_bytes(self : FrameStream) -> Int

    FrameStream::feed

    source
    fn FrameStream::feed(self : FrameStream, bytes : Bytes) -> Array[StreamItem] raise AmqpError

    Each byte is appended once; no prefix rescanning on one-byte input feeds.

    FrameStream::finish

    source
    fn FrameStream::finish(self : FrameStream) -> Unit raise AmqpError

    FrameStream::new

    source
    fn FrameStream::new(expect_header? : Bool, limits? : Limits) -> FrameStream raise AmqpError

    Limits

    source
    pub(all) struct Limits {
      max_value_bytes : Int
      max_frame_bytes : Int
      max_message_bytes : Int
      max_items : Int
      max_depth : Int
      max_links : Int
      max_pending : Int
      max_fragments : Int
    } derive(Eq, 
    Debug    )

    Limits::default

    source
    fn Limits::default() -> Limits

    Limits::equal

    source
    fn Limits::equal(Limits, Limits) -> Bool

    Limits::not_equal

    source
    fn Limits::not_equal(x : Limits, y : Limits) -> Bool

    Limits::to_repr

    source

    Message

    source
    pub struct Message {
      // private fields
    }

    Message::data

    source
    fn Message::data(bytes : Bytes, properties? : Array[(String, Value)], limits? : Limits) -> Message raise AmqpError

    Message::data_bytes

    source
    fn Message::data_bytes(self : Message) -> Bytes raise AmqpError

    Message::expiry_status

    source
    fn Message::expiry_status(self : Message, acquired_at : Int64, now : Int64) -> ExpiryStatus raise AmqpError

    Caller-supplied UTC milliseconds. TTL is measured since local acquisition; absolute-expiry-time and TTL are both enforced, without adding timestamps.

    Message::new

    source
    fn Message::new(sections : Array[Section], limits? : Limits) -> Message raise AmqpError

    Message::sections

    source
    fn Message::sections(self : Message) -> Array[Section]

    Outcome

    source
    pub(all) enum Outcome {
      Accepted
      Rejected(String, String?)
      RejectedWithoutError
      Released
      Modified(Bool, Bool)
    } derive(Eq, 
    Debug    )

    Outcome::equal

    source
    fn Outcome::equal(Outcome, Outcome) -> Bool

    Outcome::from_value

    source
    fn Outcome::from_value(value : Value) -> Outcome raise AmqpError

    Outcome::not_equal

    source
    fn Outcome::not_equal(x : Outcome, y : Outcome) -> Bool

    Outcome::to_repr

    source

    Outcome::value

    source
    fn Outcome::value(self : Outcome) -> Value

    Performative

    source
    pub struct Performative {
      // private fields
    }

    Validated transport list. Accessors return detached snapshots.

    Performative::descriptor

    source
    fn Performative::descriptor(self : Performative) -> Int

    Performative::field

    source
    fn Performative::field(self : Performative, index : Int) -> Value

    Performative::from_value

    source
    fn Performative::from_value(value : Value, limits? : Limits) -> Performative raise AmqpError

    Performative::new

    source
    fn Performative::new(code : Int, fields : Array[Value], limits? : Limits) -> Performative raise AmqpError

    Performative::value

    source
    fn Performative::value(self : Performative) -> Value

    Section

    source
    pub(all) struct Section {
      descriptor : Int
      body : Value
    }

    SendPlan

    source
    pub(all) struct SendPlan {
      frames : Int
      wire_bytes : Int
      link_credit : UInt
    } derive(
    Debug    )

    SendPlan::to_repr

    source

    SessionWindow

    source
    pub struct SessionWindow {
      // private fields
    }

    SessionWindow::can_send

    source
    fn SessionWindow::can_send(self : SessionWindow, frames? : Int) -> Bool

    SessionWindow::establish

    source
    fn SessionWindow::establish(self : SessionWindow, next_incoming : UInt, peer_incoming : UInt, peer_outgoing : UInt) -> Unit raise AmqpError

    SessionWindow::new

    source
    fn SessionWindow::new(incoming : UInt, outgoing : UInt, next_outgoing? : UInt) -> SessionWindow raise AmqpError

    SessionWindow::positions

    source
    fn SessionWindow::positions(self : SessionWindow) -> (UInt, UInt)

    SessionWindow::received

    source
    fn SessionWindow::received(self : SessionWindow) -> Unit raise AmqpError

    SessionWindow::refresh

    source
    fn SessionWindow::refresh(self : SessionWindow, incoming : UInt, outgoing : UInt) -> Unit raise AmqpError

    SessionWindow::remaining

    source
    fn SessionWindow::remaining(self : SessionWindow) -> (UInt, UInt, UInt)

    SessionWindow::sent

    source
    fn SessionWindow::sent(self : SessionWindow, frames? : Int) -> Unit raise AmqpError

    SessionWindow::update_peer

    source
    fn SessionWindow::update_peer(self : SessionWindow, next_incoming : UInt?, incoming : UInt, next_outgoing : UInt, outgoing : UInt) -> Unit raise AmqpError

    StreamItem

    source
    pub enum StreamItem {
      Header
      AmqpFrame(Frame)
    }

    Value

    source
    pub(all) enum Value {
      Null
      Boolean(Bool)
      UByte(Int)
      UShort(Int)
      UInt(UInt)
      ULong(UInt64)
      ByteValue(Int)
      Short(Int)
      IntValue(Int)
      Long(Int64)
      FloatValue(Float)
      DoubleValue(Double)
      Decimal32(Bytes)
      Decimal64(Bytes)
      Decimal128(Bytes)
      Character(UInt)
      Timestamp(Int64)
      Uuid(Bytes)
      Binary(Bytes)
      Text(String)
      Symbol(String)
      List(Array[Value])
      MapValue(Array[(Value, Value)])
      ArrayValue(Int, Value?, Array[Value])
      Described(Value, Value)
    } derive(
    Debug    )

    Exact wire domains; ULong/Long never travel through Double.

    Value::as_text

    source
    fn Value::as_text(self : Value) -> String raise AmqpError

    Value::as_uint

    source
    fn Value::as_uint(self : Value) -> UInt raise AmqpError

    Value::to_repr

    source

    decode_frame

    source
    fn decode_frame(bytes : Bytes, limits? : Limits) -> Frame raise AmqpError

    decode_message

    source
    fn decode_message(bytes : Bytes, limits? : Limits) -> Message raise AmqpError

    decode_value

    source
    fn decode_value(bytes : Bytes, limits? : Limits) -> Value raise AmqpError

    encode_frame

    source
    fn encode_frame(frame : Frame, limits? : Limits) -> Bytes raise AmqpError

    encode_message

    source
    fn encode_message(message : Message, limits? : Limits) -> Bytes raise AmqpError

    encode_value

    source
    fn encode_value(value : Value, limits? : Limits) -> Bytes raise AmqpError

    Canonical full-width forms simplify deterministic output and homogeneous arrays.

    protocol_header

    source
    fn protocol_header() -> Bytes

    serial_distance

    source
    fn serial_distance(first : UInt, last : UInt) -> UInt raise AmqpError

    Serial distance in the unambiguous half of the 32-bit sequence space.

    source

    source
    fn source(address : String) -> Value

    Non-durable, non-dynamic termini. Applications own node creation and lifetime.

    target

    source
    fn target(address : String) -> Value

    validate_protocol_header

    source
    fn validate_protocol_header(bytes : Bytes) -> Unit raise AmqpError

    Source Files

    Powered by MoonBit

    Site sourceReport issueModulesSkillsCLIBuild queueStatistics

    © 2026 mooncakes.io