bus

README

Go Message Bus using RMQ3

RPC Producer (api)

// init and consume "api:callback" for RPC replies
rmq := bus.NewBus(bus.Opts{
    Context:       appCtx,
    DSN:           "amqp://guest:guest@localhost:5672",
    CallbackQueue: "api:callback",
    Prefetch:      200,
})
go rmq.Run()

// Option 1: make RPC, you implement timeout/replies/etc yourself
msg := bus.Message{
    ToQueue:     "worker_queue",
    ToFunc:      "FoobarFunc",
    MsgTTL:      time.Hour,
    Body:        []byte(`{"Foo":"bar"}`),
}
err := rmq.Publish(&msg)

// Option 2: make RPC, wait for response
responseMsg, err := rmq.PublishAndReturn(MyRequest{Foo:"bar"}, "worker_queue", "FoobarFunc", time.Second, time.Minute, r.Context())

// Option 3: make RPC, execute closure once response comes
wg := sync.WaitGroup{}
wg.Add(1)
closure := func(resp any) {
    defer wg.Done()
    
    switch resp := resp.(type) {
    case *bus.Message:
        // response from worker
    case error:
        switch {
        case errors.Is(resp, bus.ErrRequestCancelled)
            // r,Context() cancelled
            rmq.CancelMsg(msg, "worker:callback") // have worker cancel RPC ctx
        case errors.Is(resp, bus.ErrReplyTimeout)
            // MsgDeadline reached
        case errors.Is(resp, bus.ErrPublishing):
            // rmq err, couldn't publish msg
        }
    }
}
msg := bus.Message{
    ToQueue:     "worker_queue",
    ToFunc:      "FoobarFunc",
    MsgDeadline: time.Now().Add(time.Minute),
    MsgTTL:      time.Hour,
    Body:        []byte(`{"Foo":"bar"}`),
}
rmq.PublishAndClose(&msg, closure, r.Context())
wg.Wait()

RPC Consumer (worker)

// init and consume "worker:callback" for RPC replies/ctx cancels (from publisher)
bus := bus.NewBus(bus.Opts{
    Context:       appCtx,
    DSN:           "amqp://guest:guest@localhost:5672",
    CallbackQueue: "worker:callback",
    Prefetch:      200,
})

// bind &foobar to handle RPCs from "worker_queue" and a special "worker_queue:noprefetch" queues 
err := rmq.RegisterHandler(bus.HandlerOpts{
    Queue:    "worker.FoobarService",
    Prefetch: 10,
    Handler:  &foobar,
})

go rmq.Run()

Consumer Funcs

// RPC funcs must be Exported and have bus.Response return 
type FoobarService struct {}

func (f *FoobarService) ReturnBlank() bus.Response {
    return nil
}
func (f *FoobarService) ReturnBytes() bus.Response {
    return []byte("raw data")
}
func (f *FoobarService) ReturnString() bus.Response {
    return "ok"
}
func (f *FoobarService) ReturnMap() bus.Response {
    return map[string]any{
        "str": "string",
        "int": 123,
        "bool": true,
        "struct": MyStruct{},
    }
}
func (f *FoobarService) ReturnError() bus.Response {
    return errors.New("my error")
}
func (f *FoobarService) ReturnJson() bus.Response {
    // can be struct or *struct
    return MyResponse{
        Foo: "bar",
    }
}
func (f *FoobarService) ReturnCustomResponse() bus.Response {
    return bus.CustomResponse{
        StatusCode: 201,
        Header: http.Header{"Content-Type": {"application/xml"}},
        Body: `<?xml version = "1.0" encoding = "UTF-8"?>`,
    }        
}

func (f *FoobarService) GetString(body string) bus.Response {
    //
}
func (f *FoobarService) GetBytes(body []byte) bus.Response {
    //
}
func (f *FoobarService) GetUnmarshal(request MyRequest) bus.Response {
    // can be struct or *struct
}
func (f *FoobarService) GetHttpHeader(header http.Header) bus.Rseponse {
    // passed header via bus.Message{Header: r.Header}
}
func (f *FoobarService) GetQueryVars(query url.Values) bus.Response {
    // pass r.URL.RawQuery via "RawQuery" header, ie:
    // r.Header.Set("RawQuery", r.URL.RawQuery) and bus.Message{Header: r.Header}
}
func (f *FoobarService) GetRequestCtx(ctx context.Context) bus.Response {
    // currently ONLY for ctx.Cancels (no values, deadlines, etc)
    <- ctx.Done()
}

Message Type

// bus.Message
type Message struct {
    ToQueue string // remote.worker.queue
    ToFunc  string // RemoteFunc
    ReplyTo string // api:callback
    
    MsgID       string        // msg uuid
    MsgDeadline time.Time     // used to build remote ctx, ie RemoteFunc(ctx context.Context)
    MsgTTL      time.Duration // time msg lives in RMQ until expiring
    
    // HTTP related, usually in bus.Response
    StatusCode int         // HTTP StatusCode: 200, etc
    Header     http.Header // HTTP Headers: [MyHeader: foobar, ...]
    
    Type       string      // auto pretty-printed ResponseStructName or manually via CustomResponse{Type:___}
    Body       []byte      // raw bytes, usually marshalled struct
    
    // constructed in consumeMsg() off MsgDeadline, then passed to RemoteFunc(ctx)
    // and early cancelled if got CancelMsgSpecialFuncName
    ctx    context.Context
    cancel context.CancelFunc
}

Bus Slog

By default bus will not optout anything

// bus/slog.go:
var Slog *slog.Logger = slog.New(&NilHandler{})

You can override it

import "github.com/frifox/bus"

func init() {
    bus.Slog = slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
        Level: slog.LevelDebug,
    }))
}

Documentation

Index

• Constants
• Variables
• type Bus
  • func NewBus(o Opts) *Bus
  • func (b *Bus) CallAndClose(queue string, funcName string, req any, closure func(any), ...) error
  • func (b *Bus) CallAndWait(queue string, funcName string, req any, timeout time.Duration, ...) (*Message, error)
  • func (b *Bus) CallAsync(queue string, funcName string, req any, ttl time.Duration) error
  • func (b *Bus) CancelMsg(msg Message, callbackQueue string) error
  • func (b *Bus) Connect() error
  • func (b *Bus) Consume(consumers *sync.WaitGroup, queue string)
  • func (b *Bus) HandlerFuncs() map[string][]string
  • func (b *Bus) OpenInChannel(queue string, prefetchCount int) error
  • func (b *Bus) Publish(msg *Message) error
  • func (b *Bus) PublishAndClose(msg *Message, userClosure func(any), requestCtxs ...context.Context)
  • func (b *Bus) RegisterAsyncHandleFunc(o HandleFuncOpts)
  • func (b *Bus) RegisterHandler(opts HandlerOpts) error
  • func (b *Bus) Run()
• type CustomResponse
• type HandleFuncOpts
• type Handler
• type HandlerFunc
• type HandlerOpts
• type Message
  • func (m *Message) FromDelivery(d *amqp091.Delivery)
  • func (m *Message) ToPublishing() *amqp091.Publishing
• type NilHandler
  • func (*NilHandler) Enabled(_ context.Context, _ slog.Level) bool
• type Opts
• type RMQPublishing
• type Response

Constants

const MsgTypeCtxCancel = "CtxCancel"

const MsgTypeRequest = "Request"

const MsgTypeResponse = "Response"

const NoPrefetchSuffix = ":noprefetch" // special queue for bypassing Prefetch, ie for interrupts

Variables

var ErrPublishing = errors.New("publish failed")

var ErrReplyTimeout = errors.New("bus reply timeout")

var ErrRequestCancelled = errors.New("request cancelled")

var MessageTypeError = "error"

var Slog *slog.Logger = slog.New(&NilHandler{})

Types

type Bus

type Bus struct {
	AppCtx context.Context

	context.Context
	context.CancelFunc

	DSN string

	// replies
	CallbackQueue string        // where to listen for replies
	Prefetch      int           // replies consuming Prefetch
	MsgTTL        time.Duration // TTL for replies sent to producer

	// incoming
	InConn *amqp091.Connection
	// contains filtered or unexported fields
}

func NewBus

func NewBus(o Opts) *Bus

func (*Bus) CallAndClose

func (b *Bus) CallAndClose(queue string, funcName string, req any, closure func(any), timeout time.Duration, ttl time.Duration, reqCancels ...context.Context) error

func (*Bus) CallAndWait

func (b *Bus) CallAndWait(queue string, funcName string, req any, timeout time.Duration, ttl time.Duration, reqCancels ...context.Context) (*Message, error)

func (*Bus) CallAsync

func (b *Bus) CallAsync(queue string, funcName string, req any, ttl time.Duration) error

func (*Bus) CancelMsg

func (b *Bus) CancelMsg(msg Message, callbackQueue string) error

func (*Bus) Connect

func (b *Bus) Connect() error

func (*Bus) Consume

func (b *Bus) Consume(consumers *sync.WaitGroup, queue string)

func (*Bus) HandlerFuncs

func (b *Bus) HandlerFuncs() map[string][]string

func (*Bus) OpenInChannel

func (b *Bus) OpenInChannel(queue string, prefetchCount int) error

func (*Bus) Publish

func (b *Bus) Publish(msg *Message) error

Publish pushes msg to bus. Useful if RPCs are one-way, or you want to handle replies yourself

func (*Bus) PublishAndClose

func (b *Bus) PublishAndClose(msg *Message, userClosure func(any), requestCtxs ...context.Context)

PublishAndClose extends Publish by catching replies + handing bus timeouts / request cancels

func (*Bus) RegisterAsyncHandleFunc

func (b *Bus) RegisterAsyncHandleFunc(o HandleFuncOpts)

RegisterAsyncHandleFunc registers 1 func to handle msgs from 1 queue

func (*Bus) RegisterHandler

func (b *Bus) RegisterHandler(opts HandlerOpts) error

RegisterHandler registers 1 handler (with many funcs) to handle msgs from 1 queue

func (*Bus) Run

func (b *Bus) Run()

type CustomResponse

type CustomResponse struct {
	StatusCode int
	Header     http.Header
	Type       string
	Body       Response
}

CustomResponse allows use to customize bus.Response reply

type HandleFuncOpts

type HandleFuncOpts struct {
	HandleFunc func(*Message)
	Queue      string
	Prefetch   int
}

type Handler

type Handler map[string]HandlerFunc

type HandlerFunc

type HandlerFunc struct {
	Func    reflect.Value
	Args    []reflect.Type
	Returns []reflect.Type
}

type HandlerOpts

type HandlerOpts struct {
	Handler  any
	Queue    string
	Prefetch int
}

type Message

type Message struct {
	BusMsgType string // [Request, Response, CtxCancel]
	ToQueue    string // remote.worker.queue
	ToFunc     string // RemoteFunc
	ReplyTo    string // api:callback

	MsgID       string        // msg uuid
	MsgDeadline time.Time     // used to build/cancel req ctx, ie RemoteFunc(ctx context.Context)
	MsgTTL      time.Duration // time msg lives in RMQ before dying

	StatusCode int         // HTTP StatusCode: 200, etc
	Header     http.Header // HTTP Headers: [MyHeader: foobar, ...]
	Type       string      // pretty print StructName or CustomResponse{Type:___} string
	Body       []byte      // raw bytes, or marshalled struct
	// contains filtered or unexported fields
}

func (*Message) FromDelivery

func (m *Message) FromDelivery(d *amqp091.Delivery)

func (*Message) ToPublishing

func (m *Message) ToPublishing() *amqp091.Publishing

type NilHandler

type NilHandler struct {
	slog.Handler
}

func (*NilHandler) Enabled

func (*NilHandler) Enabled(_ context.Context, _ slog.Level) bool

type Opts

type Opts struct {
	Context       context.Context
	DSN           string
	CallbackQueue string
	Prefetch      int
	//MsgTimeout time.Duration
	MsgTTL time.Duration
}

type RMQPublishing

type RMQPublishing struct {
	// rmq vars
	DeliveryMode uint8 // Transient (0 or 1) or Persistent (2)
	Expiration   string

	// bus vars
	Type            string // [Request, Response, CtxCancel]
	MessageId       string // MsgID
	CorrelationId   string // ToFuncName
	ReplyTo         string // reply queue
	ContentEncoding string // RFC3339Nano MsgTimeout for deadlines

	// request & replies
	Headers     amqp091.Table // map[string]any{nil,bool,byte,int,float,[]byte,Decimal,time.Time}; t.SetClientConnectionName() => t["connection_name"]=connName
	Body        []byte        // raw bytes
	ContentType string        // StructName (for replies)
	AppId       string        // http status code (for replies)

	// not used
	Priority  uint8     // 0 to 9
	Timestamp time.Time // truncated to 1s
	UserId    string    // creating user id - ex: "guest"
}

RMQPublishing for reference only

type Response

type Response any

Response type will be auto-detected in bus.sendHandlerReturnToBus()