pubsub

README

Package pubsub implements a simple multi-topic pub-sub library.

Install pubsub with,

go get github.com/cskr/pubsub

This repository is a go module and contains tagged releases. Please pin a version for production use.

Use of this module is governed by a BSD-style license that can be found in the LICENSE file.

Documentation

Overview

Package pubsub implements a simple multi-topic pub-sub library.

Topics must be strings and messages of any type can be published. A topic can have any number of subcribers and all of them receive messages published on the topic.

Example

package main

import (
	"fmt"

	"github.com/cskr/pubsub"
)

const topic = "topic"

func main() {
	ps := pubsub.New(0)
	ch := ps.Sub(topic)
	go publish(ps)

	for i := 1; ; i++ {
		if i == 5 {
			// See the documentation of Unsub for why it is called in a new
			// goroutine.
			go ps.Unsub(ch, "topic")
		}

		if msg, ok := <-ch; ok {
			fmt.Printf("Received %s, %d times.\n", msg, i)
		} else {
			break
		}
	}
}

func publish(ps *pubsub.PubSub) {
	for {
		ps.Pub("message", topic)
	}
}

Output:

Received message, 1 times.
Received message, 2 times.
Received message, 3 times.
Received message, 4 times.
Received message, 5 times.
Received message, 6 times.

Index

• type PubSub
  • func New(capacity int) *PubSub
  • func (ps *PubSub) AddSub(ch chan interface{}, topics ...string)
  • func (ps *PubSub) AddSubOnceEach(ch chan interface{}, topics ...string)
  • func (ps *PubSub) Close(topics ...string)
  • func (ps *PubSub) Pub(msg interface{}, topics ...string)
  • func (ps *PubSub) Shutdown()
  • func (ps *PubSub) Sub(topics ...string) chan interface{}
  • func (ps *PubSub) SubOnce(topics ...string) chan interface{}
  • func (ps *PubSub) SubOnceEach(topics ...string) chan interface{}
  • func (ps *PubSub) TryPub(msg interface{}, topics ...string)
  • func (ps *PubSub) Unsub(ch chan interface{}, topics ...string)

Examples

• Package

Types

type PubSub

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

PubSub is a collection of topics.

func New

func New(capacity int) *PubSub

New creates a new PubSub and starts a goroutine for handling operations. The capacity of the channels created by Sub and SubOnce will be as specified.

func (*PubSub) AddSub

func (ps *PubSub) AddSub(ch chan interface{}, topics ...string)

AddSub adds subscriptions to an existing channel.

func (*PubSub) AddSubOnceEach

func (ps *PubSub) AddSubOnceEach(ch chan interface{}, topics ...string)

AddSubOnceEach adds subscriptions to an existing channel with SubOnceEach behavior.

func (*PubSub) Close

func (ps *PubSub) Close(topics ...string)

Close closes all channels currently subscribed to the specified topics. If a channel is subscribed to multiple topics, some of which is not specified, it is not closed.

func (*PubSub) Pub

func (ps *PubSub) Pub(msg interface{}, topics ...string)

Pub publishes the given message to all subscribers of the specified topics.

func (*PubSub) Shutdown

func (ps *PubSub) Shutdown()

Shutdown closes all subscribed channels and terminates the goroutine.

func (*PubSub) Sub

func (ps *PubSub) Sub(topics ...string) chan interface{}

Sub returns a channel on which messages published on any of the specified topics can be received.

func (*PubSub) SubOnce

func (ps *PubSub) SubOnce(topics ...string) chan interface{}

SubOnce is similar to Sub, but only the first message published, after subscription, on any of the specified topics can be received.

func (*PubSub) SubOnceEach

func (ps *PubSub) SubOnceEach(topics ...string) chan interface{}

SubOnceEach returns a channel on which callers receive, at most, one message for each topic.

func (*PubSub) TryPub

func (ps *PubSub) TryPub(msg interface{}, topics ...string)

TryPub publishes the given message to all subscribers of the specified topics if the topic has buffer space.

func (*PubSub) Unsub

func (ps *PubSub) Unsub(ch chan interface{}, topics ...string)

Unsub unsubscribes the given channel from the specified topics. If no topic is specified, it is unsubscribed from all topics.

Unsub must be called from a goroutine that is different from the subscriber. The subscriber must consume messages from the channel until it reaches the end. Not doing so can result in a deadlock.