gomq

README

gomq

Sockets with super-powers

Motivation

ZeroMQ is a powerful communications library, making robust networking simple (retry logic, socket patterns, multiple bind/connect per socket, etc.) Go is a powerful programming language, and having go libraries written is pure Go makes it super easy to cross compile for other platforms and operating systems. Go also provides a powerful asynchronous networking library which is built seamlessly into the language (all in the net package.) This would make a ZMQ implementation in pure Go a powerful library.

Why build a new one when others are already out there?

1. It's fun

What sets this implementation apart?

1. Highly modularized. Implementing new socket types, transports, mechanisms, etc. is super simple. In fact, they need not be implemented in this repository as the code follows a dependency injection structure.
2. Support for CurveZMQ which is not found in any other Go implementation (to my searching.)
3. Tiny. This implementation does not aim to follow the same code structure as the core C++ ZMQ implementation but rather attempts to achieve a similar feature set.

Documentation

Index

• Constants
• Variables
• func FindMechanism(name string) (func() zmtp.Mechanism, bool)
• func RegisterMechanism(name string, mech func() zmtp.Mechanism) error
• func RegisterSocketType(name string, constructor SocketConstructor) error
• func RegisterTransport(name string, fac TransportFactory) error
• type Config
  • func (c *Config) ConnectTimeout() time.Duration
  • func (c *Config) Default()
  • func (c *Config) QueueLen() int
  • func (c *Config) ReconnectTimeout() time.Duration
  • func (c *Config) SetConnectTimeout(d time.Duration)
  • func (c *Config) SetQueueLen(queueLen int)
  • func (c *Config) SetReconnectTimeout(d time.Duration)
• type Context
  • func NewContext(ctx context.Context) *Context
  • func (c *Context) NewSocket(typ string, mechStr string) (*Socket, error)
• type Event
• type EventBus
• type EventType
  • func (e EventType) String() string
• type PrintBus
  • func (PrintBus) Post(ev Event)
• type Socket
  • func (s Socket) Bind(addr string) error
  • func (s Socket) Close() error
  • func (s Socket) Connect(addr string) error
  • func (s Socket) Disconnect(addr string) error
  • func (s Socket) Recv() ([][]byte, error)
  • func (s Socket) Send(data [][]byte) error
  • func (s Socket) SetOption(option string, val any) error
  • func (s Socket) SetServer(serv bool) error
  • func (s Socket) Unbind(addr string) error
• type SocketConstructor
  • func FindSocketType(name string) (SocketConstructor, bool)
• type SocketDriver
• type TransportFactory

Constants

const (
	EventTypeConnected       = EventType(0)
	EventTypeDisconnected    = EventType(1)
	EventTypeConnectFailed   = EventType(2)
	EventTypeAccepted        = EventType(3)
	EventTypeAcceptFailed    = EventType(4)
	EventTypeFailedGreeting  = EventType(5)
	EventTypeFailedHandshake = EventType(6)
	EventTypeReady           = EventType(7)
)

Variables

var ErrMechanismExists mechanismExists

var ErrMechanismNotFound mechanismNotFound

var ErrTransportExists transportExists

var ErrTransportNotFound transportNotFound

var ErrTypeExists typeExists

var ErrTypeNotFound typeNotFound

Functions

func FindMechanism

func FindMechanism(name string) (func() zmtp.Mechanism, bool)

func RegisterMechanism

func RegisterMechanism(name string, mech func() zmtp.Mechanism) error

func RegisterSocketType

func RegisterSocketType(
	name string,
	constructor SocketConstructor,
) error

func RegisterTransport

func RegisterTransport(name string, fac TransportFactory) error

Types

type Config

type Config struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func (*Config) ConnectTimeout

func (c *Config) ConnectTimeout() time.Duration

func (*Config) Default

func (c *Config) Default()

func (*Config) QueueLen

func (c *Config) QueueLen() int

func (*Config) ReconnectTimeout

func (c *Config) ReconnectTimeout() time.Duration

func (*Config) SetConnectTimeout

func (c *Config) SetConnectTimeout(d time.Duration)

func (*Config) SetQueueLen

func (c *Config) SetQueueLen(queueLen int)

func (*Config) SetReconnectTimeout

func (c *Config) SetReconnectTimeout(d time.Duration)

type Context

type Context struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewContext

func NewContext(ctx context.Context) *Context

func (*Context) NewSocket

func (c *Context) NewSocket(typ string, mechStr string) (*Socket, error)

type Event

type Event struct {
	EventType
	LocalAddr  string
	RemoteAddr string
	Notes      string
}

type EventBus

type EventBus interface {
	Post(Event)
}

type EventType

type EventType int

func (EventType) String

func (e EventType) String() string

type PrintBus

type PrintBus struct{}

func (PrintBus) Post

func (PrintBus) Post(ev Event)

type Socket

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

func (Socket) Bind

func (s Socket) Bind(addr string) error

func (Socket) Close

func (s Socket) Close() error

func (Socket) Connect

func (s Socket) Connect(addr string) error

func (Socket) Disconnect

func (s Socket) Disconnect(addr string) error

func (Socket) Recv

func (s Socket) Recv() ([][]byte, error)

func (Socket) Send

func (s Socket) Send(data [][]byte) error

func (Socket) SetOption

func (s Socket) SetOption(option string, val any) error

func (Socket) SetServer

func (s Socket) SetServer(serv bool) error

func (Socket) Unbind

func (s Socket) Unbind(addr string) error

type SocketConstructor

type SocketConstructor func(
	ctx context.Context,
	mech zmtp.Mechanism,
	conf *Config,
	eventBus EventBus,
) (SocketDriver, error)

SocketConstructor constructs a socket.

func FindSocketType

func FindSocketType(name string) (SocketConstructor, bool)

type SocketDriver

type SocketDriver interface {
	// Name of the type.
	Name() string

	// Connect to the remote address using the given transport.
	Connect(tp transport.Transport, url *url.URL) error

	// Disconnect from the address.
	Disconnect(url *url.URL) error

	// Bind to the given address using the given transport.
	Bind(tp transport.Transport, url *url.URL) error

	// Unbind from the bound address.
	Unbind(url *url.URL) error

	// Send a message over the socket.
	Send([]zmtp.Message) error

	// Recv either a command or a message on the socket.
	Recv() ([]zmtp.Message, error)

	// Close the socket
	Close() error
}

SocketDriver represents a type of socket in a communitcation pattern.

type TransportFactory

type TransportFactory func() transport.Transport

TransportFactory creates a transport given a list of key value pairs.