indexsub

package module
v0.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 14, 2026 License: MIT Imports: 5 Imported by: 0

README

CI

indexsub

indexsub.Notifier is a synchronization primitive that owns a monotonically increasing index. It allows goroutines to wait until progress reaches a target value without polling. A single subscribed index satisfies any number of goroutines waiting for the same target index.

Callers subscribing to a target index are handed a Sub, which becomes satisfied (its channel closes) once the notifier's index reaches or exceeds that target.

Features:

  • Ordered notifications. Progress never regresses, once a subscription is satisfied it remains satisfied.
  • Per-index notifications. Each subscribed index has its own signal, all waiters on the same index share it.
  • High-water mark wakeups. Advancing the notifier to idx wakes every waiter for idx and all lower indexes.
  • Cheap notification broadcast. Notifications are delivered by closing a shared channel.
  • Context-aware and select-friendly. The channel-based API can be used directly in select statements or combined with a context.Context. No polling required.
  • Convenience method. err := sub.Wait(ctx) handles common wait-with-context pattern and automatically unsubscribes when appropriate.
  • Concurrency-safe. All operations on Notifier and Sub are safe for concurrent use, with one exception: a Sub must not be copied before its first use (see docs on the type).
  • Minimal allocations. Only the first subscription to a fresh index allocates (shared channel).
  • Mutex-free inlined fast paths.
  • Safe, idempotent unsubscription.
  • Basic metrics: active subscribers, pending indexes, and container capacity.
  • Manual container compaction. Allows reclaiming memory after heavy churn.

Typical uses:

  • Waiting for a replicated log or WAL offset.
  • Waiting for a Raft commit/apply index.
  • Waiting for a sequence number or generation.
  • Waiting for a cache or snapshot version.
  • Coordinating producer/consumer pipelines.
  • Signaling progress without polling.

Install:

go get -u github.com/mnzbono/indexsub

Common usage example:

notifier := indexsub.NewNotifier()
// ...
sub := notifier.Subscribe(10)
if err := sub.Wait(ctx); err != nil {
	return err
}
workAfterIndexReached()

Further details:

Usage 
notifier := indexsub.NewNotifier()

// use channel in your selects
sub := notifier.Subscribe(10)
defer sub.Unsubscribe()

select {
// ...
case <-sub.Ch:
weGotToOurIndex()
// ...
}

// Wait(ctx) is a convenience method to wait on caller's context. Selecting on sub.Ch and ctx.Done()
// Wait also manages lifecycle of sub, unsubscribing and preventing Wait'ing on an unsubscribed sub.
// Provides richer context than selecting on Ch: Wait immeditely verifies the index to detect a notifier cancellation.
if err := sub.Wait(ctx); err != nil {
	// wait err can only be:
	// - ErrCanceledByNotifier: notifier used CancelAll.
	// - ErrUnsubscribed: the sub is already unsubscribed and Wait won't proceed as it is unsafe (channel could've been abandoned).
	// - Context.Canceled/Context.Timeout: the callers context cancel signal.
	// - nil
	return err
}

// Unsubscribe() function is safe and won't interfere with other subscribers at the same index (as long as sub wasn't copied BEFORE use)
sub := notifier.Subscribe(10)
sub2 := notifier.Subscribe(10)
sub.Unsubscribe()
sub.Unsubscribe()
sub.Unsubscribe() // sub2 still subscribed

// Stats() provide basic metrics
notifier.Stats()
Stats{
	ActiveSubscribers int64
	ActiveIndexes    int64
	Capacity int
}

// CancelAll() releases all subscribers without satisfying their index. 
// If you plan on using it, consider using Wait method over simple Ch select, as it performs index verification for you.
// You can also use CurrentIndex() to verify it wasn't a notifier cancel.
sub := notifier.Subscribe(idx)
defer sub.Unsubscribe()

<-sub.Ch
if idx > notifier.CurrentIndex() {
	panic("we got flushed!")
}

// Wait does the check automatically and safer, as the notifier is likely still holding lock of it's internal ops.
if err := sub.Wait(ctx); err == indexsub.ErrCanceledByNotifier {
	panic("we got flushed!")
}


// Sub can query the notifier's current index, even after fulfillment or cancelation.
sub := notifier.Subscribe(10)
sub.Unsubscribe()
notifier.Update(99)
sub.NotifierIndex() // 99

// Done is a simple method that tells if the notifier index is equal or higher than sub's without blocking or touching its lifecycle.
notifier.Update(9)
sub := notifier.Subscribe(10)
sub.Done() // false
notifier.Update(99)
sub.Done() // true

Update() is idempotent, monotonic and gives feedback on idx advancement

call return
notifier.Update(1) true
notifier.Update(1) false
notifier.Update(10) true
notifier.Update(2) false
notifier.CurrentIndex() 10 
if !notifier.Update(idx) {
  panic("monotonic 4life")
}

Complexity

Where n is the number of distinct pending indexes.

Operation Time Allocs/op Notes
Subscribe to a past index O(1) 0 Returns a shared already-closed subscription.
Subscribe to an existing pending index O(1) 0 Reuses the existing entry.
Subscribe to a new highest index O(1) amortized 1* Appends to the ordered queue.
Subscribe to a new out-of-order index O(n) 1* Binary search + slice insertion (copy).
Update with no pending indexes O(1) 0 Fast path.
Update advancing k pending indexes O(k) 0 Removes entries, closes channels (go scheduler wakes up subscribers).

* Only the first subscription allocates (channel). Following subscriptions to that index reuse the entry.
** Worst-case total insertion cost: inserting m distinct indexes in strictly descending order requires Θ(m²) time overall, because each insertion shifts all previously inserted elements.

Benchmark 
$ go test -bench=. -count=20
goos: linux
goarch: amd64
pkg: github.com/mnzbono/indexsub
cpu: 11th Gen Intel(R) Core(TM) i5-1135G7 @ 2.40GHz

Lower ns/op is better. SubscribeWorst intentionally exercises the pathological insertion order.

benchmark ns/op B/op allocs/op
UnsubscribeClosed-8 4.800n 0 0
SubscribeUnsubscribeChurn-8 102.8n 112 1
SubscribeExisting-8 21.79n 0 0
SubscribeClosed-8 1.919n 0 0
SubscribeNewNoGrow-8 112.0n 111 0
SubscribeNewGrowing-8 223.3n 236 1
SubscribeScatter/10-8 74.93n 67 0
SubscribeScatter/100-8 96.08n 71 0
SubscribeScatter/1000-8 127.7n 70 0
SubscribeScatter/10000-8 203.8n 70 0
SubscribeScatter/100000-8 1.224µ 70 0
SubscribeWorst/10-8 101.7n 112 1
SubscribeWorst/100-8 127.5n 112 1
SubscribeWorst/1000-8 160.8n 112 1
SubscribeWorst/10000-8 595.4n 112 1
SubscribeWorst/100000-8 5.637µ 119 1
Update/0-8 16.51n 0 0
Update/1/10-8 67.94n na na
Update/1/100-8 69.93n na na
Update/1/1000-8 74.54n na na
Update/1/10000-8 80.24n na na
Update/1/100000-8 108.3n na na
Update/10/10-8 431.0n na na
Update/10/100-8 446.2n na na
Update/10/1000-8 493.4n na na
Update/10/10000-8 540.4n na na
Update/10/100000-8 807.3n na na
geomean 131.8n

*SubscribeWorst inserts strictly descending indices, forcing every insertion to the front of the ordered queue. This is the implementation's pathological O(n²) case and is included to characterize worst-case behavior rather than expected usage.

Inlining
  • Most paths inline:
    • Update() to a lower index.
    • Unsubscribe() CAS.
    • Wait() on an unsubscribed Sub.
    • Done(), Index(), IsUnsubscribed(), CurrentIndex(), NewIndexNotifier()
    • Subscribe() could be made to inline by replacing atomic with atomic methods, but hardly worth it.

./indexsub.go:17:6: can inline init.0 with cost 6 as: func() { closedChan = make(chan struct {}); close(closedChan) }
./indexsub.go:47:6: can inline NewNotifier with cost 56 as: func(...Options) *Notifier { opts = ; switch statement; size := opts.InitialSize; n := &Notifier{...}; (*atomic.Uint64).Store(n.idx, opts.InitialIndex); return n }
./indexsub.go:148:6: can inline (*Notifier).CurrentIndex with cost 9 as: method(*Notifier) func() uint64 { return (*atomic.Uint64).Load(n.idx) }
./indexsub.go:281:6: can inline (*Notifier).newLateSub with cost 10 as: method(*Notifier) func(uint64) Sub { return Sub{...} }
./indexsub.go:276:6: can inline (*Notifier).newSubscription with cost 14 as: method(*Notifier) func(uint64, <-chan struct {}) Sub { n.subscribersTotal++; return Sub{...} }
./indexsub.go:258:6: can inline newEntry with cost 7 as: func() entry { return entry{...} }
./indexsub.go:82:6: cannot inline (*Notifier).subscribe: unhandled op DEFER
./indexsub.go:84:2: can inline (*Notifier).subscribe.deferwrap1 with cost 76 as: func() { (*sync.Mutex).Unlock(.autotmp_3) }
./indexsub.go:74:6: cannot inline (*Notifier).Subscribe: function too complex: cost 89 exceeds budget 80
./indexsub.go:120:6: cannot inline (*Notifier).update: function too complex: cost 318 exceeds budget 80
./indexsub.go:112:6: can inline (*Notifier).Update with cost 77 as: method(*Notifier) func(uint64) bool { if idx <= (*Notifier).CurrentIndex(n) { return false }; return (*Notifier).update(n, idx) }
./indexsub.go:150:6: cannot inline (*Notifier).unsubscribeOut: unhandled op DEFER
./indexsub.go:152:2: can inline (*Notifier).unsubscribeOut.deferwrap1 with cost 76 as: func() { (*sync.Mutex).Unlock(.autotmp_2) }
./indexsub.go:190:6: can inline (*Notifier).cancelAll with cost 31 as: method(*Notifier) func() { for loop; n.q = n.q[:0]; n.indexesTotal = int64(0); n.subscribersTotal = int64(0) }
./indexsub.go:207:6: can inline (*Notifier).compact with cost 58 as: method(*Notifier) func() { target := max(int(n.initSize), len(n.q), len(n.active)); if cap(n.q) <= target { return }; newMap := make(map[uint64]entry, target); for loop; n.active = newMap; newSlice := make([]uint64, 0, target); n.q = append(newSlice, n.q...) }
./indexsub.go:174:6: cannot inline (*Notifier).Flush: function too complex: cost 239 exceeds budget 80
./indexsub.go:184:6: cannot inline (*Notifier).CancelAll: function too complex: cost 179 exceeds budget 80
./indexsub.go:201:6: cannot inline (*Notifier).Compact: function too complex: cost 206 exceeds budget 80
./indexsub.go:227:6: cannot inline (*Notifier).Stats: function too complex: cost 163 exceeds budget 80
./indexsub.go:312:6: can inline (*Sub).Unsubscribe with cost 79 as: method(*Sub) func() { if atomic.SwapUint64(&s.unsubToken, uint64(0)) > (*atomic.Uint64).Load(s.notifier.idx) { (*Notifier).unsubscribeOut(s.notifier, s.idx) } }
./indexsub.go:330:6: can inline (*Sub).NotifierIndex with cost 13 as: method(*Sub) func() uint64 { return (*Notifier).CurrentIndex(s.notifier) }
./indexsub.go:297:6: cannot inline (*Sub).wait: unhandled op DEFER
./indexsub.go:298:2: can inline (*Sub).wait.deferwrap1 with cost 81 as: func() { (*Sub).Unsubscribe(.autotmp_3) }
./indexsub.go:289:6: can inline (*Sub).Wait with cost 71 as: method(*Sub) func(context.Context) error { if atomic.LoadUint64(&s.unsubToken) == uint64(0) { return ErrUnsubscribed }; return (*Sub).wait(s, ctx) }
./indexsub.go:321:6: can inline (*Sub).IsUnsubscribed with cost 8 as: method(*Sub) func() bool { return atomic.LoadUint64(&s.unsubToken) == uint64(0) }
./indexsub.go:324:6: can inline (*Sub).Done with cost 19 as: method(*Sub) func() bool { return s.idx <= (*Sub).NotifierIndex(s) }
./indexsub.go:327:6: can inline (*Sub).Index with cost 3 as: method(*Sub) func() uint64 { return s.idx }

Did you try a heap? Yes.

The monotonic nature of this primitive is very well suited for a sorted slice. But the heap was tried nevertheless, and the code left in the heap branch for future reference. The conclusion is, the heap is only better for a very specific use case: high volume of subscribers spraying the index but unsubscring before update reached them, something like time sensitive callbacks. Even though the heap can handle better adversarial subscriptions at any volume, the dominant operation for this primitive is update, and heap performs clearly worse with big queues during update. On top of that, if it was offered as opt-in it would impose type plumbing tax (interface was discarded due to worse results and boxed allocs).

(inlined slice) base sec/op sec/op vs base (heap inlined) Δ sec/op vs base (slice typed) Δ sec/op vs base (heap typed) Δ
UnsubscribeClosed-8 4.800n ± 35% 4.799n ± 35% ~ (p=0.825 n=20) 5.002n ± 37% +4.21% (p=0.005 n=20) 5.043n ± 30% +5.06% (p=0.015 n=20)
SubscribeUnsubscribeChurn-8 102.80n ± 0% 87.78n ± 2% -14.61% (p=0.000 n=20) 112.80n ± 1% +9.73% (p=0.000 n=20) 88.42n ± 2% -13.98% (p=0.000 n=20)
SubscribeExisting-8 21.79n ± 3% 22.01n ± 3% ~ (p=0.920 n=20) 23.18n ± 10% +6.38% (p=0.000 n=20) 21.80n ± 2% ~ (p=0.542 n=20)
SubscribeClosed-8 1.919n ± 2% 1.937n ± 3% ~ (p=0.482 n=20) 2.006n ± 3% +4.53% (p=0.000 n=20) 1.954n ± 4% ~ (p=0.888 n=20)
SubscribeNewNoGrow-8 112.0n ± 1% 113.4n ± 0% +1.25% (p=0.000 n=20) 113.6n ± 1% +1.43% (p=0.000 n=20) 109.8n ± 1% -2.01% (p=0.000 n=20)
SubscribeNewGrowing-8 223.3n ± 3% 219.0n ± 3% ~ (p=0.482 n=20) 242.5n ± 3% +8.57% (p=0.000 n=20) 223.0n ± 2% ~ (p=0.794 n=20)
SubscribeScatter/10-8 74.93n ± 1% 72.74n ± 1% -2.92% (p=0.000 n=20) 97.65n ± 1% +30.31% (p=0.000 n=20) 89.01n ± 1% +18.79% (p=0.000 n=20)
SubscribeScatter/100-8 96.08n ± 0% 91.22n ± 1% -5.05% (p=0.000 n=20) 97.12n ± 1% +1.09% (p=0.000 n=20) 90.78n ± 0% -5.51% (p=0.000 n=20)
SubscribeScatter/1000-8 127.7n ± 1% 102.2n ± 1% -19.93% (p=0.000 n=20) 125.8n ± 1% -1.53% (p=0.000 n=20) 100.5n ± 1% -21.30% (p=0.000 n=20)
SubscribeScatter/10000-8 203.8n ± 1% 105.8n ± 1% -48.09% (p=0.000 n=20) 209.7n ± 1% +2.87% (p=0.000 n=20) 106.5n ± 1% -47.77% (p=0.000 n=20)
SubscribeScatter/100000-8 1224.0n ± 2% 116.8n ± 1% -90.45% (p=0.000 n=20) 1334.0n ± 1% +8.99% (p=0.000 n=20) 126.7n ± 1% -89.65% (p=0.000 n=20)
SubscribeWorst/10-8 101.75n ± 1% 99.04n ± 1% -2.66% (p=0.000 n=20) 129.50n ± 1% +27.27% (p=0.000 n=20) 121.05n ± 1% +18.97% (p=0.000 n=20)
SubscribeWorst/100-8 127.5n ± 1% 118.4n ± 1% -7.17% (p=0.000 n=20) 129.8n ± 1% +1.80% (p=0.000 n=20) 114.0n ± 1% -10.58% (p=0.000 n=20)
SubscribeWorst/1000-8 160.8n ± 1% 123.8n ± 0% -23.02% (p=0.000 n=20) 162.8n ± 2% ~ (p=0.085 n=20) 119.9n ± 1% -25.41% (p=0.000 n=20)
SubscribeWorst/10000-8 595.4n ± 1% 134.1n ± 0% -77.48% (p=0.000 n=20) 614.7n ± 1% +3.25% (p=0.000 n=20) 133.6n ± 2% -77.55% (p=0.000 n=20)
SubscribeWorst/100000-8 5637.0n ± 1% 158.5n ± 1% -97.19% (p=0.000 n=20) 5774.0n ± 2% +2.43% (p=0.000 n=20) 157.7n ± 2% -97.20% (p=0.000 n=20)
Update/0-8 16.51n ± 0% 15.82n ± 0% -4.18% (p=0.000 n=20) 17.45n ± 0% +5.69% (p=0.000 n=20) 17.59n ± 3% +6.54% (p=0.000 n=20)
Update/1/10-8 67.94n ± 1% 78.33n ± 1% +15.28% (p=0.000 n=20) 73.27n ± 1% +7.84% (p=0.000 n=20) 87.70n ± 1% +29.08% (p=0.000 n=20)
Update/1/100-8 69.93n ± 1% 96.83n ± 1% +38.46% (p=0.000 n=20) 76.19n ± 1% +8.94% (p=0.000 n=20) 107.60n ± 1% +53.86% (p=0.000 n=20)
Update/1/1000-8 74.54n ± 1% 113.05n ± 2% +51.66% (p=0.000 n=20) 80.49n ± 1% +7.98% (p=0.000 n=20) 122.55n ± 1% +64.41% (p=0.000 n=20)
Update/1/10000-8 80.24n ± 1% 128.80n ± 1% +60.52% (p=0.000 n=20) 86.72n ± 1% +8.08% (p=0.000 n=20) 138.80n ± 2% +72.98% (p=0.000 n=20)
Update/1/100000-8 108.3n ± 1% 162.6n ± 1% +50.14% (p=0.000 n=20) 128.4n ± 3% +18.56% (p=0.000 n=20) 175.4n ± 1% +62.00% (p=0.000 n=20)
Update/10/10-8 431.0n ± 0% 444.2n ± 1% +3.05% (p=0.000 n=20) 462.4n ± 1% +7.28% (p=0.000 n=20) 480.1n ± 1% +11.38% (p=0.000 n=20)
Update/10/100-8 446.2n ± 1% 729.2n ± 1% +63.42% (p=0.000 n=20) 488.2n ± 1% +9.40% (p=0.000 n=20) 754.0n ± 1% +68.97% (p=0.000 n=20)
Update/10/1000-8 493.4n ± 1% 907.7n ± 0% +83.97% (p=0.000 n=20) 527.3n ± 1% +6.86% (p=0.000 n=20) 935.5n ± 1% +89.60% (p=0.000 n=20)
Update/10/10000-8 540.4n ± 1% 1128.0n ± 1% +108.75% (p=0.000 n=20) 585.5n ± 1% +8.36% (p=0.000 n=20) 1169.5n ± 1% +116.43% (p=0.000 n=20)
Update/10/100000-8 807.3n ± 1% 1424.5n ± 1% +76.44% (p=0.000 n=20) 949.0n ± 1% +17.55% (p=0.000 n=20) 1530.0n ± 1% +89.51% (p=0.000 n=20)

AI disclaimer: Design, implementation, tests, benchmarks, examples, comments and docs are hand-written. AI was used for code review feedback and to polish English phrasing in comments and docs (I'm not a native speaker).

Documentation

Overview

Package indexsub provides the Notifier.

Notifier is a synchronization primitive that owns a monotonically increasing index value. Callers can cheaply subscribe to a target index and are notified once the current index reaches or exceeds it.

Sub represents a subscription to a target index value. It becomes satisfied once the Notifier's current index reaches or exceeds that value.

Example

Subscribe and Update race in this example. The result is deterministic regardless of which happens first. 

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/mnzbono/indexsub"
)

func main() {
	notifier := indexsub.NewNotifier()

	errCh := make(chan error, 1) // helper to illustrate example

	go func() {
		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
		defer cancel()

		sub := notifier.Subscribe(10)
		errCh <- sub.Wait(ctx)
	}()

	notifier.Update(10)

	fmt.Println(<-errCh)
}

Output:
<nil>

Index

Examples

Constants

This section is empty.

Variables

View Source 
var (
	ErrCanceledByNotifier error = errors.New("subscription canceled by the notifier")
	ErrUnsubscribed       error = errors.New("subscription already unsubscribed")
)

Functions

This section is empty.

Types

type Notifier 

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

Notifier owns the index, manages subscriptions and notifies subscribers as the index advances.

func NewNotifier 

func NewNotifier(options ...Options) *Notifier

NewNotifier creates and returns a Notifier. Optional configuration can be provided.

Example 
package main

import (
	"fmt"

	"github.com/mnzbono/indexsub"
)

func main() {
	notifier := indexsub.NewNotifier(indexsub.Options{
		InitialIndex: 11,   // InitialIndex can be used to set the starting index.
		InitialSize:  1000, // InitialSize can be used to preallocate backing container size.
	})

	fmt.Println(notifier.CurrentIndex())
	fmt.Println(notifier.Stats().Capacity)
}

Output:
11
1000

func (*Notifier) CancelAll 

func (n *Notifier) CancelAll()

CancelAll releases (closing channel) all current subscribers without satisfying their requested index. Callers waiting on sub.Wait will be able to identify the cancellation. Callers directly selecting on sub.Ch should verify state to distinguish notifier cancelation from index advancement.

Example 
package main

import (
	"context"
	"fmt"

	"github.com/mnzbono/indexsub"
)

func main() {
	notifier := indexsub.NewNotifier()

	sub := notifier.Subscribe(10)

	notifier.CancelAll()

	if err := sub.Wait(context.Background()); err != nil {
		fmt.Println(err)
	}
}

Output:
subscription canceled by the notifier

func (*Notifier) Compact 

func (n *Notifier) Compact()

Compact reclaims memory from the container without interfering with subscriptions.

func (*Notifier) CurrentIndex 

func (n *Notifier) CurrentIndex() uint64

CurrentIndex returns the current index.

func (*Notifier) Flush 

func (n *Notifier) Flush()

Flush releases all current subscribers without satisfying their requested index. It also reslices the underlying map/slice. Doesn't touch the index.

Performs CancelAll+Compact under mutex.

func (*Notifier) Stats 

func (n *Notifier) Stats() Stats

Stats reports snapshot of:

  • the total subscribers that are waiting on all indexes,
  • the number of indexes that are being waited on,
  • the capacity of the underlying container.

func (*Notifier) Subscribe 

func (n *Notifier) Subscribe(idx uint64) Sub

Subscribe returns a Sub struct, which exposes the channel (Sub.Ch) for direct select, although Wait method is preferred. The channel becomes closed once the monotonic idx is equal or greater than sub's idx. Sub.Unsubscribe() allows to unsubscribe safely and idempotently.

NOTE: Sub is returned as value, not pointer, and it's not safe to copy BEFORE unsubscribing. Allocation (&sub) should be preferred over copy if moving it around before unsubscription.

func (*Notifier) Update 

func (n *Notifier) Update(idx uint64) bool

Update advances the index. Index only ever increases. Notifies all waiters up to idx and returns whether the call advanced the index or not.

Example 
package main

import (
	"fmt"

	"github.com/mnzbono/indexsub"
)

func main() {
	notifier := indexsub.NewNotifier()

	fmt.Println(notifier.Update(1))
	fmt.Println(notifier.Update(1))
	fmt.Println(notifier.Update(10))
	fmt.Println(notifier.Update(2))
	fmt.Println(notifier.CurrentIndex())
}

Output:
true
false
true
false
10

type Options 

type Options struct {
	// The index at which the notifier starts. It's equivalent to just calling Update afterwards.
	InitialIndex uint64
	// The initial size of the queue (map and slice). Provided for convenience.
	InitialSize int
}

Optional configuration for notifier initialization.

type Stats 

type Stats struct {
	ActiveSubscribers int64
	ActiveIndexes     int64
	// Capacity is equal to the capacity of the underlying container.
	// It is exposed to help operator monitor memory footprint.
	// Compact() can be called to reclaim memory.
	Capacity int
}

Stats reports snapshot of:

  • the total subscribers that are waiting on all indexes,
  • the number of indexes that are being waited on,
  • the capacity of the underlying container.

type Sub 

type Sub struct {

	// Ch is shared by all subscribers to this index.
	// It is closed when the notifier reaches the subscribed index or cancels all outstanding subscriptions.
	//
	// NOTE: An unsubscribed channel might be abandoned (without closing), blocking forever.
	// Caller is responsible for lifecycle (unsubscribe). Consider Wait method.
	Ch <-chan struct{}
	// contains filtered or unexported fields
}

Sub represents a subscription to an index.

It is not safe to copy Sub BEFORE Unsubscribe has been called (directly, or via Wait, which unsubscribes automatically), as it can overunsubscribe the index. Once unsubscribed it becomes inert and safe to copy freely.

func (*Sub) Done 

func (s *Sub) Done() bool

Done reports whether the notifier has reached or passed the subscribed index.

Example

Done can be used to cheaply peek at state. 

package main

import (
	"fmt"

	"github.com/mnzbono/indexsub"
)

func main() {
	notifier := indexsub.NewNotifier()

	sub := notifier.Subscribe(10)

	if !sub.Done() {
		fmt.Println("not satisfied yet")
	}
}

Output:
not satisfied yet

func (*Sub) Index 

func (s *Sub) Index() uint64

Index returns the Index this Sub subscribed to.

func (*Sub) IsUnsubscribed 

func (s *Sub) IsUnsubscribed() bool

IsUnsubscribed reports whether this Sub did Unsubscribe(). An unsubscribed Sub is inert and safe to copy.

func (*Sub) NotifierIndex 

func (s *Sub) NotifierIndex() uint64

NotifierIndex returns the current index the notifier is at, live.

func (*Sub) Unsubscribe 

func (s *Sub) Unsubscribe()

Unsubscribe tells the notifier to delist this subscription. If this was the last sub to that index the channel is abandoned (not closed) and not reused.

Example

Unsubscribe is used for manual lifecycle management 

package main

import (
	"github.com/mnzbono/indexsub"
)

func main() {
	notifier := indexsub.NewNotifier()

	sub := notifier.Subscribe(10)
	defer sub.Unsubscribe()
}

func (*Sub) Wait 

func (s *Sub) Wait(ctx context.Context) error

Wait is a convenience method that parks on the channel and automatically unsubs on any outcome. Returns:

  • ErrUnsubscribed if already unsubscribed.
  • ErrCanceledByNotifier If the subscribed index was flushed by the notifier.
  • context.Err() otherwise (nil on success).
Example 
package main

import (
	"context"
	"fmt"

	"github.com/mnzbono/indexsub"
)

func main() {
	notifier := indexsub.NewNotifier()

	sub := notifier.Subscribe(10)

	ctx, cancel := context.WithCancel(context.Background())
	cancel()

	notifier.Update(11) // subscribed index is satisfied...

	if err := sub.Wait(ctx); err != nil { // ... but ctx error has priority.
		fmt.Println(err)
	}
}

Output:
context canceled

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.