go-cart-actor/grain-pool.go

package main

import (
	"fmt"
	"log"
	"time"
)

type GrainPool interface {
	Process(id CartId, messages ...Message) (interface{}, error)
	Get(id CartId) (Grain, error)
}

type Ttl struct {
	Expires time.Time
	Item    *CartGrain
}

type GrainLocalPool struct {
	grains   map[CartId]*CartGrain
	expiry   []Ttl
	spawn    func(id CartId) (*CartGrain, error)
	Ttl      time.Duration
	PoolSize int
}

func NewGrainLocalPool(size int, ttl time.Duration, spawn func(id CartId) (*CartGrain, error)) *GrainLocalPool {
	ret := &GrainLocalPool{
		spawn:    spawn,
		grains:   make(map[CartId]*CartGrain),
		expiry:   make([]Ttl, 0),
		Ttl:      ttl,
		PoolSize: size,
	}
	cartPurge := time.NewTicker(time.Minute)
	go func() {
		<-cartPurge.C
		ret.Purge()
	}()
	return ret
}

func (p *GrainLocalPool) Purge() {
	lastChangeTime := time.Now().Add(-p.Ttl)
	keepChanged := lastChangeTime.Unix()
	for i := 0; i < len(p.expiry); i++ {
		item := p.expiry[i]
		if item.Expires.Before(time.Now()) {
			if item.Item.GetLastChange() > keepChanged {
				log.Printf("Changed item %s expired, keeping", item.Item.GetId())
				p.expiry = append(p.expiry[:i], p.expiry[i+1:]...)
				p.expiry = append(p.expiry, item)
			} else {
				log.Printf("Item %s expired", item.Item.GetId())
				delete(p.grains, item.Item.GetId())
				p.expiry = append(p.expiry[:i], p.expiry[i+1:]...)
			}
		} else {
			break
		}
	}
}

func (p *GrainLocalPool) GetGrains() map[CartId]*CartGrain {
	return p.grains
}

func (p *GrainLocalPool) GetGrain(id CartId) (*CartGrain, error) {
	var err error
	grain, ok := p.grains[id]
	if !ok {
		if len(p.grains) >= p.PoolSize {
			if p.expiry[0].Expires.Before(time.Now()) {
				delete(p.grains, p.expiry[0].Item.GetId())
				p.expiry = p.expiry[1:]
			} else {
				return nil, fmt.Errorf("pool is full")
			}
		}
		grain, err = p.spawn(id)

		p.grains[id] = grain
	}
	return grain, err
}

func (p *GrainLocalPool) Process(id CartId, messages ...Message) (interface{}, error) {
	grain, err := p.GetGrain(id)
	if err == nil && grain != nil {
		for _, message := range messages {
			_, err = grain.HandleMessage(&message, false)
		}
	}
	return grain, err
}

func (p *GrainLocalPool) Get(id CartId) (Grain, error) {
	return p.GetGrain(id)
}