mt

README

MT

Documentation

Index

• type Binding
  • func DefaultBinding() Binding
• type Config
  • func ParseConfig(jsonConf []byte) (Config, error)
• type Consume
  • func DefaultConsumer() Consume
• type Duration
  • func (e *Duration) UnmarshalJSON(b []byte) error
• type Exchange
  • func DefaultExchange() Exchange
  • func (e *Exchange) UnmarshalJSON(b []byte) error
• type Handler
• type HandlerFunc
• type Header
  • func (h Header) AddString(key, value string)
  • func (h Header) String(key string) string
• type Logger
• type MT
  • func NewMT(opts ...Option) *MT
  • func (t *MT) AddHandler(serviceName ServiceName, handler HandlerFunc)
  • func (t *MT) Call(serviceName ServiceName, request Request, reply ReplyFunc) (err error)
  • func (t *MT) Cast(serviceName ServiceName, request Request) (err error)
  • func (t *MT) ConnectAndServe() error
  • func (t *MT) HealthCheck() bool
  • func (t *MT) Shutdown() error
• type MassTransport
• type Option
  • func WithAMQP(dsn string) Option
  • func WithConfig(config Config) Option
  • func WithLogger(log Logger) Option
• type Queue
  • func DefaultQueue() Queue
• type ReplyFunc
• type ReplyQueue
• type Request
• type Response
• type Service
• type ServiceName

Types

type Binding

type Binding struct {
	Key       string     `json:"key,omitempty"`
	NoWait    bool       `json:"nowait,omitempty"`
	Arguments amqp.Table `json:"arguments,omitempty"`
}

func DefaultBinding

func DefaultBinding() Binding

type Config

type Config struct {
	Services map[ServiceName]Service `json:"services"`
}

Config ...

func ParseConfig

func ParseConfig(jsonConf []byte) (Config, error)

ParseConfig read config from json.

type Consume

type Consume struct {
	Tag string `json:"tag,omitempty"`
	// управление сообщением в случае невозможности обработки
	// true - сообщение вновь ставится в очередь
	// false - отбросить сообщение в отстойник, если он настроен. Иначе удалить сообщение
	Requeue   bool       `json:"requeue,omitempty"`
	NoAck     bool       `json:"noack,omitempty"`
	Exclusive bool       `json:"exclusive,omitempty"`
	NoWait    bool       `json:"nowait,omitempty"`
	Arguments amqp.Table `json:"arguments,omitempty"`
}

func DefaultConsumer

func DefaultConsumer() Consume

type Duration

type Duration time.Duration

func (*Duration) UnmarshalJSON

func (e *Duration) UnmarshalJSON(b []byte) error

type Exchange

type Exchange struct {
	Name       string     `json:"name,omitempty"`
	Type       string     `json:"type,omitempty"`
	Durable    bool       `json:"durable,omitempty"`
	AutoDelete bool       `json:"autodelete,omitempty"`
	Internal   bool       `json:"internal,omitempty"`
	NoWait     bool       `json:"nowait,omitempty"`
	Arguments  amqp.Table `json:"arguments,omitempty"`

	Binding Binding `json:"binding,omitempty"`
	Queue   Queue   `json:"queue,omitempty"`

	ReplyQueue ReplyQueue `json:"replyqueue,omitempty"`
}

func DefaultExchange

func DefaultExchange() Exchange

func (*Exchange) UnmarshalJSON

func (e *Exchange) UnmarshalJSON(b []byte) error

type Handler

type Handler interface {
	Serve(*Request)
}

Handler ...

type HandlerFunc

type HandlerFunc func(*Request, ReplyFunc) error

type Header

type Header amqp.Table

A Header represents the key-value pairs in a message header.

func (Header) AddString

func (h Header) AddString(key, value string)

AddString adds the key, string value pair to the header.

func (Header) String

func (h Header) String(key string) string

type Logger

type Logger interface {
	Infof(format string, args ...interface{})
	Errorf(format string, args ...interface{})
	Debugf(format string, args ...interface{})
}

Logger is an interface for logger to pass into mt server.

type MT

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

func NewMT

func NewMT(opts ...Option) *MT

NewMT get new instance.

func (*MT) AddHandler

func (t *MT) AddHandler(serviceName ServiceName, handler HandlerFunc)

func (*MT) Call

func (t *MT) Call(serviceName ServiceName, request Request, reply ReplyFunc) (err error)

func (*MT) Cast

func (t *MT) Cast(serviceName ServiceName, request Request) (err error)

func (*MT) ConnectAndServe

func (t *MT) ConnectAndServe() error

func (*MT) HealthCheck

func (t *MT) HealthCheck() bool

func (*MT) Shutdown

func (t *MT) Shutdown() error

type MassTransport

type MassTransport interface {
	ConnectAndServe() error
	AddHandler(ServiceName, HandlerFunc)
	Call(ServiceName, Request, ReplyFunc) error
	Cast(ServiceName, Request) error
	Shutdown() error
	HealthCheck() bool
}

MassTransport is a interface.

type Option

type Option func(*MT)

Option for MT.

func WithAMQP

func WithAMQP(dsn string) Option

WithAMQP use dsn string.

func WithConfig

func WithConfig(config Config) Option

WithConfig set config.

func WithLogger

func WithLogger(log Logger) Option

WithLogger set logger.

type Queue

type Queue struct {
	Name       string     `json:"name,omitempty"`
	Durable    bool       `json:"durable,omitempty"`
	AutoDelete bool       `json:"autodelete,omitempty"`
	NoWait     bool       `json:"nowait,omitempty"`
	Arguments  amqp.Table `json:"arguments,omitempty"`

	// @see http://www.rabbitmq.com/blog/2012/04/25/rabbitmq-performance-measurements-part-2/
	// Консумер не получит следующие n сообщений, пока не подтвердит предыдущие.
	PrefetchCount int `json:"prefetchCount,omitempty"`

	Consumer Consume `json:"consume,omitempty"`
}

func DefaultQueue

func DefaultQueue() Queue

type ReplyFunc

type ReplyFunc func(Response) error

type ReplyQueue

type ReplyQueue struct {
	BindToExchange bool     `json:"bindToExchange,omitempty"`
	Timeout        Duration `json:"timeout,omitempty"`
}

type Request

type Request struct {
	Header Header
	Body   []byte
}

Request from chan.

type Response

type Response struct {
	Body []byte
}

Response from chan.

type Service

type Service struct {
	Exchange Exchange `json:"exchange"`
}

type ServiceName

type ServiceName string