Documentation
¶
Overview ¶
Package stomp speaks STOMP 1.2 to a message broker.
This is what "JMS support" has to mean in a program that is not Java. JMS is an API rather than a protocol, so there is nothing to implement: what a site actually needs is for Perfuse to talk to the broker sitting behind their JMS applications.
STOMP rather than OpenWire, and the reasoning is worth recording because the obvious answer is wrong. OpenWire is ActiveMQ's native protocol and a binary serialisation of JMS commands - weeks of work, specific to one broker family, and undocumented except by its implementation. ActiveMQ auto-detects STOMP, AMQP and MQTT on the same port it serves OpenWire on, so a text protocol reaches the same brokers. STOMP covers ActiveMQ Classic, Artemis and RabbitMQ; AMQP 1.0 would add Azure Service Bus and IBM MQ, and is the sensible next one rather than OpenWire.
Written against the standard library. A broker client is not the place for a dependency whose failure modes are somebody else's to explain, and the framing here is small enough that owning it costs less than reading their issue tracker.
Index ¶
- Constants
- Variables
- func ErrorMessage(f *Frame) string
- func WriteFrame(w io.Writer, f Frame) error
- type Conn
- func (c *Conn) Ack(f *Frame) error
- func (c *Conn) Close() error
- func (c *Conn) Nack(f *Frame) error
- func (c *Conn) Receive(timeout time.Duration) (*Frame, error)
- func (c *Conn) Send(destination string, body []byte, headers map[string]string, ...) error
- func (c *Conn) Subscribe(id, destination, selector string) error
- type Frame
- type Options
Constants ¶
const ( CmdConnect = "CONNECT" CmdStomp = "STOMP" CmdSend = "SEND" CmdSubscribe = "SUBSCRIBE" CmdUnsubscribe = "UNSUBSCRIBE" CmdAck = "ACK" CmdNack = "NACK" CmdBegin = "BEGIN" CmdCommit = "COMMIT" CmdAbort = "ABORT" CmdDisconnect = "DISCONNECT" )
Commands sent by a client.
const ( CmdConnected = "CONNECTED" CmdMessage = "MESSAGE" CmdReceipt = "RECEIPT" CmdError = "ERROR" )
Commands sent by a broker.
Variables ¶
var ErrBrokerError = errors.New("stomp: the broker reported an error")
ErrBrokerError is returned when the broker sends an ERROR frame.
Functions ¶
func ErrorMessage ¶
ErrorMessage extracts the useful text from an ERROR frame.
The message header and the body both carry explanation and brokers use them differently: Artemis puts a summary in the header and a stack trace in the body, RabbitMQ the reverse. Taking whichever is present avoids an error that reports nothing.
Types ¶
type Conn ¶
type Conn struct {
// Version is the protocol version the broker agreed to.
Version string
// Server is what the broker calls itself, for the log.
Server string
// contains filtered or unexported fields
}
Conn is a STOMP session with a broker.
func (*Conn) Nack ¶
Nack reports that a message was not handled.
Sent rather than staying silent, because a broker holding an unacknowledged message will redeliver it after a timeout that is frequently minutes long - and a message that is going to fail again should fail again now, where the retry and the dead letter handling are the broker's own and already configured.
func (*Conn) Receive ¶
Receive waits for the next message.
Returns nil with no error when the deadline passes with nothing arriving, so a caller can poll and check for shutdown between attempts rather than blocking forever on a quiet queue.
func (*Conn) Send ¶
func (c *Conn) Send(destination string, body []byte, headers map[string]string, timeout time.Duration) error
Send publishes a message to a destination.
A receipt is requested and waited for, so that returning nil means the broker has the message rather than that the bytes left this machine. Without it a broker that is refusing messages - a full queue, a missing address, no permission - looks like a working destination, and the queue behind it drains into nothing.
func (*Conn) Subscribe ¶
Subscribe registers interest in a destination and waits for the broker to confirm it.
Waiting matters. Without a receipt this returned as soon as the frame was written, so a caller that subscribed and then expected messages published immediately afterwards could miss them - the broker had not registered the subscription yet. Found by three tests failing where an identical one passed, the difference being a few hundred milliseconds.
In production that is the shape of "we lose messages when the channel restarts", which is close to unfindable from a log.
Acknowledgement is per message rather than automatic. Automatic means the broker considers a message delivered the moment it is written to the socket, so anything that fails afterwards - a parse error, a destination being down, this process stopping - loses it silently. That is the wrong trade for clinical data even though it is faster.
type Options ¶
type Options struct {
// Addr is the broker's host and port. 61613 is the usual STOMP port.
Addr string
// Host is the virtual host. Defaults to the address's host, which is what brokers expect when there is only one.
Host string
// Login and Passcode authenticate. Both empty connects anonymously, which most brokers refuse by default.
Login string
Passcode string
// Timeout bounds connecting and each read. Defaults to thirty seconds.
Timeout time.Duration
// Heartbeat asks the broker to expect and send activity at this interval. Zero disables it.
//
// Worth enabling on a long-lived subscription. Without it a connection through a firewall that has silently dropped the
// route looks perfectly healthy from this end, and messages stop arriving with nothing logged.
Heartbeat time.Duration
// TLS wraps the connection.
TLS *tls.Config
}
Options describe how to connect.
Source Files