74 lines
1.3 KiB
Go
74 lines
1.3 KiB
Go
package reactive
|
|
|
|
import (
|
|
"context"
|
|
|
|
"codeberg.org/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()
|
|
}
|