rmap

README

Replicated Map

Replicated maps provide a mechanism for sharing data across a fleet of microservices.

Overview

Pulse replicated maps leverage Redis hashes and pub/sub to maintain replicated in-memory copies of a map across multiple nodes. Any change to the map is automatically replicated to all nodes and results in a notification that can be used to trigger actions.

Upon joining a replicated map the node receives an up-to-date snapshot of its content. The replicated map then guarantees that any change to the map results in a notification.

Usage

Creating a Replicated Map

To create a replicated map you must provide a name and a Redis client. The name is used to namespace the Redis keys and pub/sub channels used by the map. The map should later be closed to free up resources.

The Join function creates a new replicated map or joins an existing one. The Close function closes the subscription channels and the subscription to Redis.

Writing to the Map

• The Set method sets the value for a given key and returns the previous value.

• The TestAndSet method sets the value for a given key if the current value matches the expected value. It returns the previous value.

• The AppendValues and RemoveValues methods append or remove values to or from a list.

Values are stored as strings. The AppendValues method converts the values to a comma separated string before storing them.

• The Inc method increments a counter by a given amount and returns the new value.

• The Delete method deletes a key from the map and returns the previous value if any.

• Finally Reset clears the entire map.

Reading from the Map

• The Get method retrieves the value associated with a specified key and returns both the value itself and a boolean flag indicating whether the value exists in the map.

• The Keys method returns a list of all the keys in the map.

• The Len method returns the number of keys in the map.

• The Map method returns a snapshot of the current key-value pairs in the map.

Subscribing to Map Updates

• The Subscribe method returns a channel that can be used to receive notifications when the map is updated. The channel is closed when the map is closed.

Note: Notifications do not include any information about the change that triggered them. The application must query the map to retrieve the current state. Multiple updates may result in a single notification however it is guaranteed that the map is in the latest state when the notification is received.

• The Unsubscribe method unsubscribes from map updates. It also closes the subscription channel.

When to Use Replicated Maps

Replicated maps being stored in memory are not suitable for large data sets. They are also better suited for read-heavy workloads as reads are local but writes require a roundtrip to Redis.

A good use case for replicated maps is metadata or configuration information that is shared across a fleet of microservices. For example a replicated map can be used to share the list of active users across a fleet of microservices. The microservices can then use the replicated map to check if a user is active without having to query a database.

Examples

The examples/rmap directory contains a few examples that demonstrate the basic usage of the rmap package.

Documentation

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) 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) 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) 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) Unsubscribe(c <-chan EventKind)
• type MapOption
  • func WithLogger(logger pulse.Logger) 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 comma-separated list. 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 comma-separated list. 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) 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 comma separated values for the given key. This is a convenience method intended to be used in conjunction with AppendValues and RemoveValues.

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 comma-separated list 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 is the only method that can be called after the map is closed.

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 then waits for the update to be applied and the notification to be sent.

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) 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) 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.