func (uc *SubscribedCache) applyEvent(event *fes.Event) error {
	logrus.WithFields(logrus.Fields{
		fes.PubSubLabelEventID:       event.Id,
		fes.PubSubLabelEventType:     event.Type,
		fes.PubSubLabelAggregateID:   event.Aggregate.Id,
		fes.PubSubLabelAggregateType: event.Aggregate.Type,
	}).Debug("Applying event to subscribed cache.")
	if err := fes.ValidateEvent(event); err != nil {
		return err
	}
	old, err := uc.getOrCreateAggregateForEvent(event)
	if err != nil {
		return err
	}
	updated, err := uc.projector.Project(old, event)
	if err != nil {
		return err
	}
	err = uc.Put(updated)
	if err != nil {
		return err
	}
	ets, _ := ptypes.Timestamp(event.Timestamp)
	if ets.After(uc.createdAt) {
		n := fes.NewNotification(old, updated, event)
		logrus.WithFields(logrus.Fields{
			"event.id":       event.Id,
			"aggregate.id":   event.Aggregate.Id,
			"aggregate.type": event.Aggregate.Type,
			"event.type":     event.Type,
		}).Debug("SubscribedCache: publishing notification of event.")
		return uc.Publisher.Publish(n)
	}
	return nil
}
applyEvent applies an event to the cache. It retrieves the corresponding entity from the cache, copies it,
applies the event to it, and replaces the old with the new entity in the cache.
package cache
import (
	"time"
	"github.com/fission/fission-workflows/pkg/fes"
	"github.com/fission/fission-workflows/pkg/util/pubsub"
	"github.com/golang/protobuf/ptypes"
	"github.com/hashicorp/golang-lru"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/sirupsen/logrus"
)
var (
	cacheCount = prometheus.NewGaugeVec(prometheus.GaugeOpts{
		Namespace: "fes",
		Subsystem: "cache",
		Name:      "current_cache_counts",
		Help:      "The current number of entries in the caches",
	}, []string{"name"})
)
func init() {
	prometheus.MustRegister(cacheCount)
}
type LRUCache struct {
	contents *lru.Cache
}
func NewLRUCache(size int) *LRUCache {
	c, err := lru.New(size)
	if err != nil {
		panic(err)
	}
	return &LRUCache{
		contents: c,
	}
}
func (c *LRUCache) GetAggregate(a fes.Aggregate) (fes.Entity, error) {
	if err := fes.ValidateAggregate(&a); err != nil {
		return nil, err
	}
	i, ok := c.contents.Get(a)
	if !ok {
		return nil, fes.ErrEntityNotFound.WithAggregate(&a)
	}
	return i.(fes.Entity), nil
}
func (c *LRUCache) Put(entity fes.Entity) error {
	if err := fes.ValidateEntity(entity); err != nil {
		return err
	}
	a := fes.GetAggregate(entity)
	c.contents.Add(a, entity)
	return nil
}
func (c *LRUCache) List() []fes.Aggregate {
	keys := c.contents.Keys()
	results := make([]fes.Aggregate, len(keys))
	for i, key := range keys {
		results[i] = key.(fes.Aggregate)
	}
	return results
}
func (c *LRUCache) Refresh(key fes.Aggregate) {
}
func (c *LRUCache) Invalidate(a fes.Aggregate) {
	if err := fes.ValidateAggregate(&a); err != nil {
		logrus.Warnf("Failed to invalidate entry in cache: %v", err)
		return
	}
	c.contents.Remove(a)
}
type SubscribedCache struct {
	pubsub.Publisher
	fes.CacheReaderWriter
	createdAt time.Time
	projector fes.Projector
	closeC    chan struct{}
}
func NewSubscribedCache(cache fes.CacheReaderWriter, projector fes.Projector,
	sub *pubsub.Subscription) *SubscribedCache {
	c := &SubscribedCache{
		Publisher:         pubsub.NewPublisher(),
		CacheReaderWriter: cache,
		projector:         projector,
		createdAt:         time.Now(),
	}
	c.closeC = make(chan struct{})
	go func() {
		for {
			select {
			case <-c.closeC:
				logrus.Debug("SubscribedCache: listener stopped.")
				return
			case e := <-sub.Ch:
				event, ok := e.(*fes.Event)
				if !ok {
					logrus.WithField("event", e).Error("Ignoring received malformed event.")
					continue
				}
				logrus.WithField("msg", e.Labels()).Debug("SubscribedCache: received event.")
				err := c.applyEvent(event)
				if err != nil {
					logrus.WithField("event", event).Errorf("Failed to handle event: %v", err)
				}
			}
		}
	}()
	return c
}