goeventbus

README

EventBus for Golang

Description

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

• publish/subscribe 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

Simple example of publish/subscribe pattern.

var eventbus = goeventbus.NewEventBus()

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

eventbus.Channel(address).Subscriber().Listen(func(dc goeventbus.Context) {
	fmt.Printf("Message %s\n", dc.Result().Data)
})

eventbus.Channel(address).Publisher().Publish(message)

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.Channel("address").Publisher().Publish(message)

Processor

A processor works like a middleware, in fact forwards messages only if the predicate is satisfied. The method accepts a function with message and return must return a boolean.

eventbus.Channel("topic1").Processor(func(message goeventbus.Message) bool {
	return message.Options.Headers().Contains("header")
})

Network Bus

A Network bus create a tcp connection between different services.

NetworkBus is a wrapper of local eventbus.

A simple server/client example is in examples/networkbus directory.

Documentation

Index

• type Channel
  • func NewChannel(address string) Channel
• type Client
  • func NewClient(address string, eventbus EventBus) Client
• type Context
  • func NewConsumerContextWithMessage(message Message) Context
• type DefaultEventBus
  • func (e *DefaultEventBus) Channel(address string) Channel
• type EventBus
  • func NewEventBus() EventBus
• type Headers
  • func NewHeaders() Headers
  • func (h Headers) Add(key string, value string) Headers
  • func (h Headers) Contains(key string) bool
  • func (h Headers) Header(key string) string
• type Message
  • func CreateMessage() Message
  • func (m Message) SetBody(data interface{}) Message
  • func (m Message) SetOptions(options MessageOptions) Message
• type MessageOptions
  • func NewMessageOptions() MessageOptions
  • func (op MessageOptions) Headers() Headers
  • func (op MessageOptions) SetHeaders(headers Headers) MessageOptions
• type NetworkBus
  • func NewNetworkBus(bus EventBus, address, path string) NetworkBus
• type Processor
  • func NewProcessor() Processor
  • func NewProcessorWithPredicate(predicate func(message Message) bool) Processor
• type Publisher
  • func NewPublisher(ch chan Message) Publisher
• type Request
• type Server
  • func NewServer(address string) Server
• type Subscriber
  • func NewSubscriber(ch chan Message) Subscriber

Types

type Channel

type Channel interface {
	Publisher() Publisher
	Subscriber() Subscriber
	Processor(predicate func(message Message) bool) Channel
}

func NewChannel

func NewChannel(address string) Channel

type Client

type Client interface {
	Connect()
}

func NewClient

func NewClient(address string, eventbus EventBus) Client

type Context

type Context interface {
	Result() Message
}

func NewConsumerContextWithMessage

func NewConsumerContextWithMessage(message Message) Context

type DefaultEventBus

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

func (*DefaultEventBus) Channel

func (e *DefaultEventBus) Channel(address string) Channel

type EventBus

type EventBus interface {
	Channel(adress string) Channel
}

func NewEventBus

func NewEventBus() EventBus

type Headers

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

func NewHeaders

func NewHeaders() Headers

func (Headers) Add

func (h Headers) Add(key string, value string) Headers

func (Headers) Contains

func (h Headers) Contains(key string) bool

func (Headers) Header

func (h Headers) Header(key string) string

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

type MessageOptions

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

func NewMessageOptions

func NewMessageOptions() MessageOptions

func (MessageOptions) Headers

func (op MessageOptions) Headers() Headers

func (MessageOptions) SetHeaders

func (op MessageOptions) SetHeaders(headers Headers) MessageOptions

type NetworkBus

type NetworkBus interface {
	Server() Server
	Client() Client
}

func NewNetworkBus

func NewNetworkBus(bus EventBus, address, path string) NetworkBus

type Processor

type Processor interface {
	// contains filtered or unexported methods
}

func NewProcessor

func NewProcessor() Processor

func NewProcessorWithPredicate

func NewProcessorWithPredicate(predicate func(message Message) bool) Processor

type Publisher

type Publisher interface {
	Publish(message Message)
}

func NewPublisher

func NewPublisher(ch chan Message) Publisher

type Request

type Request struct {
	Channel string  `json:"channel"`
	Message Message `json:"message"`
}

type Server

type Server interface {
	Listen() (Server, error)
	Publish(channel string, message Message)
}

func NewServer

func NewServer(address string) Server

type Subscriber

type Subscriber interface {
	Listen(consumer func(context Context))
}

func NewSubscriber

func NewSubscriber(ch chan Message) Subscriber