goeventbus

README

EventBus for Golang

Description

This is a simple implementation of an event bus in golang. Actually support:

• publish/subscribe messaging.
• request/reply messaging

Get Started

To start use eventbus in your project, you can run the following command.

go get github.com/StaniPetrosyan/go-eventbus

And import

import (
	goeventbus "github.com/StaniPetrosyan/go-eventbus"
)

Publish/Subscribe

var eventbus = goeventbus.NewEventBus()

address := "topic"
options := goeventbus.NewMessageOptions().AddHeader("header", "value")
message := goeventbus.CreateMessage().SetBody("Hi Topic").SetOptions(options)

eventbus.Subscribe(address, func(dc goeventbus.ConsumerContext) {
	fmt.Printf("Message %s\n", dc.Result().Data)
})

for {
	eventbus.Publish(address, message)
	time.Sleep(time.Second)
}

Request/Reply messaging

var eventbus = goeventbus.NewEventBus()

address := "topic"

eventbus.Subscribe(address, func(dc goeventbus.ConsumerContext) {
	fmt.Printf("Message %s\n", dc.Result().Data)
	dc.Reply("Hi from topic")
})
	
eventbus.Request(address, "Hi Topic", func(dc goeventbus.ConsumerContext) {
	dc.Handle(func(message Message) {
			fmt.Printf("Message %s\n", message.Data)
	})
})

Message

For publishing, you need to create a Message object using this method.

message := goeventbus.CreateMessage().SetBody("Hi Topic")

Each message can have some options:

options := goeventbus.NewMessageOptions().AddHeader("header", "value")
message := goeventbus.CreateMessage()

message.SetOptions(options)

eventBus.Publish("address", message)

In Bound Interceptor

eventbus.AddInBoundInterceptor("topic1", func(context goeventbus.InterceptonContext) {
	if (some logic)
		context.Next()
})

Documentation

Index

• type ConsumerContext
  • func NewConsumerContext(ch chan Message) ConsumerContext
  • func (d *ConsumerContext) Handle(consume func(message Message))
  • func (d *ConsumerContext) Reply(data any)
  • func (d *ConsumerContext) Result() Message
  • func (d ConsumerContext) SetData(msg Message) ConsumerContext
• type ConsumerHandler
  • func NewConsumer(address string, callback func(context ConsumerContext)) ConsumerHandler
  • func (h ConsumerHandler) Chain() chan Message
  • func (h ConsumerHandler) Closed() bool
  • func (h ConsumerHandler) Context() ConsumerContext
  • func (h ConsumerHandler) Handle(wg *sync.WaitGroup)
• type Context
• type DefaultEventBus
  • func (e *DefaultEventBus) AddInBoundInterceptor(address string, callback func(context InterceptorContext))
  • func (e *DefaultEventBus) Publish(address string, message Message)
  • func (e *DefaultEventBus) Request(address string, message Message, callback func(context ConsumerContext))
  • func (e *DefaultEventBus) Subscribe(address string, callback func(context ConsumerContext))
• type EventBus
  • func NewEventBus() EventBus
• type Handler
• type InterceptorContext
  • func NewInterceptorContext(topic *Topic) InterceptorContext
  • func (d *InterceptorContext) Next()
  • func (d *InterceptorContext) Result() Message
  • func (d InterceptorContext) SetData(msg Message) InterceptorContext
• type InterceptorHandler
  • func NewInterceptor(address string, callback func(context InterceptorContext), ...) InterceptorHandler
  • func (h InterceptorHandler) Chain() chan Message
  • func (h InterceptorHandler) Closed() bool
  • func (h InterceptorHandler) Context() InterceptorContext
  • func (h InterceptorHandler) Handle(wg *sync.WaitGroup)
• type Message
  • func CreateMessage() Message
  • func (m Message) SetBody(data interface{}) Message
  • func (m Message) SetOptions(options MessageOptions) Message
  • func (m Message) ToJson() ([]byte, error)
• type MessageOptions
  • func NewMessageOptions() MessageOptions
  • func (op MessageOptions) AddHeader(key string, value string) MessageOptions
  • func (op MessageOptions) Header(key string) string
  • func (op MessageOptions) SetHeader(headers map[string]string) MessageOptions
• type Publisher
  • func (p *Publisher) Publish()
• type Topic
  • func NewTopic(address string) *Topic
  • func (t *Topic) AddConsumer(callback func(context ConsumerContext)) ConsumerHandler
  • func (t *Topic) AddInterceptor(callback func(context InterceptorContext)) InterceptorHandler
  • func (t *Topic) GetChannels() []chan Message
  • func (t *Topic) GetHandlers() []Handler

Types

type ConsumerContext

type ConsumerContext struct {
	Ch chan Message
	// contains filtered or unexported fields
}

func NewConsumerContext

func NewConsumerContext(ch chan Message) ConsumerContext

func (*ConsumerContext) Handle

func (d *ConsumerContext) Handle(consume func(message Message))

func (*ConsumerContext) Reply

func (d *ConsumerContext) Reply(data any)

func (*ConsumerContext) Result

func (d *ConsumerContext) Result() Message

func (ConsumerContext) SetData

func (d ConsumerContext) SetData(msg Message) ConsumerContext

type ConsumerHandler

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

func NewConsumer

func NewConsumer(address string, callback func(context ConsumerContext)) ConsumerHandler

func (ConsumerHandler) Chain

func (h ConsumerHandler) Chain() chan Message

func (ConsumerHandler) Closed

func (h ConsumerHandler) Closed() bool

func (ConsumerHandler) Context

func (h ConsumerHandler) Context() ConsumerContext

func (ConsumerHandler) Handle

func (h ConsumerHandler) Handle(wg *sync.WaitGroup)

type Context

type Context interface {
	Result() Message
}

type DefaultEventBus

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

func (*DefaultEventBus) AddInBoundInterceptor

func (e *DefaultEventBus) AddInBoundInterceptor(address string, callback func(context InterceptorContext))

func (*DefaultEventBus) Publish

func (e *DefaultEventBus) Publish(address string, message Message)

func (*DefaultEventBus) Request

func (e *DefaultEventBus) Request(address string, message Message, callback func(context ConsumerContext))

func (*DefaultEventBus) Subscribe

func (e *DefaultEventBus) Subscribe(address string, callback func(context ConsumerContext))

type EventBus

type EventBus interface {
	Subscribe(address string, callback func(context ConsumerContext))
	AddInBoundInterceptor(address string, callback func(context InterceptorContext))
	Publish(address string, message Message)
	Request(address string, message Message, callback func(context ConsumerContext))
}

func NewEventBus

func NewEventBus() EventBus

type Handler

type Handler interface {
	Handle(wg *sync.WaitGroup)
	Chain() chan Message
	Closed() bool
}

type InterceptorContext

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

func NewInterceptorContext

func NewInterceptorContext(topic *Topic) InterceptorContext

func (*InterceptorContext) Next

func (d *InterceptorContext) Next()

func (*InterceptorContext) Result

func (d *InterceptorContext) Result() Message

func (InterceptorContext) SetData

func (d InterceptorContext) SetData(msg Message) InterceptorContext

type InterceptorHandler

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

func NewInterceptor

func NewInterceptor(address string, callback func(context InterceptorContext), context InterceptorContext) InterceptorHandler

func (InterceptorHandler) Chain

func (h InterceptorHandler) Chain() chan Message

func (InterceptorHandler) Closed

func (h InterceptorHandler) Closed() bool

func (InterceptorHandler) Context

func (h InterceptorHandler) Context() InterceptorContext

func (InterceptorHandler) Handle

func (h InterceptorHandler) Handle(wg *sync.WaitGroup)

type Message

type Message struct {
	Data    interface{}
	Options MessageOptions
}

func CreateMessage

func CreateMessage() Message

func (Message) SetBody

func (m Message) SetBody(data interface{}) Message

func (Message) SetOptions

func (m Message) SetOptions(options MessageOptions) Message

func (Message) ToJson

func (m Message) ToJson() ([]byte, error)

type MessageOptions

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

func NewMessageOptions

func NewMessageOptions() MessageOptions

func (MessageOptions) AddHeader

func (op MessageOptions) AddHeader(key string, value string) MessageOptions

func (MessageOptions) Header

func (op MessageOptions) Header(key string) string

func (MessageOptions) SetHeader

func (op MessageOptions) SetHeader(headers map[string]string) MessageOptions

type Publisher

type Publisher struct {
}

func (*Publisher) Publish

func (p *Publisher) Publish()

type Topic

type Topic struct {
	Address      string
	Consumers    []Handler
	Interceptors []Handler
}

func NewTopic

func NewTopic(address string) *Topic

func (*Topic) AddConsumer

func (t *Topic) AddConsumer(callback func(context ConsumerContext)) ConsumerHandler

func (*Topic) AddInterceptor

func (t *Topic) AddInterceptor(callback func(context InterceptorContext)) InterceptorHandler

func (*Topic) GetChannels

func (t *Topic) GetChannels() []chan Message

func (*Topic) GetHandlers

func (t *Topic) GetHandlers() []Handler