observer

README

observer

observer is a Go package that aims to simplify the problem of channel-based broadcasting of events from one or more publishers to one or more observers.

Problem

The typical quick-and-dirty approach to notifying a set of observers in go is to use channels and call each in a for loop, like the following:

for _, channel := range channels {
  channel <- value
}

There are two problems with this approach:

• The broadcaster blocks every time some channel is not ready to be written to.
• If the broadcaster blocks for some channel, the remaining channels will not be written to (and therefore not receive the event) until the blocking channel is finally ready.
• It is O(N). The more observers you have, the worse this loop will behave.

Of course, this could be solved by creating one goroutine for each channel so the broadcaster doesn't block. Unfortunately, this is heavy and resource-consuming. This is especially bad if you have events being raised frequently and a considerable number of observers.

Approach

The way observer package tackles this problem is very simple. For every event, a state object containing information about the event, and a channel is created. State objects are managed using a singly linked list structure: every state points to the next. When a new event is raised, a new state object is appended to the list and the channel of the previous state is closed (this helps notify all observers that the previous state is outdated).

Package observer defines 2 concepts:

• Property: An object that is continuously updated by one or more publishers.
• Stream: The list of values a property is updated to. For every property update, that value is appended to the list in the order they happen, and is only discarded when you advance to the next value.

Memory Usage

The amount of memory used for one property is not dependent on the number of observers. It should be proportional to the number of value updates since the value last obtained by the slowest observer. As long as you keep advancing all your observers, garbage collection will take place and keep memory usage stable.

How to Use

First, you need to install the package:

go get -u github.com/imkira/go-observer

Then, you need to include it in your source:

import "github.com/imkira/go-observer"

The package will be imported with observer as name.

The following example creates one property that is updated every second by one or more publishers, and observed by one or more observers.

Documentation

For advanced usage, make sure to check the available documentation here.

Example: Creating a Property

The following code creates a property with initial value 1.

val := 1
prop := observer.NewProperty(val)

After creating the property, you can pass it around to publishers or observers as you want.

Example: Publisher

The following code represents a publisher that increments the value of the property by one every second.

val := 1
for {
  time.Sleep(time.Second)
  val += 1
  fmt.Printf("will publish value: %d\n", val)
  prop.Update(val)
}

Note:

• Property is goroutine safe: you can use it concurrently from multiple goroutines.

Example: Observer

The following code represents an observer that prints the initial value of a property and waits indefinitely for changes to its value. When there is a change, the stream is advanced and the current value of the property is printed.

stream := prop.Observe()
val := stream.Value()
fmt.Printf("initial value: %d\n", val)
for {
  select {
    // wait for changes
    case <-stream.Changes():
      // advance to next value
      stream.Next()
      // new value
      val = stream.Value()
      fmt.Printf("got new value: %d\n", val)
  }
}

Note:

• Stream is not goroutine safe: You must create one stream by calling Property.Observe() or Stream.Clone() if you want to have concurrent observers for the same property or stream.

Example

Please check examples/multiple.go for a simple example on how to use multiple observers with a single updater.

Documentation

Overview

Package observer aims to simplify the problem of channel-based broadcasting of events from one or more publishers to one or more observers.

Index

• type Property
  • func NewProperty[T any](value T) Property[T]
• type Stream

Types

type Property

type Property[T any] interface {
	// Value returns the current value for this property.
	Value() T

	// Update sets a new value for this property.
	Update(value T)

	// Observe returns a newly created Stream for this property.
	Observe() Stream[T]
}

Property is an object that is continuously updated by one or more publishers. It is completely goroutine safe: you can use Property concurrently from multiple goroutines.

func NewProperty

func NewProperty[T any](value T) Property[T]

NewProperty creates a new Property with the initial value value. It returns the created Property.

type Stream

type Stream[T any] interface {
	// Value returns the current value for this stream.
	Value() T

	// Changes returns the channel that is closed when a new value is available.
	Changes() chan struct{}

	// Next advances this stream to the next state.
	// You should never call this unless Changes channel is closed.
	Next() T

	// HasNext checks whether there is a new value available.
	HasNext() bool

	// WaitNext waits for Changes to be closed, advances the stream and returns
	// the current value.
	WaitNext() T

	// WaitNextCtx does the same as WaitNext but returns earlier with an error if the given context is cancelled first.
	WaitNextCtx(ctx context.Context) (T, error)

	// Clone creates a new independent stream from this one but sharing the same
	// Property. Updates to the property will be reflected in both streams but
	// they may have different values depending on when they advance the stream
	// with Next.
	Clone() Stream[T]

	// Peek return the value in the next state
	// You should never call this unless Changes channel is closed.
	Peek() T
}

Stream represents the list of values a property is updated to. For every property update, that value is appended to the list in the order they happen. The value is discarded once you advance the stream. Please note that Stream is not goroutine safe: you cannot use the same stream on multiple goroutines concurrently. If you want to use multiple streams for the same property, either use Property.Observe (goroutine-safe) or use Stream.Clone (before passing it to another goroutine).