rmap

Documentation

Overview

Package rmap maintains a Redis-backed replicated map whose local state is ordered by monotonic revisions and updated through pubsub notifications.

Package rmap options keep map construction small and explicit so the runtime behavior stays easy to reason about at every call site.

Package rmap keeps every replicated-map mutation in Lua so Redis updates, revision bumps, and pubsub notifications stay atomic and totally ordered.

Index

• type EventKind
• type Map
  • func Join(ctx context.Context, name string, rdb *redis.Client, opts ...MapOption) (*Map, error)
  • func (sm *Map) AppendUniqueValues(ctx context.Context, key string, items ...string) ([]string, error)
  • func (sm *Map) AppendValues(ctx context.Context, key string, items ...string) ([]string, error)
  • func (sm *Map) Close()
  • func (sm *Map) Delete(ctx context.Context, key string) (string, error)
  • func (sm *Map) DeleteEx(ctx context.Context, key string) (string, bool, error)
  • func (sm *Map) Destroy(ctx context.Context) error
  • func (sm *Map) Get(key string) (string, bool)
  • func (sm *Map) GetValues(key string) ([]string, bool)
  • func (sm *Map) Inc(ctx context.Context, key string, delta int) (int, error)
  • func (sm *Map) Keys() []string
  • func (sm *Map) Len() int
  • func (sm *Map) Map() map[string]string
  • func (sm *Map) RemoveValues(ctx context.Context, key string, items ...string) ([]string, bool, error)
  • func (sm *Map) Reset(ctx context.Context) error
  • func (sm *Map) Set(ctx context.Context, key, value string) (string, error)
  • func (sm *Map) SetAndWait(ctx context.Context, key, value string) (string, error)
  • func (sm *Map) SetEx(ctx context.Context, key, value string) (string, bool, error)
  • func (sm *Map) SetIfNotExists(ctx context.Context, key, value string) (bool, error)
  • func (sm *Map) Subscribe() <-chan EventKind
  • func (sm *Map) TestAndDelete(ctx context.Context, key, test string) (string, error)
  • func (sm *Map) TestAndDeleteEx(ctx context.Context, key, test string) (prev string, existed bool, deleted bool, err error)
  • func (sm *Map) TestAndReset(ctx context.Context, keys, tests []string) (bool, error)
  • func (sm *Map) TestAndSet(ctx context.Context, key, test, value string) (string, error)
  • func (sm *Map) TestAndSetEx(ctx context.Context, key, test, value string) (prev string, existed bool, updated bool, err error)
  • func (sm *Map) Unsubscribe(c <-chan EventKind)
• type MapOption
  • func WithLogger(logger pulse.Logger) MapOption
  • func WithSlidingTTL(ttl time.Duration) MapOption
  • func WithTTL(ttl time.Duration) MapOption

Types

type EventKind

type EventKind int

EventKind is the type of map event.

const (
	// EventChange is the event emitted when a key is added or changed.
	EventChange EventKind = iota + 1
	// EventDelete is the event emitted when a key is deleted.
	EventDelete
	// EventReset is the event emitted when the map is reset.
	EventReset
)

type Map

type Map struct {
	Name string
	// contains filtered or unexported fields
}

Map is a replicated map that emits events when elements change. Multiple processes can join the same replicated map and update it.

func Join

func Join(ctx context.Context, name string, rdb *redis.Client, opts ...MapOption) (*Map, error)

Join retrieves the content of the replicated map with the given name and subscribes to updates. The local content is eventually consistent across all nodes that join the replicated map with the same name.

Clients can call the Map method on the returned Map to retrieve a copy of its content and subscribe to its C channel to receive updates when the content changes (note that multiple remote changes may result in a single notification). The returned Map is safe for concurrent use.

Clients should call Close before exiting to stop updates and release resources resulting in a read-only point-in-time copy.

func (*Map) AppendUniqueValues

func (sm *Map) AppendUniqueValues(ctx context.Context, key string, items ...string) ([]string, error)

AppendUniqueValues appends the given items to the value for the given key if they are not already present and returns the result. The array of items is stored as a JSON array. An error is returned if: - The key is empty - The key contains an equal sign - There's an issue with the Redis operation

Example: AppendUniqueValues(ctx, "fruits", "apple", "banana") would append only unique values to the existing list of fruits and return the updated list.

func (*Map) AppendValues

func (sm *Map) AppendValues(ctx context.Context, key string, items ...string) ([]string, error)

AppendValues appends the given items to the value for the given key and returns the result. The array of items is stored as a JSON array. An error is returned if: - The key is empty - The key contains an equal sign - There's an issue with the Redis operation

Example: AppendValues(ctx, "fruits", "apple", "banana") would append "apple" and "banana" to the existing list of fruits and return the updated list.

func (*Map) Close

func (sm *Map) Close()

Close closes the connection to the map, freeing resources. It is safe to call Close multiple times.

func (*Map) Delete

func (sm *Map) Delete(ctx context.Context, key string) (string, error)

Delete deletes the value for the given key and returns the previous value. An error is returned if: - The key is empty - The key contains an equal sign - There's an issue with the Redis operation

Example: Delete(ctx, "color") would delete the "color" key and return its previous value, if any.

func (*Map) DeleteEx

func (sm *Map) DeleteEx(ctx context.Context, key string) (string, bool, error)

DeleteEx deletes the value for the given key and returns the previous value along with a flag indicating whether the key existed.

func (*Map) Destroy

func (sm *Map) Destroy(ctx context.Context) error

Destroy clears the map content from Redis and notifies subscribers while preserving the internal revision ordering used by live replicas.

func (*Map) Get

func (sm *Map) Get(key string) (string, bool)

Get returns the value for the given key.

func (*Map) GetValues

func (sm *Map) GetValues(key string) ([]string, bool)

GetValues returns the list values for the given key. This is a convenience method intended to be used in conjunction with AppendValues and RemoveValues. List values are stored as JSON arrays.

func (*Map) Inc

func (sm *Map) Inc(ctx context.Context, key string, delta int) (int, error)

Inc increments the value for the given key and returns the result. The value must represent an integer. An error is returned if: - The key is empty - The key contains an equal sign - The value does not represent an integer - There's an issue with the Redis operation

Example: Inc(ctx, "counter", 1) would increment the "counter" by 1 and return the new value.

func (*Map) Keys

func (sm *Map) Keys() []string

Keys returns a copy of the replicated map keys.

func (*Map) Len

func (sm *Map) Len() int

Len returns the number of items in the replicated map.

func (*Map) Map

func (sm *Map) Map() map[string]string

Map returns a copy of the replicated map content.

func (*Map) RemoveValues

func (sm *Map) RemoveValues(ctx context.Context, key string, items ...string) ([]string, bool, error)

RemoveValues removes the given items from the value for the given key and returns the remaining values after removal. The function behaves as follows:

1. The value for the key is expected to be a JSON array of items.
2. It removes all occurrences of the specified items from this list.
3. If the removal results in an empty list, the key is automatically deleted.
4. Returns the remaining items as a slice of strings, a boolean indicating whether any value was removed, and an error (if any).
5. If the key doesn't exist, it returns nil, false, nil.

An error is returned if: - The key is empty - The key contains an equal sign - There's an issue with the Redis operation

Example: Given a key "fruits" with value ["apple","banana","cherry","apple"] RemoveValues(ctx, "fruits", "apple", "cherry") would return (["banana"], true, nil) and update the value in Redis to ["banana"]

func (*Map) Reset

func (sm *Map) Reset(ctx context.Context) error

Reset clears the map content. Reset remains available after Close because it mutates Redis state even though the local replica is already frozen.

func (*Map) Set

func (sm *Map) Set(ctx context.Context, key, value string) (string, error)

Set sets the value for the given key and returns the previous value. An error is returned if: - The key is empty - The key contains an equal sign - There's an issue with the Redis operation

Example: Set(ctx, "color", "blue") would set the "color" key to "blue" and return the previous value, if any.

func (*Map) SetAndWait

func (sm *Map) SetAndWait(ctx context.Context, key, value string) (string, error)

SetAndWait is a convenience method that calls Set and waits until the local replica has observed the committed revision. Multiple concurrent calls with the same key and value are allowed because each waiter is tied to its own committed write rather than the final key/value pair.

The method will return an error if: - The key is empty - The key contains an equal sign - The context is cancelled - The map is stopped - There's an issue with the Redis operation

func (*Map) SetEx

func (sm *Map) SetEx(ctx context.Context, key, value string) (string, bool, error)

SetEx sets the value for the given key and returns the previous value along with a flag indicating whether the key previously existed.

func (*Map) SetIfNotExists

func (sm *Map) SetIfNotExists(ctx context.Context, key, value string) (bool, error)

SetIfNotExists sets the value for key only if it doesn't exist. Returns true if the value was set, false if the key already existed.

func (*Map) Subscribe

func (sm *Map) Subscribe() <-chan EventKind

Subscribe returns a channel that receives notifications when the map changes. The channel is closed when the map is stopped. This channel simply notifies that the map has changed, it does not provide the actual changes, instead the Map method should be used to read the current content. This allows the notification to be sent without blocking. Multiple remote updates may result in a single notification. Subscribe returns nil if the map is stopped.

func (*Map) TestAndDelete

func (sm *Map) TestAndDelete(ctx context.Context, key, test string) (string, error)

TestAndDelete tests that the value for the given key matches the test value and deletes the key if it does. It returns the previous value. An error is returned if: - The key is empty - The key contains an equal sign - There's an issue with the Redis operation

Example: TestAndDelete(ctx, "color", "blue") would delete the "color" key only if its current value is "blue", and return the previous value.

func (*Map) TestAndDeleteEx

func (sm *Map) TestAndDeleteEx(ctx context.Context, key, test string) (prev string, existed bool, deleted bool, err error)

TestAndDeleteEx deletes the given key if its current value matches the given test value. It returns the previous value, whether the key existed and whether the key was deleted.

func (*Map) TestAndReset

func (sm *Map) TestAndReset(ctx context.Context, keys, tests []string) (bool, error)

TestAndReset tests that the values for the given keys match the test values and clears the map if they do. It returns true if the map was cleared, false otherwise. An error is returned if: - Any key is empty - Any key contains an equal sign - There's an issue with the Redis operation

Example: TestAndReset(ctx, []string{"color", "size"}, []string{"blue", "large"}) would clear the map only if the "color" key has value "blue" and the "size" key has value "large", and return true if the map was cleared.

func (*Map) TestAndSet

func (sm *Map) TestAndSet(ctx context.Context, key, test, value string) (string, error)

TestAndSet sets the value for the given key if the current value matches the given test value. The previous value is returned. An error is returned if: - The key is empty - The key contains an equal sign - There's an issue with the Redis operation

Example: TestAndSet(ctx, "color", "red", "blue") would set "color" to "blue" only if its current value is "red", and return the previous value.

func (*Map) TestAndSetEx

func (sm *Map) TestAndSetEx(ctx context.Context, key, test, value string) (prev string, existed bool, updated bool, err error)

TestAndSetEx sets the value for the given key if the current value matches the given test value. It returns the previous value, whether the key existed and whether the value was updated.

func (*Map) Unsubscribe

func (sm *Map) Unsubscribe(c <-chan EventKind)

Unsubscribe removes the given channel from the list of subscribers and closes it.

type MapOption

type MapOption func(*options)

MapOption is a Map creation option.

func WithLogger

func WithLogger(logger pulse.Logger) MapOption

WithLogger sets the logger used by the map.

func WithSlidingTTL

func WithSlidingTTL(ttl time.Duration) MapOption

WithSlidingTTL sets a sliding TTL on the Redis hash backing the map. The TTL is refreshed on every write operation.

func WithTTL

func WithTTL(ttl time.Duration) MapOption

WithTTL sets an absolute TTL on the Redis hash backing the map. The TTL is set once (when the hash is created) and never extended.