1package forestdb
2
3import (
4	"fmt"
5	"sync"
6)
7
8// KVPool is a structure representing a pool of KVStores
9// inside a file.  Each has been opened with it's own
10// File handle, so they can be used concurrently safely.
11type KVPool struct {
12	closedMutex sync.RWMutex
13	closed      bool
14	stores      chan *KVStore
15}
16
17var PoolClosed = fmt.Errorf("pool already closed")
18
19func NewKVPool(filename string, config *Config, kvstore string, kvconfig *KVStoreConfig, size int) (*KVPool, error) {
20	rv := KVPool{}
21	rv.stores = make(chan *KVStore, size)
22	for i := 0; i < size; i++ {
23		db, err := Open(filename, config)
24		if err != nil {
25			// close everything else we've already opened
26			rv.Close() // ignore errors closing? and return open error?
27			return nil, err
28		}
29		kvs, err := db.OpenKVStore(kvstore, kvconfig)
30		if err != nil {
31			// close the db file we just opened
32			db.Close()
33			// close everything else we've already opened
34			rv.Close() // ignore errors closing? and return open error?
35			return nil, err
36		}
37		rv.stores <- kvs
38	}
39	return &rv, nil
40}
41
42func (p *KVPool) Get() (*KVStore, error) {
43	rv, ok := <-p.stores
44	if !ok {
45		return nil, PoolClosed
46	}
47	return rv, nil
48}
49
50func (p *KVPool) Return(kvs *KVStore) error {
51	p.closedMutex.RLock()
52	defer p.closedMutex.RUnlock()
53	if !p.closed {
54		p.stores <- kvs
55		return nil
56	}
57	return PoolClosed
58}
59
60func (p *KVPool) Close() (rverr error) {
61	p.closedMutex.Lock()
62	if !p.closed {
63		close(p.stores)
64	}
65	p.closed = true
66	p.closedMutex.Unlock()
67
68	for kvs := range p.stores {
69		err := kvs.Close()
70		if err != nil {
71			if rverr == nil {
72				rverr = err
73			}
74			// keep going try to close file
75		}
76		db := kvs.File()
77		err = db.Close()
78		if err != nil {
79			if rverr == nil {
80				rverr = err
81			}
82		}
83	}
84	return
85}
86