watch

Documentation

Overview

Package watch turns a driver's change callback into a stream of freshly loaded values.

A watchable driver announces a change by calling a func(context.Context) that carries no payload: it says the plane may have changed and nothing more. Signal is a value that records such a call, and Values is the loop that reloads through a ferry.Binding and yields a brand-new T each time.

Wiring it up is a signal, the driver's own watch option and a range, here against driver/env:

s := watch.New()

src := env.New(env.DotEnv(".env"), env.WatchFiles(ctx, s.Changed))
b, err := ferry.Bind[Config](src)
if err != nil {
	return err
}

cfg, err := b.Load(ctx) // the value to start from
if err != nil {
	return err
}
publish(cfg)

seq, errf := watch.Values(ctx, s, b)
for cfg := range seq {
	publish(cfg) // replace the pointer, never mutate the old value
}
return errf()

The ordering is the reason this package exists. A driver opens its watch when the source is built, which is before ferry.Bind has returned and long before there is a stream to range, so a change can land when there is nothing yet to load through. Nothing is lost: the Signal records it, and the stream opens with that reload.

The helper starts no goroutines. The reload runs on the goroutine doing the ranging, and there is nothing to stop and nothing to close.

One sharp edge belongs to the driver rather than to this package: a watch the driver loses - a watched directory removed, say - fires the callback one last time and then goes quiet, so the stream keeps ranging and stops reloading. The driver's own documentation says which endings it can and cannot announce.

The design records behind these decisions are in docs/adr/.

Example

The whole loop: signal first, bind once, hold a value, change the plane, receive a fresh one. The held value is untouched by the reload, which is what makes publication a replacement rather than a mutation.

plane := newMemPlane()
plane.Set(ferry.At("host"), ferry.String("db1"))

ctx := context.Background()

s := watch.New()
plane.OnChange(s.Changed) // what a driver's watch option takes

b, err := ferry.Bind[Config](plane)
if err != nil {
	fmt.Println("bind:", err)

	return
}

held, err := b.Load(ctx) // the value this goroutine is holding
if err != nil {
	fmt.Println("load:", err)

	return
}

seq, errf := watch.Values(ctx, s, b)

// Two writes land before the range reads the signal, so the two changes
// coalesce into one reload that sees both.
plane.Set(ferry.At("host"), ferry.String("db2"))
plane.Set(ferry.At("port"), ferry.Number("5432"))

for cfg := range seq {
	fmt.Printf("reloaded: %s:%d\n", cfg.Host, cfg.Port)

	break // one turn is enough for an example; a server keeps ranging
}

fmt.Printf("held:     %s:%d\n", held.Host, held.Port)
fmt.Println("stream error:", errf())

Output:
reloaded: db2:5432
held:     db1:8080
stream error: <nil>

Example (FailedReload)

A failed reload ends the stream, so a process that wants to survive one ranges again on the same signal. Nothing is lost in between: the change that fixed the plane is pending when the second stream opens.

plane := newMemPlane()
plane.Set(ferry.At("host"), ferry.String("db1"))

ctx := context.Background()

s := watch.New()
plane.OnChange(s.Changed)

b, err := ferry.Bind[Config](plane)
if err != nil {
	fmt.Println("bind:", err)

	return
}

plane.Delete(ferry.At("host")) // the plane loses a required address

for {
	seq, errf := watch.Values(ctx, s, b)
	for cfg := range seq {
		fmt.Println("reloaded:", cfg.Host)

		break // a server would keep ranging until it was told to stop
	}

	err := errf()
	if err == nil || errors.Is(err, context.Canceled) {
		return // the range ended cleanly, or the process is shutting down
	}

	fmt.Println("reload failed, address missing:", errors.Is(err, ferry.ErrMissing))

	plane.Set(ferry.At("host"), ferry.String("db2")) // somebody fixes it
}

Output:
reload failed, address missing: true
reloaded: db2

Index

• func Values[T any](ctx context.Context, s *Signal, b *ferry.Binding[T]) (seq iter.Seq[T], errf func() error)
• type Signal
  • func New() *Signal
  • func (s *Signal) Changed(context.Context)

Examples

• Package
• Package (FailedReload)

Functions

func Values

func Values[T any](ctx context.Context, s *Signal, b *ferry.Binding[T]) (seq iter.Seq[T], errf func() error)

Values streams a freshly loaded value of T for every change s records.

Range the sequence, then read the error function after the range exits:

seq, errf := watch.Values(ctx, s, b)
for cfg := range seq {
	publish(cfg)
}
if err := errf(); err != nil {
	alert(err)
}

Every value comes from a load through b, so it is brand new and a value handed out earlier never changes underneath the goroutine holding it. A change s recorded before this call - including one that landed before ferry.Bind returned - is not lost, and the stream opens with that reload. The first value otherwise arrives on the first change, never before it: load once through b for the value to start from, or call s.Changed before ranging to open the stream with the plane's current contents.

The stream ends on the first failed reload or on cancellation of ctx, and errf reports why, once, after the range exits. The load's error passes through untouched, so errors.Is against ferry's sentinels answers what went wrong; a cancelled context reports ctx.Err. Breaking out of the range is a clean ending and errf reports nil, as does calling errf before the range has exited, which reports nothing useful.

Recovery from a failure is calling Values again on the same s. Nothing is lost in between: a change that lands while no stream is ranging is pending when the next one opens.

The reload runs on the ranging goroutine and no goroutine is started here, so the stream lives exactly as long as the range does.

Sharp edges. The context a driver watches under and ctx here are different values, and passing one that outlives the driver's leaves a range waiting on a signal nothing will fire again. One range per Signal at a time: two ranges share the pending changes out rather than each seeing them, which is not policed. And a change is only ever a hint, so a coalesced or spurious one costs one load and yields a value equal to the last.

Types

type Signal

type Signal struct {
	// contains filtered or unexported fields
}

Signal records that a plane may have changed.

Build one with New, hand Signal.Changed to whatever the driver's watch option takes, and range Values to receive a freshly loaded value per change.

It holds one pending change and no more, so a burst is one change and so is a change that lands while a reload is already running. It carries no payload, because the announcement it records carries none: the reload is what reads what the plane holds now.

It is safe for use from many goroutines. One range at a time is what it is for, and two ranges over the same Signal share the pending change out between them rather than each receiving it.

func New

func New() *Signal

New returns a Signal with nothing pending.

func (*Signal) Changed

func (s *Signal) Changed(context.Context)

Changed records a change and returns immediately.

It never blocks and never fails, so a driver's watching goroutine feels no back pressure from a slow consumer, and calling it with nobody ranging is normal: the change waits, and the next Values opens with it.

Its method value has type func(context.Context), which is the shape a driver's watch option takes. The context argument is unused, because a change announcement carries nothing, including a deadline.