store.go

package fwk

import (
	"reflect"
)

type achan chan interface{}

type datastore struct {
	SvcBase
	store map[string]achan
	quit  chan struct{}
}

func (ds *datastore) Configure(ctx Context) error {
	return nil
}

func (ds *datastore) Get(k string) (interface{}, error) {
	ch, ok := ds.store[k]
	if !ok {
		return nil, Errorf("Store.Get: no such key [%v]", k)
	}
	select {
	case v, ok := <-ch:
		if !ok {
			return nil, Errorf("%s: closed channel for key [%s]", ds.Name(), k)
		}
		ch <- v
		return v, nil
	case <-ds.quit:
		return nil, Errorf("%s: timeout to get [%s]", ds.Name(), k)
	}
}

func (ds *datastore) Put(k string, v interface{}) error {
	select {
	case ds.store[k] <- v:
		return nil
	case <-ds.quit:
		return Errorf("%s: timeout to put [%s]", ds.Name(), k)
	}
}

func (ds *datastore) Has(k string) bool {
	_, ok := ds.store[k]
	return ok
}

func (ds *datastore) StartSvc(ctx Context) error {
	ds.store = make(map[string]achan)
	return nil
}

func (ds *datastore) StopSvc(ctx Context) error {
	ds.store = make(map[string]achan)
	return nil
}

// reset deletes the payload and resets the associated channel
func (ds *datastore) reset(keys []string) error {
	var err error
	for _, k := range keys {
		ch, ok := ds.store[k]
		if ok {
			select {
			case vv := <-ch:
				if vv, ok := vv.(Deleter); ok {
					err = vv.Delete()
					if err != nil {
						return err
					}
				}
			default:
			}
		}
		ds.store[k] = make(achan, 1)
	}
	ds.quit = make(chan struct{})
	return err
}

// close notifies components hanging on store.Get or .Put that event has been aborted
func (ds *datastore) close() {
	close(ds.quit)
}

func init() {
	Register(reflect.TypeOf(datastore{}),
		func(typ, name string, mgr App) (Component, error) {
			return &datastore{
				SvcBase: NewSvc(typ, name, mgr),
				store:   make(map[string]achan),
				quit:    make(chan struct{}),
			}, nil
		},
	)
}

// interface tests
var _ Store = (*datastore)(nil)

// EOF