[Pkg] feat: add channel observable

pkg/observable/interface.go

@@ -0,0 +1,16 @@
+package observable
+
+import "context"
+
+// Observable is a generic interface that allows multiple subscribers to be
+// notified of new values asynchronously.
+type Observable[V any] interface {
+	Subscribe(context.Context) Subscription[V]
+}
+
+// Subscription is a generic interface that provides access to the notified
+// channel and allows unsubscribing from an observable.
+type Subscription[V any] interface {
+	Unsubscribe()
+	Ch() <-chan V
+}

pkg/observable/notifiable/observable.go

@@ -0,0 +1,93 @@
+package notifiable
+
+import (
+	"context"
+	"sync"
+
+	"pocket/pkg/observable"
+)
+
+// Observable implements the observable.Observable interface and can be notified
+// via its corresponding notifier channel.
+type Observable[V any] struct {
+	mu          sync.RWMutex
+	ch          <-chan V // private channel that is used to emit values to subscribers
+	subscribers []chan V // subscribers is a list of channels that are subscribed to the observable
+	closed      bool
+}
+
+// NewObservable creates a new observable is notified when the notifier channel
+// receives a value.
+func NewObservable[V any](notifier chan V) (observable.Observable[V], chan V) {
+	// If the caller does not provide a notifier, create a new one and return it
+	if notifier == nil {
+		notifier = make(chan V)
+	}
+	notifee := &Observable[V]{sync.RWMutex{}, notifier, []chan V{}, false}
+
+	// Start listening to the notifier and emit values to subscribers
+	go notifee.listen(notifier)
+
+	return notifee, notifier
+}
+
+// Subscribe gets a subscription to the observable.
+func (obs *Observable[V]) Subscribe(ctx context.Context) observable.Subscription[V] {
+	obs.mu.Lock()
+	defer obs.mu.Unlock()
+
+	// Create a channel for the subscriber and append it to the subscribers list
+	ch := make(chan V, 1)
+	obs.subscribers = append(obs.subscribers, ch)
+
+	// Removal function used when unsubscribing from the observable
+	removeFromObservable := func() {
+		obs.mu.Lock()
+		defer obs.mu.Unlock()
+
+		for i, s := range obs.subscribers {
+			if ch == s {
+				obs.subscribers = append(obs.subscribers[:i], obs.subscribers[i+1:]...)
+				break
+			}
+		}
+	}
+
+	// Subscription gets its closed state from the observable
+	subscription := &Subscription[V]{ch, obs.closed, removeFromObservable}
+
+	go unsubscribeOnDone[V](ctx, subscription)
+
+	return subscription
+}
+
+// listen to the notifier and notify subscribers when values are received. This
+// function is blocking and should be run in a goroutine.
+func (obs *Observable[V]) listen(notifier <-chan V) {
+	for v := range notifier {
+		// Lock for obs.subscribers slice as it can be modified by subscribers
+		obs.mu.RLock()
+		for _, ch := range obs.subscribers {
+			ch <- v
+		}
+		obs.mu.RUnlock()
+	}
+
+	// Here we know that the notifier has been closed, all subscribers should be closed as well
+	obs.mu.Lock()
+	obs.closed = true
+	for _, ch := range obs.subscribers {
+		close(ch)
+		obs.subscribers = []chan V{}
+	}
+	obs.mu.Lock()
+}
+
+// unsubscribeOnDone unsubscribes from the subscription when the context is.
+// It is blocking and intended to be called in a goroutine.
+func unsubscribeOnDone[V any](ctx context.Context, subscription observable.Subscription[V]) {
+	if ctx != nil {
+		<-ctx.Done()
+		subscription.Unsubscribe()
+	}
+}

pkg/observable/notifiable/observable_test.go

@@ -0,0 +1,162 @@
+package notifiable_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
+
+	"pocket/pkg/observable"
+	"pocket/pkg/observable/notifiable"
+)
+
+const (
+	notifyTimeout            = 100 * time.Millisecond
+	unsubscribeSleepDuration = notifyTimeout * 2
+)
+
+func TestNewNotifiableObservable(t *testing.T) {
+	type test struct {
+		name     string
+		notifier chan int
+	}
+
+	input := 123
+	nonEmptyBufferedNotifier := make(chan int, 1)
+	nonEmptyBufferedNotifier <- input
+
+	tests := []test{
+		{name: "nil notifier", notifier: nil},
+		{name: "empty non-buffered notifier", notifier: make(chan int)},
+		{name: "empty buffered len 1 notifier", notifier: make(chan int, 1)},
+		{name: "non-empty buffered notifier", notifier: nonEmptyBufferedNotifier},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx := context.Background()
+			notifee, notifier := notifiable.NewObservable[int](tt.notifier)
+			require.NotNil(t, notifee)
+			require.NotNil(t, notifier)
+
+			// construct 3 distinct subscriptions, each with its own channel
+			subscriptions := make([]observable.Subscription[int], 3)
+			for i := range subscriptions {
+				subscriptions[i] = notifee.Subscribe(ctx)
+			}
+
+			group := errgroup.Group{}
+			notifiedOrTimedOut := func(subscriptionCh <-chan int) func() error {
+				return func() error {
+					// subscriptionCh should receive notified input
+					select {
+					case output := <-subscriptionCh:
+						require.Equal(t, input, output)
+					case <-time.After(notifyTimeout):
+						return fmt.Errorf("timed out waiting for subscription to be notified")
+					}
+					return nil
+				}
+			}
+
+			// ensure all subscription channels are notified
+			for _, subscription := range subscriptions {
+				// concurrently await notification or timeout to avoid blocking on
+				// empty and/or non-buffered notifiers.
+				group.Go(notifiedOrTimedOut(subscription.Ch()))
+			}
+
+			// notify with test input
+			notifier <- input
+
+			// wait for notifee to be notified or timeout
+			err := group.Wait()
+			require.NoError(t, err)
+
+			// unsubscribing should close subscription channel(s)
+			for _, subscription := range subscriptions {
+				subscription.Unsubscribe()
+
+				select {
+				case <-subscription.Ch():
+				default:
+					t.Fatal("subscription channel left open")
+				}
+			}
+		})
+	}
+}
+
+func TestSubscription_Unsubscribe(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	notifee, notifier := notifiable.NewObservable[int](nil)
+	require.NotNil(t, notifee)
+	require.NotNil(t, notifier)
+
+	tests := []struct {
+		name        string
+		lifecycleFn func() observable.Subscription[int]
+	}{
+		{
+			name: "nil context",
+			lifecycleFn: func() observable.Subscription[int] {
+				subscription := notifee.Subscribe(nil)
+				subscription.Unsubscribe()
+				return subscription
+			},
+		},
+		{
+			name: "only unsubscribe",
+			lifecycleFn: func() observable.Subscription[int] {
+				subscription := notifee.Subscribe(ctx)
+				subscription.Unsubscribe()
+				return subscription
+			},
+		},
+		{
+			name: "only cancel",
+			lifecycleFn: func() observable.Subscription[int] {
+				subscription := notifee.Subscribe(ctx)
+				cancel()
+				return subscription
+			},
+		},
+		{
+			name: "cancel then unsubscribe",
+			lifecycleFn: func() observable.Subscription[int] {
+				subscription := notifee.Subscribe(ctx)
+				cancel()
+				time.Sleep(unsubscribeSleepDuration)
+				subscription.Unsubscribe()
+				return subscription
+			},
+		},
+		{
+			name: "unsubscribe then cancel",
+			lifecycleFn: func() observable.Subscription[int] {
+				subscription := notifee.Subscribe(ctx)
+				subscription.Unsubscribe()
+				time.Sleep(unsubscribeSleepDuration)
+				cancel()
+				return subscription
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			subscription := tt.lifecycleFn()
+
+			select {
+			case value, ok := <-subscription.Ch():
+				require.Empty(t, value)
+				require.False(t, ok)
+			case <-time.After(notifyTimeout):
+				t.Fatal("subscription channel left open")
+			}
+		})
+	}
+}

pkg/observable/notifiable/subscription.go

@@ -0,0 +1,29 @@
+package notifiable
+
+import "pocket/pkg/observable"
+
+var _ observable.Subscription[any] = &Subscription[any]{}
+
+// Subscription implements the observable.Subscription interface.
+type Subscription[V any] struct {
+	ch                   chan V
+	closed               bool
+	removeFromObservable func()
+}
+
+// Unsubscribe closes the subscription channel and removes the subscription from
+// the observable.
+func (sub *Subscription[V]) Unsubscribe() {
+	if sub.closed {
+		return
+	}
+
+	close(sub.ch)
+	sub.closed = true
+	sub.removeFromObservable()
+}
+
+// Ch returns the subscription channel.
+func (sub *Subscription[V]) Ch() <-chan V {
+	return sub.ch
+}