Documentation
¶
Overview ¶
Package observe connects a CRDT to an application-owned reactive view.
A Store serializes mutations made through it and publishes an immutable application projection after a successful mutation. It is deliberately not a CRDT protocol, operation log, persistence layer, or transport. In particular, its Version values are process-local UI revisions and must not be sent to a peer or used as a replication acknowledgement.
The projection passed to New must be safe to retain after the function returns. For maps, slices, pointers, and byte slices that normally means returning an owned copy. Store shares one projection with all subscribers; subscribers must treat Event.Value as immutable.
Store invokes callbacks asynchronously and never while it holds the Store lock. A slow subscriber retains only its newest undelivered event. This bounds memory and lets a UI render the latest state, but it means observers must use Event.Coalesced and Version gaps when every intermediate mutation is significant. Durable replication belongs in replica and durable, not in this package.
Index ¶
- Variables
- type Callback
- type Event
- type Options
- type Origin
- type Panic
- type Store
- func (s *Store[T, V]) Close()
- func (s *Store[T, V]) Mutate(origin Origin, mutation func(T) error) error
- func (s *Store[T, V]) Snapshot() (Event[V], error)
- func (s *Store[T, V]) Subscribe(callback Callback[V]) (*Subscription[V], error)
- func (s *Store[T, V]) SubscribeFromNow(callback Callback[V]) (*Subscription[V], error)
- type Subscription
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrNilStore reports an operation on a nil Store. ErrNilStore = errors.New("observe: nil store") // ErrNilView reports a Store constructed without an immutable view function. ErrNilView = errors.New("observe: nil view function") // ErrNilMutation reports a nil mutation function. ErrNilMutation = errors.New("observe: nil mutation function") // ErrNilCallback reports a nil subscriber callback. ErrNilCallback = errors.New("observe: nil callback") // ErrInvalidOrigin reports an origin that cannot describe a mutation. ErrInvalidOrigin = errors.New("observe: invalid mutation origin") // ErrClosed reports a mutation or subscription after Store.Close. ErrClosed = errors.New("observe: store is closed") )
Functions ¶
This section is empty.
Types ¶
type Callback ¶
Callback receives one Event on a subscriber-owned goroutine. It must not retain mutable aliases from Event.Value or block indefinitely. A panic is contained, recorded on the Subscription, and unsubscribes that callback.
type Event ¶
type Event[V any] struct { Version uint64 Origin Origin Value V State crdt.StateSnapshot Coalesced uint64 }
Event is one application-visible Store revision. Value is produced by the application-supplied view function while the Store serializes a mutation. It must be treated as immutable by every callback.
Coalesced reports how many older, not-yet-delivered events this event replaced for this particular subscriber. Event versions are monotonic per Store, but a subscriber may observe gaps when it is slower than mutations.
type Options ¶
type Options struct {
OnPanic func(Panic)
}
Options controls diagnostic handling for a Store. OnPanic runs after a callback panic, on the failing callback's goroutine. It must return quickly; its own panic is contained. Callback delivery never depends on this hook.
type Origin ¶
type Origin uint8
Origin identifies why an application-visible state update was committed. It is local process metadata, not a CRDT conflict-resolution input and not a wire-protocol field.
const ( // Initial describes the current state delivered immediately after Subscribe. Initial Origin = iota // Local describes a successful local user or application mutation. Local // Remote describes a successfully applied remote delta or state update. Remote // Merge describes a successful CRDT state merge. Merge // Restore describes successful installation of recovered local state. Restore // Maintenance describes a successful local maintenance operation, such as // an authority-approved tombstone compaction. Maintenance )
type Panic ¶
Panic describes a callback panic captured by the observer dispatcher. Value is the recovered panic value and must be treated as diagnostic data.
type Store ¶
type Store[T crdt.StateReporter, V any] struct { // contains filtered or unexported fields }
Store owns one application-facing mutation path around a CRDT or another StateReporter. T should not be mutated outside Mutate while observers are active, otherwise Store cannot preserve version/event ordering.
V is an application view such as a counter value, immutable document text, or copied list of set elements. Store does not use V for CRDT merge or wire semantics.
Example ¶
package main
import (
"fmt"
"github.com/DarkInno/crdt/counter"
"github.com/DarkInno/crdt/observe"
)
func main() {
value, err := counter.NewGCounter("browser-tab")
if err != nil {
panic(err)
}
store, err := observe.New(value, func(current *counter.GCounter) uint64 {
total, err := current.Value()
if err != nil {
panic(err)
}
return total
})
if err != nil {
panic(err)
}
rendered := make(chan uint64, 2)
subscription, err := store.Subscribe(func(event observe.Event[uint64]) {
rendered <- event.Value
})
if err != nil {
panic(err)
}
<-rendered // Initial view; a UI can render it immediately.
if err := store.Mutate(observe.Local, func(current *counter.GCounter) error {
_, err := current.Increment(4)
return err
}); err != nil {
panic(err)
}
fmt.Println(<-rendered)
subscription.Unsubscribe()
<-subscription.Done()
}
Output: 4
func New ¶
func New[T crdt.StateReporter, V any](value T, view func(T) V) (*Store[T, V], error)
New creates a Store with an application-owned view function. The view runs only to serve a subscription or publish to at least one subscriber, so a Store without subscribers does not pay projection/copy costs on mutations.
func NewWithOptions ¶
func NewWithOptions[T crdt.StateReporter, V any](value T, view func(T) V, options Options) (*Store[T, V], error)
NewWithOptions creates a Store with diagnostic callback-panic handling.
func (*Store[T, V]) Close ¶
func (s *Store[T, V]) Close()
Close prevents future mutations and subscriptions, and cancels all current subscriptions. It does not wait for a callback already in progress; use a Subscription's Done channel when a caller needs to wait for quiescence.
func (*Store[T, V]) Mutate ¶
Mutate runs mutation under Store's serialization gate. A nil or failed mutation does not advance Version or notify subscribers. The wrapped CRDT operation must retain its documented all-or-nothing-on-error behavior.
mutation is intentionally not a notification callback: it may change T and must not recursively call methods on the same Store. Subscribers run only after this method releases the Store lock, so they may safely call Mutate.
func (*Store[T, V]) Snapshot ¶
Snapshot returns the current reactive view. Its Origin is Initial because it is a point-in-time read rather than a mutation notification.
func (*Store[T, V]) Subscribe ¶
func (s *Store[T, V]) Subscribe(callback Callback[V]) (*Subscription[V], error)
Subscribe atomically registers callback and queues the Store's current view before later updates may publish. This prevents a UI from missing a change between reading state and starting observation. If the callback is slow, that initial event may be coalesced into a newer event before delivery.
func (*Store[T, V]) SubscribeFromNow ¶
func (s *Store[T, V]) SubscribeFromNow(callback Callback[V]) (*Subscription[V], error)
SubscribeFromNow registers callback without an initial state event. It is for consumers that have already obtained a coherent Snapshot themselves.
type Subscription ¶
type Subscription[V any] struct { // contains filtered or unexported fields }
Subscription owns one callback registration. Unsubscribe is idempotent. Done closes after the subscription goroutine has stopped, including any callback that was already executing at the time of Unsubscribe.
func (*Subscription[V]) Done ¶
func (s *Subscription[V]) Done() <-chan struct{}
Done is closed after all delivery work for this subscription stops.
func (*Subscription[V]) Panic ¶
func (s *Subscription[V]) Panic() (Panic, bool)
Panic returns the callback panic, if one caused this subscription to stop.
func (*Subscription[V]) Unsubscribe ¶
func (s *Subscription[V]) Unsubscribe()
Unsubscribe cancels delivery to this subscription without waiting for an in-progress callback. It is safe to call from the callback itself.
Source Files