amqpwrap

Documentation

Overview

Package amqpwrap has some simple wrappers to make it easier to abstract the go-amqp types.

Index

• type AMQPClient
• type AMQPClientWrapper
  • func (w *AMQPClientWrapper) Close() error
  • func (w *AMQPClientWrapper) NewSession(ctx context.Context, opts *amqp.SessionOptions) (AMQPSession, error)
• type AMQPReceiver
• type AMQPReceiverCloser
• type AMQPReceiverWrapper
  • func (rw *AMQPReceiverWrapper) AcceptMessage(ctx context.Context, msg *amqp.Message) error
  • func (rw *AMQPReceiverWrapper) Close(ctx context.Context) error
  • func (rw *AMQPReceiverWrapper) Credits() uint32
  • func (rw *AMQPReceiverWrapper) IssueCredit(credit uint32) error
  • func (rw *AMQPReceiverWrapper) LinkName() string
  • func (rw *AMQPReceiverWrapper) LinkSourceFilterValue(name string) interface{}
  • func (rw *AMQPReceiverWrapper) ModifyMessage(ctx context.Context, msg *amqp.Message, options *amqp.ModifyMessageOptions) error
  • func (rw *AMQPReceiverWrapper) Prefetched() *amqp.Message
  • func (rw *AMQPReceiverWrapper) Receive(ctx context.Context) (*amqp.Message, error)
  • func (rw *AMQPReceiverWrapper) RejectMessage(ctx context.Context, msg *amqp.Message, e *amqp.Error) error
  • func (rw *AMQPReceiverWrapper) ReleaseMessage(ctx context.Context, msg *amqp.Message) error
• type AMQPSender
• type AMQPSenderCloser
• type AMQPSession
• type AMQPSessionWrapper
  • func (w *AMQPSessionWrapper) Close(ctx context.Context) error
  • func (w *AMQPSessionWrapper) NewReceiver(ctx context.Context, source string, opts *amqp.ReceiverOptions) (AMQPReceiverCloser, error)
  • func (w *AMQPSessionWrapper) NewSender(ctx context.Context, target string, opts *amqp.SenderOptions) (AMQPSenderCloser, error)

Types

type AMQPClient

type AMQPClient interface {
	Close() error
	NewSession(ctx context.Context, opts *amqp.SessionOptions) (AMQPSession, error)
}

type AMQPClientWrapper

type AMQPClientWrapper struct {
	Inner *amqp.Client
}

AMQPClientWrapper is a simple interface, implemented by *AMQPClientWrapper It exists only so we can return AMQPSession, which itself only exists so we can return interfaces for AMQPSender and AMQPReceiver from AMQPSession.

func (*AMQPClientWrapper) Close

func (w *AMQPClientWrapper) Close() error

func (*AMQPClientWrapper) NewSession

func (w *AMQPClientWrapper) NewSession(ctx context.Context, opts *amqp.SessionOptions) (AMQPSession, error)

type AMQPReceiver

type AMQPReceiver interface {
	IssueCredit(credit uint32) error
	Receive(ctx context.Context) (*amqp.Message, error)
	Prefetched() *amqp.Message

	// settlement functions
	AcceptMessage(ctx context.Context, msg *amqp.Message) error
	RejectMessage(ctx context.Context, msg *amqp.Message, e *amqp.Error) error
	ReleaseMessage(ctx context.Context, msg *amqp.Message) error
	ModifyMessage(ctx context.Context, msg *amqp.Message, options *amqp.ModifyMessageOptions) error

	LinkName() string
	LinkSourceFilterValue(name string) interface{}

	// Credits returns the # of credits still active on this link.
	Credits() uint32
}

AMQPReceiver is implemented by *amqp.Receiver

type AMQPReceiverCloser

type AMQPReceiverCloser interface {
	AMQPReceiver
	Close(ctx context.Context) error
}

AMQPReceiverCloser is implemented by *amqp.Receiver

type AMQPReceiverWrapper

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

func (*AMQPReceiverWrapper) AcceptMessage

func (rw *AMQPReceiverWrapper) AcceptMessage(ctx context.Context, msg *amqp.Message) error

settlement functions

func (*AMQPReceiverWrapper) Close

func (rw *AMQPReceiverWrapper) Close(ctx context.Context) error

func (*AMQPReceiverWrapper) Credits

func (rw *AMQPReceiverWrapper) Credits() uint32

func (*AMQPReceiverWrapper) IssueCredit

func (rw *AMQPReceiverWrapper) IssueCredit(credit uint32) error

func (*AMQPReceiverWrapper) LinkName

func (rw *AMQPReceiverWrapper) LinkName() string

func (*AMQPReceiverWrapper) LinkSourceFilterValue

func (rw *AMQPReceiverWrapper) LinkSourceFilterValue(name string) interface{}

func (*AMQPReceiverWrapper) ModifyMessage

func (rw *AMQPReceiverWrapper) ModifyMessage(ctx context.Context, msg *amqp.Message, options *amqp.ModifyMessageOptions) error

func (*AMQPReceiverWrapper) Prefetched

func (rw *AMQPReceiverWrapper) Prefetched() *amqp.Message

func (*AMQPReceiverWrapper) Receive

func (rw *AMQPReceiverWrapper) Receive(ctx context.Context) (*amqp.Message, error)

func (*AMQPReceiverWrapper) RejectMessage

func (rw *AMQPReceiverWrapper) RejectMessage(ctx context.Context, msg *amqp.Message, e *amqp.Error) error

func (*AMQPReceiverWrapper) ReleaseMessage

func (rw *AMQPReceiverWrapper) ReleaseMessage(ctx context.Context, msg *amqp.Message) error

type AMQPSender

type AMQPSender interface {
	Send(ctx context.Context, msg *amqp.Message) error
	MaxMessageSize() uint64
	LinkName() string
}

AMQPSender is implemented by *amqp.Sender

type AMQPSenderCloser

type AMQPSenderCloser interface {
	AMQPSender
	Close(ctx context.Context) error
}

AMQPSenderCloser is implemented by *amqp.Sender

type AMQPSession

type AMQPSession interface {
	Close(ctx context.Context) error
	NewReceiver(ctx context.Context, source string, opts *amqp.ReceiverOptions) (AMQPReceiverCloser, error)
	NewSender(ctx context.Context, target string, opts *amqp.SenderOptions) (AMQPSenderCloser, error)
}

AMQPSession is a simple interface, implemented by *AMQPSessionWrapper. It exists only so we can return AMQPReceiver/AMQPSender interfaces.

type AMQPSessionWrapper

type AMQPSessionWrapper struct {
	Inner *amqp.Session
}

func (*AMQPSessionWrapper) Close

func (w *AMQPSessionWrapper) Close(ctx context.Context) error

func (*AMQPSessionWrapper) NewReceiver

func (w *AMQPSessionWrapper) NewReceiver(ctx context.Context, source string, opts *amqp.ReceiverOptions) (AMQPReceiverCloser, error)

func (*AMQPSessionWrapper) NewSender

func (w *AMQPSessionWrapper) NewSender(ctx context.Context, target string, opts *amqp.SenderOptions) (AMQPSenderCloser, error)