rxxr

README

rxxr

Allows values to be published and subscribed to. Subscribers will receive the most recently published value, if one exists.

Subscribed functions are run in their own goroutine. Calling methods on the pipe are thread safe, but published values will not be locked.

Constructor

// Provide no options for default pipe.
pipe := rxxr.New[int]()

Configure

// To use the config:
pipe := rxxr.New[int](
    // Create a pipe already containing a value.
	// Will also imply SendOnSubscribe.
    UseInitialValue(42),
    // When a function is subscribed, call it with the current value if one exists.
	SendOnSubscribe,
	// Catch panics from subscribed functions.
    PanicHandler(func(r any) {
	    // Handle panic
    }),
)

Subscribe

pipe := rxxr.New[int]()

unsubscribe := pipe.Subscribe(func(i int) {
	// Do something
})

// Do something

unsubscribe()

Publish

pipe := rxxr.New[int]()
pipe.Subscribe(func (i int) {
	fmt.Printf("received: %d\n", i)
})
pipe.Publish(1)
pipe.Publish(2)
pipe.Publish(3)
pipe.Publish(5)
pipe.Publish(8)
pipe.Publish(13)

// received: 1
// received: 2
// received: 3
// received: 5
// received: 8
// received: 13

Get Value

pipe := rxxr.New[int]()
fmt.Println(pipe.Value())

pipe.Publish(42)
fmt.Println(pipe.Value())

// 0 false
// 42 true

Close

pipe := rxxr.New[int]()
pipe.Close()

// NOP
pipe.Publish(5)

Documentation

Index

• func SendOnSubscribe[T any](p *Pipe[T])
• type Option
  • func PanicHandler[T any](fn func(any)) Option[T]
  • func UseInitialValue[T any](t T) Option[T]
• type Pipe
  • func New[T any](options ...Option[T]) *Pipe[T]
  • func (p *Pipe[T]) Close()
  • func (p *Pipe[T]) Publish(t T)
  • func (p *Pipe[T]) Subscribe(fn func(T)) (cancel func())
  • func (p *Pipe[T]) Value() (t T, ok bool)

Functions

func SendOnSubscribe

func SendOnSubscribe[T any](p *Pipe[T])

SendOnSubscribe sets if the current value should be sent to new subscribers, if a value is not sent the subscriber will receive no values.

Types

type Option

type Option[T any] func(*Pipe[T])

func PanicHandler

func PanicHandler[T any](fn func(any)) Option[T]

PanicHandler sets the function to receive panics from running subscribed funcs

func UseInitialValue

func UseInitialValue[T any](t T) Option[T]

UseInitialValue Initializes a pipe with the provided value.

type Pipe

type Pipe[T any] struct {
	// contains filtered or unexported fields
}

func New

func New[T any](options ...Option[T]) *Pipe[T]

New creates a simple pipe that transmits received values to all subscribers

func (*Pipe[T]) Close

func (p *Pipe[T]) Close()

func (*Pipe[T]) Publish

func (p *Pipe[T]) Publish(t T)

func (*Pipe[T]) Subscribe

func (p *Pipe[T]) Subscribe(fn func(T)) (cancel func())

func (*Pipe[T]) Value

func (p *Pipe[T]) Value() (t T, ok bool)