zmqutil

README

zmqutil

This project is no longer being maintained, nor is its dependency github.com/alecthomas/gozmq.

Package zmqutil implements some ØMQ (http://www.zeromq.org) abstractions and utilities.

A context from this package remembers its sockets and has its own Linger option. When a context is closed, it will set the Linger option on each socket that would linger longer and then close them all.

An additonal type, Poller, provides a convenient way to attach event handlers to sockets.

Building

Build tags are used to distinguish versions of ØMQ. Version 2.1 is zmq_2_1, 2.2 is zmq_2_x, and 3.2 is zmq_3_x. If no build tags are specified then the default is 2.2 and this is reflected in the documentation generated by godoc.

Status

This package is in early development and is not widely used so the public API may change significantly at any time. Please let me know if you are using this package in production.

License

Use of this source code is governed by a BSD-style license that can be found in the LICENSE file.

Documentation

Overview

Package zmqutil implements some ØMQ (http://www.zeromq.org) abstractions and utilities.

A context from this package remembers its sockets and has its own Linger option. When a context is closed, it will set the Linger option on each socket that would linger longer and then close them all.

An additonal type, Poller, provides a convenient way to attach event handlers to sockets.

Index

• type Context
  • func NewContext() *Context
  • func (c *Context) Close() error
  • func (c *Context) NewSocket(t zmq.SocketType) *Socket
  • func (c *Context) SetLinger(linger time.Duration) error
  • func (c *Context) SetLogger(logger *log.Logger)
  • func (c *Context) SetVerbose(verbose bool) error
• type Event
• type Poller
  • func NewPoller(context *Context) *Poller
  • func (p *Poller) HandleErr(s *Socket, h func())
  • func (p *Poller) HandleIn(s *Socket, h func([][]byte))
  • func (p *Poller) HandleOut(s *Socket, h func())
  • func (p *Poller) Poll(timeout time.Duration) (err error)
  • func (p *Poller) Run() error
  • func (p *Poller) SetLogger(logger *log.Logger)
  • func (p *Poller) Unhandle(s *Socket)
• type Socket
  • func (s *Socket) MustBind(addr string)
  • func (s *Socket) MustConnect(addr string)

Examples

• Poller

Types

type Context

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

A Context corresponds to a ØMQ context.

A Context is essentially a socket factory that can be closed only after all the sockets it has created are also closed.

func NewContext

func NewContext() *Context

NewContext returns a new context or panics.

func (*Context) Close

func (c *Context) Close() error

Close closes the context, blocking until the job is done.

This also propagates the context's LINGER option to all sockets that would linger longer and then closes each of them.

func (*Context) NewSocket

func (c *Context) NewSocket(t zmq.SocketType) *Socket

NewSocket creates a new socket and registers it to be closed when the context is closed.

NewSocket will panic if the specified socket type is not valid, if the context is nil, or if there is not enough memory.

func (*Context) SetLinger

func (c *Context) SetLinger(linger time.Duration) error

SetLinger adjusts the amount of time that Close() will wait for queued messages to be sent. The default is to wait forever.

func (*Context) SetLogger

func (c *Context) SetLogger(logger *log.Logger)

SetLogger sets the logger that will be used for trace logging.

func (*Context) SetVerbose

func (c *Context) SetVerbose(verbose bool) error

SetVerbose enables (or disables) logging to os.Stdout.

When verbose is true and a logger has already been set through SetLogger, this will have no effect.

type Event

type Event struct {
	Socket *Socket        // socket on which events occurred
	Events zmq.PollEvents // bitmask of events that occurred
	Fault  error          // handlers may set this to halt the poller
}

A Event represents a set of events on a socket.

type Poller

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

A Poller is a ZeroMQ poller running in a goroutine.

The poller will respond to events on sockets by calling handlers that have been associated with those events on those sockets through Handle() and HandleFunc().

Note: since a Socket is not thread-safe, a Socket being polled by a Poller should not be operated on outside the scope of a handler.

Example

context := NewContext()
defer context.Close()

poller := NewPoller(context)

push := context.NewSocket(zmq.PUSH)
pull := context.NewSocket(zmq.PULL)
push.MustBind("tcp://127.0.0.1:5555")
pull.MustConnect("tcp://127.0.0.1:5555")

recv := make(chan string, 1)

poller.HandleIn(pull, func(m [][]byte) { recv <- string(m[0]) })

push.Send([]byte("Hello!"), 0)

poller.Poll(-1)

fmt.Println(<-recv)

Output:

Hello!

func NewPoller

func NewPoller(context *Context) *Poller

NewPoller creates a new poller.

func (*Poller) HandleErr

func (p *Poller) HandleErr(s *Socket, h func())

HandleErr sets a function to be called when an error occurs on s.

func (*Poller) HandleIn

func (p *Poller) HandleIn(s *Socket, h func([][]byte))

HandleIn sets the function that will be called for each message that arrives on s.

func (*Poller) HandleOut

func (p *Poller) HandleOut(s *Socket, h func())

HandleOut sets the function that will be called when a message can be sent on s with no delay.

func (*Poller) Poll

func (p *Poller) Poll(timeout time.Duration) (err error)

Poll polls, with the specified timeout, all sockets for all events that have been registered with event handlers.

A negative timeout means forever; otherwise, timeout wll be truncated to millisecond precision.

Execution will halt and return first error encountered from polling or handling.

func (*Poller) Run

func (p *Poller) Run() error

Run repeatedly calls Poll with an infinite timeout until an error is returned, then returns that error.

func (*Poller) SetLogger

func (p *Poller) SetLogger(logger *log.Logger)

SetLogger sets the logger that detailed messages will be sent to.

func (*Poller) Unhandle

func (p *Poller) Unhandle(s *Socket)

Unhandle removes any handlers for s and stops polling s.

type Socket

type Socket struct {
	*zmq.Socket
	// contains filtered or unexported fields
}

func (*Socket) MustBind

func (s *Socket) MustBind(addr string)

func (*Socket) MustConnect

func (s *Socket) MustConnect(addr string)