zmq

package module
v0.0.0-...-71a3e77 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 17, 2014 License: MIT Imports: 7 Imported by: 0

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

View Source 
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
)
View Source 
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
)
View Source 
const (
	Queue     DeviceType = C.ZMQ_QUEUE
	Forwarder            = C.ZMQ_FORWARDER
	Streamer             = C.ZMQ_STREAMER
)

Variables

View Source 
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")
)
View Source 
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.

func Version 

func Version() (major, minor, patch int)

Version reports 0MQ library version.

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).

func (*Channels) Close 

func (c *Channels) Close()

Closes the Channels object. This will ensure that a number of internal sockets are closed, and that worker goroutines are stopped cleanly.

func (*Channels) Errors 

func (c *Channels) Errors() <-chan error

func (*Channels) In 

func (c *Channels) In() <-chan [][]byte

func (*Channels) Out 

func (c *Channels) Out() chan<- [][]byte

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 

func NewContext() (*Context, error)

Creates a new Context with the default number of IO threads (one).

func NewContextThreads 

func NewContextThreads(nthreads int) (ctx *Context, err error)

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.

func (*Context) MakePair 

func (c *Context) MakePair() (a *Socket, b *Socket)

Creates a pair of connected inproc sockets that can be used for safe inter-thread communication. Returns both sockets.

func (*Context) Socket 

func (c *Context) Socket(socktype SocketType) (sock *Socket, err error)

Creates a new Socket of the specified type.

type DeviceType 

type DeviceType int

type EventSet 

type EventSet int

An EventSet is a bitmask of IO events.

func (EventSet) CanRecv 

func (e EventSet) CanRecv() bool

Returns true if the associated socket can receive immediately, or if the associated file can be read from.

func (EventSet) CanSend 

func (e EventSet) CanSend() bool

Returns true if the associated socket can send immediately, or if the associated file can be written to.

func (EventSet) HasError 

func (e EventSet) HasError() bool

Returns true if the associated file has an error condition. HasError will always return false for sockets.

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) Events 

func (p *PollSet) Events(index int) EventSet

Returns the triggered events for a socket/file of the given index.

func (*PollSet) Fd 

func (p *PollSet) Fd(fd uintptr, events EventSet) (index int)

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) File 

func (p *PollSet) File(f *os.File, events EventSet) (index int)

Shortcut that calls Fd() with the file's file descriptor.

func (*PollSet) Monitored 

func (p *PollSet) Monitored(index int) (events EventSet)

Returns the monitored events for the socket/file of the given index.

func (*PollSet) Poll 

func (p *PollSet) Poll(timeout time.Duration) (n int, err error)

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.

func (*PollSet) Socket 

func (p *PollSet) Socket(sock *Socket, events EventSet) (index int)

Adds a Socket to the PollSet along with a set of events to monitor. Returns the index in the PollSet of the added socket.

func (*PollSet) Update 

func (p *PollSet) Update(index int, events EventSet)

Updates the set of monitored events for the socket/file of the given index.

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) Bind 

func (s *Socket) Bind(endpoint string) (err error)

Binds the socket to the specified local endpoint address.

func (*Socket) Channels 

func (s *Socket) Channels() *Channels

Creates a new Channels object with the default channel buffer size (zero).

func (*Socket) ChannelsBuffer 

func (s *Socket) ChannelsBuffer(chanbuf int) (c *Channels)

Creates a new Channels object with the given channel buffer size.

func (*Socket) Close 

func (s *Socket) Close()

Closes the socket.

func (*Socket) Connect 

func (s *Socket) Connect(endpoint string) (err error)

Connects the socket to the specified remote endpoint.

func (*Socket) Disconnect 

func (s *Socket) Disconnect(endpoint string) (err error)

Disconnects the socket from the specified remote endpoint.

func (*Socket) GetAffinity 

func (s *Socket) GetAffinity() uint64

func (*Socket) GetBacklog 

func (s *Socket) GetBacklog() int

func (*Socket) GetDelayAttachOnConnect 

func (s *Socket) GetDelayAttachOnConnect() bool

func (*Socket) GetEvents 

func (s *Socket) GetEvents() EventSet

func (*Socket) GetFD 

func (s *Socket) GetFD() uintptr

func (*Socket) GetIPv4Only 

func (s *Socket) GetIPv4Only() bool

func (*Socket) GetIdentity 

func (s *Socket) GetIdentity() []byte

func (*Socket) GetLastEndpoint 

func (s *Socket) GetLastEndpoint() string

func (*Socket) GetLinger 

func (s *Socket) GetLinger() time.Duration

func (*Socket) GetMaxMsgSize 

func (s *Socket) GetMaxMsgSize() int64

func (*Socket) GetMulticastHops 

func (s *Socket) GetMulticastHops() int

func (*Socket) GetRate 

func (s *Socket) GetRate() (kbits int)

func (*Socket) GetReconnectIVL 

func (s *Socket) GetReconnectIVL() time.Duration

func (*Socket) GetReconnectIVLMax 

func (s *Socket) GetReconnectIVLMax() time.Duration

func (*Socket) GetRecoveryIVL 

func (s *Socket) GetRecoveryIVL() time.Duration

func (*Socket) GetRecvBuffer 

func (s *Socket) GetRecvBuffer() (bytes int)

func (*Socket) GetRecvHWM 

func (s *Socket) GetRecvHWM() uint64

func (*Socket) GetRecvTimeout 

func (s *Socket) GetRecvTimeout() time.Duration

func (*Socket) GetSendBuffer 

func (s *Socket) GetSendBuffer() (bytes int)

func (*Socket) GetSendHWM 

func (s *Socket) GetSendHWM() uint64

func (*Socket) GetSendTimeout 

func (s *Socket) GetSendTimeout() time.Duration

func (*Socket) GetTCPKeepAlive 

func (s *Socket) GetTCPKeepAlive() int

func (*Socket) GetTCPKeepAliveCount 

func (s *Socket) GetTCPKeepAliveCount() int

func (*Socket) GetTCPKeepAliveIdle 

func (s *Socket) GetTCPKeepAliveIdle() int

func (*Socket) GetTCPKeepAliveInterval 

func (s *Socket) GetTCPKeepAliveInterval() int

func (*Socket) GetType 

func (s *Socket) GetType() SocketType

func (*Socket) Recv 

func (s *Socket) Recv() (parts [][]byte, err error)

Receives a multi-part message.

func (*Socket) RecvPart 

func (s *Socket) RecvPart() (part []byte, more bool, err error)

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) Send 

func (s *Socket) Send(parts [][]byte) (err error)

Sends a message containing a number of parts.

func (*Socket) SendPart 

func (s *Socket) SendPart(part []byte, more bool) (err error)

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 (s *Socket) SetAffinity(affinity uint64)

func (*Socket) SetBacklog 

func (s *Socket) SetBacklog(backlog int)

func (*Socket) SetDelayAttachOnConnect 

func (s *Socket) SetDelayAttachOnConnect(delay bool)

func (*Socket) SetIPv4Only 

func (s *Socket) SetIPv4Only(ipv4only bool)

func (*Socket) SetIdentitiy 

func (s *Socket) SetIdentitiy(ident []byte)

func (*Socket) SetLinger 

func (s *Socket) SetLinger(linger time.Duration)

func (*Socket) SetMaxMsgSize 

func (s *Socket) SetMaxMsgSize(bytes int64)

func (*Socket) SetMulticastHops 

func (s *Socket) SetMulticastHops(ttl int)

func (*Socket) SetRate 

func (s *Socket) SetRate(kbits int)

func (*Socket) SetReconnectIVL 

func (s *Socket) SetReconnectIVL(ivl time.Duration)

func (*Socket) SetReconnectIVLMax 

func (s *Socket) SetReconnectIVLMax(max time.Duration)

func (*Socket) SetRecoveryIVL 

func (s *Socket) SetRecoveryIVL(ivl time.Duration)

func (*Socket) SetRecvBuffer 

func (s *Socket) SetRecvBuffer(bytes int)

func (*Socket) SetRecvHWM 

func (s *Socket) SetRecvHWM(hwm uint64)

func (*Socket) SetRecvTimeout 

func (s *Socket) SetRecvTimeout(timeo time.Duration)

func (*Socket) SetRouterMandatory 

func (s *Socket) SetRouterMandatory(errorUnroutable bool)

func (*Socket) SetSendBuffer 

func (s *Socket) SetSendBuffer(bytes int)

func (*Socket) SetSendHWM 

func (s *Socket) SetSendHWM(hwm uint64)

func (*Socket) SetSendTimeout 

func (s *Socket) SetSendTimeout(timeo time.Duration)

func (*Socket) SetTCPAcceptFilter 

func (s *Socket) SetTCPAcceptFilter(filter string)

func (*Socket) SetTCPKeepAlive 

func (s *Socket) SetTCPKeepAlive(keepalive int)

func (*Socket) SetTCPKeepAliveCount 

func (s *Socket) SetTCPKeepAliveCount(count int)

func (*Socket) SetTCPKeepAliveIdle 

func (s *Socket) SetTCPKeepAliveIdle(idle int)

func (*Socket) SetTCPKeepAliveInterval 

func (s *Socket) SetTCPKeepAliveInterval(interval int)

func (*Socket) SetXPubVerbose 

func (s *Socket) SetXPubVerbose(verbose bool)

func (*Socket) Subscribe 

func (s *Socket) Subscribe(filter []byte)

Subscribe sets up a filter for incoming messages on Sub sockets.

func (*Socket) Unbind 

func (s *Socket) Unbind(endpoint string) (err error)

Unbinds the socket from the specified local endpoint address.

func (*Socket) Unsubscribe 

func (s *Socket) Unsubscribe(filter []byte)

Unsubscribes from a filter on a Sub socket.

type SocketType 

type SocketType int

Source Files

View all 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.
helpers

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.