pebblequeue/set.go

125 lines
2.9 KiB
Go

// Copyright 2020 Andrew Thornton. All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.
package levelqueue
import (
"sync"
"github.com/cockroachdb/pebble"
)
const (
setPrefixStr = "set"
)
// Set defines a set struct
type Set struct {
db *pebble.DB
closeUnderlyingDB bool
lock sync.Mutex
prefix []byte
}
// OpenSet opens a set from the db path or creates a set if it doesn't exist.
// The keys will be prefixed with "set-" by default
func OpenSet(dataDir string) (*Set, error) {
db, err := pebble.Open(dataDir, &pebble.Options{})
if err != nil {
return nil, err
}
return NewSet(db, []byte(setPrefixStr), true)
}
// NewSet creates a set from a db. The keys will be prefixed with prefix
// and at close the db will be closed as per closeUnderlyingDB
func NewSet(db *pebble.DB, prefix []byte, closeUnderlyingDB bool) (*Set, error) {
set := &Set{
db: db,
closeUnderlyingDB: closeUnderlyingDB,
}
set.prefix = make([]byte, len(prefix))
copy(set.prefix, prefix)
return set, nil
}
// Add adds a member string to a key set, returns true if the member was not already present
func (set *Set) Add(value []byte) (bool, error) {
set.lock.Lock()
defer set.lock.Unlock()
setKey := withPrefix(set.prefix, value)
_, closer, err := set.db.Get(setKey)
if err == pebble.ErrNotFound {
return true, set.db.Set(setKey, []byte(""), pebble.Sync)
}
if err == nil {
closer.Close()
}
return false, err
}
// Members returns the current members of the set
func (set *Set) Members() ([][]byte, error) {
set.lock.Lock()
defer set.lock.Unlock()
var members [][]byte
prefix := withPrefix(set.prefix, []byte{})
iter := set.db.NewIter(&pebble.IterOptions{}) // FIXME:
for iter.Next() {
slice := iter.Key()[len(prefix):]
value := make([]byte, len(slice))
copy(value, slice)
members = append(members, value)
}
iter.Close()
return members, iter.Error()
}
// Has returns if the member is in the set
func (set *Set) Has(value []byte) (bool, error) {
set.lock.Lock()
defer set.lock.Unlock()
setKey := withPrefix(set.prefix, value)
_, closer, err := set.db.Get(setKey)
if err == nil {
closer.Close()
}
if err == pebble.ErrNotFound {
return false, nil
}
return err == nil, err
}
// Remove removes a member from the set, returns true if the member was present
func (set *Set) Remove(value []byte) (bool, error) {
set.lock.Lock()
defer set.lock.Unlock()
setKey := withPrefix(set.prefix, value)
_, closer, err := set.db.Get(setKey)
if err == nil {
closer.Close()
return true, set.db.Delete(setKey, pebble.Sync)
}
if err == pebble.ErrNotFound {
err = nil
}
return false, err
}
// Close closes the set (and the underlying db if set to closeUnderlyingDB)
func (set *Set) Close() error {
set.lock.Lock()
defer set.lock.Unlock()
if !set.closeUnderlyingDB {
set.db = nil
return nil
}
err := set.db.Close()
set.db = nil
return err
}