README
¶
go-zmq
Go (golang) bindings for ZeroMQ (zmq, 0mq), currently built for ZeroMQ 3.2.x. Pull requests welcome.
View the API docs here.
Basic Usage
Import go-zmq:
import "github.com/vaughan0/go-zmq"
Create a new ZeroMQ Context and Socket:
ctx, err := zmq.NewContext()
if err != nil {
panic(err)
}
defer ctx.Close()
Bind to a local endpoint:
sock, err := ctx.Socket(zmq.Rep)
if err != nil {
panic(err)
}
defer sock.Close()
if err = sock.Bind("tcp://*:5555"); err != nil {
panic(err)
}
Receive and send messages:
for {
parts, err := sock.Recv()
if err != nil {
panic(err)
}
response := fmt.Sprintf("Received %d message parts", len(parts))
if err = sock.Send([][]byte{
[]byte(response),
}); err != nil {
panic(err)
}
}
Using Channels
ZeroMQ sockets are not thread-safe, which would make any ambitious ZeroMQ-using gopher sad. Luckily go-zmq provides a (thread-safe) way to use sockets with native Go channels. This also allows one to use the select construct with ZeroMQ sockets.
Start by using the Channels() method of a socket:
chans := sock.Channels()
defer chans.Close()
Now you can send and receive messages using the channels returned by chans.Out() and chans.In(), respectively. Don't forget to also check chans.Errors() to see if any errors occur.
for {
select {
case msg := <-chans.In():
go func() {
resp := doSomething(msg)
chans.Out() <- resp
}()
case err := <-chans.Errors():
panic(err)
}
}
Documentation
¶
Overview ¶
Package zmq provides ZeroMQ bindings for Go.
Index ¶
- Constants
- Variables
- func Device(deviceType DeviceType, frontend, backend *Socket)
- func Version() (major, minor, patch int)
- type Channels
- type Context
- type DeviceType
- type EventSet
- type PollSet
- func (p *PollSet) Events(index int) EventSet
- func (p *PollSet) Fd(fd uintptr, events EventSet) (index int)
- func (p *PollSet) File(f *os.File, events EventSet) (index int)
- func (p *PollSet) Monitored(index int) (events EventSet)
- func (p *PollSet) Poll(timeout time.Duration) (n int, err error)
- func (p *PollSet) Socket(sock *Socket, events EventSet) (index int)
- func (p *PollSet) Update(index int, events EventSet)
- type Socket
- func (s *Socket) Bind(endpoint string) (err error)
- func (s *Socket) Channels() *Channels
- func (s *Socket) ChannelsBuffer(chanbuf int) (c *Channels)
- func (s *Socket) Close()
- func (s *Socket) Connect(endpoint string) (err error)
- func (s *Socket) Disconnect(endpoint string) (err error)
- func (s *Socket) GetAffinity() uint64
- func (s *Socket) GetBacklog() int
- func (s *Socket) GetDelayAttachOnConnect() bool
- func (s *Socket) GetEvents() EventSet
- func (s *Socket) GetFD() uintptr
- func (s *Socket) GetIPv4Only() bool
- func (s *Socket) GetIdentity() []byte
- func (s *Socket) GetLastEndpoint() string
- func (s *Socket) GetLinger() time.Duration
- func (s *Socket) GetMaxMsgSize() int64
- func (s *Socket) GetMulticastHops() int
- func (s *Socket) GetRate() (kbits int)
- func (s *Socket) GetReconnectIVL() time.Duration
- func (s *Socket) GetReconnectIVLMax() time.Duration
- func (s *Socket) GetRecoveryIVL() time.Duration
- func (s *Socket) GetRecvBuffer() (bytes int)
- func (s *Socket) GetRecvHWM() uint64
- func (s *Socket) GetRecvTimeout() time.Duration
- func (s *Socket) GetSendBuffer() (bytes int)
- func (s *Socket) GetSendHWM() uint64
- func (s *Socket) GetSendTimeout() time.Duration
- func (s *Socket) GetTCPKeepAlive() int
- func (s *Socket) GetTCPKeepAliveCount() int
- func (s *Socket) GetTCPKeepAliveIdle() int
- func (s *Socket) GetTCPKeepAliveInterval() int
- func (s *Socket) GetType() SocketType
- func (s *Socket) Recv() (parts [][]byte, err error)
- func (s *Socket) RecvPart() (part []byte, more bool, err error)
- func (s *Socket) Send(parts [][]byte) (err error)
- func (s *Socket) SendPart(part []byte, more bool) (err error)
- func (s *Socket) SetAffinity(affinity uint64)
- func (s *Socket) SetBacklog(backlog int)
- func (s *Socket) SetDelayAttachOnConnect(delay bool)
- func (s *Socket) SetIPv4Only(ipv4only bool)
- func (s *Socket) SetIdentitiy(ident []byte)
- func (s *Socket) SetLinger(linger time.Duration)
- func (s *Socket) SetMaxMsgSize(bytes int64)
- func (s *Socket) SetMulticastHops(ttl int)
- func (s *Socket) SetRate(kbits int)
- func (s *Socket) SetReconnectIVL(ivl time.Duration)
- func (s *Socket) SetReconnectIVLMax(max time.Duration)
- func (s *Socket) SetRecoveryIVL(ivl time.Duration)
- func (s *Socket) SetRecvBuffer(bytes int)
- func (s *Socket) SetRecvHWM(hwm uint64)
- func (s *Socket) SetRecvTimeout(timeo time.Duration)
- func (s *Socket) SetRouterMandatory(errorUnroutable bool)
- func (s *Socket) SetSendBuffer(bytes int)
- func (s *Socket) SetSendHWM(hwm uint64)
- func (s *Socket) SetSendTimeout(timeo time.Duration)
- func (s *Socket) SetTCPAcceptFilter(filter string)
- func (s *Socket) SetTCPKeepAlive(keepalive int)
- func (s *Socket) SetTCPKeepAliveCount(count int)
- func (s *Socket) SetTCPKeepAliveIdle(idle int)
- func (s *Socket) SetTCPKeepAliveInterval(interval int)
- func (s *Socket) SetXPubVerbose(verbose bool)
- func (s *Socket) Subscribe(filter []byte)
- func (s *Socket) Unbind(endpoint string) (err error)
- func (s *Socket) Unsubscribe(filter []byte)
- type SocketType
Constants ¶
const ( // An input event. Corresponds to receiving on sockets and reading from files. In EventSet = C.ZMQ_POLLIN // An output event. Corresponds to sending on sockets and writing to files. Out = C.ZMQ_POLLOUT // An error event. Corresponds to errors on files. The Error event does not apply to sockets. Error = C.ZMQ_POLLERR // No events None = 0 )
const ( Req SocketType = C.ZMQ_REQ Rep = C.ZMQ_REP Dealer = C.ZMQ_DEALER Router = C.ZMQ_ROUTER Pub = C.ZMQ_PUB Sub = C.ZMQ_SUB XPub = C.ZMQ_XPUB XSub = C.ZMQ_XSUB Push = C.ZMQ_PUSH Pull = C.ZMQ_PULL Pair = C.ZMQ_PAIR )
const ( Queue DeviceType = C.ZMQ_QUEUE Forwarder = C.ZMQ_FORWARDER Streamer = C.ZMQ_STREAMER )
Variables ¶
var ( // ErrTerminated is returned when a socket's context has been closed. ErrTerminated = errors.New("zmq context has been terminated") // ErrTimeout is returned when an operation times out or a non-blocking operation cannot run immediately. ErrTimeout = errors.New("zmq timeout") ErrInterrupted = errors.New("system call interrupted") )
var PairPrefix = "socket-pair-"
Prefix used for socket-pair endpoint names in MakePair.
Functions ¶
func Device ¶
func Device(deviceType DeviceType, frontend, backend *Socket)
Creates and runs a ZeroMQ Device. See zmq_device(3) for more details.
Types ¶
type Channels ¶
type Channels struct {
// contains filtered or unexported fields
}
Channels provides a method for using Go channels to send and receive messages on a Socket. This is useful not only because it allows one to use select for sockets, but also because Sockets by themselves are not thread-safe (ie. one should not Send and Recv on the same socket from different threads).
type Context ¶
type Context struct {
// contains filtered or unexported fields
}
A Context manages multiple Sockets. Contexts are thread-safe.
func DefaultContext ¶
func DefaultContext() *Context
Returns the default Context. Note that the context will not be created until the first call to DefaultContext.
func NewContext ¶
Creates a new Context with the default number of IO threads (one).
func NewContextThreads ¶
Creates a new Context with the given number of dedicated IO threads.
func (*Context) Close ¶
func (c *Context) Close()
Closes the Context. Close will block until all related Sockets are closed, and all pending messages are either physically transferred to the network or the socket's linger period expires.
type DeviceType ¶
type DeviceType int
type EventSet ¶
type EventSet int
An EventSet is a bitmask of IO events.
func (EventSet) CanRecv ¶
Returns true if the associated socket can receive immediately, or if the associated file can be read from.
type PollSet ¶
type PollSet struct {
// contains filtered or unexported fields
}
A PollSet represents a set of sockets and/or file descriptors, along with monitored and triggered EventSets. The zero PollSet is valid (and empty).
func (*PollSet) Fd ¶
Adds a file descriptor to the PollSet along with a set of events to monitor. Returns the index in the PollSet of the added file.
func (*PollSet) Poll ¶
Poll waits for activity on the monitored set of sockets and/or files. If the timeout is zero, Poll will return immediately. If it is negative, Poll will wait forever until an event is triggered. Poll returns the number of sockets/files for which events were triggered, or a non-nil error.
type Socket ¶
type Socket struct {
// contains filtered or unexported fields
}
A ZeroMQ Socket.
func NewSocket ¶
func NewSocket(socktype SocketType) (*Socket, error)
Creates a new socket using the default context (see DefaultContext).
func (*Socket) Channels ¶
Creates a new Channels object with the default channel buffer size (zero).
func (*Socket) ChannelsBuffer ¶
Creates a new Channels object with the given channel buffer size.
func (*Socket) Disconnect ¶
Disconnects the socket from the specified remote endpoint.
func (*Socket) GetAffinity ¶
func (*Socket) GetBacklog ¶
func (*Socket) GetDelayAttachOnConnect ¶
func (*Socket) GetIPv4Only ¶
func (*Socket) GetIdentity ¶
func (*Socket) GetLastEndpoint ¶
func (*Socket) GetMaxMsgSize ¶
func (*Socket) GetMulticastHops ¶
func (*Socket) GetReconnectIVL ¶
func (*Socket) GetReconnectIVLMax ¶
func (*Socket) GetRecoveryIVL ¶
func (*Socket) GetRecvBuffer ¶
func (*Socket) GetRecvHWM ¶
func (*Socket) GetRecvTimeout ¶
func (*Socket) GetSendBuffer ¶
func (*Socket) GetSendHWM ¶
func (*Socket) GetSendTimeout ¶
func (*Socket) GetTCPKeepAlive ¶
func (*Socket) GetTCPKeepAliveCount ¶
func (*Socket) GetTCPKeepAliveIdle ¶
func (*Socket) GetTCPKeepAliveInterval ¶
func (*Socket) GetType ¶
func (s *Socket) GetType() SocketType
func (*Socket) RecvPart ¶
Receives a single part along with a boolean flag (more) indicating whether more parts of the same message follow (true), or this is the last part of the message (false). As with Send/SendPart, this is fairly low-level and Recv should generally be used instead.
func (*Socket) SendPart ¶
Sends a single message part. The `more` flag is used to specify whether this is the last part of the message (false), or if there are more parts to follow (true). SendPart is fairly low-level, and usually Send will be the preferred method to use.
func (*Socket) SetAffinity ¶
func (*Socket) SetBacklog ¶
func (*Socket) SetDelayAttachOnConnect ¶
func (*Socket) SetIPv4Only ¶
func (*Socket) SetIdentitiy ¶
func (*Socket) SetMaxMsgSize ¶
func (*Socket) SetMulticastHops ¶
func (*Socket) SetReconnectIVL ¶
func (*Socket) SetReconnectIVLMax ¶
func (*Socket) SetRecoveryIVL ¶
func (*Socket) SetRecvBuffer ¶
func (*Socket) SetRecvHWM ¶
func (*Socket) SetRecvTimeout ¶
func (*Socket) SetRouterMandatory ¶
func (*Socket) SetSendBuffer ¶
func (*Socket) SetSendHWM ¶
func (*Socket) SetSendTimeout ¶
func (*Socket) SetTCPAcceptFilter ¶
func (*Socket) SetTCPKeepAlive ¶
func (*Socket) SetTCPKeepAliveCount ¶
func (*Socket) SetTCPKeepAliveIdle ¶
func (*Socket) SetTCPKeepAliveInterval ¶
func (*Socket) SetXPubVerbose ¶
func (*Socket) Unsubscribe ¶
Unsubscribes from a filter on a Sub socket.
type SocketType ¶
type SocketType int
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Hello World client in Go Connects REQ socket to tcp://localhost:5555 Sends "Hello" to server, expects "World" back Hello World server in Go Binds REP socket to tcp://*:5555 Expects "Hello" from client, replies with "World" Demonstrate identities as used by the request-reply pattern.
|
Hello World client in Go Connects REQ socket to tcp://localhost:5555 Sends "Hello" to server, expects "World" back Hello World server in Go Binds REP socket to tcp://*:5555 Expects "Hello" from client, replies with "World" Demonstrate identities as used by the request-reply pattern. |