Documentation
¶
Overview ¶
Package chanx provides paradigmatic goroutine communication: observable values, subjects, and cancelable subscriptions over channels.
This is the communication axis of the concx toolkit (alongside nest for lifecycle and schedx for orchestration). Prefer this module path over the archived github.com/xoctopus/x/chanx package.
Key types:
- NotifiableObserver: send and receive values until CancelCause
- Subject: fan-out Send to subscribers; Observe returns an Observer
- Observer / Subscriber: value stream plus Done/Err lifecycle
A nil cancel cause is normalized to ErrCompleted.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
View Source
var (
ErrCompleted = errors.New("completed")
)
Functions ¶
This section is empty.
Types ¶
type Cancelable ¶
type Cancelable interface {
CancelCause(err error)
}
type NotifiableObserver ¶
type NotifiableObserver[T any] interface { Observer[T] ValueNotifier[T] }
Example ¶
package main
import (
"fmt"
"sync"
"time"
"github.com/xoctopus/concx/pkg/chanx"
)
func main() {
// Create a NotifiableObserver
obs := chanx.NewNotifiableObserver[int]()
var wg sync.WaitGroup
// Start a producer goroutine
wg.Go(func() {
obs.Send(1)
obs.Send(2)
obs.Send(3)
time.Sleep(10 * time.Millisecond)
// Cancel after sending all data. A nil error will be converted to ErrCompleted.
obs.CancelCause(nil)
})
// Consume data in the main goroutine until the channel is closed
for v := range obs.Value() {
fmt.Println("Received:", v)
}
wg.Wait()
// Check the final error state
fmt.Println("Error:", obs.Err())
}
Output: Received: 1 Received: 2 Received: 3 Error: completed
func NewNotifiableObserver ¶
func NewNotifiableObserver[T any]() NotifiableObserver[T]
type Observable ¶
type Observer ¶
type Observer[T any] interface { ValueObserver[T] Cancelable Done() <-chan struct{} Err() error }
type ObserverFunc ¶
func (ObserverFunc[T]) Observe ¶
func (f ObserverFunc[T]) Observe() Observer[T]
type Subject ¶
type Subject[T any] struct { // contains filtered or unexported fields }
Example ¶
package main
import (
"fmt"
"slices"
"sync"
"time"
"github.com/xoctopus/x/iterx"
"github.com/xoctopus/concx/pkg/chanx"
)
func main() {
// Create a Subject
subject := &chanx.Subject[int]{}
// Create two observers
obs1 := subject.Observe()
obs2 := subject.Observe()
var (
wg sync.WaitGroup
results = make(chan string, 4)
)
consuming := func(name string, observer chanx.Observer[int]) func() {
return func() {
for v := range observer.Value() {
results <- fmt.Sprintf("%s received: %d", name, v)
}
}
}
// Consumer 1
wg.Go(consuming("obs1", obs1))
// Consumer 2
wg.Go(consuming("obs2", obs2))
// Producer broadcasts data via the Subject
subject.Send(1)
time.Sleep(10 * time.Millisecond)
subject.Send(2)
time.Sleep(10 * time.Millisecond)
// Close the Subject, which cascades the cancellation to all subscribers
subject.CancelCause(nil)
// Wait for all consumers to finish
wg.Wait()
close(results)
for _, s := range slices.Sorted(iterx.Recv(results)) {
fmt.Println(s)
}
// Verify the error state of the Subject and its Observers
fmt.Println("Subject Error:", subject.Err())
fmt.Println("Obs1 Error:", obs1.Err())
fmt.Println("Obs2 Error:", obs2.Err())
}
Output: obs1 received: 1 obs1 received: 2 obs2 received: 1 obs2 received: 2 Subject Error: completed Obs1 Error: completed Obs2 Error: completed
func (*Subject[T]) CancelCause ¶
func (*Subject[T]) Subscribe ¶
func (s *Subject[T]) Subscribe(o Subscriber[T])
type Subscriber ¶
type Subscriber[T any] interface { ValueNotifier[T] Cancelable Done() <-chan struct{} Err() error }
type ValueNotifier ¶
type ValueNotifier[T any] interface { Send(x T) }
type ValueObserver ¶
type ValueObserver[T any] interface { Value() <-chan T }
Source Files
Click to show internal directories.
Click to hide internal directories.