observable.go

package observable

import (
	"sync"

	"github.com/twinj/uuid"

	log "github.com/sirupsen/logrus"
)

// Observable is a variable that can be observed for changes.
type Observable struct {
	value     []byte
	listeners map[uuid.UUID]chan []byte
	lock      sync.Mutex
}

// New creates a new observable.
func New() *Observable {
	return &Observable{
		listeners: map[uuid.UUID]chan []byte{},
	}
}

// NewWithValue creates a new observable with an initial value.
func NewWithValue(value []byte) *Observable {
	o := New()

	o.value = value

	return o
}

// SetValue sets the value of the observable. All listeners will be informed.
func (o *Observable) SetValue(value []byte) {
	o.value = value

	for _, v := range o.listeners {
		v <- value
	}
}

// GetValue gets the value of the observable.
func (o *Observable) GetValue() []byte {
	return o.value
}

// Register interest in value changes.
func (o *Observable) Register() (uuid.UUID, chan []byte) {
	c := make(chan []byte)
	id := uuid.NewV4()

	o.lock.Lock()
	defer o.lock.Unlock()

	o.listeners[id] = c

	log.Debugf("Registered listener %s.", id)

	return id, c
}

// Unregister interest in value changes.
func (o *Observable) Unregister(id uuid.UUID) {
	o.lock.Lock()
	defer o.lock.Unlock()

	c, ok := o.listeners[id]

	if !ok {
		log.Warnf("Listener %s does not exist.", id)
		return
	}

	delete(o.listeners, id)

	close(c)

	log.Debugf("Unregistered listener %s.", id)
}