ultrago/reactive/subject.go

package reactive

import (
	"context"

	"git.minkystudios.ru/noth/ultrago/reactive/multicast"
)

type Subject[T any] struct {
	multi *multicast.Chan[Notification[T]]
	err error
}

func NewSubject[T any]() *Subject[T] {
	return &Subject[T]{
		multi: multicast.NewChan[Notification[T]](0, 0),
	}
}

func (s *Subject[T]) Observe(ctx context.Context) <-chan Notification[T] {
	if s.err != nil {
		ch := make(chan Notification[T])
		ch <- MakeError[T](s.err)
		return ch
	}

	if s.multi.Closed() {
		return nil
	}

	endpoint, err := s.multi.NewEndpoint(0)
	if err != nil {
		// revisit behavior? maybe return a channel with MakeError(err)?
		panic(err)
	}

	ch := make(chan Notification[T])
	endpoint.Range(func(value Notification[T], closed bool) bool {
		ch <- value
		return true
	}, func(err error, closed bool) bool {
		if err != nil {
			ch <- MakeError[T](err)
		}
		if closed {
			close(ch)
		}
		return true
	}, 0)

	go func() {
		<-ctx.Done()
		endpoint.Cancel()
	}()

	return ch
}

func (s *Subject[T]) OnNext(value T) {
	s.multi.FastSend(MakeNext[T](value))
}

func (s *Subject[T]) OnError(err error) {
	s.multi.FastSend(MakeError[T](err))
	s.multi.Close(err)
}

func (s *Subject[T]) OnCompleted() {
	s.multi.Close(nil)
}

func (s *Subject[T]) Closed() bool {
	return s.multi.Closed()
}