broker

Documentation

Index

• type Broker
  • func NewBroker(provider BrokerProvider) *Broker
  • func (b *Broker) GracefulStop(ctx context.Context) error
  • func (b *Broker) Init(opts *InitOptions)
• type BrokerInstance
  • func InstanceFromCtx(ctx web.Context) BrokerInstance
• type BrokerInstanceDefault
  • func (bid *BrokerInstanceDefault) Publish(ctx context.Context, topic string, m interface{}) error
• type BrokerPlugin
  • func NewBrokerPlugin(opts Options) *BrokerPlugin
  • func (b *BrokerPlugin) AppendMiddlewares() []web.MiddlewareFunc
  • func (b *BrokerPlugin) Instance() interface{}
  • func (b *BrokerPlugin) Name() string
  • func (b *BrokerPlugin) OnCreated(z *zepto.Zepto)
  • func (b *BrokerPlugin) OnZeptoInit(z *zepto.Zepto)
  • func (b *BrokerPlugin) OnZeptoStart(z *zepto.Zepto)
  • func (b *BrokerPlugin) OnZeptoStop(z *zepto.Zepto)
  • func (b *BrokerPlugin) PrependMiddlewares() []web.MiddlewareFunc
• type BrokerProvider
• type DefaultSubscriptionContext
  • func (sc *DefaultSubscriptionContext) Broker() BrokerInstance
• type InitOptions
• type Message
• type Options
• type SubscribeHandler
• type SubscriptionContext
• type SubscriptionsMap

Types

type Broker

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

BrokerWrapper is a struct that wrap the broker provider (gcp, rabbitmq, etc) and handle with message encode/decode

func NewBroker

func NewBroker(provider BrokerProvider) *Broker

func (*Broker) GracefulStop

func (b *Broker) GracefulStop(ctx context.Context) error

GracefulStop graceful stop all subscriptions

func (*Broker) Init

func (b *Broker) Init(opts *InitOptions)

type BrokerInstance

type BrokerInstance interface {
	Publish(ctx context.Context, topic string, m interface{}) error
}

func InstanceFromCtx

func InstanceFromCtx(ctx web.Context) BrokerInstance

type BrokerInstanceDefault

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

func (*BrokerInstanceDefault) Publish

func (bid *BrokerInstanceDefault) Publish(ctx context.Context, topic string, m interface{}) error

type BrokerPlugin

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

func NewBrokerPlugin

func NewBrokerPlugin(opts Options) *BrokerPlugin

func (*BrokerPlugin) AppendMiddlewares

func (b *BrokerPlugin) AppendMiddlewares() []web.MiddlewareFunc

func (*BrokerPlugin) Instance

func (b *BrokerPlugin) Instance() interface{}

func (*BrokerPlugin) Name

func (b *BrokerPlugin) Name() string

func (*BrokerPlugin) OnCreated

func (b *BrokerPlugin) OnCreated(z *zepto.Zepto)

func (*BrokerPlugin) OnZeptoInit

func (b *BrokerPlugin) OnZeptoInit(z *zepto.Zepto)

func (*BrokerPlugin) OnZeptoStart

func (b *BrokerPlugin) OnZeptoStart(z *zepto.Zepto)

func (*BrokerPlugin) OnZeptoStop

func (b *BrokerPlugin) OnZeptoStop(z *zepto.Zepto)

func (*BrokerPlugin) PrependMiddlewares

func (b *BrokerPlugin) PrependMiddlewares() []web.MiddlewareFunc

type BrokerProvider

type BrokerProvider interface {
	Init(opts *InitOptions)
	Publish(ctx context.Context, topic string, msg *Message) error
	Subscribe(ctx context.Context, topic string, handler SubscribeHandler) error
	GracefulStop(ctx context.Context) error
}

type DefaultSubscriptionContext

type DefaultSubscriptionContext struct {
	context.Context
	BrokerInstance BrokerInstance
}

func (*DefaultSubscriptionContext) Broker

func (sc *DefaultSubscriptionContext) Broker() BrokerInstance

type InitOptions

type InitOptions struct {
	Logger   logger.Logger
	Instance BrokerInstance
}

type Message

type Message struct {
	Header map[string]string
	Body   []byte
}

type Options

type Options struct {
	Provider      BrokerProvider
	Subscriptions SubscriptionsMap
}

type SubscribeHandler

type SubscribeHandler func(ctx SubscriptionContext, msg *Message)

type SubscriptionContext

type SubscriptionContext interface {
	context.Context
	Broker() BrokerInstance
}

type SubscriptionsMap

type SubscriptionsMap map[string]interface{}