🥮 mooncakes.io

    moon-nsq

    A native MoonBit client library for the NSQ messaging platform.

    nsq
    message-queue
    client
    async
    Download zip
    Author
    Version
    0.1.0
    License
    Apache-2.0
    Repository
    Last updated
    20 days ago
    Downloads
    28

    #moon-nsq

    使用 MoonBit 编写的 NSQ TCP V2 客户端实验性实现，作为 MoonBit 开源生态比赛项目持续开发。

    当前状态：开发中。 已实现基础协议、异步 TCP API 和部分辅助组件，并有单元测试及模拟 TCP 服务测试。尚未完成真实 nsqd / nsqlookupd 联调，不能承诺生产环境可用性。截至 2026-10-16，尚未确认发布到 Mooncakes，请从源码使用。

    #已实现的功能

    范围当前实现
    协议NSQ V2 握手、帧编解码、消息解析、增量帧解析
    发布PUB、延迟发布 DPUB、批量发布 MPUB、绑定 Topic 的 Producer
    消费SUB、RDY、FIN、REQ、TOUCH，以及 Consumer 启动、暂停、恢复和单条处理
    连接明文 TCP、基础 IDENTIFY、读取帧时回复心跳、CLS 关闭请求
    发现nsqlookupd 查询、端点轮询、健康状态和冷却路由
    辅助组件批量缓冲、重试决策、计数器、延迟直方图、投递窗口、拓扑配置数据模型
    管理操作操作描述、参数检查及 URL 构造；没有管理 HTTP 请求执行 API

    ConnectionPool 是有容量限制的端点租约管理器，不建立、持有或复用 TCP 连接。重试策略、健康度和指标是独立组件，尚未组成自动重连或自动故障转移流程。死信组件仅提供决策和统计，不会自动发布到死信 Topic。

    #从源码开始

    需要安装 MoonBit 工具链和 Git。本地验证使用 moonc v0.10.14；GitHub Actions 在 Ubuntu 上验证 wasm 目标。

    git clone https://github.com/clhhhhhh/moon-nsq.git
cd moon-nsq
moon update
moon check --target wasm --deny-warn
moon build --target wasm --deny-warn
moon test --target wasm --deny-warn

    默认目标为 wasm，直接依赖 moonbitlang/async@0.22.1。网络功能使用 MoonBit 运行器提供的能力，不是可直接用于浏览器的客户端。Native 目标曾通过本地检查，但本机缺少 C 编译器，未完成 Native 构建和测试；其他目标未验证。

    包名为 clhhhhhh/moon-nsq。由于尚未确认发布，不应把 moon add clhhhhhh/moon-nsq 当作已经验证可用的安装方式。

    #命令行发布示例

    仓库中的 cmd/main 使用低层 Connection 发布一条 UTF-8 消息，固定连接目标主机的 4150 TCP 端口。需要自行安装 NSQ 并启动服务：

    nsqd --tcp-address=127.0.0.1:4150 --http-address=127.0.0.1:4151

    在项目目录的另一终端运行：

    moon run cmd/main --target wasm -- 127.0.0.1 orders.created "hello"

    收到 broker 的 OK 后输出 published to orders.created，表示发布请求得到确认，不表示消费者已经处理消息。以上是真实服务的预期运行步骤，本项目尚未完成该端到端验证。

    不提供参数可查看用法，不需要 NSQ 服务：

    moon run cmd/main --target wasm

    #API 使用顺序

    包导入配置可参考 cmd/main/moon.pkg，公开接口见 pkg.generated.mbti。

    • 发布：Producer::connect(host, port, topic) 完成连接和 IDENTIFY，随后调用 publish 或 publish_deferred，最后 close。
    • 批量发布：创建 new_batch，逐条 push 并处理 Added、Full、TooLarge 结果，再调用 publish_batch。成功确认后清空批次，失败保留数据；网络断开时 broker 是否已接收可能无法判断，重试可能产生重复消息。
    • 消费：Consumer::connect 后调用 start，再 receive，并以 finish、requeue 或 discard 结算消息；长任务可手动 touch。discard 实际发送 FIN，不会转发到死信队列。
    • 回调消费：process_one 每次只处理一条消息，由调用者循环并处理异常。达到 max_attempts 后的重投递决策会转为丢弃确认。
    • 服务发现：创建 Lookupd，调用 lookup(topic) 获取端点，再由调用者建立连接。发现刷新和重新连接需要自行调度。

    目前仓库仅提供发布 CLI，没有独立的消费者命令行示例。

    #测试和 CI

    截至 2026-10-16，最近一次验证为 54 个测试通过。覆盖协议编解码、命令、参数校验、批次、路由、租约等组件；部分连接流程使用本地模拟 TCP 服务验证，包括心跳应答、NOP 无响应和发布失败时保留批次。

    这些测试不依赖真实 NSQ 服务，不能替代真实 broker 兼容性和可靠性验证。目前没有真实 NSQ 端到端测试、压力测试或性能基准结果。

    GitHub Actions 在 Ubuntu 上执行依赖更新及 WASM 目标的检查、构建、测试。最近已确认成功的代码提交为 48b20db。

    维护接口与格式：

    moon info --target wasm
moon fmt
git diff --check

    #来源和许可证

    项目采用根目录 LICENSE 中的 Apache-2.0 许可证。协议实现参考 NSQ TCP 协议规范，管理操作定义参考 NSQ HTTP API 文档。本仓库为 MoonBit 实现，包含 AI 辅助开发与审查工作，不宣称移植了完整的官方客户端。

    NSQ 项目采用 MIT 许可证，是协议和服务端参考项目，不随本库源码分发。使用本库及其依赖时还应遵守依赖自身的许可证。

    ProtocolError

    source
    pub suberror ProtocolError {
      InvalidFrameLength(Int)
      UnknownFrameType(Int)
      TruncatedFrame(Int)
      InvalidCommand(String)
      TruncatedMessage(Int)
      ServerError(String)
      UnexpectedFrame(String)
      InvalidOption(String)
      InvalidState(String)
    } derive(
    Debug    )

    AdminClient

    source
    pub struct AdminClient {
      host : String
      port : Int
      base_path : String
    } derive(
    Debug    )

    Simple URL builder for an nsqd administration endpoint.

    AdminClient::health_url

    source
    fn AdminClient::health_url(self : AdminClient) -> String

    Builds a health endpoint URL.

    AdminClient::host

    source
    fn AdminClient::host(self : AdminClient) -> String

    AdminClient::new

    source
    fn AdminClient::new(host : String, port? : Int, base_path? : String) -> AdminClient raise ProtocolError

    AdminClient::port

    source
    fn AdminClient::port(self : AdminClient) -> Int

    AdminClient::stats_url

    source
    fn AdminClient::stats_url(self : AdminClient, format? : String) -> String raise ProtocolError

    Builds a stats endpoint URL with the requested format.

    AdminClient::to_repr

    source

    AdminClient::url

    source
    fn AdminClient::url(self : AdminClient, operation : AdminOperation) -> String raise ProtocolError

    Builds an absolute HTTP URL without performing network I/O.

    AdminOperation

    source
    pub enum AdminOperation {
      CreateTopic(String)
      DeleteTopic(String)
      CreateChannel(String, String)
      DeleteChannel(String, String)
      Tombstone(String)
      Pause(String)
      Unpause(String)
    } derive(Eq, 
    Debug    )

    HTTP operation used by an nsqd/nsqlookupd administration client.

    AdminOperation::equal

    source

    AdminOperation::http_method

    source
    fn AdminOperation::http_method(self : AdminOperation) -> String

    Returns the HTTP method used for an administrative operation. NSQ v1 administration endpoints require POST.

    AdminOperation::is_mutating

    source
    fn AdminOperation::is_mutating(self : AdminOperation) -> Bool

    Returns whether the operation changes broker state.

    AdminOperation::label

    source
    fn AdminOperation::label(self : AdminOperation) -> String

    Returns a concise operation label for logs and metrics.

    AdminOperation::not_equal

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

    AdminOperation::path

    source
    fn AdminOperation::path(self : AdminOperation) -> String raise ProtocolError

    Produces the relative endpoint path used by the NSQ HTTP API.

    AdminOperation::to_repr

    source

    AdminOperation::validate

    source
    fn AdminOperation::validate(self : AdminOperation) -> Unit raise ProtocolError

    Validates an admin operation before it becomes an HTTP path.

    BackoffOptions

    source
    pub struct BackoffOptions {
      initial_delay : Int
      max_delay : Int
      multiplier : Int
    } derive(
    Debug    )

    A bounded exponential delay policy used after disconnected sessions.

    BackoffOptions::delay_for

    source
    fn BackoffOptions::delay_for(self : BackoffOptions, attempt : Int) -> Int

    Returns the capped delay for a zero-based reconnect attempt.

    BackoffOptions::new

    source
    fn BackoffOptions::new(initial_delay? : Int, max_delay? : Int, multiplier? : Int) -> BackoffOptions raise ProtocolError

    BackoffOptions::to_repr

    source

    BatchPush

    source
    pub enum BatchPush {
      Added
      Full
      TooLarge
    } derive(Eq, 
    Debug    )

    Result of attempting to add a message to a bounded publish batch.

    BatchPush::equal

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

    BatchPush::not_equal

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

    BatchPush::to_repr

    source

    BrokerSet

    source
    pub struct BrokerSet {
      producers : Array[ProducerEndpoint]
      next_index : Int
    }

    A nonempty round-robin set of broker TCP endpoints.

    BrokerSet::length

    source
    fn BrokerSet::length(self : BrokerSet) -> Int

    BrokerSet::new

    source
    fn BrokerSet::new(producers : Array[ProducerEndpoint]) -> BrokerSet raise ProtocolError

    BrokerSet::next

    source

    Selects a broker and advances the round-robin cursor.

    BrokerSet::refresh

    source
    async fn BrokerSet::refresh(self : BrokerSet, lookupd : Lookupd, topic : String) -> Unit

    Refreshes this broker set from the provided nsqlookupd endpoint.

    BrokerSet::replace

    source
    fn BrokerSet::replace(self : BrokerSet, producers : Array[ProducerEndpoint]) -> Unit raise ProtocolError

    Replaces all endpoints atomically after a discovery refresh.

    ChannelSpec

    source
    pub struct ChannelSpec {
      topic : String
      name : String
      max_in_flight : Int
      max_attempts : Int
    } derive(
    Debug    )

    Channel configuration paired with a topic.

    ChannelSpec::max_attempts

    source
    fn ChannelSpec::max_attempts(self : ChannelSpec) -> Int

    ChannelSpec::max_in_flight

    source
    fn ChannelSpec::max_in_flight(self : ChannelSpec) -> Int

    ChannelSpec::name

    source
    fn ChannelSpec::name(self : ChannelSpec) -> String

    ChannelSpec::new

    source
    fn ChannelSpec::new(topic : String, name : String, max_in_flight? : Int, max_attempts? : Int) -> ChannelSpec raise ProtocolError

    ChannelSpec::to_repr

    source

    ChannelSpec::topic

    source
    fn ChannelSpec::topic(self : ChannelSpec) -> String

    Connection

    source
    pub struct Connection {
      // private fields
    }

    A single NSQ TCP connection. Concurrent reads on one connection are unsupported.

    Connection::close

    source
    fn Connection::close(self : Connection) -> Unit

    Connection::close_gracefully

    source
    async fn Connection::close_gracefully(self : Connection) -> Unit

    Stops new delivery and waits for NSQ's CLOSE_WAIT acknowledgement.

    Connection::finish

    source
    async fn Connection::finish(self : Connection, id : Bytes) -> Unit

    Marks a message as processed.

    Connection::identify

    source
    async fn Connection::identify(self : Connection, heartbeat_interval? : Int) -> Unit

    Sends IDENTIFY with feature negotiation disabled.

    Connection::identify_with_options

    source
    async fn Connection::identify_with_options(self : Connection, options : ConnectionOptions) -> Unit

    Sends all supported IDENTIFY settings and waits for the broker response.

    Connection::ping

    source
    async fn Connection::ping(self : Connection) -> Unit

    Writes NOP. NSQ sends no response; this does not confirm broker liveness.

    Connection::publish

    source
    async fn Connection::publish(self : Connection, topic : String, body : Bytes) -> Unit

    Publishes one binary message and waits for the broker acknowledgement.

    Connection::publish_batch

    source
    async fn Connection::publish_batch(self : Connection, batch : PublishBatch) -> Unit

    Sends the current contents of a batch as one MPUB request and clears it on success.

    Connection::publish_deferred

    source
    async fn Connection::publish_deferred(self : Connection, topic : String, delay_ms : Int, body : Bytes) -> Unit

    Publishes a message after the requested broker-side delay.

    Connection::ready

    source
    async fn Connection::ready(self : Connection, count : Int) -> Unit

    Changes the number of messages the broker may deliver to this connection.

    Connection::receive

    source
    async fn Connection::receive(self : Connection) -> Message

    Waits for the next delivered message, automatically answering heartbeats.

    Connection::requeue

    source
    async fn Connection::requeue(self : Connection, id : Bytes, delay_ms : Int) -> Unit

    Requeues a message after the requested delay in milliseconds.

    Connection::subscribe

    source
    async fn Connection::subscribe(self : Connection, topic : String, channel : String) -> Unit

    Subscribes this connection to one topic/channel pair.

    Connection::touch

    source
    async fn Connection::touch(self : Connection, id : Bytes) -> Unit

    Extends the server-side timeout for an in-flight message.

    ConnectionHealth

    source
    pub struct ConnectionHealth {
      opened_at : Int64
      last_activity : Int64
      heartbeats_sent : Int
      heartbeats_received : Int
      requests : Int
      request_failures : Int
      bytes_read : Int64
      bytes_written : Int64
    } derive(
    Debug    )

    Snapshot of connection-level health and request outcomes.

    ConnectionHealth::age

    source
    fn ConnectionHealth::age(self : ConnectionHealth, now_ms : Int64) -> Int64

    Connection age measured from the supplied monotonic clock.

    ConnectionHealth::bytes_read

    source
    fn ConnectionHealth::bytes_read(self : ConnectionHealth) -> Int64

    ConnectionHealth::bytes_written

    source
    fn ConnectionHealth::bytes_written(self : ConnectionHealth) -> Int64

    ConnectionHealth::failure_rate_basis_points

    source
    fn ConnectionHealth::failure_rate_basis_points(self : ConnectionHealth) -> Int

    Failure ratio in basis points, avoiding floating-point metrics.

    ConnectionHealth::heartbeats_received

    source
    fn ConnectionHealth::heartbeats_received(self : ConnectionHealth) -> Int

    ConnectionHealth::heartbeats_sent

    source
    fn ConnectionHealth::heartbeats_sent(self : ConnectionHealth) -> Int

    ConnectionHealth::idle_for

    source
    fn ConnectionHealth::idle_for(self : ConnectionHealth, now_ms : Int64) -> Int64

    Duration since the last observed activity.

    ConnectionHealth::is_alive

    source
    fn ConnectionHealth::is_alive(self : ConnectionHealth, now_ms : Int64, timeout_ms : Int64) -> Bool

    Returns true when activity is within the supplied liveness window.

    ConnectionHealth::last_activity

    source
    fn ConnectionHealth::last_activity(self : ConnectionHealth) -> Int64

    ConnectionHealth::new

    source
    fn ConnectionHealth::new(now_ms : Int64) -> ConnectionHealth

    Creates a health tracker at a supplied monotonic timestamp.

    ConnectionHealth::opened_at

    source
    fn ConnectionHealth::opened_at(self : ConnectionHealth) -> Int64

    ConnectionHealth::record_failure

    source
    fn ConnectionHealth::record_failure(self : ConnectionHealth, now_ms : Int64) -> Unit

    ConnectionHealth::record_heartbeat_received

    source
    fn ConnectionHealth::record_heartbeat_received(self : ConnectionHealth, now_ms : Int64) -> Unit

    ConnectionHealth::record_heartbeat_sent

    source
    fn ConnectionHealth::record_heartbeat_sent(self : ConnectionHealth, now_ms : Int64) -> Unit

    ConnectionHealth::record_read

    source
    fn ConnectionHealth::record_read(self : ConnectionHealth, now_ms : Int64, bytes : Int) -> Unit raise ProtocolError

    ConnectionHealth::record_request

    source
    fn ConnectionHealth::record_request(self : ConnectionHealth, now_ms : Int64) -> Unit

    ConnectionHealth::record_success

    source
    fn ConnectionHealth::record_success(self : ConnectionHealth, now_ms : Int64) -> Unit

    Records a successful request and updates its activity timestamp.

    ConnectionHealth::record_write

    source
    fn ConnectionHealth::record_write(self : ConnectionHealth, now_ms : Int64, bytes : Int) -> Unit raise ProtocolError

    ConnectionHealth::request_failures

    source
    fn ConnectionHealth::request_failures(self : ConnectionHealth) -> Int

    ConnectionHealth::requests

    source
    fn ConnectionHealth::requests(self : ConnectionHealth) -> Int

    ConnectionHealth::success_rate_basis_points

    source
    fn ConnectionHealth::success_rate_basis_points(self : ConnectionHealth) -> Int

    Successful request ratio in basis points.

    ConnectionHealth::to_repr

    source

    ConnectionLease

    source
    pub struct ConnectionLease {
      endpoint : ProducerEndpoint
      generation : Int
      released : Bool
      // private fields
    } derive(
    Debug    )

    A lease returned by the endpoint pool before opening a TCP connection.

    ConnectionLease::endpoint

    source

    ConnectionLease::generation

    source
    fn ConnectionLease::generation(self : ConnectionLease) -> Int

    ConnectionLease::is_released

    source
    fn ConnectionLease::is_released(self : ConnectionLease) -> Bool

    ConnectionLease::to_repr

    source

    ConnectionOptions

    source
    pub struct ConnectionOptions {
      client_id : String
      hostname : String
      heartbeat_interval : Int
      output_buffer_size : Int
      output_buffer_timeout : Int
      message_timeout : Int
      sample_rate : Int
      tls_v1 : Bool
      feature_negotiation : Bool
    } derive(
    Debug    )

    Settings sent by IDENTIFY when a connection is opened.

    ConnectionOptions::new

    source
    fn ConnectionOptions::new(client_id? : String, hostname? : String, heartbeat_interval? : Int, output_buffer_size? : Int, output_buffer_timeout? : Int, message_timeout? : Int, sample_rate? : Int, tls_v1? : Bool, feature_negotiation? : Bool) -> ConnectionOptions raise ProtocolError

    Creates conservative settings compatible with NSQ 1.x brokers.

    ConnectionOptions::to_repr

    source

    ConnectionOptions::validate

    source
    fn ConnectionOptions::validate(self : ConnectionOptions) -> Unit raise ProtocolError

    Checks ranges before any bytes are sent to a broker.

    ConnectionPool

    source
    pub struct ConnectionPool {
      routing : RoutingTable
      generation : Int
      max_connections : Int
      active : Int
      stats : PoolStats
    } derive(
    Debug    )

    A bounded endpoint selector suitable for producers and lookup refreshes.

    ConnectionPool::active

    source
    fn ConnectionPool::active(self : ConnectionPool) -> Int

    ConnectionPool::capacity

    source
    fn ConnectionPool::capacity(self : ConnectionPool) -> Int

    ConnectionPool::lease

    source
    fn ConnectionPool::lease(self : ConnectionPool, now_ms : Int64) -> ConnectionLease raise ProtocolError

    Reserves an endpoint, rejecting leases once the configured bound is full.

    ConnectionPool::mark_failure

    source
    fn ConnectionPool::mark_failure(self : ConnectionPool, lease : ConnectionLease, now_ms : Int64) -> Bool

    ConnectionPool::mark_success

    source
    fn ConnectionPool::mark_success(self : ConnectionPool, lease : ConnectionLease) -> Bool

    ConnectionPool::new

    source
    fn ConnectionPool::new(producers : Array[ProducerEndpoint], max_connections? : Int) -> ConnectionPool raise ProtocolError

    ConnectionPool::refresh

    source
    fn ConnectionPool::refresh(self : ConnectionPool, producers : Array[ProducerEndpoint]) -> Unit raise ProtocolError

    Replaces discovery results and invalidates old lease generations.

    ConnectionPool::release

    source
    fn ConnectionPool::release(self : ConnectionPool, lease : ConnectionLease) -> Unit

    Releases a lease exactly once.

    ConnectionPool::stats

    source

    ConnectionPool::to_repr

    source

    Consumer

    source
    pub struct Consumer {
      topic : String
      channel : String
      options : ConsumerOptions
      started : Bool
      received : Int
      finished : Int
      requeued : Int
      discarded : Int
      touched : Int
      // private fields
    }

    A stateful NSQ subscription built on one TCP connection.

    Consumer::close

    source
    async fn Consumer::close(self : Consumer) -> Unit

    Stops delivery, waits for CLOSE_WAIT, then closes the underlying socket.

    Consumer::connect

    source
    async fn Consumer::connect(host : String, port : Int, topic : String, channel : String, options? : ConsumerOptions, connection_options? : ConnectionOptions) -> Consumer

    Opens, identifies, and prepares a consumer without starting delivery.

    Consumer::continue_delivery

    source
    async fn Consumer::continue_delivery(self : Consumer) -> Unit

    Restores the configured in-flight allowance after a pause.

    Consumer::discard

    source
    async fn Consumer::discard(self : Consumer, message : Message) -> Unit

    Permanently acknowledges a message that the application intentionally discards.

    Consumer::finish

    source
    async fn Consumer::finish(self : Consumer, message : Message) -> Unit

    Acknowledges a message as processed successfully.

    Consumer::pause

    source
    async fn Consumer::pause(self : Consumer) -> Unit

    Pauses delivery without closing the TCP connection.

    Consumer::process_one

    source
    async fn Consumer::process_one(self : Consumer, handler : async (Message) -> Disposition) -> Unit

    Applies one handler decision, including automatic discard after too many attempts.

    Consumer::receive

    source
    async fn Consumer::receive(self : Consumer) -> Message

    Reads one message. The caller must settle it through finish, requeue, or discard.

    Consumer::requeue

    source
    async fn Consumer::requeue(self : Consumer, message : Message) -> Unit

    Requeues a message unless the configured retry ceiling was reached.

    Consumer::start

    source
    async fn Consumer::start(self : Consumer) -> Unit

    Sends SUB and the initial RDY count exactly once.

    Consumer::stats

    source
    fn Consumer::stats(self : Consumer) -> ConsumerStats

    Returns a snapshot of counters collected by this consumer session.

    Consumer::touch

    source
    async fn Consumer::touch(self : Consumer, message : Message) -> Unit

    Extends a long-running message's timeout.

    ConsumerOptions

    source
    pub struct ConsumerOptions {
      max_in_flight : Int
      requeue_delay : Int
      max_attempts : Int
      touch_interval : Int
    } derive(
    Debug    )

    Consumer-local settings that do not alter broker topology.

    ConsumerOptions::new

    source
    fn ConsumerOptions::new(max_in_flight? : Int, requeue_delay? : Int, max_attempts? : Int, touch_interval? : Int) -> ConsumerOptions raise ProtocolError

    ConsumerOptions::to_repr

    source

    ConsumerOptions::validate

    source
    fn ConsumerOptions::validate(self : ConsumerOptions) -> Unit raise ProtocolError

    ConsumerStats

    source
    pub struct ConsumerStats {
      received : Int
      finished : Int
      requeued : Int
      discarded : Int
      touched : Int
    } derive(
    Debug    )

    Counters for one consumer session.

    ConsumerStats::to_repr

    source

    DeadLetterReason

    source
    pub enum DeadLetterReason {
      AttemptsExceeded
      HandlerRejected
      InvalidPayload
      Shutdown
    } derive(Eq, 
    Debug    )

    Why a consumer stopped retrying a message.

    DeadLetterReason::equal

    source

    DeadLetterReason::not_equal

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

    DeadLetterReason::to_repr

    source

    DeadLetterStats

    source
    pub struct DeadLetterStats {
      sent : Int
      failed : Int
      last_reason : DeadLetterReason?
    } derive(
    Debug    )

    A compact counter used by dead-letter sinks.

    DeadLetterStats::failed

    source
    fn DeadLetterStats::failed(self : DeadLetterStats) -> Int

    DeadLetterStats::new

    source

    DeadLetterStats::record_failure

    source
    fn DeadLetterStats::record_failure(self : DeadLetterStats) -> Unit

    DeadLetterStats::record_sent

    source
    fn DeadLetterStats::record_sent(self : DeadLetterStats, reason : DeadLetterReason) -> Unit

    DeadLetterStats::sent

    source
    fn DeadLetterStats::sent(self : DeadLetterStats) -> Int

    DeadLetterStats::to_repr

    source

    DeliveryWindow

    source
    pub struct DeliveryWindow {
      limit : Int
      in_flight : Int
      accepted : Int
      settled : Int
    } derive(
    Debug    )

    Consumer-side in-flight window independent of broker connection details.

    DeliveryWindow::accepted

    source
    fn DeliveryWindow::accepted(self : DeliveryWindow) -> Int

    DeliveryWindow::available

    source
    fn DeliveryWindow::available(self : DeliveryWindow) -> Int

    Returns available capacity for issuing an RDY command.

    DeliveryWindow::in_flight

    source
    fn DeliveryWindow::in_flight(self : DeliveryWindow) -> Int

    DeliveryWindow::limit

    source
    fn DeliveryWindow::limit(self : DeliveryWindow) -> Int

    DeliveryWindow::new

    source
    fn DeliveryWindow::new(limit : Int) -> DeliveryWindow raise ProtocolError

    DeliveryWindow::reset

    source
    fn DeliveryWindow::reset(self : DeliveryWindow) -> Int

    Clears in-flight state after a connection is lost; accepted totals remain.

    DeliveryWindow::settle

    source
    fn DeliveryWindow::settle(self : DeliveryWindow) -> Bool

    Settles one delivery exactly when a slot is occupied.

    DeliveryWindow::settled

    source
    fn DeliveryWindow::settled(self : DeliveryWindow) -> Int

    DeliveryWindow::to_repr

    source

    DeliveryWindow::try_accept

    source
    fn DeliveryWindow::try_accept(self : DeliveryWindow) -> Bool

    Reserves one delivery slot, returning false when the window is full.

    DeliveryWindow::utilization_basis_points

    source
    fn DeliveryWindow::utilization_basis_points(self : DeliveryWindow) -> Int

    Current utilization in basis points for dashboards.

    Disposition

    source
    pub enum Disposition {
      Finish
      Requeue(Int)
      Discard
    } derive(
    Debug    )

    The disposition chosen after a consumer handler sees one message.

    Disposition::to_repr

    source

    Frame

    source
    pub enum Frame {
      Response(Bytes)
      Error(Bytes)
      Message(Bytes)
    } derive(Eq, 
    Debug    )

    Frame::equal

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

    Frame::not_equal

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

    Frame::to_repr

    source

    FrameStream

    source
    pub struct FrameStream {
      buffered : Bytes
    }

    Incrementally assembles complete NSQ V2 frames from arbitrary TCP chunks.

    FrameStream::buffered_bytes

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

    Number of bytes retained while waiting for a complete frame.

    FrameStream::clear

    source
    fn FrameStream::clear(self : FrameStream) -> Unit

    Drops all retained partial bytes.

    FrameStream::new

    source

    FrameStream::next

    source
    fn FrameStream::next(self : FrameStream) -> Frame? raise ProtocolError

    Removes and returns one complete frame, or None when more bytes are needed.

    FrameStream::push

    source
    fn FrameStream::push(self : FrameStream, chunk : Bytes) -> Unit raise ProtocolError

    Appends bytes from one transport read without attempting to decode them.

    IdentifyPayload

    source
    type IdentifyPayload derive(ToJson)

    IdentifyPayload::to_json

    source

    LatencyHistogram

    source
    pub struct LatencyHistogram {
      samples : Int
      total_ms : Int64
      minimum_ms : Int
      maximum_ms : Int
      buckets : Array[Int]
    } derive(
    Debug    )

    A small fixed-bucket latency histogram for producer and consumer metrics.

    LatencyHistogram::count

    source
    fn LatencyHistogram::count(self : LatencyHistogram) -> Int

    LatencyHistogram::maximum

    source
    fn LatencyHistogram::maximum(self : LatencyHistogram) -> Int

    LatencyHistogram::mean

    source
    fn LatencyHistogram::mean(self : LatencyHistogram) -> Int

    Returns the arithmetic mean, or zero for an empty histogram.

    LatencyHistogram::merge

    source
    fn LatencyHistogram::merge(self : LatencyHistogram, other : LatencyHistogram) -> Unit

    Combines another histogram into this one.

    LatencyHistogram::minimum

    source
    fn LatencyHistogram::minimum(self : LatencyHistogram) -> Int

    LatencyHistogram::new

    source

    Creates an empty histogram with eleven practical latency buckets.

    LatencyHistogram::observe

    source
    fn LatencyHistogram::observe(self : LatencyHistogram, latency_ms : Int) -> Unit raise ProtocolError

    Adds a non-negative observation to the histogram.

    LatencyHistogram::percentile

    source
    fn LatencyHistogram::percentile(self : LatencyHistogram, percentile : Int) -> Int raise ProtocolError

    Returns the upper bound of the bucket containing the requested percentile.

    LatencyHistogram::to_repr

    source

    LatencyHistogram::total

    source
    fn LatencyHistogram::total(self : LatencyHistogram) -> Int64

    LookupResponse

    source
    type LookupResponse derive(
    FromJson    )

    Lookupd

    source
    pub struct Lookupd {
      base_url : String
    } derive(
    Debug    )

    An HTTP endpoint of an nsqlookupd process.

    Lookupd::lookup

    source
    async fn Lookupd::lookup(self : Lookupd, topic : String) -> Array[ProducerEndpoint]

    Looks up brokers responsible for a topic through the nsqlookupd HTTP API.

    Lookupd::new

    source
    fn Lookupd::new(host : String, port? : Int) -> Lookupd raise ProtocolError

    Lookupd::ping

    source
    async fn Lookupd::ping(self : Lookupd) -> Bool

    Queries /ping and returns whether this lookupd endpoint is reachable.

    Lookupd::to_repr

    source

    Message

    source
    pub struct Message {
      timestamp : Int64
      attempts : Int
      id : Bytes
      body : Bytes
    } derive(
    Debug    )

    A delivered NSQ message after removal of the V2 frame envelope.

    Message::to_repr

    source

    PoolStats

    source
    pub struct PoolStats {
      leases : Int
      releases : Int
      failures : Int
      successes : Int
      rejected : Int
    } derive(
    Debug    )

    Statistics for a pool's endpoint selection decisions.

    PoolStats::failures

    source
    fn PoolStats::failures(self : PoolStats) -> Int

    PoolStats::leases

    source
    fn PoolStats::leases(self : PoolStats) -> Int

    PoolStats::new

    source
    fn PoolStats::new() -> PoolStats

    PoolStats::rejected

    source
    fn PoolStats::rejected(self : PoolStats) -> Int

    PoolStats::releases

    source
    fn PoolStats::releases(self : PoolStats) -> Int

    PoolStats::successes

    source
    fn PoolStats::successes(self : PoolStats) -> Int

    PoolStats::to_repr

    source

    Producer

    source
    pub struct Producer {
      topic : String
      published : Int
      deferred : Int
      batches : Int
      batch_messages : Int
      // private fields
    }

    A topic-bound producer over one already identified NSQ connection.

    Producer::close

    source
    fn Producer::close(self : Producer) -> Unit

    Closes the underlying connection immediately.

    Producer::connect

    source
    async fn Producer::connect(host : String, port : Int, topic : String, options? : ConnectionOptions) -> Producer

    Opens and identifies a topic-bound producer.

    Producer::new_batch

    source
    fn Producer::new_batch(self : Producer, max_messages? : Int, max_bytes? : Int) -> PublishBatch raise ProtocolError

    Creates a bounded batch that targets this producer's topic.

    Producer::ping

    source
    async fn Producer::ping(self : Producer) -> Unit

    Sends a one-way NOP; no acknowledgement is provided by NSQ.

    Producer::publish

    source
    async fn Producer::publish(self : Producer, body : Bytes) -> Unit

    Publishes one binary message to the producer's configured topic.

    Producer::publish_batch

    source
    async fn Producer::publish_batch(self : Producer, batch : PublishBatch) -> Unit

    Sends all queued messages in a batch and records the total count.

    Producer::publish_deferred

    source
    async fn Producer::publish_deferred(self : Producer, delay_ms : Int, body : Bytes) -> Unit

    Publishes one binary message after a broker-side delay.

    Producer::stats

    source
    fn Producer::stats(self : Producer) -> ProducerStats

    ProducerEndpoint

    source
    pub struct ProducerEndpoint {
      broadcast_address : String
      tcp_port : Int
      http_port : Int
      version : String
    } derive(
    Debug    , 
    FromJson    )

    A TCP endpoint advertised by nsqlookupd.

    ProducerEndpoint::address

    source
    fn ProducerEndpoint::address(self : ProducerEndpoint) -> String

    Returns a host-and-port string accepted by connect.

    ProducerEndpoint::from_json

    source

    ProducerEndpoint::new

    source
    fn ProducerEndpoint::new(broadcast_address : String, tcp_port : Int, http_port? : Int, version? : String) -> ProducerEndpoint raise ProtocolError

    ProducerEndpoint::to_repr

    source

    ProducerHealth

    source
    pub struct ProducerHealth {
      endpoint : ProducerEndpoint
      failures : Int
      successes : Int
      last_failure_ms : Int64
      healthy : Bool
    } derive(
    Debug    )

    Health state tracked for one discovered NSQ producer.

    ProducerHealth::endpoint

    source

    ProducerHealth::failures

    source
    fn ProducerHealth::failures(self : ProducerHealth) -> Int

    ProducerHealth::is_healthy

    source
    fn ProducerHealth::is_healthy(self : ProducerHealth) -> Bool

    ProducerHealth::mark_failure

    source
    fn ProducerHealth::mark_failure(self : ProducerHealth, now_ms : Int64) -> Unit

    Marks a failed connection and records the time for cooldown decisions.

    ProducerHealth::mark_success

    source
    fn ProducerHealth::mark_success(self : ProducerHealth) -> Unit

    Marks a successful request and clears the failure state.

    ProducerHealth::new

    source

    Creates a healthy endpoint record.

    ProducerHealth::recover_if_ready

    source
    fn ProducerHealth::recover_if_ready(self : ProducerHealth, now_ms : Int64, cooldown_ms : Int64) -> Bool

    Re-enables an endpoint after its cooldown has elapsed.

    ProducerHealth::to_repr

    source

    ProducerMetrics

    source
    pub struct ProducerMetrics {
      published : Int
      failed : Int
      bytes : Int64
      latency : LatencyHistogram
    } derive(
    Debug    )

    Counters for an NSQ producer session.

    ProducerMetrics::bytes

    source
    fn ProducerMetrics::bytes(self : ProducerMetrics) -> Int64

    ProducerMetrics::failed

    source
    fn ProducerMetrics::failed(self : ProducerMetrics) -> Int

    ProducerMetrics::latency

    source

    ProducerMetrics::new

    source

    ProducerMetrics::published

    source
    fn ProducerMetrics::published(self : ProducerMetrics) -> Int

    ProducerMetrics::record_failure

    source
    fn ProducerMetrics::record_failure(self : ProducerMetrics) -> Unit

    ProducerMetrics::record_success

    source
    fn ProducerMetrics::record_success(self : ProducerMetrics, payload_bytes : Int, latency_ms : Int) -> Unit raise ProtocolError

    ProducerMetrics::to_repr

    source

    ProducerStats

    source
    pub struct ProducerStats {
      published : Int
      deferred : Int
      batches : Int
      batch_messages : Int
    } derive(
    Debug    )

    Counters for a producer connection.

    ProducerStats::to_repr

    source

    PublishBatch

    source
    pub struct PublishBatch {
      topic : String
      max_messages : Int
      max_bytes : Int
      bodies : Array[Bytes]
      payload_bytes : Int
    } derive(
    Debug    )

    A reusable, bounded collection that produces one NSQ MPUB command.

    PublishBatch::clear

    source
    fn PublishBatch::clear(self : PublishBatch) -> Unit

    Empties a batch without allocating a command.

    PublishBatch::command

    source
    fn PublishBatch::command(self : PublishBatch) -> Bytes raise ProtocolError

    Returns the binary MPUB command without changing the batch.

    PublishBatch::encoded_size

    source
    fn PublishBatch::encoded_size(self : PublishBatch) -> Int

    Total bytes that the MPUB payload would consume.

    PublishBatch::is_empty

    source
    fn PublishBatch::is_empty(self : PublishBatch) -> Bool

    PublishBatch::length

    source
    fn PublishBatch::length(self : PublishBatch) -> Int

    Number of queued bodies, excluding the MPUB count prefix.

    PublishBatch::new

    source
    fn PublishBatch::new(topic : String, max_messages? : Int, max_bytes? : Int) -> PublishBatch raise ProtocolError

    PublishBatch::push

    source
    fn PublishBatch::push(self : PublishBatch, body : Bytes) -> BatchPush

    Attempts to append one body without exceeding either configured bound.

    PublishBatch::take_command

    source
    fn PublishBatch::take_command(self : PublishBatch) -> Bytes raise ProtocolError

    Takes the current command and resets the batch for reuse.

    PublishBatch::to_repr

    source

    RetryAction

    source
    pub enum RetryAction {
      Retry(Int)
      DeadLetter(DeadLetterReason)
      Finish
    } derive(Eq, 
    Debug    )

    The action selected by a retry policy.

    RetryAction::equal

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

    RetryAction::not_equal

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

    RetryAction::to_repr

    source

    RetryPolicy

    source
    pub struct RetryPolicy {
      max_attempts : Int
      initial_delay : Int
      max_delay : Int
      multiplier : Int
      jitter_window : Int
    } derive(
    Debug    )

    Retry policy for application handlers that may fail transiently.

    RetryPolicy::decide

    source
    fn RetryPolicy::decide(self : RetryPolicy, message : Message, reason : DeadLetterReason, seed : Int) -> RetryAction

    Chooses retry or dead-letter based on the broker's attempt counter.

    RetryPolicy::delay_for

    source
    fn RetryPolicy::delay_for(self : RetryPolicy, attempt : Int) -> Int

    Calculates a capped delay before the next retry.

    RetryPolicy::delay_with_seed

    source
    fn RetryPolicy::delay_with_seed(self : RetryPolicy, attempt : Int, seed : Int) -> Int

    Adds deterministic jitter to a retry delay using a caller-provided seed.

    RetryPolicy::new

    source
    fn RetryPolicy::new(max_attempts? : Int, initial_delay? : Int, max_delay? : Int, multiplier? : Int, jitter_window? : Int) -> RetryPolicy raise ProtocolError

    Creates a bounded exponential policy. Jitter is deterministic when supplied by the caller.

    RetryPolicy::to_repr

    source

    RetryPolicy::validate

    source
    fn RetryPolicy::validate(self : RetryPolicy) -> Unit raise ProtocolError

    RoutingTable

    source
    pub struct RoutingTable {
      entries : Array[ProducerHealth]
      cursor : Int
      cooldown_ms : Int64
    } derive(
    Debug    )

    Round-robin routing table that skips unhealthy producers when possible.

    RoutingTable::choose

    source
    fn RoutingTable::choose(self : RoutingTable, now_ms : Int64) -> ProducerEndpoint

    Chooses the next healthy endpoint and advances the cursor.

    RoutingTable::length

    source
    fn RoutingTable::length(self : RoutingTable) -> Int

    RoutingTable::mark_failure

    source
    fn RoutingTable::mark_failure(self : RoutingTable, address : String, now_ms : Int64) -> Bool

    RoutingTable::mark_success

    source
    fn RoutingTable::mark_success(self : RoutingTable, address : String) -> Bool

    RoutingTable::new

    source
    fn RoutingTable::new(producers : Array[ProducerEndpoint], cooldown_ms? : Int64) -> RoutingTable raise ProtocolError

    RoutingTable::replace

    source
    fn RoutingTable::replace(self : RoutingTable, producers : Array[ProducerEndpoint]) -> Unit raise ProtocolError

    Replaces endpoints while retaining no stale health state.

    RoutingTable::to_repr

    source

    TopicSpec

    source
    pub struct TopicSpec {
      name : String
      partitions : Int
      retention_days : Int
      max_message_bytes : Int
    } derive(
    Debug    )

    Durable topic configuration used by provisioning tools.

    TopicSpec::max_message_bytes

    source
    fn TopicSpec::max_message_bytes(self : TopicSpec) -> Int

    TopicSpec::name

    source
    fn TopicSpec::name(self : TopicSpec) -> String

    TopicSpec::new

    source
    fn TopicSpec::new(name : String, partitions? : Int, retention_days? : Int, max_message_bytes? : Int) -> TopicSpec raise ProtocolError

    TopicSpec::partitions

    source
    fn TopicSpec::partitions(self : TopicSpec) -> Int

    TopicSpec::retention_days

    source
    fn TopicSpec::retention_days(self : TopicSpec) -> Int

    TopicSpec::to_repr

    source

    TopologyPlan

    source
    pub struct TopologyPlan {
      topic : TopicSpec
      channels : Array[ChannelSpec]
    } derive(
    Debug    )

    A validated provisioning plan for one topic and its channels.

    TopologyPlan::channels

    source
    fn TopologyPlan::channels(self : TopologyPlan) -> Array[ChannelSpec]

    TopologyPlan::contains_channel

    source
    fn TopologyPlan::contains_channel(self : TopologyPlan, name : String) -> Bool

    TopologyPlan::new

    source
    fn TopologyPlan::new(topic : TopicSpec, channels : Array[ChannelSpec]) -> TopologyPlan raise ProtocolError

    TopologyPlan::to_repr

    source

    TopologyPlan::topic

    source
    fn TopologyPlan::topic(self : TopologyPlan) -> TopicSpec

    attempts_exceeded

    source
    fn attempts_exceeded() -> DeadLetterReason

    close_command

    source
    fn close_command() -> Bytes

    Encodes CLS, asking a consumer connection to stop receiving new messages.

    connect

    source
    async fn connect(host : String, port : Int) -> Connection

    Opens a TCP connection and writes the mandatory NSQ V2 magic preamble.

    create_channel_operation

    source
    fn create_channel_operation(topic : String, channel : String) -> AdminOperation

    create_topic_operation

    source
    fn create_topic_operation(topic : String) -> AdminOperation

    Constructors kept explicit so callers can build operations without matching internals.

    decode_frame

    source
    fn decode_frame(data : Bytes) -> Frame raise ProtocolError

    Decodes a complete NSQ V2 frame, including its length prefix.

    decode_message

    source
    fn decode_message(data : Bytes) -> Message raise ProtocolError

    Decodes the timestamp, retry count, identifier, and body in an NSQ message.

    decode_u32

    source
    fn decode_u32(data : Bytes, offset : Int) -> Int

    deferred_publish_command

    source
    fn deferred_publish_command(topic : String, delay_ms : Int, body : Bytes) -> Bytes raise ProtocolError

    Encodes DPUB topic delay, requesting delayed delivery in milliseconds.

    delete_channel_operation

    source
    fn delete_channel_operation(topic : String, channel : String) -> AdminOperation

    delete_topic_operation

    source
    fn delete_topic_operation(topic : String) -> AdminOperation

    discard_disposition

    source
    fn discard_disposition() -> Disposition

    Creates a permanent discard decision.

    encode_frame

    source
    fn encode_frame(kind : Int, body : Bytes) -> Bytes raise ProtocolError

    Encodes one NSQ V2 frame. body does not include the four-byte type field.

    encode_u32

    source
    fn encode_u32(value : Int) -> Bytes

    finish_action

    source
    fn finish_action() -> RetryAction

    finish_disposition

    source
    fn finish_disposition() -> Disposition

    Creates a successful processing decision.

    handler_rejected

    source
    fn handler_rejected() -> DeadLetterReason

    invalid_payload

    source
    fn invalid_payload() -> DeadLetterReason

    magic_v2

    source
    let magic_v2 : Bytes

    The fixed four-byte preamble required before every NSQ TCP session.

    max_frame_size

    source
    let max_frame_size : Int

    The NSQ V2 frame header is a 32-bit, big-endian length followed by a type.

    message_command

    source
    fn message_command(verb : String, id : Bytes) -> Bytes raise ProtocolError

    Encodes a message disposition command such as FIN or TOUCH.

    multi_publish_command

    source
    fn multi_publish_command(topic : String, bodies : Array[Bytes]) -> Bytes raise ProtocolError

    Encodes MPUB topic with a count-prefixed collection of binary bodies.

    nop_command

    source
    fn nop_command() -> Bytes

    Encodes NOP, useful for testing a TCP session without creating a topic.

    pause_operation

    source
    fn pause_operation(topic : String) -> AdminOperation

    publish_command

    source
    fn publish_command(topic : String, body : Bytes) -> Bytes raise ProtocolError

    Encodes PUB topic, including the payload length prefix.

    ready_command

    source
    fn ready_command(count : Int) -> Bytes raise ProtocolError

    Encodes a bounded RDY command. Zero pauses delivery.

    requeue_command

    source
    fn requeue_command(id : Bytes, delay_ms : Int) -> Bytes raise ProtocolError

    Encodes REQ without a network connection, for deterministic testing.

    requeue_disposition

    source
    fn requeue_disposition(delay_ms : Int) -> Disposition raise ProtocolError

    Creates a retry decision with an application-selected delay.

    shutdown_reason

    source
    fn shutdown_reason() -> DeadLetterReason

    subscribe_command

    source
    fn subscribe_command(topic : String, channel : String) -> Bytes raise ProtocolError

    Encodes SUB topic channel.

    tombstone_operation

    source
    fn tombstone_operation(topic : String) -> AdminOperation

    touch_command

    source
    fn touch_command(id : Bytes) -> Bytes raise ProtocolError

    Encodes TOUCH to extend a message's configured timeout.

    unpause_operation

    source
    fn unpause_operation(topic : String) -> AdminOperation

    validate_name

    source
    fn validate_name(name : String) -> Unit raise ProtocolError

    Validates the topic and channel grammar used by NSQ command lines.

    Source Files

    Powered by MoonBit

    Site sourceReport issueModulesSkillsCLIBuild queueStatistics

    © 2026 mooncakes.io