LoadStream function

mongox/common/loadstream.go

@@ -0,0 +1,67 @@
+package common
+
+import (
+	"reflect"
+
+	"github.com/mainnika/mongox-go-driver/mongox"
+	"github.com/mainnika/mongox-go-driver/mongox/errors"
+	"github.com/mainnika/mongox-go-driver/mongox/query"
+	"github.com/mongodb/mongo-go-driver/mongo/options"
+)
+
+// LoadStream function loads documents one by one into a target channel
+func LoadStream(db *mongox.Database, target interface{}, composed *query.Query) error {
+
+	targetV := reflect.ValueOf(target)
+	targetT := targetV.Type()
+
+	targetK := targetV.Kind()
+	if targetK != reflect.Chan {
+		panic(errors.InternalErrorf("target is not a chan"))
+	}
+	if targetT.Elem().Kind() != reflect.Ptr {
+		panic(errors.InternalErrorf("chan element should be a document ptr"))
+	}
+
+	dummy := reflect.Zero(targetT.Elem())
+	collection := db.GetCollectionOf(dummy.Interface())
+	opts := &options.FindOptions{}
+
+	if composed.Sorter() != nil {
+		opts.Sort = composed.Sorter().Sort()
+	}
+	if composed.Limiter() != nil {
+		limit := int64(composed.Limiter().Limit())
+		opts.Limit = &limit
+	}
+
+	result, err := collection.Find(db.Context(), composed.M(), opts)
+	if err != nil {
+		return errors.InternalErrorf("can't create find result: %s", err)
+	}
+
+	go func() {
+		defer result.Close(db.Context())
+
+		for {
+			elem, ok := targetV.Recv()
+			if !ok {
+				break
+			}
+
+			if result.Next(db.Context()) != true {
+				targetV.Send(dummy)
+				break
+			}
+
+			if result.Decode(elem.Interface()) != nil {
+				targetV.Send(dummy)
+				break
+			}
+
+			targetV.Send(elem)
+		}
+	}()
+
+	return nil
+}

mongox/query/limit.go

@@ -0,0 +1,17 @@
+package query
+
+// Limiter is a filter to limit the result
+type Limiter interface {
+	Limit() int
+}
+
+// Limit is a simple implementation of the Limiter filter
+type Limit int
+
+var _ Limiter = Limit(0)
+
+// Limit returns a limit
+func (l Limit) Limit() int {
+
+	return int(l)
+}