Chore: use generics as possible

common/observable/observable.go

@@ -5,14 +5,14 @@ import (
 	"sync"
 )
 
-type Observable struct {
-	iterable Iterable
-	listener map[Subscription]*Subscriber
+type Observable[T any] struct {
+	iterable Iterable[T]
+	listener map[Subscription[T]]*Subscriber[T]
 	mux      sync.Mutex
 	done     bool
 }
 
-func (o *Observable) process() {
+func (o *Observable[T]) process() {
 	for item := range o.iterable {
 		o.mux.Lock()
 		for _, sub := range o.listener {
@@ -23,7 +23,7 @@ func (o *Observable) process() {
 	o.close()
 }
 
-func (o *Observable) close() {
+func (o *Observable[T]) close() {
 	o.mux.Lock()
 	defer o.mux.Unlock()
 
@@ -33,18 +33,18 @@ func (o *Observable) close() {
 	}
 }
 
-func (o *Observable) Subscribe() (Subscription, error) {
+func (o *Observable[T]) Subscribe() (Subscription[T], error) {
 	o.mux.Lock()
 	defer o.mux.Unlock()
 	if o.done {
-		return nil, errors.New("Observable is closed")
+		return nil, errors.New("observable is closed")
 	}
-	subscriber := newSubscriber()
+	subscriber := newSubscriber[T]()
 	o.listener[subscriber.Out()] = subscriber
 	return subscriber.Out(), nil
 }
 
-func (o *Observable) UnSubscribe(sub Subscription) {
+func (o *Observable[T]) UnSubscribe(sub Subscription[T]) {
 	o.mux.Lock()
 	defer o.mux.Unlock()
 	subscriber, exist := o.listener[sub]
@@ -55,10 +55,10 @@ func (o *Observable) UnSubscribe(sub Subscription) {
 	subscriber.Close()
 }
 
-func NewObservable(any Iterable) *Observable {
-	observable := &Observable{
-		iterable: any,
-		listener: map[Subscription]*Subscriber{},
+func NewObservable[T any](iter Iterable[T]) *Observable[T] {
+	observable := &Observable[T]{
+		iterable: iter,
+		listener: map[Subscription[T]]*Subscriber[T]{},
 	}
 	go observable.process()
 	return observable