pubsub

README

pubsub

Simple PubSub interface for golang apps with plugable drivers.

Supported drivers

• in-memory implementation based on go channels
• nats.io
• redis using redigo
• nsq.io - draft, not completed!

API

API is pretty simple

// Hub interface of pubsub system.
type Hub interface {
	// Publish sends input message to specified channels.
	Publish(channels []string, msg interface{})
	// Subscribe opens channel to listen specified channels.
	Subscribe(channels []string) (Channel, error)
	// Close stops the pubsub hub.
	Close() error
}

// Channel to listen pubsub events.
type Channel interface {
	// Read returns channel to receive events.
	Read() <-chan interface{}
	// Close stops listening underlying pubsub channels.
	Close() error
	// CloseNotify returns channel to receive event when this channel is closed.
	CloseNotify() <-chan bool
}

Server-sent events

See code of built-in package

TODO

• continuous integration
• (in progress) unit tests
• example for websocket event streams
• remove dependency on "github.com/drone/config" module
• get working nsq driver
• find and implement more drivers maybe faster then nats.io
  • sarama client for Apache Kafka
  • zmq4 client for zeromq

Documentation

Index

• func Cleanup()
• func Init(config HubConfig) (err error)
• func Marshal(value interface{}) ([]byte, error)
• func Publish(channels []string, msg interface{}) error
• func RegisterDriver(d Driver, knownNames ...string)
• func Unmarshal(data []byte) (interface{}, error)
• type Channel
  • func Subscribe(channels []string) (Channel, error)
• type Driver
• type Event
• type Hub
  • func MakeHub(config HubConfig) (Hub, error)
  • func NewHub() Hub
• type HubConfig
  • func (c HubConfig) GetInt(key string, defval int) int
  • func (c HubConfig) GetString(key string, defval string) string

Functions

func Cleanup

func Cleanup()

Cleanup pubsub facilities.

func Init

func Init(config HubConfig) (err error)

func Marshal

func Marshal(value interface{}) ([]byte, error)

Marshal message to byte array.

func Publish

func Publish(channels []string, msg interface{}) error

Publish message to given channels.

func RegisterDriver

func RegisterDriver(d Driver, knownNames ...string)

RegisterDriver registers driver with given names.

func Unmarshal

func Unmarshal(data []byte) (interface{}, error)

Unmarshal message from byte array.

Types

type Channel

type Channel interface {
	// Read returns channel to receive events.
	Read() <-chan interface{}
	// Close stops listening underlying pubsub channels.
	Close() error
	// CloseNotify returns channel to receive event when this channel is closed.
	CloseNotify() <-chan bool
}

Channel to listen pubsub events.

func Subscribe

func Subscribe(channels []string) (Channel, error)

Subscribe on given channels.

type Driver

type Driver interface {
	Create(config HubConfig) (Hub, error)
}

Driver defines interface for pubsub modules

type Event

type Event struct {
	ID           string      `json:"id,omitempty"`     // event id
	Action       string      `json:"action,omitempty"` // http method or specific action
	Method       string      `json:"method,omitempty"` // http method
	URL          string      `json:"url,omitempty"`
	ResourceID   string      `json:"resource_id,omitempty"`   // resource id
	ResourceType string      `json:"resource_type,omitempty"` // resource type
	Payload      interface{} `json:"payload,omitempty"`       // input payload
	CreatedBy    string      `json:"created_by,omitempty"`
	CreatedAt    time.Time   `json:"created_at,omitempty"`
	Result       interface{} `json:"result,omitempty"` // any result of mutation
}

Event defines abstract event usually about HTTP update

type Hub

type Hub interface {
	// Publish sends input message to specified channels.
	Publish(channels []string, msg interface{})
	// Subscribe opens channel to listen specified channels.
	Subscribe(channels []string) (Channel, error)
	// Close stops the pubsub hub.
	Close() error
}

Hub interface of pubsub system.

func MakeHub

func MakeHub(config HubConfig) (Hub, error)

MakeHub returns new instance of the pubsub hub.

func NewHub

func NewHub() Hub

NewHub creates new in-process pubsub hub.

type HubConfig

type HubConfig map[string]interface{}

HubConfig defines input for Driver.Create function.

func (HubConfig) GetInt

func (c HubConfig) GetInt(key string, defval int) int

GetInt property.

func (HubConfig) GetString

func (c HubConfig) GetString(key string, defval string) string

GetString property.