subscriber

README

subscriber

import "github.com/reactivego/subscriber"

Package subscriber provides a subscription tree implementation.

The New function creates a subscription and returns a Subscriber interface. It allows a publisher and subscribing client to communicate about a subscription. The publisher can see whether the client is still subscribed by polling the Closed method. The client can control the subscription by either calling Unsubscribe itself or when it has defered subscription management to other code call Wait to wait for other code to cancel the subscription.

Once the client has called Unsubscribe on the subscription, that subscription will then call Unsubscribe on all of its children created through the Add method. After that it will become lifeless, meaning no events will ever be triggered by it anymore.

In case the publisher decides it wants to stop publishing, it MUST NEVER call Unsubscribe itself on a subscriber, but should indicate the desire through other (out-of-band) means to the subscribing client who must then call Unsubscribe itself.

The client that receives a Subcriber interface can call Wait to wait for the subscription to be canceled. That may be done by the code that e.g. created the subscription in responese to some external event.

The client can also keep a reference to the subscription and call Unsubscribe itself to indicate to the publisher it is no longer interested in receiving data associated with the subscription.

The implementation is designed to be used from concurrently running goroutines. It uses WaitGroups, Mutexes and atomic reference counting.

Documentation

Overview

Package subscriber provides a subscription tree implementation.

The New function creates a subscription and returns a Subscriber interface. It allows a publisher and subscribing client to communicate about a subscription. The publisher can see whether the client is still subscribed by polling the Closed method. The client can control the subscription by either calling Unsubscribe itself or when it has defered subscription management to other code call Wait to wait for other code to cancel the subscription.

Once the client has called Unsubscribe on the subscription, that subscription will then call Unsubscribe on all of its children created through the Add method. After that it will become lifeless, meaning no events will ever be triggered by it anymore.

In case the publisher decides it wants to stop publishing, it MUST NEVER call Unsubscribe itself on a subscriber, but should indicate the desire through other (out-of-band) means to the subscribing client who must then call Unsubscribe itself.

The client that receives a Subcriber interface can call Wait to wait for the subscription to be canceled. That may be done by the code that e.g. created the subscription in responese to some external event.

The client can also keep a reference to the subscription and call Unsubscribe itself to indicate to the publisher it is no longer interested in receiving data associated with the subscription.

The implementation is designed to be used from concurrently running goroutines. It uses WaitGroups, Mutexes and atomic reference counting.

Example (Basic)

var s subscriber
if s.Subscribed() == s.Canceled() {
	fmt.Println("subscriber can't be both subscribed and canceled")
}
if s.Canceled() {
	fmt.Println("subscriber should be subscribed by default")
}
s.Unsubscribe()
if s.Subscribed() {
	fmt.Println("subscriber should be canceled after Unsubscribe call")
}

fmt.Println("OK")

Output:

OK

Example (SubscriberLoop)

parent := &subscriber{}

child1 := parent.Add()
child2 := parent.Add()
child2.OnUnsubscribe(parent.Unsubscribe)
child3 := parent.Add()

if parent.Canceled() || child1.Canceled() || child2.Canceled() || child3.Canceled() {
	fmt.Println("none of the subscribers should be cancelled here")
}

child2.Unsubscribe()

if parent.Subscribed() || child1.Subscribed() || child2.Subscribed() || child3.Subscribed() {
	fmt.Println("all of the subscribers should be cancelled here")
}

fmt.Println("OK")

Output:

OK

Index

• Constants
• type Error
  • func (e Error) Error() string
• type Subscriber
  • func New() Subscriber
• type Subscription

Examples

• Package (Basic)
• Package (SubscriberLoop)

Constants

const ErrUnsubscribed = Error("subscriber unsubscribed")

Unsubscribed is the error returned by wait when the Unsubscribe method is called on the subscription.

Types

type Error

type Error string

Error type to allow constant error declarations.

func (Error) Error

func (e Error) Error() string

Error method implements Error interface.

type Subscriber

type Subscriber interface {
	// A Subscriber is also a Subscription.
	Subscription

	// Add will create and return a new child Subscriber setup in such a way that
	// calling Unsubscribe on the parent will also call Unsubscribe on the child.
	// Calling the Unsubscribe method on the child will NOT propagate to the
	// parent!
	Add() Subscriber

	// OnUnsubscribe will add the given callback function to the subscriber.
	// The callback will be called when either the Unsubscribe of the parent
	// or of the subscriber itself is called. If the subscription was already
	// canceled, then the callback function will just be called immediately.
	OnUnsubscribe(callback func())

	// OnWait will register a callback to  call when subscription Wait is called.
	OnWait(callback func())

	// Done will set the error internally and then cancel the subscription by
	// calling the Unsubscribe method. A nil value for error indicates success.
	Done(err error)

	// Error returns the error set by calling the Done(err) method. As long as
	// the subscriber is still subscribed Error will return nil.
	Error() error
}

Subscriber is a Subscription with management functionality.

func New

func New() Subscriber

New will create and return a new Subscriber.

type Subscription

type Subscription interface {
	// Subscribed returns true when the subscription is currently active.
	Subscribed() bool

	// Unsubscribe will do nothing if the subscription is not active. If the
	// state is still active however, it will be changed to canceled.
	// Subsequently, it will call Unsubscribe on all child subscriptions added
	// through Add, along with all methods added through OnUnsubscribe. When the
	// subscription is canceled by calling Unsubscribe a call to the Wait method
	// will return the error ErrUnsubscribed.
	Unsubscribe()

	// Canceled returns true when the subscription state is canceled.
	Canceled() bool

	// Wait will by default block the calling goroutine and wait for the
	// Unsubscribe method to be called on this subscription.
	// However, when OnWait was called with a callback wait function it will
	// call that instead. Calling Wait on a subscription that has already been
	// canceled will return immediately. If the subscriber was canceled by
	// calling Unsubscribe, then the error returned is ErrUnsubscribed.
	// If the subscriber was terminated by calling Done, then the error
	// returned here is the one passed to Done.
	Wait() error
}

Subscription is an interface that allows code to monitor and control a subscription it received.