koinosmq

README

Koinos MQ Golang

Golang wrapper to interact with AMQP

Documentation

Index

• Constants
• type BroadcastHandlerFunc
• type CheckRetryResult
• type Client
  • func NewClient(addr string, rpcRetryPolicy RetryPolicy) *Client
  • func (c *Client) Broadcast(ctx context.Context, contentType ContentType, topic string, args []byte) error
  • func (c *Client) RPC(ctx context.Context, contentType ContentType, rpcService string, args []byte) ([]byte, error)
  • func (c *Client) SetNumConsumers(rpcReturnNumConsumers int)
  • func (c *Client) Start(ctx context.Context) <-chan struct{}
• type ContentType
• type ExponentialBackoffOptions
• type RPCCallResult
• type RPCHandlerFunc
• type RequestHandler
  • func NewRequestHandler(addr string, consumers uint, replyRetryPolicy RetryPolicy) *RequestHandler
  • func (r *RequestHandler) SetBroadcastHandler(topic string, handler BroadcastHandlerFunc)
  • func (r *RequestHandler) SetRPCHandler(rpcType string, handler RPCHandlerFunc)
  • func (r *RequestHandler) Start(ctx context.Context) <-chan struct{}
• type RetryPolicy

Constants

const (
	OctetStream = "application/octet-stream"
)

Types

type BroadcastHandlerFunc

type BroadcastHandlerFunc = func(topic string, data []byte)

BroadcastHandlerFunc Function type to handle a broadcast message

type CheckRetryResult

type CheckRetryResult struct {
	DoRetry bool
	Timeout time.Duration
}

CheckRetryResult represents describes whether a retry is requested, and how long to timeout first

type Client

type Client struct {
	/**
	 * Remote address to connect to.
	 */
	Address string
	// contains filtered or unexported fields
}

Client AMPQ Golang Wrapper

- Each RPC message has an rpcService - Queue for RPC of type T is kept in queue named `koins_rpc_T` - RPC messages per node type - Single global exchange for events, all node types

func NewClient

func NewClient(addr string, rpcRetryPolicy RetryPolicy) *Client

NewClient factory method.

func (*Client) Broadcast

func (c *Client) Broadcast(ctx context.Context, contentType ContentType, topic string, args []byte) error

Broadcast a message via AMQP

func (*Client) RPC

func (c *Client) RPC(ctx context.Context, contentType ContentType, rpcService string, args []byte) ([]byte, error)

RPC makes an RPC call

func (*Client) SetNumConsumers

func (c *Client) SetNumConsumers(rpcReturnNumConsumers int)

SetNumConsumers sets the number of consumers for queues.

This sets the number of parallel goroutines that consume the respective AMQP queues. Must be called before Connect().

func (*Client) Start

func (c *Client) Start(ctx context.Context) <-chan struct{}

Start begins the connection loop. Blocks until first connected to AMQP

type ContentType

type ContentType string

type ExponentialBackoffOptions

type ExponentialBackoffOptions struct {
	MaxTimeout  time.Duration
	Exponent    float32
	NextTimeout time.Duration
}

ExponentialBackoffOptions are the options for the exponential backoff policy

type RPCCallResult

type RPCCallResult struct {
	Result []byte
	Error  error
}

RPCCallResult is the result of an rpc call

type RPCHandlerFunc

type RPCHandlerFunc = func(rpcType string, data []byte) ([]byte, error)

RPCHandlerFunc Function type to handle an RPC message

type RequestHandler

type RequestHandler struct {
	/**
	 * Remote address to connect to.
	 */
	Address string
	// contains filtered or unexported fields
}

RequestHandler AMPQ Golang Wrapper

- Each RPC message has an rpcType - Queue for RPC of type T is kept in queue named `koins_rpc_T` - RPC messages per node type - Single global exchange for events, all node types

func NewRequestHandler

func NewRequestHandler(addr string, consumers uint, replyRetryPolicy RetryPolicy) *RequestHandler

NewRequestHandler factory method.

func (*RequestHandler) SetBroadcastHandler

func (r *RequestHandler) SetBroadcastHandler(topic string, handler BroadcastHandlerFunc)

SetBroadcastHandler sets the broadcast handler for a type.

func (*RequestHandler) SetRPCHandler

func (r *RequestHandler) SetRPCHandler(rpcType string, handler RPCHandlerFunc)

SetRPCHandler sets the RPC handler for an RPC type.

func (*RequestHandler) Start

func (r *RequestHandler) Start(ctx context.Context) <-chan struct{}

Start begins the connection loop.

type RetryPolicy

type RetryPolicy int

RetryPolicy is an enum

const (
	// NoRetry does not retry
	NoRetry RetryPolicy = iota
	// ExponentialBackoff retires forever, with exponentially increading sleep
	ExponentialBackoff
)