diff --git a/refreshable/async.go b/refreshable/async.go index 4ccd6f75..d386ad8a 100644 --- a/refreshable/async.go +++ b/refreshable/async.go @@ -71,7 +71,7 @@ func Wait[T any](ctx context.Context, ready Ready[T]) (T, bool) { // ready is an Updatable which exposes a channel that is closed when a value is first available. // Current returns the zero value before Update is called, marking the value ready. type ready[T any] struct { - in Updatable[T] + in *defaultRefreshable[T] readyC <-chan struct{} cancel context.CancelFunc } diff --git a/refreshable/refreshable.go b/refreshable/refreshable.go index b48dbb96..17329e52 100644 --- a/refreshable/refreshable.go +++ b/refreshable/refreshable.go @@ -5,7 +5,10 @@ package refreshable import ( + "bytes" "context" + "maps" + "slices" "sync" ) @@ -59,9 +62,80 @@ type Ready[T any] interface { // It is safe to call multiple times. type UnsubscribeFunc func() -// New returns a new Updatable that begins with the given value. +// New returns a new Updatable that begins with the given value and uses reflect.DeepEqual for debouncing. func New[T any](val T) Updatable[T] { - return newDefault(val) + return newDefault(val, nil) +} + +// NewComparable returns a new Updatable using the == operator for debouncing. +// Use for primitive and comparable types like string, int, or structs with only comparable fields. +// Convert an existing refreshable with CacheWith(NewComparable, original). +func NewComparable[T comparable](val T) *defaultRefreshable[T] { + return newDefault(val, func(x, y T) bool { return x == y }) +} + +// NewComparableMap returns a new Updatable for maps with comparable keys and values, +// using maps.Equal for debouncing. +// Convert an existing refreshable with CacheWith(NewComparableMap, original). +func NewComparableMap[T ~map[K]V, K comparable, V comparable](val T) *defaultRefreshable[T] { + return newDefault(val, maps.Equal[T, T, K, V]) +} + +// NewComparableSlice returns a new Updatable for slices with comparable elements, +// using slices.Equal for debouncing. +// Convert an existing refreshable with CacheWith(NewComparableSlice, original). +func NewComparableSlice[T ~[]E, E comparable](val T) *defaultRefreshable[T] { + return newDefault(val, slices.Equal[T, E]) +} + +// NewBytes returns a new Updatable for byte slices (or named types with underlying type []byte), +// using bytes.Equal for debouncing. +// Convert an existing refreshable with CacheWith(NewBytes, original). +func NewBytes[T ~[]byte](val T) *defaultRefreshable[T] { + return newDefault(val, func(old T, val T) bool { return bytes.Equal(old, val) }) +} + +// selfEqual is a type that can compare itself to another value of the same type. +// Examples include *x509.CertPool, *x509.Certificate, slog.Attr, slog.Value, net.IP, reflect.Value, regexp.Regexp, and time.Time. +// Can also be implemented by any type that requires custom comparison. +type selfEqual[T any] interface { + Equal(T) bool +} + +// NewEqualMethod returns a new Updatable for types implementing Equal(T) bool, +// using that method for debouncing. Compatible with types like time.Time and net.IP. +// Convert an existing refreshable with CacheWith(NewEqualMethod, original). +func NewEqualMethod[T selfEqual[T]](val T) *defaultRefreshable[T] { + return newDefault(val, T.Equal) +} + +// NewEqualMethodMap returns a new Updatable for maps whose values implement Equal(V) bool, +// comparing entries element-wise for debouncing. +// Convert an existing refreshable with CacheWith(NewEqualMethodMap, original). +func NewEqualMethodMap[T ~map[K]V, K comparable, V selfEqual[V]](val T) *defaultRefreshable[T] { + return newDefault(val, func(old T, val T) bool { return maps.EqualFunc[T, T, K, V](old, val, V.Equal) }) +} + +// NewEqualMethodSlice returns a new Updatable for slices whose elements implement Equal(E) bool, +// comparing elements pairwise for debouncing. +// Convert an existing refreshable with CacheWith(NewEqualMethodSlice, original). +func NewEqualMethodSlice[T ~[]E, E selfEqual[E]](val T) *defaultRefreshable[T] { + return newDefault(val, func(old T, val T) bool { return slices.EqualFunc[T, T, E](old, val, E.Equal) }) +} + +// NewEqualFunc returns a new Updatable using a custom equality function for debouncing. +// Use for any type where you can provide an appropriate comparison function. +// If equals is nil, the default equality function (reflect.DeepEqual) is used. +// Convert an existing refreshable with CacheWithFunc. +func NewEqualFunc[T any](val T, equal func(T, T) bool) *defaultRefreshable[T] { + return newDefault(val, equal) +} + +// CacheWithFunc returns a new Refreshable that subscribes to the original Refreshable and caches its value. +// This is useful in combination with View to avoid recomputing an expensive mapped value +// each time it is retrieved. The returned refreshable is read-only (does not implement Update). +func CacheWithFunc[T any](equals func(old T, val T) bool, original Refreshable[T]) *readOnlyRefreshable[T] { + return CacheWith(func(val T) *defaultRefreshable[T] { return NewEqualFunc(val, equals) }, original) } // Cached returns a new Refreshable that subscribes to the original Refreshable and caches its value. @@ -73,6 +147,16 @@ func Cached[T any](original Refreshable[T]) (Refreshable[T], UnsubscribeFunc) { return out.readOnly(), stop } +// CacheWith returns a new Refreshable that subscribes to the original Refreshable and caches its value +// using the provided constructor to determine an "equality function" used to debounce new values. +// This is useful in combination with View to avoid recomputing an expensive mapped value +// each time it is retrieved. The returned refreshable is read-only (does not implement Update). +func CacheWith[T any](constructor func(val T) *defaultRefreshable[T], original Refreshable[T]) *readOnlyRefreshable[T] { + out := constructor(*new(T)) + original.Subscribe(out.Update) + return out.readOnly() +} + // View returns a Refreshable implementation that converts the original Refreshable value to a new value using mapFn. // Current() and Subscribe() invoke mapFn as needed on the current value of the original Refreshable. // Subscription callbacks are invoked with the mapped value each time the original value changes diff --git a/refreshable/refreshable_constructors_test.go b/refreshable/refreshable_constructors_test.go new file mode 100644 index 00000000..7c46f503 --- /dev/null +++ b/refreshable/refreshable_constructors_test.go @@ -0,0 +1,176 @@ +// Copyright (c) 2021 Palantir Technologies. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package refreshable_test + +import ( + "testing" + "time" + + refreshable "github.com/palantir/pkg/refreshable/v2" + "github.com/stretchr/testify/assert" +) + +// equalLenString implements selfEqual for use with NewEqualMethod. +// Its Equal method compares string lengths rather than contents, so two +// strings of the same length are "equal" even if they differ—making the +// behavior clearly distinct from reflect.DeepEqual. +type equalLenString struct{ val string } + +func (e equalLenString) Equal(other equalLenString) bool { return len(e.val) == len(other.val) } + +// testUpdatable verifies debouncing: updating with an equal value should not +// notify subscribers, while updating with a different value should. +func testUpdatable[T any](t *testing.T, r refreshable.Updatable[T], same, different T) { + t.Helper() + updates := 0 + r.Subscribe(func(T) { updates++ }) + assert.Equal(t, 1, updates, "subscribe should fire immediately") + + r.Update(same) + assert.Equal(t, 1, updates, "equal value should be debounced") + + r.Update(different) + assert.Equal(t, 2, updates, "different value should notify") +} + +func TestNewComparable(t *testing.T) { + t.Run("string", func(t *testing.T) { + r := refreshable.NewComparable("hello") + assert.Equal(t, "hello", r.Current()) + testUpdatable(t, r, "hello", "world") + }) + t.Run("int", func(t *testing.T) { + r := refreshable.NewComparable(42) + assert.Equal(t, 42, r.Current()) + testUpdatable(t, r, 42, 99) + }) + t.Run("bool", func(t *testing.T) { + testUpdatable(t, refreshable.NewComparable(true), true, false) + }) + t.Run("struct", func(t *testing.T) { + type kv struct{ K, V string } + testUpdatable(t, refreshable.NewComparable(kv{"a", "b"}), kv{"a", "b"}, kv{"c", "d"}) + }) +} + +func TestNewComparableMap(t *testing.T) { + r := refreshable.NewComparableMap(map[string]int{"a": 1}) + assert.Equal(t, map[string]int{"a": 1}, r.Current()) + testUpdatable(t, r, map[string]int{"a": 1}, map[string]int{"b": 2}) +} + +func TestNewComparableSlice(t *testing.T) { + r := refreshable.NewComparableSlice([]string{"a", "b"}) + assert.Equal(t, []string{"a", "b"}, r.Current()) + testUpdatable(t, r, []string{"a", "b"}, []string{"c"}) +} + +func TestNewEqualMethod(t *testing.T) { + t.Run("time.Time", func(t *testing.T) { + now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + // time.Equal treats the same instant in different zones as equal. + sameInstant := now.In(time.FixedZone("UTC+1", 3600)) + testUpdatable(t, refreshable.NewEqualMethod(now), sameInstant, now.Add(time.Second)) + }) + t.Run("custom", func(t *testing.T) { + // "hi" and "ab" have the same length (Equal returns true), but "bye" has a different length. + testUpdatable(t, refreshable.NewEqualMethod(equalLenString{"hi"}), equalLenString{"ab"}, equalLenString{"bye"}) + }) +} + +func TestNewEqualFunc(t *testing.T) { + // NewEqualFunc works with any type given a custom equality function. + type point struct{ X, Y int } + r := refreshable.NewEqualFunc(point{1, 2}, func(a, b point) bool { return a == b }) + assert.Equal(t, point{1, 2}, r.Current()) + testUpdatable(t, r, point{1, 2}, point{3, 4}) +} + +func TestNewEqualMethodMap(t *testing.T) { + now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + sameInstant := now.In(time.FixedZone("UTC+1", 3600)) + r := refreshable.NewEqualMethodMap(map[string]time.Time{"t": now}) + testUpdatable(t, r, map[string]time.Time{"t": sameInstant}, map[string]time.Time{"t": now.Add(time.Hour)}) +} + +func TestNewEqualMethodSlice(t *testing.T) { + now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + sameInstant := now.In(time.FixedZone("UTC+1", 3600)) + r := refreshable.NewEqualMethodSlice([]time.Time{now}) + testUpdatable(t, r, []time.Time{sameInstant}, []time.Time{now.Add(time.Hour)}) +} + +func TestNewBytes(t *testing.T) { + r := refreshable.NewBytes([]byte("hello")) + assert.Equal(t, []byte("hello"), r.Current()) + testUpdatable(t, r, []byte("hello"), []byte("world")) +} + +func TestNewBytes_NamedType(t *testing.T) { + type blob []byte + r := refreshable.NewBytes(blob("data")) + assert.Equal(t, blob("data"), r.Current()) + testUpdatable(t, r, blob("data"), blob("other")) +} + +func TestCacheWith(t *testing.T) { + t.Run("propagates values from source", func(t *testing.T) { + source := refreshable.NewComparable("hello") + cached := refreshable.CacheWith[string](refreshable.NewComparable, source) + assert.Equal(t, "hello", cached.Current()) + + source.Update("world") + assert.Equal(t, "world", cached.Current()) + }) + + t.Run("debounces with constructor equality", func(t *testing.T) { + // Use NewEqualMethod so that time.Time.Equal is used for debouncing, + // which treats the same instant in different zones as equal. + now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + source := refreshable.New(now) + sourceUpdates := 0 + source.Subscribe(func(t time.Time) { sourceUpdates++ }) + assert.Equal(t, 1, sourceUpdates, "subscribe should fire immediately") + + cached := refreshable.CacheWith[time.Time](refreshable.NewEqualMethod, source) + cacheUpdates := 0 + cached.Subscribe(func(time.Time) { cacheUpdates++ }) + assert.Equal(t, 1, cacheUpdates, "subscribe should fire immediately") + + // Same instant in a different zone: time.Equal considers them equal. + sameInstant := now.In(time.FixedZone("UTC+1", 3600)) + source.Update(sameInstant) + assert.Equal(t, 1, cacheUpdates, "equal time should be debounced") + assert.Equal(t, 2, sourceUpdates, "expected reflect-based source not to debounce equal time") + + source.Update(now.Add(time.Second)) + assert.Equal(t, 2, cacheUpdates, "different time should notify") + }) + + t.Run("debounces map with element equality", func(t *testing.T) { + // Source uses reflect.DeepEqual, which compares time.Time zone pointers. + // CacheWith uses NewEqualMethodMap, which compares values with time.Time.Equal. + // An update with the same instant in a different zone passes through the + // source (not DeepEqual) but is debounced by the cached refreshable. + now := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) + source := refreshable.New(map[string]time.Time{"t": now}) + sourceUpdates := 0 + source.Subscribe(func(t map[string]time.Time) { sourceUpdates++ }) + cached := refreshable.CacheWith[map[string]time.Time](refreshable.NewEqualMethodMap, source) + assert.Equal(t, map[string]time.Time{"t": now}, cached.Current()) + + cacheUpdates := 0 + cached.Subscribe(func(map[string]time.Time) { cacheUpdates++ }) + assert.Equal(t, 1, cacheUpdates) + + sameInstant := now.In(time.FixedZone("UTC+1", 3600)) + source.Update(map[string]time.Time{"t": sameInstant}) + assert.Equal(t, 1, cacheUpdates, "same instant in different zone should be debounced by CacheWith") + assert.Equal(t, 2, sourceUpdates, "expected reflect-based source not to debounce equal time") + + source.Update(map[string]time.Time{"t": now.Add(time.Hour)}) + assert.Equal(t, 2, cacheUpdates, "different time should notify") + }) +} diff --git a/refreshable/refreshable_default.go b/refreshable/refreshable_default.go index a4132aaf..011556cf 100644 --- a/refreshable/refreshable_default.go +++ b/refreshable/refreshable_default.go @@ -12,18 +12,20 @@ import ( type defaultRefreshable[T any] struct { mux sync.Mutex - current atomic.Value + current atomic.Pointer[T] subscribers []*func(T) + equals func(T, T) bool } -func newDefault[T any](val T) *defaultRefreshable[T] { +func newDefault[T any](val T, equals func(T, T) bool) *defaultRefreshable[T] { d := new(defaultRefreshable[T]) + d.equals = equals d.current.Store(&val) return d } func newZero[T any]() *defaultRefreshable[T] { - return newDefault(*new(T)) + return newDefault(*new(T), nil) } // Update changes the value of the Refreshable, then blocks while subscribers are executed. @@ -31,16 +33,16 @@ func (d *defaultRefreshable[T]) Update(val T) { d.mux.Lock() defer d.mux.Unlock() old := d.current.Swap(&val) - if reflect.DeepEqual(*(old.(*T)), val) { - return - } - for _, sub := range d.subscribers { - (*sub)(val) + equal := (d.equals != nil && d.equals(*old, val)) || (d.equals == nil && reflect.DeepEqual(*old, val)) + if !equal { + for _, sub := range d.subscribers { + (*sub)(val) + } } } func (d *defaultRefreshable[T]) Current() T { - return *(d.current.Load().(*T)) + return *(d.current.Load()) } func (d *defaultRefreshable[T]) Subscribe(consumer func(T)) UnsubscribeFunc { diff --git a/refreshable/refreshable_validating.go b/refreshable/refreshable_validating.go index c8cee90e..d0313271 100644 --- a/refreshable/refreshable_validating.go +++ b/refreshable/refreshable_validating.go @@ -38,7 +38,8 @@ func (v *validRefreshable[T]) Validation() (T, error) { func newValidRefreshable[M any]() *validRefreshable[M] { valid := &validRefreshable[M]{ - r: newDefault(validRefreshableContainer[M]{}), + // TODO: Wire equality + r: newDefault(validRefreshableContainer[M]{}, nil), } return valid } @@ -94,7 +95,8 @@ func identity[T any](validatingFn func(context.Context, T) error) func(ctx conte func validatedFromRefreshable[M any](original Refreshable[M]) Validated[M] { valid := &validRefreshable[M]{ - r: newDefault(validRefreshableContainer[M]{}), + // TODO: Wire equality + r: newDefault(validRefreshableContainer[M]{}, nil), } original.Subscribe(func(m M) { valid.r.Update(validRefreshableContainer[M]{