pubsubwrap

Documentation

Index

• type FilterMiddleware
• type MessageLockerByType
  • func NewMessageLockerByType(ctx context.Context) *MessageLockerByType
  • func (ml *MessageLockerByType) Filter(m *message.Message) (*message.Message, error)
  • func (ml *MessageLockerByType) GetMessagesFilters(ctx context.Context, in *introproto.EmptyArgs) (*introproto.AllMessageFilterStats, error)
  • func (ml *MessageLockerByType) SetMessagesFilter(ctx context.Context, in *introproto.MessageFilterByType) (*introproto.MessageFilterByType, error)
• type MessageStatByType
  • func NewMessageStatByType() *MessageStatByType
  • func (ms *MessageStatByType) Filter(m *message.Message) (*message.Message, error)
  • func (ms *MessageStatByType) GetMessagesStat(context.Context, *introproto.EmptyArgs) (*introproto.AllMessageStatByType, error)
• type PublisherService
  • func NewPublisherService(ml *MessageLockerByType, ms *MessageStatByType) PublisherService
• type PublisherWrapper
  • func NewPublisherWrapper(pb message.Publisher) *PublisherWrapper
  • func (p *PublisherWrapper) Close() error
  • func (p *PublisherWrapper) Middleware(fm ...FilterMiddleware)
  • func (p *PublisherWrapper) Publish(topic string, messages ...*message.Message) error

Types

type FilterMiddleware

type FilterMiddleware interface {
	Filter(m *message.Message) (*message.Message, error)
}

FilterMiddleware an interface for message filtering and modification.

type MessageLockerByType

type MessageLockerByType struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func NewMessageLockerByType

func NewMessageLockerByType(ctx context.Context) *MessageLockerByType

NewMessageLockerByType is a constructor for MessageLockerByType.

func (*MessageLockerByType) Filter

func (ml *MessageLockerByType) Filter(m *message.Message) (*message.Message, error)

Filter returns nil for filtered message types.

func (*MessageLockerByType) GetMessagesFilters

func (ml *MessageLockerByType) GetMessagesFilters(ctx context.Context, in *introproto.EmptyArgs) (*introproto.AllMessageFilterStats, error)

GetMessagesFilters returns filter state and statistic per message type (as map).

func (*MessageLockerByType) SetMessagesFilter

func (ml *MessageLockerByType) SetMessagesFilter(ctx context.Context, in *introproto.MessageFilterByType) (*introproto.MessageFilterByType, error)

SetMessagesFilter sets filter for provided message type.

type MessageStatByType

type MessageStatByType struct {
	sync.Mutex
	// contains filtered or unexported fields
}

MessageStatByType holds publish statistic per message type.

func NewMessageStatByType

func NewMessageStatByType() *MessageStatByType

NewMessageStatByType is a constructor for MessageStatByType.

func (*MessageStatByType) Filter

func (ms *MessageStatByType) Filter(m *message.Message) (*message.Message, error)

Filter counts published messages by type.

func (*MessageStatByType) GetMessagesStat

func (ms *MessageStatByType) GetMessagesStat(context.Context, *introproto.EmptyArgs) (*introproto.AllMessageStatByType, error)

GetMessagesStat returns publish statistic per message type.

type PublisherService

type PublisherService struct {
	*MessageLockerByType
	*MessageStatByType
}

PublisherService implements introproto.PublisherServer.

func NewPublisherService

func NewPublisherService(ml *MessageLockerByType, ms *MessageStatByType) PublisherService

NewPublisherService creates PublisherService.

type PublisherWrapper

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

PublisherWrapper wraps message Publisher.

func NewPublisherWrapper

func NewPublisherWrapper(pb message.Publisher) *PublisherWrapper

NewPublisherWrapper creates new message.Publisher wrapper.

func (*PublisherWrapper) Close

func (p *PublisherWrapper) Close() error

Close wraps message.Close method.

func (*PublisherWrapper) Middleware

func (p *PublisherWrapper) Middleware(fm ...FilterMiddleware)

Middleware adds middleware filters (order matters!).

func (*PublisherWrapper) Publish

func (p *PublisherWrapper) Publish(topic string, messages ...*message.Message) error

Publish wraps message.Publish method, i.e. applies all middleware filters for every message.