README
¶
pipe
Channel transformers for Golang.
Documentation
¶
Index ¶
- func Converge2[A, B any](ch1 <-chan A, ch2 <-chan B) <-chan any
- func Converge3[A, B, C any](ch1 <-chan A, ch2 <-chan B, ch3 <-chan C) <-chan any
- func ConvergeN(chans ...any) <-chan any
- func Until[T comparable, P Listenable[T]](b P, targets ...T)
- func UntilContext[T comparable, P Listenable[T]](ctx context.Context, b P, targets ...T)
- type Broadcaster
- type BroadcasterC
- type BroadcasterCM
- type BroadcasterM
- type CancelFunc
- type Controller
- type ControllerC
- type ControllerCM
- type ControllerM
- type Listenable
- type ListenableC
- type ListenableCM
- type ListenableM
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func ConvergeN ¶
ConvergeN converges values from arbitary number of channels. Each of chans should be of type <-chan T for some T.
func Until ¶
func Until[T comparable, P Listenable[T]](b P, targets ...T)
Until blocks until one of the conditions satisfies: 1) one of the value from b shows up in targets; 2) b does not accept new listeners (either b is detached or upstream channel closed).
func UntilContext ¶
func UntilContext[T comparable, P Listenable[T]](ctx context.Context, b P, targets ...T)
UntilContext blocks until one of the conditions satisfies: 1) one of the value from b shows up in targets; 2) b does not accept new listeners (either b is detached or upstream channel closed); 3) ctx is canceled.
Types ¶
type Broadcaster ¶
type Broadcaster[T any] struct { // contains filtered or unexported fields }
func Broadcast ¶
func Broadcast[T any](upstream <-chan T) *Broadcaster[T]
Broadcast returns a Broadcaster that pipes values from upstream channel into listeners. Broadcaster gaurantees upstream <- val from outside will NOT block, but if it's detached prematurely, upstream <- val will block again.
func (*Broadcaster) Bind ¶
func (b *Broadcaster) Bind(out chan<- T) CancelFunc
Bind registers out as a new listener, which receives subsequent values from the upstream channel. If the input channel closed or the broadcaster detached, out will be closed immediately. A canceller is returned for canceling the subscription. When called, out will be unregistered and closed.
func (*Broadcaster) Detach ¶
func (b *Broadcaster) Detach()
Detach prematurely detaches the broadcaster from the upstream channel. No more values from upstream channel would be broadcasted, and no more new listeners should be registered.
func (*Broadcaster) Listen ¶
func (b *Broadcaster) Listen() (<-chan T, CancelFunc)
Listen creates a new output channel and registers it as a new listener. The output channel and corresponding canceller is returned.
type BroadcasterC ¶
type BroadcasterC[T comparable] struct { // contains filtered or unexported fields }
func BroadcastC ¶
func BroadcastC[T comparable](in <-chan T) *BroadcasterC[T]
BroadcastC returns a broadcaster with a comparable type T as element type. This allows methods like b.Until(targets...) to be called instead of Until(b, targets...), which helps auto type inference and sometimes saves the typing of type variables.
func (*BroadcasterC) Until ¶
func (b *BroadcasterC) Until(targets ...T)
Shorthand for Until(b, targets...)
func (*BroadcasterC) UntilCh ¶
func (b *BroadcasterC) UntilCh(targets ...T) (<-chan struct{}, CancelFunc)
Shorthand for UntilCh(b, targets...)
func (*BroadcasterC) UntilContext ¶
Shorthand for UntilContext(ctx, b, targets...)
type BroadcasterCM ¶
type BroadcasterCM[T comparable] struct { // contains filtered or unexported fields }
func BroadcastCM ¶
func BroadcastCM[T comparable](in <-chan T, initial T) *BroadcasterCM[T]
BroadcastCM returns a broadcaster with comparable element type and also is able to memorize the latest value.
func (*BroadcasterCM[T]) Current ¶
func (b *BroadcasterCM[T]) Current() T
Current returns the latest value that the broadcaster memorizes.
func (*BroadcasterCM) Until ¶
func (b *BroadcasterCM) Until(targets ...T)
Shorthand for Until(b, targets...)
func (*BroadcasterCM) UntilCh ¶
func (b *BroadcasterCM) UntilCh(targets ...T) (<-chan struct{}, CancelFunc)
Shorthand for UntilCh(b, targets...)
func (*BroadcasterCM) UntilContext ¶
Shorthand for UntilContext(ctx, b, targets...)
type BroadcasterM ¶
type BroadcasterM[T any] struct { // contains filtered or unexported fields }
func BroadcastM ¶
func BroadcastM[T any](upstream <-chan T, initial T) *BroadcasterM[T]
BroadcastM returns a broadcaster that memorizes the latest value from upstream. Newly registered listener will be firstly fed with the memorized latest value, then subsequent values from upstream. If no value coming out of upstream yet, initial is fed. The latest value is stored by value (instead of by reference).
func (*BroadcasterM) Bind ¶
func (b *BroadcasterM) Bind(out chan<- T) CancelFunc
Bind registers out as a new listener, which receives subsequent values from the upstream channel. If the input channel closed or the broadcaster detached, out will be closed immediately. A canceller is returned for canceling the subscription. When called, out will be unregistered and closed.
func (*BroadcasterM[T]) Current ¶
func (b *BroadcasterM[T]) Current() T
Current returns the latest value that the broadcaster memorizes.
func (*BroadcasterM) Detach ¶
func (b *BroadcasterM) Detach()
Detach prematurely detaches the broadcaster from the upstream channel. No more values from upstream channel would be broadcasted, and no more new listeners should be registered.
func (*BroadcasterM) Listen ¶
func (b *BroadcasterM) Listen() (<-chan T, CancelFunc)
Listen creates a new output channel and registers it as a new listener. The output channel and corresponding canceller is returned.
type CancelFunc ¶
type CancelFunc func()
func UntilCh ¶
func UntilCh[T comparable, P Listenable[T]](b P, targets ...T) (signalCh <-chan struct{}, canceller CancelFunc)
UntilCh is the asynchronous version of Until. The returned signalCh will be closed when one of the conditions satisfies: 1) one of the value from b shows up in targets; 2) b does not accept new listeners (either b is detached or upstream channel closed); 3) canceller is called.
type Controller ¶
type Controller[T any] struct { // contains filtered or unexported fields }
A Controller bundles a upstream channel and a broadcaster.
func NewController ¶
func NewController[T any]() *Controller[T]
func (*Controller) Bind ¶
func (b *Controller) Bind(out chan<- T) CancelFunc
Bind registers out as a new listener, which receives subsequent values from the upstream channel. If the input channel closed or the broadcaster detached, out will be closed immediately. A canceller is returned for canceling the subscription. When called, out will be unregistered and closed.
func (*Controller) Detach ¶
func (b *Controller) Detach()
Detach prematurely detaches the broadcaster from the upstream channel. No more values from upstream channel would be broadcasted, and no more new listeners should be registered.
func (*Controller) Listen ¶
func (b *Controller) Listen() (<-chan T, CancelFunc)
Listen creates a new output channel and registers it as a new listener. The output channel and corresponding canceller is returned.
type ControllerC ¶
type ControllerC[T comparable] struct { // contains filtered or unexported fields }
A Controller with a comparable element type.
func NewControllerC ¶
func NewControllerC[T comparable]() *ControllerC[T]
func (ControllerC) Sink ¶
func (s ControllerC) Sink() chan<- T
Sink returns the upstream channel of the controller.
func (*ControllerC) Until ¶
func (b *ControllerC) Until(targets ...T)
Shorthand for Until(b, targets...)
func (*ControllerC) UntilCh ¶
func (b *ControllerC) UntilCh(targets ...T) (<-chan struct{}, CancelFunc)
Shorthand for UntilCh(b, targets...)
func (*ControllerC) UntilContext ¶
Shorthand for UntilContext(ctx, b, targets...)
type ControllerCM ¶
type ControllerCM[T comparable] struct { // contains filtered or unexported fields }
func NewControllerCM ¶
func NewControllerCM[T comparable](initial T) *ControllerCM[T]
A Controller with a comparable element type and memorizable broadcaster.
func (*ControllerCM[T]) Current ¶
func (c *ControllerCM[T]) Current() T
Current returns the latest value that the broadcaster memorizes.
func (ControllerCM) Sink ¶
func (s ControllerCM) Sink() chan<- T
Sink returns the upstream channel of the controller.
func (*ControllerCM) Until ¶
func (b *ControllerCM) Until(targets ...T)
Shorthand for Until(b, targets...)
func (*ControllerCM) UntilCh ¶
func (b *ControllerCM) UntilCh(targets ...T) (<-chan struct{}, CancelFunc)
Shorthand for UntilCh(b, targets...)
func (*ControllerCM) UntilContext ¶
Shorthand for UntilContext(ctx, b, targets...)
type ControllerM ¶
type ControllerM[T any] struct { // contains filtered or unexported fields }
A Controller with a memorizable broadcaster.
func NewControllerM ¶
func NewControllerM[T any](initial T) *ControllerM[T]
func (*ControllerM) Bind ¶
func (b *ControllerM) Bind(out chan<- T) CancelFunc
Bind registers out as a new listener, which receives subsequent values from the upstream channel. If the input channel closed or the broadcaster detached, out will be closed immediately. A canceller is returned for canceling the subscription. When called, out will be unregistered and closed.
func (*ControllerM[T]) Current ¶
func (c *ControllerM[T]) Current() T
Current returns the latest value that the broadcaster memorizes.
func (*ControllerM) Detach ¶
func (b *ControllerM) Detach()
Detach prematurely detaches the broadcaster from the upstream channel. No more values from upstream channel would be broadcasted, and no more new listeners should be registered.
func (*ControllerM) Listen ¶
func (b *ControllerM) Listen() (<-chan T, CancelFunc)
Listen creates a new output channel and registers it as a new listener. The output channel and corresponding canceller is returned.
type Listenable ¶
type Listenable[T any] interface { Bind(out chan<- T) CancelFunc Listen() (out <-chan T, cancel CancelFunc) }
A listenable object that one can bind listeners to.
type ListenableC ¶
type ListenableC[T comparable] interface { Listenable[T] Until(...T) UntilCh(...T) (<-chan struct{}, CancelFunc) UntilContext(context.Context, ...T) }
A listenable object with comparable element type. This allows additional methods Until, UntilCh and UntilContext to be called.
type ListenableCM ¶
type ListenableCM[T comparable] interface { ListenableC[T] Current() T }
A listenable object with comparable element type and memorizes the latest value.
type ListenableM ¶
type ListenableM[T any] interface { Listenable[T] Current() T }
A listenable object that also memorizes the latest value.
Source Files