bump go-pkgz/lcw to v0.7.1

2020-06-25 17:28:33 -05:00
parent df9c05b490
commit ebf379f4d6
13 changed files with 180 additions and 15 deletions
@@ -27,8 +27,11 @@ Main features:

 ## Usage

-```
-cache := lcw.NewLruCache(lcw.MaxKeys(500), lcw.MaxCacheSize(65536), lcw.MaxValSize(200), lcw.MaxKeySize(32))
+```go
+cache, err := lcw.NewLruCache(lcw.MaxKeys(500), lcw.MaxCacheSize(65536), lcw.MaxValSize(200), lcw.MaxKeySize(32))
+if err != nil {
+    panic("failed to create cache")
+}
 defer cache.Close()

 val, err := cache.Get("key123", func() (lcw.Value, error) {
@@ -59,7 +62,7 @@ Cache can be created with URIs:
 1. Key is not a string, but a composed type made from partition, key-id and list of scopes (tags). 
 1. Value type limited to `[]byte`
 1. Added `Flush` method for scoped/tagged invalidation of multiple records in a given partition
-1. A simplified interface with Get, Stat and Flush only.
+1. A simplified interface with Get, Stat, Flush and Close only.

 ## Details

@@ -6,6 +6,8 @@
 // 3 flavors of cache provided - NoP (do-nothing cache), ExpirableCache (TTL based), and LruCache
 package lcw

+//go:generate sh -c "mockery -inpkg -name LoadingCache -print > /tmp/cache-mock.tmp && mv /tmp/cache-mock.tmp cache_mock.go"
+
 import (
 	"fmt"
 )
@@ -81,3 +83,4 @@ func (n *Nop) Stat() CacheStat {
 func (n *Nop) Close() error {
 	return nil
 }
+
@@ -0,0 +1,24 @@
+// Package eventbus provides PubSub interface used for distributed cache invalidation,
+// as well as NopPubSub and RedisPubSub implementations.
+package eventbus
+
+// PubSub interface is used for distributed cache invalidation.
+// Publish is called on each entry invalidation,
+// Subscribe is used for subscription for these events.
+type PubSub interface {
+	Publish(fromID, key string) error
+	Subscribe(fn func(fromID, key string)) error
+}
+
+// NopPubSub implements default do-nothing pub-sub (event bus)
+type NopPubSub struct{}
+
+// Subscribe does nothing for NopPubSub
+func (n *NopPubSub) Subscribe(fn func(fromID, key string)) error {
+	return nil
+}
+
+// Publish does nothing for NopPubSub
+func (n *NopPubSub) Publish(fromID, key string) error {
+	return nil
+}
@@ -0,0 +1,72 @@
+package eventbus
+
+import (
+	"strings"
+	"time"
+
+	"github.com/go-redis/redis/v7"
+	"github.com/hashicorp/go-multierror"
+	"github.com/pkg/errors"
+)
+
+// NewRedisPubSub creates new RedisPubSub with given parameters.
+// Returns an error in case of problems with creating PubSub client for specified channel.
+func NewRedisPubSub(addr, channel string) (*RedisPubSub, error) {
+	client := redis.NewClient(&redis.Options{Addr: addr})
+	pubSub := client.Subscribe(channel)
+	// wait for subscription to be created and ignore the message
+	if _, err := pubSub.Receive(); err != nil {
+		_ = client.Close()
+		return nil, errors.Wrapf(err, "problem subscribing to channel %s on address %s", channel, addr)
+	}
+	return &RedisPubSub{client: client, pubSub: pubSub, channel: channel, done: make(chan struct{})}, nil
+}
+
+// RedisPubSub provides Redis implementation for PubSub interface
+type RedisPubSub struct {
+	client  *redis.Client
+	pubSub  *redis.PubSub
+	channel string
+
+	done chan struct{}
+}
+
+// Subscribe calls provided function on subscription channel provided on new RedisPubSub instance creation.
+// Should not be called more than once. Spawns a goroutine and does not return an error.
+func (m *RedisPubSub) Subscribe(fn func(fromID, key string)) error {
+	go func(done <-chan struct{}, pubsub *redis.PubSub) {
+		for {
+			select {
+			case <-done:
+				return
+			default:
+			}
+			msg, err := pubsub.ReceiveTimeout(time.Second * 10)
+			if err != nil {
+				continue
+			}
+
+			// Process the message
+			if msg, ok := msg.(*redis.Message); ok {
+				payload := strings.Split(msg.Payload, "$")
+				fn(payload[0], strings.Join(payload[1:], "$"))
+			}
+		}
+	}(m.done, m.pubSub)
+
+	return nil
+}
+
+// Publish publishes provided message to channel provided on new RedisPubSub instance creation
+func (m *RedisPubSub) Publish(fromID, key string) error {
+	return m.client.Publish(m.channel, fromID+"$"+key).Err()
+}
+
+// Close cleans up running goroutines and closes Redis clients
+func (m *RedisPubSub) Close() error {
+	close(m.done)
+	errs := new(multierror.Error)
+	errs = multierror.Append(errs, errors.Wrap(m.pubSub.Close(), "problem closing pubSub client"))
+	errs = multierror.Append(errs, errors.Wrap(m.client.Close(), "problem closing redis client"))
+	return errs.ErrorOrNil()
+}
@@ -4,8 +4,10 @@ import (
 	"sync/atomic"
 	"time"

+	"github.com/google/uuid"
 	"github.com/pkg/errors"

+	"github.com/go-pkgz/lcw/eventbus"
 	"github.com/go-pkgz/lcw/internal/cache"
 )

@@ -14,6 +16,7 @@ type ExpirableCache struct {
 	options
 	CacheStat
 	currentSize int64
+	id          string
 	backend     *cache.LoadingCache
 }

@@ -24,7 +27,9 @@ func NewExpirableCache(opts ...Option) (*ExpirableCache, error) {
 			maxKeys:      1000,
 			maxValueSize: 0,
 			ttl:          5 * time.Minute,
+			eventBus:     &eventbus.NopPubSub{},
 		},
+		id: uuid.New().String(),
 	}

 	for _, opt := range opts {
@@ -33,6 +38,10 @@ func NewExpirableCache(opts ...Option) (*ExpirableCache, error) {
 		}
 	}

+	if err := res.eventBus.Subscribe(res.onBusEvent); err != nil {
+		return nil, errors.Wrapf(err, "can't subscribe to event bus")
+	}
+
 	backend, err := cache.NewLoadingCache(
 		cache.MaxKeys(res.maxKeys),
 		cache.TTL(res.ttl),
@@ -45,6 +54,10 @@ func NewExpirableCache(opts ...Option) (*ExpirableCache, error) {
 				size := s.Size()
 				atomic.AddInt64(&res.currentSize, -1*int64(size))
 			}
+			// ignore the error on Publish as we don't have log inside the module and
+			// there is no other way to handle it: we publish the cache invalidation
+			// and hope for the best
+			_ = res.eventBus.Publish(res.id, key)
 		}),
 	)
 	if err != nil {
@@ -128,6 +141,13 @@ func (c *ExpirableCache) Close() error {
 	return nil
 }

+// onBusEvent reacts on invalidation message triggered by event bus from another cache instance
+func (c *ExpirableCache) onBusEvent(id, key string) {
+	if id != c.id {
+		c.backend.Invalidate(key)
+	}
+}
+
 func (c *ExpirableCache) size() int64 {
 	return atomic.LoadInt64(&c.currentSize)
 }
@@ -3,6 +3,7 @@ module github.com/go-pkgz/lcw
 require (
 	github.com/alicebob/miniredis/v2 v2.11.4
 	github.com/go-redis/redis/v7 v7.2.0
+	github.com/google/uuid v1.1.1
 	github.com/hashicorp/go-multierror v1.1.0
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/pkg/errors v0.9.1
@@ -16,6 +16,8 @@ github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/gomodule/redigo v1.7.1-0.20190322064113-39e2c31b7ca3 h1:6amM4HsNPOvMLVc2ZnyqrjeQ92YAVWn7T4WBKK87inY=
 github.com/gomodule/redigo v1.7.1-0.20190322064113-39e2c31b7ca3/go.mod h1:B4C85qUVwatsJoIUNIfCRsp7qO0iAmpGFZ4EELWSbC4=
+github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
+github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/go-multierror v1.1.0 h1:B9UzwGQJehnUY1yNrnwREHc3fGbC2xefo8g4TbElacI=
@@ -3,8 +3,11 @@ package lcw
 import (
 	"sync/atomic"

+	"github.com/google/uuid"
 	lru "github.com/hashicorp/golang-lru"
 	"github.com/pkg/errors"
+
+	"github.com/go-pkgz/lcw/eventbus"
 )

 // LruCache wraps lru.LruCache with loading cache Get and size limits
@@ -13,6 +16,7 @@ type LruCache struct {
 	CacheStat
 	backend     *lru.Cache
 	currentSize int64
+	id          string // uuid identifying cache instance
 }

 // NewLruCache makes LRU LoadingCache implementation, 1000 max keys by default
@@ -21,7 +25,9 @@ func NewLruCache(opts ...Option) (*LruCache, error) {
 		options: options{
 			maxKeys:      1000,
 			maxValueSize: 0,
+			eventBus:     &eventbus.NopPubSub{},
 		},
+		id: uuid.New().String(),
 	}
 	for _, opt := range opts {
 		if err := opt(&res.options); err != nil {
@@ -29,23 +35,33 @@ func NewLruCache(opts ...Option) (*LruCache, error) {
 		}
 	}

+	err := res.init()
+	return &res, err
+}
+
+func (c *LruCache) init() error {
+	if err := c.eventBus.Subscribe(c.onBusEvent); err != nil {
+		return errors.Wrapf(err, "can't subscribe to event bus")
+	}
+
 	onEvicted := func(key interface{}, value interface{}) {
-		if res.onEvicted != nil {
-			res.onEvicted(key.(string), value)
+		if c.onEvicted != nil {
+			c.onEvicted(key.(string), value)
 		}
 		if s, ok := value.(Sizer); ok {
 			size := s.Size()
-			atomic.AddInt64(&res.currentSize, -1*int64(size))
+			atomic.AddInt64(&c.currentSize, -1*int64(size))
 		}
+		_ = c.eventBus.Publish(c.id, key.(string)) // signal invalidation to other nodes
 	}

 	var err error
 	// OnEvicted called automatically for expired and manually deleted
-	if res.backend, err = lru.NewWithEvict(res.maxKeys, onEvicted); err != nil {
-		return nil, errors.Wrap(err, "failed to make lru cache backend")
+	if c.backend, err = lru.NewWithEvict(c.maxKeys, onEvicted); err != nil {
+		return errors.Wrap(err, "failed to make lru cache backend")
 	}

-	return &res, nil
+	return nil
 }

 // Get gets value by key or load with fn if not found in cache
@@ -131,6 +147,13 @@ func (c *LruCache) Close() error {
 	return nil
 }

+// onBusEvent reacts on invalidation message triggered by event bus from another cache instance
+func (c *LruCache) onBusEvent(id, key string) {
+	if id != c.id && c.backend.Contains(key) { // prevent reaction on event from this cache
+		c.backend.Remove(key)
+	}
+}
+
 func (c *LruCache) size() int64 {
 	return atomic.LoadInt64(&c.currentSize)
 }
@@ -3,6 +3,8 @@ package lcw
 import (
 	"errors"
 	"time"
+
+	"github.com/go-pkgz/lcw/eventbus"
 )

 type options struct {
@@ -12,6 +14,7 @@ type options struct {
 	maxCacheSize int64
 	ttl          time.Duration
 	onEvicted    func(key string, value Value)
+	eventBus     eventbus.PubSub
 }

 // Option func type
@@ -84,3 +87,11 @@ func OnEvicted(fn func(key string, value Value)) Option {
 		return nil
 	}
 }
+
+// EventBus sets PubSub for distributed cache invalidation
+func EventBus(pubSub eventbus.PubSub) Option {
+	return func(o *options) error {
+		o.eventBus = pubSub
+		return nil
+	}
+}