stream

Documentation

Overview

Package stream implements a generic reliable, ordered, streaming client-server protocol over AMQP.

Index

• func Dial(ctx context.Context, urlString string, option ...Option) (nc net.Conn, err error)
• func Listen(ctx context.Context, urlString string, option ...Option) (l net.Listener, err error)
• type Addr
  • func NewAddr(urlString string) (*Addr, error)
  • func (a *Addr) Network() string
  • func (a *Addr) ServerQueueName() (string, error)
  • func (a *Addr) String() string
• type Connection
  • func Connect(ctx context.Context, serverURL string, option ...Option) (*Connection, error)
  • func (c *Connection) Addr() *Addr
  • func (c *Connection) Close() error
  • func (c *Connection) Dial(ctx context.Context, serverQueueName string) (nc net.Conn, err error)
  • func (c *Connection) IsClosed() bool
  • func (c *Connection) Listen(ctx context.Context, serverQueueName string) (l net.Listener, err error)
• type Option
  • func WithExternalAuth() Option
  • func WithInsecure() Option
  • func WithTLSConfig(config *tls.Config) Option

Functions

func Dial

func Dial(
	ctx context.Context, urlString string, option ...Option,
) (nc net.Conn, err error)

Dial establishes a connection to an AMQP stream server. The URL should be a standard amqp(s) URL, which, in addition, must have a server_queue parameter set to the server control queue name. If tlsConfig is nil but a secure connection was requested, an empty config with the server name taken from the URL will be used.

Dial is thus a combination of Connect, followed by Connection.Dial. It is a useful shortcut if there is only a single AMQP client stream.

func Listen

func Listen(
	ctx context.Context, urlString string, option ...Option,
) (l net.Listener, err error)

Listen creates an AMQP stream server listener. It is a combination of calls to Connect and Connect.Listen.

The given context only pertains to creation of the listener. Once Listen returns, the context can be safely cancelled without impairing the operation of the returned listener.

The URL should be a standard amqp(s) URL, which, in addition, must have a server_queue parameter set to the server control queue name.

Types

type Addr

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

Addr is an AMQP stream address.

func NewAddr

func NewAddr(urlString string) (*Addr, error)

NewAddr creates a new address from the given URL string.

func (*Addr) Network

func (a *Addr) Network() string

Network implements net.Addr.Network.

func (*Addr) ServerQueueName

func (a *Addr) ServerQueueName() (string, error)

ServerQueueName returns the server queue name. If this address does not have a server queue name, a net.InvalidAddrError is returned instead.

func (*Addr) String

func (a *Addr) String() string

String implements net.Addr.String.

type Connection

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

Connection is an AMQP connection underlying a listener or a client.

func Connect

func Connect(
	ctx context.Context, serverURL string, option ...Option,
) (*Connection, error)

Connect connects to an AMQP server without starting a Listener or dialling to establish an AMQP stream yet. This is useful if multiple servers and/or clients should use the same AMQP connection.

The URL should be a standard amqp(s) URL.

func (*Connection) Addr

func (c *Connection) Addr() *Addr

Addr returns the address of this connection.

func (*Connection) Close

func (c *Connection) Close() error

Close closes this connections. This will cause any remaining AMQP streams to be severed, their Accept, Close, Read, and Write methods returning errors.

func (*Connection) Dial

func (c *Connection) Dial(
	ctx context.Context, serverQueueName string,
) (nc net.Conn, err error)

Dial establishes an AMQP client stream on this connection. The serverQueueName is the server control queue name.

func (*Connection) IsClosed

func (c *Connection) IsClosed() bool

IsClosed reports whether this connection is closed (either via a call to c.Close or due to some error).

func (*Connection) Listen

func (c *Connection) Listen(
	ctx context.Context, serverQueueName string,
) (l net.Listener, err error)

Listen creates a new listener on the given server queue using this connection.

type Option

type Option func(o *opts) error

Option is an AMQP stream option. Options are returned by the With… functions.

func WithExternalAuth

func WithExternalAuth() Option

WithExternalAuth enables external authentication (e. g., via TLS credentials) with the AMQP server. By default, plain authentication is used.

func WithInsecure

func WithInsecure() Option

WithInsecure indicates that no TLS configuration should be used when connecting to the AMQP server. This means either a plaintext connection (for amqp:// URLs) or server verification only (for amqps:// URLs).

func WithTLSConfig

func WithTLSConfig(config *tls.Config) Option

WithTLSConfig sets the TLS configuration for the connection to the AMQP server. After a call to WithTLSConfig, config must not be changed.