tyomq

README

tyo-mq-client-go

A Go client for tyo-mq — the distributed pub/sub messaging service with durable delivery (ACK / retry / dead-letter queue), MQTT-style topic wildcards, consumer groups, and multi-tenant auth realms.

The client is a dependency-light Socket.IO v4 implementation (only gorilla/websocket) with typed protocol structs and a small convenience layer.

Install

go get github.com/tyolab/tyo-mq-client-go

Requires Go 1.22+ and a running tyo-mq server (npm install tyo-mq && npx tyo-mq-server, or Docker — see the server repo).

Quick start

import tyomq "github.com/tyolab/tyo-mq-client-go"

c := tyomq.NewClient("http://localhost:17352", nil)
ready := make(chan struct{})
go c.Connect(ctx, ready)
<-ready

// with auth enabled on the server:
// err := c.Authenticate(ctx, "my-token")

// produce
c.RegisterProducer("order-service")
c.Produce("order-service", "order-placed", map[string]any{"orderId": 1001})

// subscribe (durable + auto-ACK)
c2 := tyomq.NewClient("http://localhost:17352", nil)
// ... Connect as above ...
c2.RegisterConsumer("email-service")
c2.Subscribe(tyomq.SubscribeReq{
    Producer: "order-service",
    Event:    "order-placed",
    Consumer: "email-service",
    Durable:  true,
    Ack:      true,
    Retry:    &tyomq.RetryPolicy{MaxAttempts: 3, Delay: "5s", Backoff: "exponential"},
}, func(msg tyomq.ConsumedMessage) {
    fmt.Printf("order event: %s\n", msg.Message)
})

Run the complete example against a local server:

go run ./examples/pubsub -server http://localhost:17352

Feature coverage

Feature	How
Fire-and-forget pub/sub	Produce / Subscribe
Durable delivery + ACK/retry	SubscribeReq{Durable, Ack, ManualAck, AckTimeout, Retry}; auto-ACK or c.Ack(msgID)
Topic wildcards (+, #)	SubscribeReq{Mode: "topic", Event: "orders/#"}
Consumer groups	SubscribeReq{Group: "workers"}
Broadcast / TTL / guaranteed	emit a full ProduceReq
Authentication (tokens)	Authenticate(ctx, token)
Authorization request flow	AuthorizationRequest + signed NewAuthNextReq / NewAuthDecideReq
Signed manager commands	CreateAdminProof (HMAC-SHA256, matches the server exactly)
Custom namespaces (e.g. /remote)	NewClientNS(url, "/remote", log)

Anything not covered by a helper is one c.Emit(event, payload) + c.On(event, handler) away — the full wire protocol is documented in the server repo.

Reconnection

Connect runs one connection and returns when it drops; reconnection policy belongs to the caller (a simple retry loop) so that services control their own backoff. Durable subscriptions survive: reconnect with the same consumer name, re-Subscribe, and queued messages replay.

Other clients

Node.js (and browsers) ships with the server package; see also Python, Rust, C/C++, Ruby, Java, and C#.

All clients are exercised together by the cross-language conformance suite, which runs the same pub/sub, durable-delivery, topic, group, and auth scenarios against every client (and every producer/consumer language pair) and publishes the resulting matrix.

License

Apache-2.0. Built by TYO Lab.

Documentation

Overview

Package tyomq provides a Socket.IO v4 client for the tyo-mq message broker.

Protocol: Engine.IO v4 over WebSocket, then Socket.IO v4 events. Wire format: `42["event_name",{...payload...}]`

Index

• Constants
• func ConsumeEventName(producer, event, scope string) string
• func NewAuthDecideReq(adminToken, requestID string, approved bool, role, reason string) map[string]interface{}
• func NewAuthNextReq(adminToken, realmFilter string) map[string]interface{}
• type AckReq
• type AdminProof
  • func CreateAdminProof(adminToken, action string, body interface{}) AdminProof
• type AuthFail
• type AuthOK
• type AuthenticationReq
• type AuthorizationApproved
• type AuthorizationDecideResult
• type AuthorizationNextResult
• type AuthorizationRejected
• type AuthorizationRequest
• type AuthorizationRequestOK
• type Client
  • func NewClient(serverURL string, log *slog.Logger) *Client
  • func NewClientNS(serverURL, namespace string, log *slog.Logger) *Client
  • func (c *Client) Ack(msgID string) error
  • func (c *Client) Authenticate(ctx context.Context, token string) error
  • func (c *Client) Close()
  • func (c *Client) Connect(ctx context.Context, ready chan<- struct{}) error
  • func (c *Client) Emit(event string, payload interface{}) error
  • func (c *Client) On(event string, h Handler)
  • func (c *Client) Produce(from, event string, message interface{}) error
  • func (c *Client) RegisterConsumer(name string) error
  • func (c *Client) RegisterProducer(name string) error
  • func (c *Client) Subscribe(req SubscribeReq, handler func(ConsumedMessage)) error
  • func (c *Client) WaitFor(event string) <-chan json.RawMessage
• type ConsumedMessage
• type Handler
• type PendingAuthRequest
• type ProduceReq
• type RegisterConsumer
• type RegisterProducer
• type RetryPolicy
• type ServerError
• type SubscribeReq

Constants

const (
	DefaultPort = 17352

	// AllProducers subscribes to an event (or topic pattern) from any producer.
	AllProducers = "TYO-MQ-ALL"

	// EventAll is the suffix used when subscribing to all events of a producer.
	EventAll = "TM-ALL"
)

Well-known tyo-mq protocol constants.

Functions

func ConsumeEventName

func ConsumeEventName(producer, event, scope string) string

ConsumeEventName returns the Socket.IO event name that tyo-mq emits when delivering a message from the given producer with the given event name.

scope="" (default) → "CONSUME-<lower(producer-event)>"
scope="all"        → "CONSUME-<lower(producer)>-TM-ALL"

func NewAuthDecideReq

func NewAuthDecideReq(adminToken, requestID string, approved bool, role, reason string) map[string]interface{}

NewAuthDecideReq builds the signed payload for AUTHORIZATION_DECIDE.

func NewAuthNextReq

func NewAuthNextReq(adminToken, realmFilter string) map[string]interface{}

NewAuthNextReq builds the signed payload for AUTHORIZATION_NEXT. Pass realmFilter="" to get the next request across all realms.

Types

type AckReq

type AckReq struct {
	MsgID string `json:"msgId"`
}

AckReq is emitted as "ACK" to acknowledge one delivered message.

type AdminProof

type AdminProof struct {
	Timestamp int64  `json:"timestamp"`
	Nonce     string `json:"nonce"`
	Signature string `json:"signature"`
}

AdminProof is the HMAC-SHA256 signed proof required for manager actions. Build one with CreateAdminProof.

func CreateAdminProof

func CreateAdminProof(adminToken, action string, body interface{}) AdminProof

CreateAdminProof creates an HMAC-SHA256 signed proof for a tyo-mq manager action. Matches the JS adminSignature.createAdminProof implementation exactly.

type AuthFail

type AuthFail struct {
	Code    int    `json:"code"`
	Message string `json:"message"`
}

type AuthOK

type AuthOK struct {
	Realm string `json:"realm"`
	Role  string `json:"role"`
}

type AuthenticationReq

type AuthenticationReq struct {
	Token string `json:"token"`
}

AuthenticationReq is emitted as "AUTHENTICATION" when the server has auth enabled. The server answers with "AUTH_OK" or "AUTH_FAIL".

type AuthorizationApproved

type AuthorizationApproved struct {
	RequestID string `json:"request_id"`
	Realm     string `json:"realm"`
	Role      string `json:"role"`
}

type AuthorizationDecideResult

type AuthorizationDecideResult struct {
	OK      bool                `json:"ok"`
	Request *PendingAuthRequest `json:"request"`
}

AuthorizationDecideResult is the server response to "AUTHORIZATION_DECIDE".

type AuthorizationNextResult

type AuthorizationNextResult struct {
	OK      bool                `json:"ok"`
	Request *PendingAuthRequest `json:"request"` // nil when the queue is empty
}

AuthorizationNextResult is the server response to "AUTHORIZATION_NEXT".

type AuthorizationRejected

type AuthorizationRejected struct {
	RequestID string `json:"request_id"`
	Reason    string `json:"reason"`
}

type AuthorizationRequest

type AuthorizationRequest struct {
	Realm             string      `json:"realm"`
	Role              string      `json:"role"`
	ClientID          string      `json:"client_id"`
	ClientName        string      `json:"client_name"`
	ClientToken       string      `json:"client_token"`
	ChallengeResponse interface{} `json:"challenge_response,omitempty"`
}

AuthorizationRequest is emitted as "AUTHORIZATION_REQUEST" by a new client asking to be admitted to a realm. The client generates its own ClientToken; once an operator approves the request, that token authenticates normally.

type AuthorizationRequestOK

type AuthorizationRequestOK struct {
	OK        bool   `json:"ok"`
	RequestID string `json:"request_id"`
	Status    string `json:"status"`
}

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client is a Socket.IO v4 client for tyo-mq.

Usage:

c := tyomq.NewClient("http://localhost:17352", logger)
c.On("AUTH_OK", func(p json.RawMessage) { ... })
ready := make(chan struct{})
go c.Connect(ctx, ready)
<-ready  // wait for namespace "/" to be joined
c.Emit("AUTHENTICATION", tyomq.AuthenticationReq{Token: "..."})

func NewClient

func NewClient(serverURL string, log *slog.Logger) *Client

NewClient returns an unconnected client for the default namespace "/".

func NewClientNS

func NewClientNS(serverURL, namespace string, log *slog.Logger) *Client

NewClientNS returns an unconnected client for a specific Socket.IO namespace (e.g. "/remote"). Call Connect to establish the connection.

func (*Client) Ack

func (c *Client) Ack(msgID string) error

Ack acknowledges one ACK-enabled delivery by its MsgID.

func (*Client) Authenticate

func (c *Client) Authenticate(ctx context.Context, token string) error

Authenticate sends the AUTHENTICATION message and waits for AUTH_OK or AUTH_FAIL. Call it right after Connect signals ready, before registering as a producer or consumer. Not needed when the server runs with auth disabled.

func (*Client) Close

func (c *Client) Close()

Close sends a WebSocket close frame and tears down the connection.

func (*Client) Connect

func (c *Client) Connect(ctx context.Context, ready chan<- struct{}) error

Connect dials the server, performs the Engine.IO + Socket.IO handshake, and runs the read loop until ctx is cancelled or the connection drops.

ready is closed exactly once when namespace "/" is joined and the client is ready to Emit. The caller must create ready before starting this goroutine:

ready := make(chan struct{})
go client.Connect(ctx, ready)
<-ready

func (*Client) Emit

func (c *Client) Emit(event string, payload interface{}) error

Emit sends a Socket.IO event to the server. Returns an error if the client is not currently connected (caller should wait for Connected() first, or retry).

func (*Client) On

func (c *Client) On(event string, h Handler)

On registers a handler for the given Socket.IO event name. Safe to call concurrently. Multiple handlers for one event run in order.

func (*Client) Produce

func (c *Client) Produce(from, event string, message interface{}) error

Produce publishes one fire-and-forget message. For durable/broadcast/TTL options emit a full ProduceReq: c.Emit("PRODUCE", tyomq.ProduceReq{...}).

func (*Client) RegisterConsumer

func (c *Client) RegisterConsumer(name string) error

RegisterConsumer announces this connection as a consumer named name. The name doubles as the durable consumer identity: reconnect with the same name to replay queued messages of a durable subscription.

func (*Client) RegisterProducer

func (c *Client) RegisterProducer(name string) error

RegisterProducer announces this connection as a producer named name.

func (*Client) Subscribe

func (c *Client) Subscribe(req SubscribeReq, handler func(ConsumedMessage)) error

Subscribe sends a SUBSCRIBE request and dispatches matching deliveries to handler. req.Consumer defaults to the name passed to RegisterConsumer being required — set it explicitly. For topic-pattern subscriptions set Mode: "topic"; req.Producer then defaults to AllProducers.

When req.Ack is true and req.ManualAck is false, Subscribe acknowledges each delivery automatically after handler returns. With ManualAck, call c.Ack(msg.MsgID) yourself once the work has truly succeeded.

func (*Client) WaitFor

func (c *Client) WaitFor(event string) <-chan json.RawMessage

WaitFor registers a one-shot handler for event and returns a channel that receives its first payload. Useful for handshake-style events:

ok := c.WaitFor("AUTH_OK")
c.Emit("AUTHENTICATION", tyomq.AuthenticationReq{Token: token})
select { case <-ok: ... case <-ctx.Done(): ... }

type ConsumedMessage

type ConsumedMessage struct {
	Event   string          `json:"event"`
	Message json.RawMessage `json:"message"`
	From    string          `json:"from"`
	MsgID   string          `json:"msgId"`
}

ConsumedMessage is the payload of a "CONSUME-…" event. MsgID is present only on ACK-enabled deliveries; send AckReq with it (or use Subscribe, which can do so automatically).

type Handler

type Handler func(payload json.RawMessage)

Handler is called when a matching Socket.IO event is received.

type PendingAuthRequest

type PendingAuthRequest struct {
	RequestID      string `json:"request_id"`
	Status         string `json:"status"`
	Realm          string `json:"realm"`
	Role           string `json:"role"`
	ClientID       string `json:"client_id"`
	ClientName     string `json:"client_name"`
	CreatedAt      string `json:"created_at"`
	DecisionReason string `json:"decision_reason,omitempty"`
}

PendingAuthRequest is a pending authorization request returned by "AUTHORIZATION_NEXT_RESULT".

type ProduceReq

type ProduceReq struct {
	Event      string      `json:"event"`
	Message    interface{} `json:"message"`
	From       string      `json:"from"`
	Guaranteed bool        `json:"guaranteed,omitempty"` // persist until consumed
	TTL        interface{} `json:"ttl,omitempty"`        // e.g. "1h" or ms
	Method     string      `json:"method,omitempty"`     // "broadcast" when broadcasting
	Broadcast  string      `json:"broadcast,omitempty"`  // "realm" | "group"
	Group      string      `json:"group,omitempty"`
}

ProduceReq is emitted as "PRODUCE".

Broadcast "realm" delivers one copy to every connected member of the producer's realm; "group" (with Group set) delivers one copy to every member of that consumer group. TTL bounds how long a durable copy may wait.

type RegisterConsumer

type RegisterConsumer struct {
	Name       string `json:"name"`
	ID         string `json:"id,omitempty"`
	ConsumerID string `json:"consumer_id,omitempty"`
}

RegisterConsumer is emitted as "CONSUMER" after connecting.

type RegisterProducer

type RegisterProducer struct {
	Name string `json:"name"`
}

RegisterProducer is emitted as "PRODUCER" after connecting.

type RetryPolicy

type RetryPolicy struct {
	MaxAttempts int    `json:"max_attempts,omitempty"`
	Delay       string `json:"delay,omitempty"`
	Backoff     string `json:"backoff,omitempty"` // "" | "exponential"
}

RetryPolicy configures server-side re-delivery for ACK-enabled durable subscriptions. Delay accepts duration strings like "5s", "200ms".

type ServerError

type ServerError struct {
	Code    int    `json:"code"`
	Message string `json:"message"`
}

type SubscribeReq

type SubscribeReq struct {
	Event      string       `json:"event"`
	Producer   string       `json:"producer"`
	Consumer   string       `json:"consumer"`
	Scope      string       `json:"scope,omitempty"` // "all" or "" (default)
	Durable    bool         `json:"durable,omitempty"`
	Ack        bool         `json:"ack,omitempty"`
	ManualAck  bool         `json:"manual_ack,omitempty"`
	AckTimeout string       `json:"ack_timeout,omitempty"` // e.g. "30s"
	Retry      *RetryPolicy `json:"retry,omitempty"`
	Mode       string       `json:"mode,omitempty"` // "" | "topic"
	Group      string       `json:"group,omitempty"`
	ConsumerID string       `json:"consumer_id,omitempty"`
}

SubscribeReq is emitted as "SUBSCRIBE".

The zero value of the optional fields gives plain fire-and-forget delivery. Durable + Ack turn on guaranteed delivery: the server queues messages while the consumer is offline, waits for "ACK" per message, retries on the RetryPolicy schedule, and dead-letters messages that exhaust their attempts. Mode "topic" treats Event as an MQTT-style pattern ("orders/+/status", "factory/#") matched against events from any producer. Group makes the subscription part of a consumer group that load-balances deliveries.