pubsub: github.com/juju/pubsub Index | Files

package pubsub

import "github.com/juju/pubsub"

Package pubsub provides publish and subscribe functionality within a single process.

A message as far as a hub is concerned is defined by a topic, and a data blob. All subscribers that match the published topic are notified, and have their callback function called with both the topic and the data blob.

All subscribers get their own goroutine. This way slow consumers do not slow down the act of publishing, and slow consumers do not inferfere with other consumers. Subscribers are guaranteed to get the messages that match their topic matcher in the order that the messages were published to the hub.

This package defines two types of hubs. * Simple hubs * Structured hubs

Simple hubs just pass the datablob to the subscribers untouched. Structuctured hubs will serialize the datablob into a `map[string]interface{}` using the marshaller that was defined to create it. The subscription handler functions for structured hubs allow the handlers to define a structure for the datablob to be marshalled into.

Hander functions for a structured hub can get all the published data available by defining a callback with the signature:

func (Topic, map[string]interface{})

Or alternatively, define a struct type, and use that type as the second argument.

func (Topic, SomeStruct, error)

The structured hub will try to serialize the published information into the struct specified. If there is an error marshalling, that error is passed to the callback as the error parameter.

Index

Package Files

doc.go logger.go matcher.go multiplexer.go simplehub.go simplemultiplexer.go structuredcallback.go structuredhub.go subscriber.go

Variables

var JSONMarshaller = &jsonMarshaller{}

JSONMarshaller simply wraps the json.Marshal and json.Unmarshal calls for the Marshaller interface.

func MatchAll Uses

func MatchAll(_ string) bool

MatchAll is a topic matcher that matches all topics.

func MatchRegexp Uses

func MatchRegexp(expression string) func(topic string) bool

MatchRegexp expects a valid regular expression. If the expression passed in is not valid, the function panics. The expected use of this is to be able to do something like:

hub.SubscribeMatch(pubsub.MatchRegex("prefix.*suffix"), handler)

type Logger Uses

type Logger interface {
    Debugf(format string, values ...interface{})
    Tracef(format string, values ...interface{})
}

Logger represents methods called by this package to a logging system.

type Marshaller Uses

type Marshaller interface {
    // Marshal converts the argument into a byte streem that it can then Unmarshal.
    Marshal(interface{}) ([]byte, error)

    // Unmarshal attempts to convert the byte stream into type passed in as the
    // second arg.
    Unmarshal([]byte, interface{}) error
}

Marshaller defines the Marshal and Unmarshal methods used to serialize and deserialize the structures used in Publish and Subscription handlers of the structured hub.

type Multiplexer Uses

type Multiplexer interface {
    Add(topic string, handler interface{}) error
    AddMatch(matcher func(string) bool, handler interface{}) error
    Unsubscribe()
}

Multiplexer allows multiple subscriptions to be made sharing a single message queue from the hub. This means that all the messages for the various subscriptions are called back in the order that the messages were published. If more than one handler is added to the Multiplexer that matches any given topic, the handlers are called back one after the other in the order that they were added.

type SimpleHub Uses

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

SimpleHub provides the base functionality of dealing with subscribers, and the notification of subscribers of events.

func NewSimpleHub Uses

func NewSimpleHub(config *SimpleHubConfig) *SimpleHub

NewSimpleHub returns a new SimpleHub instance.

A simple hub does not touch the data that is passed through to Publish. This data is passed through to each Subscriber. Note that all subscribers are notified in parallel, and that no modification should be done to the data or data races will occur.

func (*SimpleHub) NewMultiplexer Uses

func (h *SimpleHub) NewMultiplexer() SimpleMultiplexer

NewMultiplexer creates a new multiplexer for the hub and subscribes it. Unsubscribing the multiplexer stops calls for all handlers added.

func (*SimpleHub) Publish Uses

func (h *SimpleHub) Publish(topic string, data interface{}) <-chan struct{}

Publish will notifiy all the subscribers that are interested by calling their handler function.

The data is passed through to each Subscriber untouched. Note that all subscribers are notified in parallel, and that no modification should be done to the data or data races will occur.

The channel return value is closed when all the subscribers have been notified of the event.

func (*SimpleHub) Subscribe Uses

func (h *SimpleHub) Subscribe(topic string, handler func(string, interface{})) func()

Subscribe to a topic with a handler function. If the topic is the same as the published topic, the handler function is called with the published topic and the associated data.

The return value is a function that will unsubscribe the caller from the hub, for this subscription.

func (*SimpleHub) SubscribeMatch Uses

func (h *SimpleHub) SubscribeMatch(matcher func(string) bool, handler func(string, interface{})) func()

SubscribeMatch takes a function that determins whether the topic matches, and a handler function. If the matcher matches the published topic, the handler function is called with the published topic and the associated data.

The return value is a function that will unsubscribe the caller from the hub, for this subscription.

type SimpleHubConfig Uses

type SimpleHubConfig struct {
    // Logger allows specifying a logging implementation for debug
    // and trace level messages emitted from the hub.
    Logger Logger
}

SimpleHubConfig is the argument struct for NewSimpleHub.

type SimpleMultiplexer Uses

type SimpleMultiplexer interface {
    Add(topic string, handler func(string, interface{}))
    AddMatch(matcher func(string) bool, handler func(string, interface{}))
    Unsubscribe()
}

SimpleMultiplexer allows multiple subscriptions to be made sharing a single message queue from the hub. This means that all the messages for the various subscriptions are called back in the order that the messages were published. If more than one handler is added to the SimpleMultiplexer that matches any given topic, the handlers are called back one after the other in the order that they were added.

type StructuredHub Uses

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

StructuredHub allows the hander functions to accept either structures or map[string]interface{}. The published structure does not need to match the structures of the subscribers. The structures are marshalled using the Marshaller defined in the StructuredHubConfig. If one is not specified, the marshalling is handled by the standard json library.

func NewStructuredHub Uses

func NewStructuredHub(config *StructuredHubConfig) *StructuredHub

NewStructuredHub returns a new StructuredHub instance.

func (*StructuredHub) NewMultiplexer Uses

func (h *StructuredHub) NewMultiplexer() (Multiplexer, error)

NewMultiplexer creates a new multiplexer for the hub and subscribes it. Unsubscribing the multiplexer stops calls for all handlers added. Only structured hubs support multiplexer.

func (*StructuredHub) Publish Uses

func (h *StructuredHub) Publish(topic string, data interface{}) (<-chan struct{}, error)

Publish will notifiy all the subscribers that are interested by calling their handler function.

The data is serialized out using the marshaller and then back into a map[string]interface{}. If there is an error marshalling the data, Publish fails with an error. The resulting map is then updated with any annotations provided. The annotated values are only set if the specified field is missing or empty. After the annotations are set, the PostProcess function is called if one was specified. The resulting map is then passed to each of the subscribers.

Subscribers are notified in parallel, and that no modification should be done to the data or data races will occur.

The channel return value is closed when all the subscribers have been notified of the event.

func (*StructuredHub) Subscribe Uses

func (h *StructuredHub) Subscribe(topic string, handler interface{}) (func(), error)

Subscribe takes a topic with a handler function. If the topic is the same as the published topic, the handler function is called with the published topic and the associated data.

The function return value is a function that will unsubscribe the caller from the hub, for this subscription.

The hander function must have the signature:

`func(string, map[string]interface{})`

or

`func(string, SomeStruct, error)`

where `SomeStruct` is any structure.

If the hander function does not match one of these signatures, the Subscribe function returns an error.

The map[string]interface{} from the Publish call is unmarshalled into the `SomeStruct` structure. If there is an error unmarshalling the handler is called with a zerod structure and an error with the marshalling error.

func (*StructuredHub) SubscribeMatch Uses

func (h *StructuredHub) SubscribeMatch(matcher func(string) bool, handler interface{}) (func(), error)

SubscribeMatch takes a function that determins whether the topic matches, and a handler function. If the matcher matches the published topic, the handler function is called with the published topic and the associated data.

All other aspects of the function are the same as the `Subscribe` method.

type StructuredHubConfig Uses

type StructuredHubConfig struct {
    // Logger allows specifying a logging implementation for debug
    // and trace level messages emitted from the hub.
    Logger Logger

    // Marshaller defines how the structured hub will convert from structures to
    // a map[string]interface{} and back. If this is not specified, the
    // `JSONMarshaller` is used.
    Marshaller Marshaller

    // Annotations are added to each message that is published if and only if
    // the values are not already set.
    Annotations map[string]interface{}

    // PostProcess allows the caller to modify the resulting
    // map[string]interface{}. This is useful when a dynamic value, such as a
    // timestamp is added to the map, or when other type conversions are
    // necessary across all the values in the map.
    PostProcess func(map[string]interface{}) (map[string]interface{}, error)
}

StructuredHubConfig is the argument struct for NewStructuredHub.

Package pubsub imports 6 packages (graph) and is imported by 84 packages. Updated 2019-05-04. Refresh now. Tools for package owners.