rabbit

Documentation

Index

• Constants
• Variables
• type Delay
  • func NewDelay(min int, max int, randomIncrement bool) (d *Delay)
  • func (delay *Delay) Delay(after ...func(curDelayDuration int))
  • func (delay *Delay) Reset()
• type Message
  • func (message *Message) SetOption(option ...PublishOption)
• type PublishOption
• type Route
• type RxHandler
• type ServerContext
• type Session
  • func New(ctx context.Context, addr string, options ...SessionOption) Session
• type SessionOption
• type TxHandler

Constants

const DefaultExchange = "amq.direct"

Variables

var (
	ErrNotConnected = errors.New("not connected to a server")
	ErrTxBufferFull = errors.New("tx queue is full")
)

var MaxTxBuffer = 1024

var TxWorkerOption = func(workNum int) SessionOption {
	return func(session *sessionRabbit) {
		log.Fine("set tx workers:%d", workNum)
		session.multiTx = workNum
	}
}

var WithPublishConfirm = func(usePublishConfirm bool) SessionOption {
	return func(session *sessionRabbit) {
		session.publishConfirm = usePublishConfirm
	}
}

WithPublishConfirm 启用publish confirm模式

Types

type Delay

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

func NewDelay

func NewDelay(min int, max int, randomIncrement bool) (d *Delay)

func (*Delay) Delay

func (delay *Delay) Delay(after ...func(curDelayDuration int))

func (*Delay) Reset

func (delay *Delay) Reset()

type Message

type Message struct {
	Payload interface{}
	Router  Route
	Options []PublishOption
}

func (*Message) SetOption

func (message *Message) SetOption(option ...PublishOption)

type PublishOption

type PublishOption func(m amqp.Publishing, payload interface{}) (amqp.Publishing, error)

var (
	JsonEncodeOption PublishOption = func(m amqp.Publishing, payload interface{}) (p amqp.Publishing, err error) {
		bs, err := json.Marshal(payload)
		if err != nil {
			return
		}
		p = m
		p.ContentType = "application/json"
		p.Body = bs
		return
	}
	TextEncodeOption PublishOption = func(m amqp.Publishing, payload interface{}) (p amqp.Publishing, err error) {
		var bs []byte
		switch payload.(type) {
		case string:
			s := payload.(string)
			bs = []byte(s)
		case []byte:
			bs = payload.([]byte)
		default:
			err = fmt.Errorf("need txt payload")
			log.Fine(err)
			return
		}
		p = m
		p.ContentType = "text/plain"
		p.Body = bs
		return
	}
	UUIDMsgIdOption PublishOption = func(m amqp.Publishing, payload interface{}) (p amqp.Publishing, err error) {
		p = m
		id, err := uuid.NewRandom()
		if err != nil {
			return
		}
		p.MessageId = id.String()
		return
	}
)

type Route

type Route struct {
	Exchange string
	Key      string
	Queue    string
}

type RxHandler

type RxHandler func(msg amqp.Delivery, txHandler TxHandler, serverContext ServerContext) (err error)

type ServerContext

type ServerContext struct {
	Consumers int
	Messages  int
	Queue     string
	Router    Route
}

type Session

type Session interface {
	Publish(m Message) error
	Close() error
	Subscribe(r Route, rxHandler RxHandler)
}

func New

func New(ctx context.Context, addr string, options ...SessionOption) Session

type SessionOption

type SessionOption func(session *sessionRabbit)

type TxHandler

type TxHandler func(m Message) (err error)