Merge 317fe3a into ccb01ed

.travis.yml

@@ -1,4 +1,6 @@
 language: go
+go:
+  - "1.13.5"
 
 install:
   - go get ./...

README.md

@@ -234,6 +234,23 @@ where := bolthold.Where("Id").In(bolthold.Slice(t)...)
 
 ```
 
+### ForEach
+
+When working with large datasets, you may not want to have to store the entire dataset in memory.  It's be much more
+efficient to work with a single record at a time rather than grab all the records and loop through them, which is
+what cursors are used for in databases.  In BoltHold you can accomplish the same thing by calling ForEach:
+
+```Go
+err := store.ForEach(boltholdWhere("Id").Gt(4), func(record *Item) error {
+	// do stuff with record
+
+	// if you return an error, then the query will stop iterating through records
+
+	return nil
+})
+
+```
+
 ### Aggregate Queries
 
 Aggregate queries are queries that group results by a field.  For example, lets say you had a collection of employees:

foreach_test.go

@@ -0,0 +1,50 @@
+// Copyright 2019 Tim Shannon. All rights reserved.
+// Use of this source code is governed by the MIT license
+// that can be found in the LICENSE file.
+
+package bolthold_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/timshannon/bolthold"
+)
+
+func TestForEach(t *testing.T) {
+	testWrap(t, func(store *bolthold.Store, t *testing.T) {
+		insertTestData(t, store)
+		for _, tst := range testResults {
+			t.Run(tst.name, func(t *testing.T) {
+				count := 0
+				err := store.ForEach(tst.query, func(record *ItemTest) error {
+					count++
+
+					found := false
+					for i := range tst.result {
+						if record.equal(&testData[tst.result[i]]) {
+							found = true
+							break
+						}
+					}
+
+					if !found {
+						if testing.Verbose() {
+							return fmt.Errorf("%v was not found in the result set! Full results: %v",
+								record, tst.result)
+						}
+						return fmt.Errorf("%v was not found in the result set!", record)
+					}
+
+					return nil
+				})
+				if count != len(tst.result) {
+					t.Fatalf("ForEach count is %d wanted %d.", count, len(tst.result))
+				}
+				if err != nil {
+					t.Fatalf("Error during ForEach iteration: %s", err)
+				}
+			})
+		}
+	})
+}

get.go

@@ -85,6 +85,7 @@ func (s *Store) TxFindOne(tx *bolt.Tx, result interface{}, query *Query) error {
 	return s.findOneQuery(tx, result, query)
 }
 
+// FindOneInBucket allows you to pass in your own bucket to retrieve a single record from the bolthold
 func (s *Store) FindOneInBucket(parent *bolt.Bucket, result interface{}, query *Query) error {
 	return s.findOneQuery(parent, result, query)
 }
@@ -109,3 +110,23 @@ func (s *Store) TxCount(tx *bolt.Tx, dataType interface{}, query *Query) (int, e
 func (s *Store) CountInBucket(parent *bolt.Bucket, dataType interface{}, query *Query) (int, error) {
 	return s.countQuery(parent, dataType, query)
 }
+
+// ForEach runs the function fn against every record that matches the query
+// Useful for when working with large sets of data that you don't want to hold the entire result
+// set in memory, similar to database cursors
+// Return an error from fn, will stop the cursor from iterating
+func (s *Store) ForEach(query *Query, fn interface{}) error {
+	return s.Bolt().View(func(tx *bolt.Tx) error {
+		return s.TxForEach(tx, query, fn)
+	})
+}
+
+// TxForEach is the same as ForEach but you get to specify your transaction
+func (s *Store) TxForEach(tx *bolt.Tx, query *Query, fn interface{}) error {
+	return s.forEach(tx, query, fn)
+}
+
+// ForEachInBucket is the same as ForEach but you get to specify your parent bucket
+func (s *Store) ForEachInBucket(parent *bolt.Bucket, query *Query, fn interface{}) error {
+	return s.forEach(parent, query, fn)
+}

query.go

@@ -550,3 +550,31 @@ func (s *Store) findOneQuery(source bucketSource, result interface{}, query *Que
 
 	return nil
 }
+
+func (s *Store) forEach(source bucketSource, query *Query, fn interface{}) error {
+	if query == nil {
+		query = &Query{}
+	}
+
+	fnVal := reflect.ValueOf(fn)
+	argType := reflect.TypeOf(fn).In(0)
+
+	if argType.Kind() == reflect.Ptr {
+		argType = argType.Elem()
+	}
+
+	dataType := reflect.New(argType).Interface()
+
+	return s.runQuery(source, dataType, query, nil, query.skip, func(r *record) error {
+		out := fnVal.Call([]reflect.Value{r.value})
+		if len(out) != 1 {
+			return fmt.Errorf("foreach function does not return an error")
+		}
+
+		if out[0].IsNil() {
+			return nil
+		}
+
+		return out[0].Interface().(error)
+	})
+}