diff --git a/prefix_queue.go b/prefix_queue.go index 9d609ca..0fe04e7 100644 --- a/prefix_queue.go +++ b/prefix_queue.go @@ -10,6 +10,7 @@ import ( "github.com/syndtr/goleveldb/leveldb" "github.com/syndtr/goleveldb/leveldb/errors" + "github.com/syndtr/goleveldb/leveldb/opt" ) // prefixDelimiter defines the delimiter used to separate a prefix from an @@ -71,7 +72,7 @@ func OpenPrefixQueue(dataDir string) (*PrefixQueue, error) { } // Enqueue adds an item to the queue. -func (pq *PrefixQueue) Enqueue(prefix, value []byte) (*Item, error) { +func (pq *PrefixQueue) Enqueue(prefix, value []byte, opts ...*opt.WriteOptions) (*Item, error) { pq.Lock() defer pq.Unlock() @@ -94,7 +95,7 @@ func (pq *PrefixQueue) Enqueue(prefix, value []byte) (*Item, error) { } // Add it to the queue. - if err := pq.db.Put(item.Key, item.Value, nil); err != nil { + if err := pq.db.Put(item.Key, item.Value, getOpts(opts)); err != nil { return nil, err } @@ -248,7 +249,7 @@ func (pq *PrefixQueue) PeekByIDString(prefix string, id uint64) (*Item, error) { } // Update updates an item in the given queue without changing its position. -func (pq *PrefixQueue) Update(prefix []byte, id uint64, newValue []byte) (*Item, error) { +func (pq *PrefixQueue) Update(prefix []byte, id uint64, newValue []byte, opts ...*opt.WriteOptions) (*Item, error) { pq.Lock() defer pq.Unlock() @@ -276,7 +277,7 @@ func (pq *PrefixQueue) Update(prefix []byte, id uint64, newValue []byte) (*Item, } // Update this item in the queue. - if err := pq.db.Put(item.Key, item.Value, nil); err != nil { + if err := pq.db.Put(item.Key, item.Value, getOpts(opts)); err != nil { return nil, err } @@ -392,7 +393,7 @@ func (pq *PrefixQueue) getOrCreateQueue(prefix []byte) (*queue, error) { } // savePrefixQueue saves the given queue for the given prefix. -func (pq *PrefixQueue) saveQueue(prefix []byte, q *queue) error { +func (pq *PrefixQueue) saveQueue(prefix []byte, q *queue, opts ...*opt.WriteOptions) error { // Encode the queue using gob. var buffer bytes.Buffer enc := gob.NewEncoder(&buffer) @@ -401,14 +402,14 @@ func (pq *PrefixQueue) saveQueue(prefix []byte, q *queue) error { } // Save it to the database. - return pq.db.Put(generateKeyPrefixData(prefix), buffer.Bytes(), nil) + return pq.db.Put(generateKeyPrefixData(prefix), buffer.Bytes(), getOpts(opts)) } // save saves the main prefix queue data. -func (pq *PrefixQueue) save() error { +func (pq *PrefixQueue) save(opts ...*opt.WriteOptions) error { val := make([]byte, 8) binary.BigEndian.PutUint64(val, pq.size) - return pq.db.Put(pq.getDataKey(), val, nil) + return pq.db.Put(pq.getDataKey(), val, getOpts(opts)) } // getDataKey generates the main prefix queue data key. diff --git a/priority_queue.go b/priority_queue.go index 937621f..2344ed5 100644 --- a/priority_queue.go +++ b/priority_queue.go @@ -8,6 +8,7 @@ import ( "sync" "github.com/syndtr/goleveldb/leveldb" + "github.com/syndtr/goleveldb/leveldb/opt" "github.com/syndtr/goleveldb/leveldb/util" ) @@ -82,7 +83,7 @@ func OpenPriorityQueue(dataDir string, order order) (*PriorityQueue, error) { } // Enqueue adds an item to the priority queue. -func (pq *PriorityQueue) Enqueue(priority uint8, value []byte) (*PriorityItem, error) { +func (pq *PriorityQueue) Enqueue(priority uint8, value []byte, opts ...*opt.WriteOptions) (*PriorityItem, error) { pq.Lock() defer pq.Unlock() @@ -103,7 +104,7 @@ func (pq *PriorityQueue) Enqueue(priority uint8, value []byte) (*PriorityItem, e } // Add it to the priority queue. - if err := pq.db.Put(item.Key, item.Value, nil); err != nil { + if err := pq.db.Put(item.Key, item.Value, getOpts(opts)); err != nil { return nil, err } @@ -264,7 +265,7 @@ func (pq *PriorityQueue) PeekByPriorityID(priority uint8, id uint64) (*PriorityI // Update updates an item in the priority queue without changing its // position. -func (pq *PriorityQueue) Update(priority uint8, id uint64, newValue []byte) (*PriorityItem, error) { +func (pq *PriorityQueue) Update(priority uint8, id uint64, newValue []byte, opts ...*opt.WriteOptions) (*PriorityItem, error) { pq.Lock() defer pq.Unlock() @@ -287,7 +288,7 @@ func (pq *PriorityQueue) Update(priority uint8, id uint64, newValue []byte) (*Pr } // Update this item in the queue. - if err := pq.db.Put(item.Key, item.Value, nil); err != nil { + if err := pq.db.Put(item.Key, item.Value, getOpts(opts)); err != nil { return nil, err } diff --git a/queue.go b/queue.go index a170db3..5198cd1 100644 --- a/queue.go +++ b/queue.go @@ -8,6 +8,7 @@ import ( "sync" "github.com/syndtr/goleveldb/leveldb" + "github.com/syndtr/goleveldb/leveldb/opt" ) // Queue is a standard FIFO (first in, first out) queue. @@ -55,7 +56,7 @@ func OpenQueue(dataDir string) (*Queue, error) { } // Enqueue adds an item to the queue. -func (q *Queue) Enqueue(value []byte) (*Item, error) { +func (q *Queue) Enqueue(value []byte, opts ...*opt.WriteOptions) (*Item, error) { q.Lock() defer q.Unlock() @@ -72,7 +73,7 @@ func (q *Queue) Enqueue(value []byte) (*Item, error) { } // Add it to the queue. - if err := q.db.Put(item.Key, item.Value, nil); err != nil { + if err := q.db.Put(item.Key, item.Value, getOpts(opts)); err != nil { return nil, err } @@ -188,7 +189,7 @@ func (q *Queue) PeekByID(id uint64) (*Item, error) { } // Update updates an item in the queue without changing its position. -func (q *Queue) Update(id uint64, newValue []byte) (*Item, error) { +func (q *Queue) Update(id uint64, newValue []byte, opts ...*opt.WriteOptions) (*Item, error) { q.Lock() defer q.Unlock() @@ -210,7 +211,7 @@ func (q *Queue) Update(id uint64, newValue []byte) (*Item, error) { } // Update this item in the queue. - if err := q.db.Put(item.Key, item.Value, nil); err != nil { + if err := q.db.Put(item.Key, item.Value, getOpts(opts)); err != nil { return nil, err } diff --git a/stack.go b/stack.go index 9b4090f..e950870 100644 --- a/stack.go +++ b/stack.go @@ -8,6 +8,7 @@ import ( "sync" "github.com/syndtr/goleveldb/leveldb" + "github.com/syndtr/goleveldb/leveldb/opt" ) // Stack is a standard LIFO (last in, first out) stack. @@ -55,7 +56,7 @@ func OpenStack(dataDir string) (*Stack, error) { } // Push adds an item to the stack. -func (s *Stack) Push(value []byte) (*Item, error) { +func (s *Stack) Push(value []byte, opts ...*opt.WriteOptions) (*Item, error) { s.Lock() defer s.Unlock() @@ -72,7 +73,7 @@ func (s *Stack) Push(value []byte) (*Item, error) { } // Add it to the stack. - if err := s.db.Put(item.Key, item.Value, nil); err != nil { + if err := s.db.Put(item.Key, item.Value, getOpts(opts)); err != nil { return nil, err } @@ -188,7 +189,7 @@ func (s *Stack) PeekByID(id uint64) (*Item, error) { } // Update updates an item in the stack without changing its position. -func (s *Stack) Update(id uint64, newValue []byte) (*Item, error) { +func (s *Stack) Update(id uint64, newValue []byte, opts ...*opt.WriteOptions) (*Item, error) { s.Lock() defer s.Unlock() @@ -210,7 +211,7 @@ func (s *Stack) Update(id uint64, newValue []byte) (*Item, error) { } // Update this item in the stack. - if err := s.db.Put(item.Key, item.Value, nil); err != nil { + if err := s.db.Put(item.Key, item.Value, getOpts(opts)); err != nil { return nil, err }