Chore: use generics as possible

common/observable/iterable.go

@@ -1,3 +1,3 @@
 package observable
 
-type Iterable <-chan any
+type Iterable[T any] <-chan T

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

common/observable/observable_test.go

@@ -9,8 +9,8 @@ import (
 	"go.uber.org/atomic"
 )
 
-func iterator(item []any) chan any {
-	ch := make(chan any)
+func iterator[T any](item []T) chan T {
+	ch := make(chan T)
 	go func() {
 		time.Sleep(100 * time.Millisecond)
 		for _, elm := range item {
@@ -22,8 +22,8 @@ func iterator(item []any) chan any {
 }
 
 func TestObservable(t *testing.T) {
-	iter := iterator([]any{1, 2, 3, 4, 5})
-	src := NewObservable(iter)
+	iter := iterator[int]([]int{1, 2, 3, 4, 5})
+	src := NewObservable[int](iter)
 	data, err := src.Subscribe()
 	assert.Nil(t, err)
 	count := 0
@@ -34,15 +34,15 @@ func TestObservable(t *testing.T) {
 }
 
 func TestObservable_MultiSubscribe(t *testing.T) {
-	iter := iterator([]any{1, 2, 3, 4, 5})
-	src := NewObservable(iter)
+	iter := iterator[int]([]int{1, 2, 3, 4, 5})
+	src := NewObservable[int](iter)
 	ch1, _ := src.Subscribe()
 	ch2, _ := src.Subscribe()
 	count := atomic.NewInt32(0)
 
 	var wg sync.WaitGroup
 	wg.Add(2)
-	waitCh := func(ch <-chan any) {
+	waitCh := func(ch <-chan int) {
 		for range ch {
 			count.Inc()
 		}
@@ -55,8 +55,8 @@ func TestObservable_MultiSubscribe(t *testing.T) {
 }
 
 func TestObservable_UnSubscribe(t *testing.T) {
-	iter := iterator([]any{1, 2, 3, 4, 5})
-	src := NewObservable(iter)
+	iter := iterator[int]([]int{1, 2, 3, 4, 5})
+	src := NewObservable[int](iter)
 	data, err := src.Subscribe()
 	assert.Nil(t, err)
 	src.UnSubscribe(data)
@@ -65,8 +65,8 @@ func TestObservable_UnSubscribe(t *testing.T) {
 }
 
 func TestObservable_SubscribeClosedSource(t *testing.T) {
-	iter := iterator([]any{1})
-	src := NewObservable(iter)
+	iter := iterator[int]([]int{1})
+	src := NewObservable[int](iter)
 	data, _ := src.Subscribe()
 	<-data
 
@@ -75,18 +75,18 @@ func TestObservable_SubscribeClosedSource(t *testing.T) {
 }
 
 func TestObservable_UnSubscribeWithNotExistSubscription(t *testing.T) {
-	sub := Subscription(make(chan any))
-	iter := iterator([]any{1})
-	src := NewObservable(iter)
+	sub := Subscription[int](make(chan int))
+	iter := iterator[int]([]int{1})
+	src := NewObservable[int](iter)
 	src.UnSubscribe(sub)
 }
 
 func TestObservable_SubscribeGoroutineLeak(t *testing.T) {
-	iter := iterator([]any{1, 2, 3, 4, 5})
-	src := NewObservable(iter)
+	iter := iterator[int]([]int{1, 2, 3, 4, 5})
+	src := NewObservable[int](iter)
 	max := 100
 
-	var list []Subscription
+	var list []Subscription[int]
 	for i := 0; i < max; i++ {
 		ch, _ := src.Subscribe()
 		list = append(list, ch)
@@ -94,7 +94,7 @@ func TestObservable_SubscribeGoroutineLeak(t *testing.T) {
 
 	var wg sync.WaitGroup
 	wg.Add(max)
-	waitCh := func(ch <-chan any) {
+	waitCh := func(ch <-chan int) {
 		for range ch {
 		}
 		wg.Done()
@@ -115,11 +115,11 @@ func TestObservable_SubscribeGoroutineLeak(t *testing.T) {
 }
 
 func Benchmark_Observable_1000(b *testing.B) {
-	ch := make(chan any)
-	o := NewObservable(ch)
+	ch := make(chan int)
+	o := NewObservable[int](ch)
 	num := 1000
 
-	subs := []Subscription{}
+	subs := []Subscription[int]{}
 	for i := 0; i < num; i++ {
 		sub, _ := o.Subscribe()
 		subs = append(subs, sub)
@@ -130,7 +130,7 @@ func Benchmark_Observable_1000(b *testing.B) {
 
 	b.ResetTimer()
 	for _, sub := range subs {
-		go func(s Subscription) {
+		go func(s Subscription[int]) {
 			for range s {
 			}
 			wg.Done()

common/observable/subscriber.go

@@ -4,30 +4,30 @@ import (
 	"sync"
 )
 
-type Subscription <-chan any
+type Subscription[T any] <-chan T
 
-type Subscriber struct {
-	buffer chan any
+type Subscriber[T any] struct {
+	buffer chan T
 	once   sync.Once
 }
 
-func (s *Subscriber) Emit(item any) {
+func (s *Subscriber[T]) Emit(item T) {
 	s.buffer <- item
 }
 
-func (s *Subscriber) Out() Subscription {
+func (s *Subscriber[T]) Out() Subscription[T] {
 	return s.buffer
 }
 
-func (s *Subscriber) Close() {
+func (s *Subscriber[T]) Close() {
 	s.once.Do(func() {
 		close(s.buffer)
 	})
 }
 
-func newSubscriber() *Subscriber {
-	sub := &Subscriber{
-		buffer: make(chan any, 200),
+func newSubscriber[T any]() *Subscriber[T] {
+	sub := &Subscriber[T]{
+		buffer: make(chan T, 200),
 	}
 	return sub
 }