pubsub

README

Simple in-memory pubsub library

Requirements

• Golang 1.11+
• dep 0.5.0
• golangci-lint
  or
• nix - just run nix-shell in project's folder and it will drop you into development environment

Usage

package main

import (
	"fmt"
	"log"
)

func main() {
	s := pubsub.NewServer()
	topic := "topic"
	subscriber := "sub"
	msg := []byte(`{"foo":"bar"}`)
	s.Subscribe(topic, subscriber)
	s.Publish(topic, msg)
	msg, err := s.Poll(topic, subscriber)
	if err != nil {
		log.Fatalf("Unexpected error: %v", err)
	}
	s.Unsubscribe(topic, subscriber)

	fmt.Println(string(msg))
	// Output: {"foo":"bar"}
}

There is also executable example in example_test.go.

Testing

Run:

golangci-ling run && go test -v -timeout=30s --cover --race ./...

or with nix:

nix-shell --run "pubsub-lint-run-tests"

Ideas to test

Current implementation uses mutexes for concurrency control and linked list from stdlib to store messages. It would be great to implement and compare performance with next implementations:

• channel based communication
• stm instead of mutexes
• Review possible implementations for concurrent queues (right now we have RWMutex on linked list for all operations)

Documentation

Index

• Variables
• type Server
  • func NewServer() *Server
  • func (s *Server) Poll(topic, subscriber string) ([]byte, error)
  • func (s *Server) Publish(topic string, msg []byte)
  • func (s *Server) Subscribe(topic string, subscriber string)
  • func (s *Server) Unsubscribe(topic, subscriber string)

Examples

• Server

Variables

var ErrSubscriptionNotFound = errors.New("pubsub: subscription not found")

Types

type Server

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

Server should be created via pubsub.NewServer fuction

Example

s := NewServer()
topic := "topic"
subscriber := "sub"
msg := []byte(`{"foo":"bar"}`)
s.Subscribe(topic, subscriber)
s.Publish(topic, msg)
msg, err := s.Poll(topic, subscriber)
if err != nil {
	log.Fatalf("Unexpected error: %v", err)
}
s.Unsubscribe(topic, subscriber)

fmt.Println(string(msg))

Output:

{"foo":"bar"}

func NewServer

func NewServer() *Server

NewServer initialize pubsub server. Please create Server only via NewServer

func (*Server) Poll

func (s *Server) Poll(topic, subscriber string) ([]byte, error)

Poll - returns next message on topic of subscriber and removes it from subscribers queue. If there are no new messages it will return nil, nil. If there is no subscription it will return pubsub.ErrSubscriptionNotFound

func (*Server) Publish

func (s *Server) Publish(topic string, msg []byte)

Publish - deliver msg to all topic subscribers This function traverse all topic subscribers, so it is better to call it in async manner: go s.Publish(topic, msg)

func (*Server) Subscribe

func (s *Server) Subscribe(topic string, subscriber string)

Subscribe - subscribes subscriber to topic. Subscriber can receive messages received after subscription on this topic with Poll

func (*Server) Unsubscribe

func (s *Server) Unsubscribe(topic, subscriber string)

Unsubscribe - remove topic's subscription for client