zmq4

package module
v1.2.11 Latest Latest
Warning

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

 Go to latest Published: Oct 31, 2023 License: BSD-2-Clause Imports: 11 Imported by: 980

README

A Go interface to ZeroMQ version 4.

Warning

Starting with Go 1.14, on Unix-like systems, you will get a lot of interrupted signal calls. See the top of a package documentation for a fix.

Go Report Card GoDoc

This requires ZeroMQ version 4.0.1 or above. To use CURVE security in versions prior to 4.2, ZeroMQ must be installed with libsodium enabled.

Partial support for ZeroMQ 4.2 DRAFT is available in the alternate version of zmq4 draft. The API pertaining to this is subject to change. To use this:

import (
    zmq "github.com/pebbe/zmq4/draft"
)

For ZeroMQ version 3, see: http://github.com/pebbe/zmq3

For ZeroMQ version 2, see: http://github.com/pebbe/zmq2

Including all examples of ØMQ - The Guide.

Keywords: zmq, zeromq, 0mq, networks, distributed computing, message passing, fanout, pubsub, pipeline, request-reply

See also

  • go-zeromq/zmq4 — A pure-Go implementation of ØMQ (ZeroMQ), version 4
  • go-nanomsg — Language bindings for nanomsg in Go
  • goczmq — A Go interface to CZMQ
  • Mangos — An implementation in pure Go of the SP ("Scalable Protocols") protocols

Requirements

zmq4 is just a wrapper for the ZeroMQ library. It doesn't include the library itself. So you need to have ZeroMQ installed, including its development files. On Linux and Darwin you can check this with ($ is the command prompt):

$ pkg-config --modversion libzmq
4.3.1

The Go compiler must be able to compile C code. You can check this with:

$ go env CGO_ENABLED
1

You can't do cross-compilation. That would disable C.

Windows

Build with CGO_CFLAGS and CGO_LDFLAGS environment variables, for example:

$env:CGO_CFLAGS='-ID:/dev/vcpkg/installed/x64-windows/include'
$env:CGO_LDFLAGS='-LD:/dev/vcpkg/installed/x64-windows/lib -l:libzmq-mt-4_3_4.lib'

Deploy result program with libzmq-mt-4_3_4.dll

Install

go get github.com/pebbe/zmq4

Docs

API change

There has been an API change in commit 0bc5ab465849847b0556295d9a2023295c4d169e of 2014-06-27, 10:17:55 UTC in the functions AuthAllow and AuthDeny.

Old:

func AuthAllow(addresses ...string)
func AuthDeny(addresses ...string)

New:

func AuthAllow(domain string, addresses ...string)
func AuthDeny(domain string, addresses ...string)

If domain can be parsed as an IP address, it will be interpreted as such, and it and all remaining addresses are added to all domains.

So this should still work as before:

zmq.AuthAllow("127.0.0.1", "123.123.123.123")

But this won't compile:

a := []string{"127.0.0.1", "123.123.123.123"}
zmq.AuthAllow(a...)

And needs to be rewritten as:

a := []string{"127.0.0.1", "123.123.123.123"}
zmq.AuthAllow("*", a...)

Furthermore, an address can now be a single IP address, as well as an IP address and mask in CIDR notation, e.g. "123.123.123.0/24".

Documentation

Overview

A Go interface to ZeroMQ (zmq, 0mq) version 4.

For ZeroMQ version 3, see: http://github.com/pebbe/zmq3

For ZeroMQ version 2, see: http://github.com/pebbe/zmq2

http://www.zeromq.org/

See also the wiki: https://github.com/pebbe/zmq4/wiki

----

A note on the use of a context:

This package provides a default context. This is what will be used by the functions without a context receiver, that create a socket or manipulate the context. Package developers that import this package should probably not use the default context with its associated functions, but create their own context(s). See: type Context.

----

Since Go 1.14 you will get a lot of interrupted system calls.

See: https://golang.org/doc/go1.14#runtime

There are two options to prevent this.

The first option is to build your program with the environment variable: 

GODEBUG=asyncpreemptoff=1

The second option is to let the program retry after an interrupted system call.

Initially, this is set to true, for the global context, and for contexts created with NewContext().

When you install a signal handler, for instance to handle Ctrl-C, you should probably clear this option in your signal handler. For example: 

zctx, _ := zmq.NewContext()

ctx, cancel := context.WithCancel(context.Background())

go func() {
    chSignal := make(chan os.Signal, 1)
    signal.Notify(chSignal, syscall.SIGHUP, syscall.SIGINT, syscall.SIGQUIT, syscall.SIGTERM)
    <-chSignal
    zmq4.SetRetryAfterEINTR(false)
    zctx.SetRetryAfterEINTR(false)
    cancel()
}()

----

Index

Constants

View Source 
const (

	// On Windows platform some of the standard POSIX errnos are not defined.
	EADDRINUSE      = Errno(C.EADDRINUSE)
	EADDRNOTAVAIL   = Errno(C.EADDRNOTAVAIL)
	EAFNOSUPPORT    = Errno(C.EAFNOSUPPORT)
	ECONNABORTED    = Errno(C.ECONNABORTED)
	ECONNREFUSED    = Errno(C.ECONNREFUSED)
	ECONNRESET      = Errno(C.ECONNRESET)
	EHOSTUNREACH    = Errno(C.EHOSTUNREACH)
	EINPROGRESS     = Errno(C.EINPROGRESS)
	EMSGSIZE        = Errno(C.EMSGSIZE)
	ENETDOWN        = Errno(C.ENETDOWN)
	ENETRESET       = Errno(C.ENETRESET)
	ENETUNREACH     = Errno(C.ENETUNREACH)
	ENOBUFS         = Errno(C.ENOBUFS)
	ENOTCONN        = Errno(C.ENOTCONN)
	ENOTSOCK        = Errno(C.ENOTSOCK)
	ENOTSUP         = Errno(C.ENOTSUP)
	EPROTONOSUPPORT = Errno(C.EPROTONOSUPPORT)
	ETIMEDOUT       = Errno(C.ETIMEDOUT)

	// Native 0MQ error codes.
	EFSM           = Errno(C.EFSM)
	EMTHREAD       = Errno(C.EMTHREAD)
	ENOCOMPATPROTO = Errno(C.ENOCOMPATPROTO)
	ETERM          = Errno(C.ETERM)
)
View Source 
const (
	MaxSocketsDflt = int(C.ZMQ_MAX_SOCKETS_DFLT)
	IoThreadsDflt  = int(C.ZMQ_IO_THREADS_DFLT)
)
View Source 
const (
	// Constants for NewSocket()
	// See: http://api.zeromq.org/4-1:zmq-socket#toc3
	REQ    = Type(C.ZMQ_REQ)
	REP    = Type(C.ZMQ_REP)
	DEALER = Type(C.ZMQ_DEALER)
	ROUTER = Type(C.ZMQ_ROUTER)
	PUB    = Type(C.ZMQ_PUB)
	SUB    = Type(C.ZMQ_SUB)
	XPUB   = Type(C.ZMQ_XPUB)
	XSUB   = Type(C.ZMQ_XSUB)
	PUSH   = Type(C.ZMQ_PUSH)
	PULL   = Type(C.ZMQ_PULL)
	PAIR   = Type(C.ZMQ_PAIR)
	STREAM = Type(C.ZMQ_STREAM)
)
View Source 
const (
	// Flags for (*Socket)Send(), (*Socket)Recv()
	// For Send, see: http://api.zeromq.org/4-1:zmq-send#toc2
	// For Recv, see: http://api.zeromq.org/4-1:zmq-msg-recv#toc2
	DONTWAIT = Flag(C.ZMQ_DONTWAIT)
	SNDMORE  = Flag(C.ZMQ_SNDMORE)
)
View Source 
const (
	// Flags for (*Socket)Monitor() and (*Socket)RecvEvent()
	// See: http://api.zeromq.org/4-3:zmq-socket-monitor#toc3
	EVENT_ALL                        = Event(C.ZMQ_EVENT_ALL)
	EVENT_CONNECTED                  = Event(C.ZMQ_EVENT_CONNECTED)
	EVENT_CONNECT_DELAYED            = Event(C.ZMQ_EVENT_CONNECT_DELAYED)
	EVENT_CONNECT_RETRIED            = Event(C.ZMQ_EVENT_CONNECT_RETRIED)
	EVENT_LISTENING                  = Event(C.ZMQ_EVENT_LISTENING)
	EVENT_BIND_FAILED                = Event(C.ZMQ_EVENT_BIND_FAILED)
	EVENT_ACCEPTED                   = Event(C.ZMQ_EVENT_ACCEPTED)
	EVENT_ACCEPT_FAILED              = Event(C.ZMQ_EVENT_ACCEPT_FAILED)
	EVENT_CLOSED                     = Event(C.ZMQ_EVENT_CLOSED)
	EVENT_CLOSE_FAILED               = Event(C.ZMQ_EVENT_CLOSE_FAILED)
	EVENT_DISCONNECTED               = Event(C.ZMQ_EVENT_DISCONNECTED)
	EVENT_MONITOR_STOPPED            = Event(C.ZMQ_EVENT_MONITOR_STOPPED)
	EVENT_HANDSHAKE_FAILED_NO_DETAIL = Event(C.ZMQ_EVENT_HANDSHAKE_FAILED_NO_DETAIL)
	EVENT_HANDSHAKE_SUCCEEDED        = Event(C.ZMQ_EVENT_HANDSHAKE_SUCCEEDED)
	EVENT_HANDSHAKE_FAILED_PROTOCOL  = Event(C.ZMQ_EVENT_HANDSHAKE_FAILED_PROTOCOL)
	EVENT_HANDSHAKE_FAILED_AUTH      = Event(C.ZMQ_EVENT_HANDSHAKE_FAILED_AUTH)
)
View Source 
const (
	// Flags for (*Socket)GetEvents()
	// See: http://api.zeromq.org/4-1:zmq-getsockopt#toc8
	POLLIN  = State(C.ZMQ_POLLIN)
	POLLOUT = State(C.ZMQ_POLLOUT)
)
View Source 
const (
	// Constants for (*Socket)GetMechanism()
	// See: http://api.zeromq.org/4-1:zmq-getsockopt#toc22
	NULL   = Mechanism(C.ZMQ_NULL)
	PLAIN  = Mechanism(C.ZMQ_PLAIN)
	CURVE  = Mechanism(C.ZMQ_CURVE)
	GSSAPI = Mechanism(C.ZMQ_GSSAPI)
)
View Source 
const CURVE_ALLOW_ANY = "*"

Variables

View Source 
var (
	ErrorContextClosed         = errors.New("Context is closed")
	ErrorSocketClosed          = errors.New("Socket is closed")
	ErrorMoreExpected          = errors.New("More expected")
	ErrorNotImplemented405     = errors.New("Not implemented, requires 0MQ version 4.0.5")
	ErrorNotImplemented41      = errors.New("Not implemented, requires 0MQ version 4.1")
	ErrorNotImplemented42      = errors.New("Not implemented, requires 0MQ version 4.2")
	ErrorNotImplementedWindows = errors.New("Not implemented on Windows")
	ErrorNoSocket              = errors.New("No such socket")
)

Functions

func AuthAllow 

func AuthAllow(domain string, addresses ...string)

Allow (whitelist) some addresses for a domain.

An address can be a single IP address, or an IP address and mask in CIDR notation.

For NULL, all clients from these addresses will be accepted.

For PLAIN and CURVE, they will be allowed to continue with authentication.

You can call this method multiple times to whitelist multiple IP addresses.

If you whitelist a single address for a domain, any non-whitelisted addresses for that domain are treated as blacklisted.

Use domain "*" for all domains.

For backward compatibility: if domain can be parsed as an IP address, it will be interpreted as another address, and it and all remaining addresses will be added to all domains.

func AuthCurveAdd 

func AuthCurveAdd(domain string, pubkeys ...string)

Add public user keys for CURVE authentication for a given domain.

To cover all domains, use "*".

Public keys are in Z85 printable text format.

To allow all client keys without checking, specify CURVE_ALLOW_ANY for the key.

func AuthCurvePublic 

func AuthCurvePublic(z85SecretKey string) (z85PublicKey string, err error)

Helper function to derive z85 public key from secret key

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

func AuthCurveRemove 

func AuthCurveRemove(domain string, pubkeys ...string)

Remove user keys from CURVE authentication for a given domain.

func AuthCurveRemoveAll 

func AuthCurveRemoveAll(domain string)

Remove all user keys from CURVE authentication for a given domain.

func AuthDeny 

func AuthDeny(domain string, addresses ...string)

Deny (blacklist) some addresses for a domain.

An address can be a single IP address, or an IP address and mask in CIDR notation.

For all security mechanisms, this rejects the connection without any further authentication.

Use either a whitelist for a domain, or a blacklist for a domain, not both. If you define both a whitelist and a blacklist for a domain, only the whitelist takes effect.

Use domain "*" for all domains.

For backward compatibility: if domain can be parsed as an IP address, it will be interpreted as another address, and it and all remaining addresses will be added to all domains.

func AuthMetaBlob 

func AuthMetaBlob(key, value string) (blob []byte, err error)

This encodes a key/value pair into the format used by a ZAP handler.

Returns an error if key is more then 255 characters long.

func AuthPlainAdd 

func AuthPlainAdd(domain, username, password string)

Add a user for PLAIN authentication for a given domain.

Set `domain` to "*" to apply to all domains.

func AuthPlainRemove 

func AuthPlainRemove(domain string, usernames ...string)

Remove users from PLAIN authentication for a given domain.

func AuthPlainRemoveAll 

func AuthPlainRemoveAll(domain string)

Remove all users from PLAIN authentication for a given domain.

func AuthSetMetadataHandler 

func AuthSetMetadataHandler(
	handler func(
		version, request_id, domain, address, identity, mechanism string, credentials ...string) (metadata map[string]string))

This function sets the metadata handler that is called by the ZAP handler to retrieve key/value properties that should be set on reply messages in case of a status code "200" (succes).

Default properties are `Socket-Type`, which is already set, and `Identity` and `User-Id` that are empty by default. The last two can be set, and more properties can be added.

The `User-Id` property is used for the `user id` frame of the reply message. All other properties are stored in the `metadata` frame of the reply message.

The default handler returns an empty map.

For the meaning of the handler arguments, and other details, see: http://rfc.zeromq.org/spec:27#toc10

func AuthSetVerbose 

func AuthSetVerbose(verbose bool)

Enable verbose tracing of commands and activity.

func AuthStart 

func AuthStart() (err error)

Start authentication.

Note that until you add policies, all incoming NULL connections are allowed (classic ZeroMQ behaviour), and all PLAIN and CURVE connections are denied.

func AuthStop 

func AuthStop()

Stop authentication.

func Error 

func Error(e int) string

Get 0MQ error message string.

func GetBlocky 

func GetBlocky() (bool, error)

Returns the blocky setting in the default context.

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

func GetIoThreads 

func GetIoThreads() (int, error)

Returns the size of the 0MQ thread pool in the default context.

func GetIpv6 

func GetIpv6() (bool, error)

Returns the IPv6 option in the default context.

func GetMaxMsgsz 

func GetMaxMsgsz() (int, error)

Returns the maximum message size in the default context.

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

func GetMaxSockets 

func GetMaxSockets() (int, error)

Returns the maximum number of sockets allowed in the default context.

func GetRetryAfterEINTR added in v1.1.0 

func GetRetryAfterEINTR() bool

Returns the retry after EINTR setting in the default context.

func HasCurve 

func HasCurve() bool

Returns false for ZeroMQ version < 4.1.0

Else: returns true if the library supports the CURVE security mechanism

func HasGssapi 

func HasGssapi() bool

Returns false for ZeroMQ version < 4.1.0

Else: returns true if the library supports the GSSAPI security mechanism

func HasIpc 

func HasIpc() bool

Returns false for ZeroMQ version < 4.1.0

Else: returns true if the library supports the ipc:// protocol

func HasNorm 

func HasNorm() bool

Returns false for ZeroMQ version < 4.1.0

Else: returns true if the library supports the norm:// protocol

func HasPgm 

func HasPgm() bool

Returns false for ZeroMQ version < 4.1.0

Else: returns true if the library supports the pgm:// protocol

func HasTipc 

func HasTipc() bool

Returns false for ZeroMQ version < 4.1.0

Else: returns true if the library supports the tipc:// protocol

func NewCurveKeypair 

func NewCurveKeypair() (z85_public_key, z85_secret_key string, err error)

Generate a new CURVE keypair

See: http://api.zeromq.org/4-1:zmq-curve-keypair#toc2

func Proxy 

func Proxy(frontend, backend, capture *Socket) error

Start built-in ØMQ proxy

See: http://api.zeromq.org/4-1:zmq-proxy#toc2

func ProxySteerable 

func ProxySteerable(frontend, backend, capture, control *Socket) error

Start built-in ØMQ proxy with PAUSE/RESUME/TERMINATE control flow

Returns ErrorNotImplemented405 with ZeroMQ version < 4.0.5

See: http://api.zeromq.org/4-1:zmq-proxy-steerable#toc2

func SetBlocky 

func SetBlocky(i bool) error

Sets the blocky behavior in the default context.

See: http://api.zeromq.org/4-2:zmq-ctx-set#toc3

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

func SetIoThreads 

func SetIoThreads(n int) error

Specifies the size of the 0MQ thread pool to handle I/O operations in the default context. If your application is using only the inproc transport for messaging you may set this to zero, otherwise set it to at least one. This option only applies before creating any sockets.

Default value: 1

func SetIpv6 

func SetIpv6(i bool) error

Sets the IPv6 value for all sockets created in the default context from this point onwards. A value of true means IPv6 is enabled, while false means the socket will use only IPv4. When IPv6 is enabled, a socket will connect to, or accept connections from, both IPv4 and IPv6 hosts.

Default value: false

func SetMaxMsgsz 

func SetMaxMsgsz(n int) error

Set maximum message size in the default context.

Default value: INT_MAX

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

func SetMaxSockets 

func SetMaxSockets(n int) error

Sets the maximum number of sockets allowed in the default context.

Default value: 1024

func SetRetryAfterEINTR added in v1.1.0 

func SetRetryAfterEINTR(retry bool)

Sets the retry after EINTR setting in the default context.

Initital value is true.

func SetThreadPriority 

func SetThreadPriority(n int) error

Sets scheduling priority for default context’s thread pool.

This option requires ZeroMQ version 4.1, and is not available on Windows.

Supported values for this option depend on chosen scheduling policy. Details can be found in sched.h file, or at http://man7.org/linux/man-pages/man2/sched_setscheduler.2.html

This option only applies before creating any sockets on the context.

Default value: -1

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

Returns ErrorNotImplementedWindows on Windows

func SetThreadSchedPolicy 

func SetThreadSchedPolicy(n int) error

Sets the scheduling policy for default context’s thread pool.

This option requires ZeroMQ version 4.1, and is not available on Windows.

Supported values for this option can be found in sched.h file, or at http://man7.org/linux/man-pages/man2/sched_setscheduler.2.html

This option only applies before creating any sockets on the context.

Default value: -1

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

Returns ErrorNotImplementedWindows on Windows

func Term 

func Term() error

Terminates the default context.

For linger behavior, see: http://api.zeromq.org/4-1:zmq-ctx-term

func Version 

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

Report 0MQ library version.

func Z85decode 

func Z85decode(s string) string

Decode a binary key from Z85 printable text

See: http://api.zeromq.org/4-1:zmq-z85-decode

func Z85encode 

func Z85encode(data string) string

Encode a binary key as Z85 printable text

See: http://api.zeromq.org/4-1:zmq-z85-encode

Types

type Context 

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

A context that is not the default context.

func NewContext 

func NewContext() (ctx *Context, err error)

Create a new context.

func (*Context) GetBlocky 

func (ctx *Context) GetBlocky() (bool, error)

Returns the blocky setting.

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

func (*Context) GetIoThreads 

func (ctx *Context) GetIoThreads() (int, error)

Returns the size of the 0MQ thread pool.

func (*Context) GetIpv6 

func (ctx *Context) GetIpv6() (bool, error)

Returns the IPv6 option.

func (*Context) GetMaxMsgsz 

func (ctx *Context) GetMaxMsgsz() (int, error)

Returns the maximum message size.

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

func (*Context) GetMaxSockets 

func (ctx *Context) GetMaxSockets() (int, error)

Returns the maximum number of sockets allowed.

func (*Context) GetRetryAfterEINTR added in v1.1.0 

func (ctx *Context) GetRetryAfterEINTR() bool

Returns the retry after EINTR setting.

func (*Context) NewSocket 

func (ctx *Context) NewSocket(t Type) (soc *Socket, err error)

Create 0MQ socket in the given context.

WARNING: The Socket is not thread safe. This means that you cannot access the same Socket from different goroutines without using something like a mutex.

For a description of socket types, see: http://api.zeromq.org/4-1:zmq-socket#toc3

func (*Context) SetBlocky 

func (ctx *Context) SetBlocky(i bool) error

Sets the blocky behavior.

See: http://api.zeromq.org/4-2:zmq-ctx-set#toc3

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

func (*Context) SetIoThreads 

func (ctx *Context) SetIoThreads(n int) error

Specifies the size of the 0MQ thread pool to handle I/O operations. If your application is using only the inproc transport for messaging you may set this to zero, otherwise set it to at least one. This option only applies before creating any sockets.

Default value: 1

func (*Context) SetIpv6 

func (ctx *Context) SetIpv6(i bool) error

Sets the IPv6 value for all sockets created in the context from this point onwards. A value of true means IPv6 is enabled, while false means the socket will use only IPv4. When IPv6 is enabled, a socket will connect to, or accept connections from, both IPv4 and IPv6 hosts.

Default value: false

func (*Context) SetMaxMsgsz 

func (ctx *Context) SetMaxMsgsz(n int) error

Set maximum message size.

Default value: INT_MAX

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

func (*Context) SetMaxSockets 

func (ctx *Context) SetMaxSockets(n int) error

Sets the maximum number of sockets allowed.

Default value: 1024

func (*Context) SetRetryAfterEINTR added in v1.1.0 

func (ctx *Context) SetRetryAfterEINTR(retry bool)

Sets the retry after EINTR setting.

Initital value is true.

func (*Context) SetThreadPriority 

func (ctx *Context) SetThreadPriority(n int) error

Sets scheduling priority for internal context’s thread pool.

This option requires ZeroMQ version 4.1, and is not available on Windows.

Supported values for this option depend on chosen scheduling policy. Details can be found in sched.h file, or at http://man7.org/linux/man-pages/man2/sched_setscheduler.2.html

This option only applies before creating any sockets on the context.

Default value: -1

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

Returns ErrorNotImplementedWindows on Windows

func (*Context) SetThreadSchedPolicy 

func (ctx *Context) SetThreadSchedPolicy(n int) error

Sets the scheduling policy for internal context’s thread pool.

This option requires ZeroMQ version 4.1, and is not available on Windows.

Supported values for this option can be found in sched.h file, or at http://man7.org/linux/man-pages/man2/sched_setscheduler.2.html

This option only applies before creating any sockets on the context.

Default value: -1

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

Returns ErrorNotImplementedWindows on Windows

func (*Context) Term 

func (ctx *Context) Term() error

Terminates the context.

For linger behavior, see: http://api.zeromq.org/4-1:zmq-ctx-term

type Errno 

type Errno uintptr

An Errno is an unsigned number describing an error condition as returned by a call to ZeroMQ. It implements the error interface. The number is either a standard system error, or an error defined by the C library of ZeroMQ.

func AsErrno 

func AsErrno(err error) Errno

Convert error to Errno.

Example usage: 

switch AsErrno(err) {

case zmq.Errno(syscall.EINTR):
    // standard system error

    // call was interrupted

case zmq.ETERM:
    // error defined by ZeroMQ

    // context was terminated

}

See also: examples/interrupt.go

func (Errno) Error 

func (errno Errno) Error() string

Return Errno as string.

type Event 

type Event int

Used by (*Socket)Monitor() and (*Socket)RecvEvent()

func (Event) String 

func (e Event) String() string

Socket event as string.

type Flag 

type Flag int

Used by (*Socket)Send() and (*Socket)Recv()

func (Flag) String 

func (f Flag) String() string

Socket flag as string.

type Mechanism 

type Mechanism int

Specifies the security mechanism, used by (*Socket)GetMechanism()

func (Mechanism) String 

func (m Mechanism) String() string

Security mechanism as string.

type Polled 

type Polled struct {
	Socket *Socket // socket with matched event(s)
	Events State   // actual matched event(s)
}

Return type for (*Poller)Poll

type Poller 

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

func NewPoller 

func NewPoller() *Poller

Create a new Poller

func (*Poller) Add 

func (p *Poller) Add(soc *Socket, events State) int

Add items to the poller

Events is a bitwise OR of zmq.POLLIN and zmq.POLLOUT

Returns the id of the item, which can be used as a handle to (*Poller)Update and as an index into the result of (*Poller)PollAll

func (*Poller) Poll 

func (p *Poller) Poll(timeout time.Duration) ([]Polled, error)

Input/output multiplexing

If timeout < 0, wait forever until a matching event is detected

Only sockets with matching socket events are returned in the list.

Example: 

poller := zmq.NewPoller()
poller.Add(socket0, zmq.POLLIN)
poller.Add(socket1, zmq.POLLIN)
//  Process messages from both sockets
for {
    sockets, _ := poller.Poll(-1)
    for _, socket := range sockets {
        switch s := socket.Socket; s {
        case socket0:
            msg, _ := s.Recv(0)
            //  Process msg
        case socket1:
            msg, _ := s.Recv(0)
            //  Process msg
        }
    }
}

func (*Poller) PollAll 

func (p *Poller) PollAll(timeout time.Duration) ([]Polled, error)

This is like (*Poller)Poll, but it returns a list of all sockets, in the same order as they were added to the poller, not just those sockets that had an event.

For each socket in the list, you have to check the Events field to see if there was actually an event.

When error is not nil, the return list contains no sockets.

func (*Poller) Remove 

func (p *Poller) Remove(id int) error

Remove a socket from the poller

Returns ErrorNoSocket if the id was out of range

func (*Poller) RemoveBySocket 

func (p *Poller) RemoveBySocket(soc *Socket) error

Remove a socket from the poller

Returns ErrorNoSocket if the socket didn't match

func (*Poller) String 

func (p *Poller) String() string

Poller as string.

func (*Poller) Update 

func (p *Poller) Update(id int, events State) (previous State, err error)

Update the events mask of a socket in the poller

Replaces the Poller's bitmask for the specified id with the events parameter passed

Returns the previous value, or ErrorNoSocket if the id was out of range

func (*Poller) UpdateBySocket 

func (p *Poller) UpdateBySocket(soc *Socket, events State) (previous State, err error)

Update the events mask of a socket in the poller

Replaces the Poller's bitmask for the specified socket with the events parameter passed

Returns the previous value, or ErrorNoSocket if the socket didn't match

type Reactor 

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

func NewReactor 

func NewReactor() *Reactor

Create a reactor to mix the handling of sockets and channels (timers or other channels).

Example: 

reactor := zmq.NewReactor()
reactor.AddSocket(socket1, zmq.POLLIN, socket1_handler)
reactor.AddSocket(socket2, zmq.POLLIN, socket2_handler)
reactor.AddChannelTime(time.Tick(time.Second), 1, ticker_handler)
reactor.Run(time.Second)

Warning:

There are problems with the reactor showing up with Go 1.14 (and later) such as data race occurrences and code lock-up. Using SetRetryAfterEINTR seems an effective fix, but at the moment there is no guaranty.

func (*Reactor) AddChannel 

func (r *Reactor) AddChannel(ch <-chan interface{}, limit int, handler func(interface{}) error) (id uint64)

Add channel handler to the reactor.

Returns id of added handler, that can be used later to remove it.

If limit is positive, at most this many items will be handled in each run through the main loop, otherwise it will process as many items as possible.

The handler function receives the value received from the channel.

func (*Reactor) AddChannelTime 

func (r *Reactor) AddChannelTime(ch <-chan time.Time, limit int, handler func(interface{}) error) (id uint64)

This function wraps AddChannel, using a channel of type time.Time instead of type interface{}.

func (*Reactor) AddSocket 

func (r *Reactor) AddSocket(soc *Socket, events State, handler func(State) error)

Add socket handler to the reactor.

You can have only one handler per socket. Adding a second one will remove the first.

The handler receives the socket state as an argument: POLLIN, POLLOUT, or both.

func (*Reactor) RemoveChannel 

func (r *Reactor) RemoveChannel(id uint64)

Remove a channel from the reactor.

Closed channels are removed automatically.

func (*Reactor) RemoveSocket 

func (r *Reactor) RemoveSocket(soc *Socket)

Remove a socket handler from the reactor.

func (*Reactor) Run 

func (r *Reactor) Run(interval time.Duration) (err error)

Run the reactor.

The interval determines the time-out on the polling of sockets. Interval must be positive if there are channels. If there are no channels, you can set interval to -1.

The run alternates between polling/handling sockets (using the interval as timeout), and reading/handling channels. The reading of channels is without time-out: if there is no activity on any channel, the run continues to poll sockets immediately.

The run exits when any handler returns an error, returning that same error.

func (*Reactor) SetVerbose 

func (r *Reactor) SetVerbose(verbose bool)

type Socket 

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

Socket functions starting with `Set` or `Get` are used for setting and getting socket options.

func NewSocket 

func NewSocket(t Type) (soc *Socket, err error)

Create 0MQ socket in the default context.

WARNING: The Socket is not thread safe. This means that you cannot access the same Socket from different goroutines without using something like a mutex.

For a description of socket types, see: http://api.zeromq.org/4-1:zmq-socket#toc3

func (*Socket) Bind 

func (soc *Socket) Bind(endpoint string) error

Accept incoming connections on a socket.

For a description of endpoint, see: http://api.zeromq.org/4-1:zmq-bind#toc2

func (*Socket) ClientAuthCurve 

func (client *Socket) ClientAuthCurve(server_public_key, client_public_key, client_secret_key string) error

Set CURVE client role.

func (*Socket) ClientAuthPlain 

func (client *Socket) ClientAuthPlain(username, password string) error

Set PLAIN client role.

func (*Socket) Close 

func (soc *Socket) Close() error

If not called explicitly, the socket will be closed on garbage collection

func (*Socket) Connect 

func (soc *Socket) Connect(endpoint string) error

Create outgoing connection from socket.

For a description of endpoint, see: http://api.zeromq.org/4-1:zmq-connect#toc2

func (*Socket) Context 

func (soc *Socket) Context() (*Context, error)

Return the context associated with a socket

func (*Socket) Disconnect 

func (soc *Socket) Disconnect(endpoint string) error

Disconnect a socket.

For a description of endpoint, see: http://api.zeromq.org/4-1:zmq-disconnect#toc2

func (*Socket) GetAffinity 

func (soc *Socket) GetAffinity() (uint64, error)

ZMQ_AFFINITY: Retrieve I/O thread affinity

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc3

func (*Socket) GetBacklog 

func (soc *Socket) GetBacklog() (int, error)

ZMQ_BACKLOG: Retrieve maximum length of the queue of outstanding connections

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc4

func (*Socket) GetConnectTimeout 

func (soc *Socket) GetConnectTimeout() (time.Duration, error)

ZMQ_CONNECT_TIMEOUT: Retrieve connect() timeout

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-getsockopt#toc5

func (*Socket) GetCurvePublickeyRaw 

func (soc *Socket) GetCurvePublickeyRaw() (string, error)

ZMQ_CURVE_PUBLICKEY: Retrieve current CURVE public key

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc5

func (*Socket) GetCurvePublickeykeyZ85 

func (soc *Socket) GetCurvePublickeykeyZ85() (string, error)

ZMQ_CURVE_PUBLICKEY: Retrieve current CURVE public key

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc5

func (*Socket) GetCurveSecretkeyRaw 

func (soc *Socket) GetCurveSecretkeyRaw() (string, error)

ZMQ_CURVE_SECRETKEY: Retrieve current CURVE secret key

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc6

func (*Socket) GetCurveSecretkeyZ85 

func (soc *Socket) GetCurveSecretkeyZ85() (string, error)

ZMQ_CURVE_SECRETKEY: Retrieve current CURVE secret key

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc6

func (*Socket) GetCurveServerkeyRaw 

func (soc *Socket) GetCurveServerkeyRaw() (string, error)

ZMQ_CURVE_SERVERKEY: Retrieve current CURVE server key

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc7

func (*Socket) GetCurveServerkeyZ85 

func (soc *Socket) GetCurveServerkeyZ85() (string, error)

ZMQ_CURVE_SERVERKEY: Retrieve current CURVE server key

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc7

func (*Socket) GetEvents 

func (soc *Socket) GetEvents() (State, error)

ZMQ_EVENTS: Retrieve socket event state

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc8

func (*Socket) GetFd 

func (soc *Socket) GetFd() (int, error)

ZMQ_FD: Retrieve file descriptor associated with the socket

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc9

func (*Socket) GetGssapiPlaintext 

func (soc *Socket) GetGssapiPlaintext() (bool, error)

ZMQ_GSSAPI_PLAINTEXT: Retrieve GSSAPI plaintext or encrypted status

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc10

func (*Socket) GetGssapiPrincipal 

func (soc *Socket) GetGssapiPrincipal() (string, error)

ZMQ_GSSAPI_PRINCIPAL: Retrieve the name of the GSSAPI principal

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc11

func (*Socket) GetGssapiServer 

func (soc *Socket) GetGssapiServer() (bool, error)

ZMQ_GSSAPI_SERVER: Retrieve current GSSAPI server role

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc12

func (*Socket) GetGssapiServicePrincipal 

func (soc *Socket) GetGssapiServicePrincipal() (string, error)

ZMQ_GSSAPI_SERVICE_PRINCIPAL: Retrieve the name of the GSSAPI service principal

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc13

func (*Socket) GetHandshakeIvl 

func (soc *Socket) GetHandshakeIvl() (time.Duration, error)

ZMQ_HANDSHAKE_IVL: Retrieve maximum handshake interval

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc14

func (*Socket) GetIdentity 

func (soc *Socket) GetIdentity() (string, error)

ZMQ_IDENTITY: Retrieve socket identity

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc15

func (*Socket) GetImmediate 

func (soc *Socket) GetImmediate() (bool, error)

ZMQ_IMMEDIATE: Retrieve attach-on-connect value

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc16

func (*Socket) GetInvertMatching 

func (soc *Socket) GetInvertMatching() (int, error)

ZMQ_INVERT_MATCHING: Retrieve inverted filtering status

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-getsockopt#toc18

func (*Socket) GetIpv6 

func (soc *Socket) GetIpv6() (bool, error)

ZMQ_IPV6: Retrieve IPv6 socket status

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc18

func (*Socket) GetLastEndpoint 

func (soc *Socket) GetLastEndpoint() (string, error)

ZMQ_LAST_ENDPOINT: Retrieve the last endpoint set

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc19

func (*Socket) GetLinger 

func (soc *Socket) GetLinger() (time.Duration, error)

ZMQ_LINGER: Retrieve linger period for socket shutdown

Returns time.Duration(-1) for infinite

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc20

func (*Socket) GetMaxmsgsize 

func (soc *Socket) GetMaxmsgsize() (int64, error)

ZMQ_MAXMSGSIZE: Maximum acceptable inbound message size

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc21

func (*Socket) GetMechanism 

func (soc *Socket) GetMechanism() (Mechanism, error)

ZMQ_MECHANISM: Retrieve current security mechanism

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc22

func (*Socket) GetMulticastHops 

func (soc *Socket) GetMulticastHops() (int, error)

ZMQ_MULTICAST_HOPS: Maximum network hops for multicast packets

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc23

func (*Socket) GetMulticastMaxtpdu 

func (soc *Socket) GetMulticastMaxtpdu() (int, error)

ZMQ_MULTICAST_MAXTPDU: Maximum transport data unit size for multicast packets

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-getsockopt#toc26

func (*Socket) GetPlainPassword 

func (soc *Socket) GetPlainPassword() (string, error)

ZMQ_PLAIN_PASSWORD: Retrieve current password

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc24

func (*Socket) GetPlainServer 

func (soc *Socket) GetPlainServer() (int, error)

ZMQ_PLAIN_SERVER: Retrieve current PLAIN server role

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc25

func (*Socket) GetPlainUsername 

func (soc *Socket) GetPlainUsername() (string, error)

ZMQ_PLAIN_USERNAME: Retrieve current PLAIN username

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc26

func (*Socket) GetRate 

func (soc *Socket) GetRate() (int, error)

ZMQ_RATE: Retrieve multicast data rate

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc27

func (*Socket) GetRcvbuf 

func (soc *Socket) GetRcvbuf() (int, error)

ZMQ_RCVBUF: Retrieve kernel receive buffer size

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc28

func (*Socket) GetRcvhwm 

func (soc *Socket) GetRcvhwm() (int, error)

ZMQ_RCVHWM: Retrieve high water mark for inbound messages

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc29

func (*Socket) GetRcvmore 

func (soc *Socket) GetRcvmore() (bool, error)

ZMQ_RCVMORE: More message data parts to follow

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc30

func (*Socket) GetRcvtimeo 

func (soc *Socket) GetRcvtimeo() (time.Duration, error)

ZMQ_RCVTIMEO: Maximum time before a socket operation returns with EAGAIN

Returns time.Duration(-1) for infinite

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc31

func (*Socket) GetReconnectIvl 

func (soc *Socket) GetReconnectIvl() (time.Duration, error)

ZMQ_RECONNECT_IVL: Retrieve reconnection interval

Returns time.Duration(-1) for no reconnection

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc32

func (*Socket) GetReconnectIvlMax 

func (soc *Socket) GetReconnectIvlMax() (time.Duration, error)

ZMQ_RECONNECT_IVL_MAX: Retrieve maximum reconnection interval

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc33

func (*Socket) GetRecoveryIvl 

func (soc *Socket) GetRecoveryIvl() (time.Duration, error)

ZMQ_RECOVERY_IVL: Get multicast recovery interval

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc34

func (*Socket) GetSndbuf 

func (soc *Socket) GetSndbuf() (int, error)

ZMQ_SNDBUF: Retrieve kernel transmit buffer size

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc35

func (*Socket) GetSndhwm 

func (soc *Socket) GetSndhwm() (int, error)

ZMQ_SNDHWM: Retrieves high water mark for outbound messages

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc36

func (*Socket) GetSndtimeo 

func (soc *Socket) GetSndtimeo() (time.Duration, error)

ZMQ_SNDTIMEO: Maximum time before a socket operation returns with EAGAIN

Returns time.Duration(-1) for infinite

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc37

func (*Socket) GetSocksProxy 

func (soc *Socket) GetSocksProxy() (string, error)

ZMQ_SOCKS_PROXY: NOT DOCUMENTED

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

func (*Socket) GetTcpKeepalive 

func (soc *Socket) GetTcpKeepalive() (int, error)

ZMQ_TCP_KEEPALIVE: Override SO_KEEPALIVE socket option

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc38

func (*Socket) GetTcpKeepaliveCnt 

func (soc *Socket) GetTcpKeepaliveCnt() (int, error)

ZMQ_TCP_KEEPALIVE_CNT: Override TCP_KEEPCNT socket option

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc39

func (*Socket) GetTcpKeepaliveIdle 

func (soc *Socket) GetTcpKeepaliveIdle() (int, error)

ZMQ_TCP_KEEPALIVE_IDLE: Override TCP_KEEPCNT(or TCP_KEEPALIVE on some OS)

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc40

func (*Socket) GetTcpKeepaliveIntvl 

func (soc *Socket) GetTcpKeepaliveIntvl() (int, error)

ZMQ_TCP_KEEPALIVE_INTVL: Override TCP_KEEPINTVL socket option

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc41

func (*Socket) GetTcpMaxrt 

func (soc *Socket) GetTcpMaxrt() (time.Duration, error)

ZMQ_TCP_MAXRT: Retrieve Max TCP Retransmit Timeout

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-getsockopt#toc44

func (*Socket) GetThreadSafe 

func (soc *Socket) GetThreadSafe() (bool, error)

ZMQ_THREAD_SAFE: Retrieve socket thread safety

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-getsockopt#toc45

func (*Socket) GetTos 

func (soc *Socket) GetTos() (int, error)

ZMQ_TOS: Retrieve the Type-of-Service socket override status

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc42

func (*Socket) GetType 

func (soc *Socket) GetType() (Type, error)

ZMQ_TYPE: Retrieve socket type

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc43

func (*Socket) GetVmciBufferMaxSize 

func (soc *Socket) GetVmciBufferMaxSize() (uint64, error)

ZMQ_VMCI_BUFFER_MAX_SIZE: Retrieve max buffer size of the VMCI socket

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-getsockopt#toc51

func (*Socket) GetVmciBufferMinSize 

func (soc *Socket) GetVmciBufferMinSize() (uint64, error)

ZMQ_VMCI_BUFFER_MIN_SIZE: Retrieve min buffer size of the VMCI socket

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-getsockopt#toc50

func (*Socket) GetVmciBufferSize 

func (soc *Socket) GetVmciBufferSize() (uint64, error)

ZMQ_VMCI_BUFFER_SIZE: Retrieve buffer size of the VMCI socket

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-getsockopt#toc49

func (*Socket) GetVmciConnectTimeout 

func (soc *Socket) GetVmciConnectTimeout() (time.Duration, error)

ZMQ_VMCI_CONNECT_TIMEOUT: Retrieve connection timeout of the VMCI socket

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-getsockopt#toc52

func (*Socket) GetZapDomain 

func (soc *Socket) GetZapDomain() (string, error)

ZMQ_ZAP_DOMAIN: Retrieve RFC 27 authentication domain

See: http://api.zeromq.org/4-1:zmq-getsockopt#toc44

func (*Socket) Getusefd 

func (soc *Socket) Getusefd() (int, error)

ZMQ_USE_FD: Retrieve the pre-allocated socket file descriptor

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-getsockopt#toc29

func (*Socket) Monitor 

func (soc *Socket) Monitor(addr string, events Event) error

Register a monitoring callback.

See: http://api.zeromq.org/4-1:zmq-socket-monitor#toc2

WARNING: Closing a context with a monitoring callback will lead to random crashes. This is a bug in the ZeroMQ library. The monitoring callback has the same context as the socket it was created for.

Example: 

package main

import (
    zmq "github.com/pebbe/zmq4"
    "log"
    "time"
)

func rep_socket_monitor(addr string) {
    s, err := zmq.NewSocket(zmq.PAIR)
    if err != nil {
        log.Fatalln(err)
    }
    err = s.Connect(addr)
    if err != nil {
        log.Fatalln(err)
    }
    for {
        a, b, c, err := s.RecvEvent(0)
        if err != nil {
            log.Println(err)
            break
        }
        log.Println(a, b, c)
    }
    s.Close()
}

func main() {

    // REP socket
    rep, err := zmq.NewSocket(zmq.REP)
    if err != nil {
        log.Fatalln(err)
    }

    // REP socket monitor, all events
    err = rep.Monitor("inproc://monitor.rep", zmq.EVENT_ALL)
    if err != nil {
        log.Fatalln(err)
    }
    go rep_socket_monitor("inproc://monitor.rep")

    // Generate an event
    rep.Bind("tcp://*:5555")
    if err != nil {
        log.Fatalln(err)
    }

    // Allow some time for event detection
    time.Sleep(time.Second)

    rep.Close()
    zmq.Term()
}

func (*Socket) Recv 

func (soc *Socket) Recv(flags Flag) (string, error)

Receive a message part from a socket.

For a description of flags, see: http://api.zeromq.org/4-1:zmq-msg-recv#toc2

func (*Socket) RecvBytes 

func (soc *Socket) RecvBytes(flags Flag) ([]byte, error)

Receive a message part from a socket.

For a description of flags, see: http://api.zeromq.org/4-1:zmq-msg-recv#toc2

func (*Socket) RecvBytesWithMetadata 

func (soc *Socket) RecvBytesWithMetadata(flags Flag, properties ...string) (msg []byte, metadata map[string]string, err error)

Receive a message part with metadata.

This requires ZeroMQ version 4.1.0. Lower versions will return the message part without metadata.

The returned metadata map contains only those properties that exist on the message.

For a description of flags, see: http://api.zeromq.org/4-1:zmq-msg-recv#toc2

For a description of metadata, see: http://api.zeromq.org/4-1:zmq-msg-gets#toc3

func (*Socket) RecvEvent 

func (soc *Socket) RecvEvent(flags Flag) (event_type Event, addr string, value int, err error)

Receive a message part from a socket interpreted as an event.

For a description of flags, see: http://api.zeromq.org/4-1:zmq-msg-recv#toc2

For a description of event_type, see: http://api.zeromq.org/4-1:zmq-socket-monitor#toc3

For an example, see: func (*Socket) Monitor

func (*Socket) RecvMessage 

func (soc *Socket) RecvMessage(flags Flag) (msg []string, err error)

Receive parts as message from socket.

Returns last non-nil error code.

func (*Socket) RecvMessageBytes 

func (soc *Socket) RecvMessageBytes(flags Flag) (msg [][]byte, err error)

Receive parts as message from socket.

Returns last non-nil error code.

func (*Socket) RecvMessageBytesWithMetadata 

func (soc *Socket) RecvMessageBytesWithMetadata(flags Flag, properties ...string) (msg [][]byte, metadata map[string]string, err error)

Receive parts as message from socket, including metadata.

Metadata is picked from the first message part.

For details about metadata, see RecvBytesWithMetadata().

Returns last non-nil error code.

func (*Socket) RecvMessageWithMetadata 

func (soc *Socket) RecvMessageWithMetadata(flags Flag, properties ...string) (msg []string, metadata map[string]string, err error)

Receive parts as message from socket, including metadata.

Metadata is picked from the first message part.

For details about metadata, see RecvWithMetadata().

Returns last non-nil error code.

func (*Socket) RecvWithMetadata 

func (soc *Socket) RecvWithMetadata(flags Flag, properties ...string) (msg string, metadata map[string]string, err error)

Receive a message part with metadata.

This requires ZeroMQ version 4.1.0. Lower versions will return the message part without metadata.

The returned metadata map contains only those properties that exist on the message.

For a description of flags, see: http://api.zeromq.org/4-1:zmq-msg-recv#toc2

For a description of metadata, see: http://api.zeromq.org/4-1:zmq-msg-gets#toc3

func (*Socket) Send 

func (soc *Socket) Send(data string, flags Flag) (int, error)

Send a message part on a socket.

For a description of flags, see: http://api.zeromq.org/4-1:zmq-send#toc2

func (*Socket) SendBytes 

func (soc *Socket) SendBytes(data []byte, flags Flag) (int, error)

Send a message part on a socket.

For a description of flags, see: http://api.zeromq.org/4-1:zmq-send#toc2

func (*Socket) SendMessage 

func (soc *Socket) SendMessage(parts ...interface{}) (total int, err error)

Send multi-part message on socket.

Any `[]string' or `[][]byte' is split into separate `string's or `[]byte's

Any other part that isn't a `string' or `[]byte' is converted to `string' with `fmt.Sprintf("%v", part)'.

Returns total bytes sent.

func (*Socket) SendMessageDontwait 

func (soc *Socket) SendMessageDontwait(parts ...interface{}) (total int, err error)

Like SendMessage(), but adding the DONTWAIT flag.

func (*Socket) ServerAuthCurve 

func (server *Socket) ServerAuthCurve(domain, secret_key string) error

Set CURVE server role.

func (*Socket) ServerAuthNull 

func (server *Socket) ServerAuthNull(domain string) error

Set NULL server role.

func (*Socket) ServerAuthPlain 

func (server *Socket) ServerAuthPlain(domain string) error

Set PLAIN server role.

func (*Socket) SetAffinity 

func (soc *Socket) SetAffinity(value uint64) error

ZMQ_AFFINITY: Set I/O thread affinity

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc3

func (*Socket) SetBacklog 

func (soc *Socket) SetBacklog(value int) error

ZMQ_BACKLOG: Set maximum length of the queue of outstanding connections

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc4

func (*Socket) SetConflate 

func (soc *Socket) SetConflate(value bool) error

ZMQ_CONFLATE: Keep only last message

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc6

func (*Socket) SetConnectRid 

func (soc *Socket) SetConnectRid(value string) error

ZMQ_CONNECT_RID: Assign the next outbound connection id

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc5

func (*Socket) SetConnectTimeout 

func (soc *Socket) SetConnectTimeout(value time.Duration) error

ZMQ_CONNECT_TIMEOUT: Set connect() timeout

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc7

func (*Socket) SetCurvePublickey 

func (soc *Socket) SetCurvePublickey(key string) error

ZMQ_CURVE_PUBLICKEY: Set CURVE public key

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc7

func (*Socket) SetCurveSecretkey 

func (soc *Socket) SetCurveSecretkey(key string) error

ZMQ_CURVE_SECRETKEY: Set CURVE secret key

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc8

func (*Socket) SetCurveServer 

func (soc *Socket) SetCurveServer(value int) error

ZMQ_CURVE_SERVER: Set CURVE server role

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc9

func (*Socket) SetCurveServerkey 

func (soc *Socket) SetCurveServerkey(key string) error

ZMQ_CURVE_SERVERKEY: Set CURVE server key

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc10

func (*Socket) SetGssapiPlaintext 

func (soc *Socket) SetGssapiPlaintext(value bool) error

ZMQ_GSSAPI_PLAINTEXT: Disable GSSAPI encryption

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc11

func (*Socket) SetGssapiPrincipal 

func (soc *Socket) SetGssapiPrincipal(value string) error

ZMQ_GSSAPI_PRINCIPAL: Set name of GSSAPI principal

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc12

func (*Socket) SetGssapiServer 

func (soc *Socket) SetGssapiServer(value bool) error

ZMQ_GSSAPI_SERVER: Set GSSAPI server role

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc13

func (*Socket) SetGssapiServicePrincipal 

func (soc *Socket) SetGssapiServicePrincipal(value string) error

ZMQ_GSSAPI_SERVICE_PRINCIPAL: Set name of GSSAPI service principal

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc14

func (*Socket) SetHandshakeIvl 

func (soc *Socket) SetHandshakeIvl(value time.Duration) error

ZMQ_HANDSHAKE_IVL: Set maximum handshake interval

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc15

func (*Socket) SetHeartbeatIvl 

func (soc *Socket) SetHeartbeatIvl(value time.Duration) error

ZMQ_HEARTBEAT_IVL: Set interval between sending ZMTP heartbeats

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc17

func (*Socket) SetHeartbeatTimeout 

func (soc *Socket) SetHeartbeatTimeout(value time.Duration) error

ZMQ_HEARTBEAT_TIMEOUT: Set timeout for ZMTP heartbeats

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc18

func (*Socket) SetHeartbeatTtl 

func (soc *Socket) SetHeartbeatTtl(value time.Duration) error

ZMQ_HEARTBEAT_TTL: Set the TTL value for ZMTP heartbeats

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc19

func (*Socket) SetIdentity 

func (soc *Socket) SetIdentity(value string) error

ZMQ_IDENTITY: Set socket identity

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc16

func (*Socket) SetImmediate 

func (soc *Socket) SetImmediate(value bool) error

ZMQ_IMMEDIATE: Queue messages only to completed connections

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc17

func (*Socket) SetInvertMatching 

func (soc *Socket) SetInvertMatching(value int) error

ZMQ_INVERT_MATCHING: Invert message filtering

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc22

func (*Socket) SetIpv6 

func (soc *Socket) SetIpv6(value bool) error

ZMQ_IPV6: Enable IPv6 on socket

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc18

func (*Socket) SetLinger 

func (soc *Socket) SetLinger(value time.Duration) error

ZMQ_LINGER: Set linger period for socket shutdown

For infinite, use -1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc19

func (*Socket) SetMaxmsgsize 

func (soc *Socket) SetMaxmsgsize(value int64) error

ZMQ_MAXMSGSIZE: Maximum acceptable inbound message size

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc20

func (*Socket) SetMulticastHops 

func (soc *Socket) SetMulticastHops(value int) error

ZMQ_MULTICAST_HOPS: Maximum network hops for multicast packets

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc21

func (*Socket) SetMulticastMaxtpdu 

func (soc *Socket) SetMulticastMaxtpdu(value int) error

ZMQ_MULTICAST_MAXTPDU: Maximum transport data unit size for multicast packets

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc27

func (*Socket) SetPlainPassword 

func (soc *Socket) SetPlainPassword(password string) error

ZMQ_PLAIN_PASSWORD: Set PLAIN security password

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc22

func (*Socket) SetPlainServer 

func (soc *Socket) SetPlainServer(value int) error

ZMQ_PLAIN_SERVER: Set PLAIN server role

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc23

func (*Socket) SetPlainUsername 

func (soc *Socket) SetPlainUsername(username string) error

ZMQ_PLAIN_USERNAME: Set PLAIN security username

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc24

func (*Socket) SetProbeRouter 

func (soc *Socket) SetProbeRouter(value int) error

ZMQ_PROBE_ROUTER: bootstrap connections to ROUTER sockets

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc25

func (*Socket) SetRate 

func (soc *Socket) SetRate(value int) error

ZMQ_RATE: Set multicast data rate

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc26

func (*Socket) SetRcvbuf 

func (soc *Socket) SetRcvbuf(value int) error

ZMQ_RCVBUF: Set kernel receive buffer size

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc27

func (*Socket) SetRcvhwm 

func (soc *Socket) SetRcvhwm(value int) error

ZMQ_RCVHWM: Set high water mark for inbound messages

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc28

func (*Socket) SetRcvtimeo 

func (soc *Socket) SetRcvtimeo(value time.Duration) error

ZMQ_RCVTIMEO: Maximum time before a recv operation returns with EAGAIN

For infinite, use -1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc29

func (*Socket) SetReconnectIvl 

func (soc *Socket) SetReconnectIvl(value time.Duration) error

ZMQ_RECONNECT_IVL: Set reconnection interval

For no reconnection, use -1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc30

func (*Socket) SetReconnectIvlMax 

func (soc *Socket) SetReconnectIvlMax(value time.Duration) error

ZMQ_RECONNECT_IVL_MAX: Set maximum reconnection interval

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc31

func (*Socket) SetRecoveryIvl 

func (soc *Socket) SetRecoveryIvl(value time.Duration) error

ZMQ_RECOVERY_IVL: Set multicast recovery interval

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc32

func (*Socket) SetReqCorrelate 

func (soc *Socket) SetReqCorrelate(value int) error

ZMQ_REQ_CORRELATE: match replies with requests

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc33

func (*Socket) SetReqRelaxed 

func (soc *Socket) SetReqRelaxed(value int) error

ZMQ_REQ_RELAXED: relax strict alternation between request and reply

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc34

func (*Socket) SetRouterHandover 

func (soc *Socket) SetRouterHandover(value bool) error

ZMQ_ROUTER_HANDOVER: handle duplicate client identities on ROUTER sockets

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc35

func (*Socket) SetRouterMandatory 

func (soc *Socket) SetRouterMandatory(value int) error

ZMQ_ROUTER_MANDATORY: accept only routable messages on ROUTER sockets

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc36

func (*Socket) SetRouterRaw 

func (soc *Socket) SetRouterRaw(value int) error

ZMQ_ROUTER_RAW: switch ROUTER socket to raw mode

This option is deprecated since ZeroMQ version 4.1, please use ZMQ_STREAM sockets instead.

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc37

func (*Socket) SetSndbuf 

func (soc *Socket) SetSndbuf(value int) error

ZMQ_SNDBUF: Set kernel transmit buffer size

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc38

func (*Socket) SetSndhwm 

func (soc *Socket) SetSndhwm(value int) error

ZMQ_SNDHWM: Set high water mark for outbound messages

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc39

func (*Socket) SetSndtimeo 

func (soc *Socket) SetSndtimeo(value time.Duration) error

ZMQ_SNDTIMEO: Maximum time before a send operation returns with EAGAIN

For infinite, use -1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc40

func (*Socket) SetSocksProxy 

func (soc *Socket) SetSocksProxy(value string) error

ZMQ_SOCKS_PROXY: NOT DOCUMENTED

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

func (*Socket) SetStreamNotify 

func (soc *Socket) SetStreamNotify(value int) error

ZMQ_STREAM_NOTIFY: send connect and disconnect notifications

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc48

func (*Socket) SetSubscribe 

func (soc *Socket) SetSubscribe(filter string) error

ZMQ_SUBSCRIBE: Establish message filter

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc41

func (*Socket) SetTcpAcceptFilter 

func (soc *Socket) SetTcpAcceptFilter(filter string) error

ZMQ_TCP_ACCEPT_FILTER: Assign filters to allow new TCP connections

This option is deprecated since ZeroMQ version 4.1, please use authentication via the ZAP API and IP address whitelisting / blacklisting.

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc50

func (*Socket) SetTcpKeepalive 

func (soc *Socket) SetTcpKeepalive(value int) error

ZMQ_TCP_KEEPALIVE: Override SO_KEEPALIVE socket option

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc42

func (*Socket) SetTcpKeepaliveCnt 

func (soc *Socket) SetTcpKeepaliveCnt(value int) error

ZMQ_TCP_KEEPALIVE_CNT: Override TCP_KEEPCNT socket option

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc43

func (*Socket) SetTcpKeepaliveIdle 

func (soc *Socket) SetTcpKeepaliveIdle(value int) error

ZMQ_TCP_KEEPALIVE_IDLE: Override TCP_KEEPCNT(or TCP_KEEPALIVE on some OS)

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc44

func (*Socket) SetTcpKeepaliveIntvl 

func (soc *Socket) SetTcpKeepaliveIntvl(value int) error

ZMQ_TCP_KEEPALIVE_INTVL: Override TCP_KEEPINTVL socket option

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc45

func (*Socket) SetTcpMaxrt 

func (soc *Socket) SetTcpMaxrt(value time.Duration) error

ZMQ_TCP_MAXRT: Set TCP Maximum Retransmit Timeout

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc54

func (*Socket) SetTos 

func (soc *Socket) SetTos(value int) error

ZMQ_TOS: Set the Type-of-Service on socket

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc46

func (*Socket) SetUnsubscribe 

func (soc *Socket) SetUnsubscribe(filter string) error

ZMQ_UNSUBSCRIBE: Remove message filter

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc47

func (*Socket) SetUseFd 

func (soc *Socket) SetUseFd(value int) error

ZMQ_USE_FD: Set the pre-allocated socket file descriptor

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc31

func (*Socket) SetVmciBufferMaxSize 

func (soc *Socket) SetVmciBufferMaxSize(value uint64) error

ZMQ_VMCI_BUFFER_MAX_SIZE: Set max buffer size of the VMCI socket

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc70

func (*Socket) SetVmciBufferMinSize 

func (soc *Socket) SetVmciBufferMinSize(value uint64) error

ZMQ_VMCI_BUFFER_MIN_SIZE: Set min buffer size of the VMCI socket

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc69

func (*Socket) SetVmciBufferSize 

func (soc *Socket) SetVmciBufferSize(value uint64) error

ZMQ_VMCI_BUFFER_SIZE: Set buffer size of the VMCI socket

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc68

func (*Socket) SetVmciConnectTimeout 

func (soc *Socket) SetVmciConnectTimeout(value time.Duration) error

ZMQ_VMCI_CONNECT_TIMEOUT: Set connection timeout of the VMCI socket

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc71

func (*Socket) SetXpubManual 

func (soc *Socket) SetXpubManual(value int) error

ZMQ_XPUB_MANUAL: change the subscription handling to manual

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc59

func (*Socket) SetXpubNodrop 

func (soc *Socket) SetXpubNodrop(value bool) error

ZMQ_XPUB_NODROP: do not silently drop messages if SENDHWM is reached

Returns ErrorNotImplemented41 with ZeroMQ version < 4.1

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc60

func (*Socket) SetXpubVerbose 

func (soc *Socket) SetXpubVerbose(value int) error

ZMQ_XPUB_VERBOSE: provide all subscription messages on XPUB sockets

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc48

func (*Socket) SetXpubVerboser 

func (soc *Socket) SetXpubVerboser(value int) error

ZMQ_XPUB_VERBOSER: pass subscribe and unsubscribe messages on XPUB socket

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc58

func (*Socket) SetXpubWelcomeMsg 

func (soc *Socket) SetXpubWelcomeMsg(value string) error

ZMQ_XPUB_WELCOME_MSG: set welcome message that will be received by subscriber when connecting

Returns ErrorNotImplemented42 with ZeroMQ version < 4.2

See: http://api.zeromq.org/4-2:zmq-setsockopt#toc61

func (*Socket) SetZapDomain 

func (soc *Socket) SetZapDomain(domain string) error

ZMQ_ZAP_DOMAIN: Set RFC 27 authentication domain

See: http://api.zeromq.org/4-1:zmq-setsockopt#toc49

func (Socket) String 

func (soc Socket) String() string

Socket as string.

func (*Socket) Unbind 

func (soc *Socket) Unbind(endpoint string) error

Stop accepting connections on a socket.

For a description of endpoint, see: http://api.zeromq.org/4-1:zmq-unbind#toc2

type State 

type State int

Used by (soc *Socket)GetEvents()

func (State) String 

func (s State) String() string

Socket state as string.

type Type 

type Type int

Specifies the type of a socket, used by NewSocket()

func (Type) String 

func (t Type) String() string

Socket type as string.

Source Files

View all Source files

Directories

Path Synopsis
A Go interface to ZeroMQ (zmq, 0mq) version 4.
 A Go interface to ZeroMQ (zmq, 0mq) version 4.
examples module

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.