query: avoid ResultsWithChan

It's impossible to cancel this correctly. This patch also tries to avoid
goroutines whenever possible.

query/query.go

@@ -204,12 +204,12 @@ func NewResultBuilder(q Query) *ResultBuilder {
 }
 
 // ResultsWithChan returns a Results object from a channel
-// of Result entries. Respects its own Close()
+// of Result entries.
+//
+// DEPRECATED: This iterator is impossible to cancel correctly. Canceling it
+// will leave anything trying to write to the result channel hanging.
 func ResultsWithChan(q Query, res <-chan Result) Results {
-	b := NewResultBuilder(q)
-
-	// go consume all the entries and add them to the results.
-	b.Process.Go(func(worker goprocess.Process) {
+	return ResultsWithProcess(q, func(worker goprocess.Process, out chan<- Result) {
 		for {
 			select {
 			case <-worker.Closing(): // client told us to close early
@@ -220,13 +220,24 @@ func ResultsWithChan(q Query, res <-chan Result) Results {
 				}
 
 				select {
-				case b.Output <- e:
+				case out <- e:
 				case <-worker.Closing(): // client told us to close early
 					return
 				}
 			}
 		}
 	})
+}
+
+// ResultsWithProcess returns a Results object with the results generated by the
+// passed subprocess.
+func ResultsWithProcess(q Query, proc func(goprocess.Process, chan<- Result)) Results {
+	b := NewResultBuilder(q)
+
+	// go consume all the entries and add them to the results.
+	b.Process.Go(func(worker goprocess.Process) {
+		proc(worker, b.Output)
+	})
 
 	go b.Process.CloseAfterChildren()
 	return b.Results()

query/query_impl.go

 package query
 
-import "sort"
+import (
+	"sort"
 
-func DerivedResults(qr Results, ch <-chan Result) Results {
-	return &results{
-		query: qr.Query(),
-		proc:  qr.Process(),
-		res:   ch,
-	}
-}
+	goprocess "github.com/jbenet/goprocess"
+)
 
 // NaiveFilter applies a filter to the results.
 func NaiveFilter(qr Results, filter Filter) Results {
-	ch := make(chan Result)
-	go func() {
-		defer close(ch)
-		defer qr.Close()
-
-		for e := range qr.Next() {
+	return ResultsFromIterator(qr.Query(), Iterator{
+		Next: func() (Result, bool) {
+			for {
+				e, ok := qr.NextSync()
+				if !ok {
+					return Result{}, false
+				}
 				if e.Error != nil || filter.Filter(e.Entry) {
-				ch <- e
+					return e, true
 				}
 			}
-	}()
-
-	return ResultsWithChan(qr.Query(), ch)
+		},
+		Close: func() error {
+			return qr.Close()
+		},
+	})
 }
 
 // NaiveLimit truncates the results to a given int limit
 func NaiveLimit(qr Results, limit int) Results {
-	ch := make(chan Result)
-	go func() {
-		defer close(ch)
-		defer qr.Close()
-
-		l := 0
-		for e := range qr.Next() {
-			if e.Error != nil {
-				ch <- e
-				continue
-			}
-			ch <- e
-			l++
-			if limit > 0 && l >= limit {
-				break
-			}
+	if limit == 0 {
+		// 0 means no limit
+		return qr
 	}
-	}()
-
-	return ResultsWithChan(qr.Query(), ch)
+	closed := false
+	return ResultsFromIterator(qr.Query(), Iterator{
+		Next: func() (Result, bool) {
+			if limit == 0 {
+				if !closed {
+					closed = true
+					err := qr.Close()
+					if err != nil {
+						return Result{Error: err}, true
+					}
+				}
+				return Result{}, false
+			}
+			limit--
+			return qr.NextSync()
+		},
+		Close: func() error {
+			if closed {
+				return nil
+			}
+			closed = true
+			return qr.Close()
+		},
+	})
 }
 
 // NaiveOffset skips a given number of results
 func NaiveOffset(qr Results, offset int) Results {
-	ch := make(chan Result)
-	go func() {
-		defer close(ch)
-		defer qr.Close()
-
-		sent := 0
-		for e := range qr.Next() {
-			if e.Error != nil {
-				ch <- e
-			}
-
-			if sent < offset {
-				sent++
-				continue
-			}
-			ch <- e
-		}
-	}()
-
-	return ResultsWithChan(qr.Query(), ch)
+	return ResultsFromIterator(qr.Query(), Iterator{
+		Next: func() (Result, bool) {
+			for ; offset > 0; offset-- {
+				res, ok := qr.NextSync()
+				if !ok || res.Error != nil {
+					return res, ok
+				}
+			}
+			return qr.NextSync()
+		},
+		Close: func() error {
+			return qr.Close()
+		},
+	})
 }
 
 // NaiveOrder reorders results according to given orders.
@@ -83,29 +84,37 @@ func NaiveOrder(qr Results, orders ...Order) Results {
 		return qr
 	}
 
-	ch := make(chan Result)
-	var entries []Entry
-	go func() {
-		defer close(ch)
+	return ResultsWithProcess(qr.Query(), func(worker goprocess.Process, out chan<- Result) {
 		defer qr.Close()
-
-		for e := range qr.Next() {
+		var entries []Entry
+	collect:
+		for {
+			select {
+			case <-worker.Closing():
+				return
+			case e, ok := <-qr.Next():
+				if !ok {
+					break collect
+				}
 				if e.Error != nil {
-				ch <- e
+					out <- e
+					continue
 				}
-
 				entries = append(entries, e.Entry)
 			}
+		}
+
 		sort.Slice(entries, func(i int, j int) bool {
 			return Less(orders, entries[i], entries[j])
 		})
-
 		for _, e := range entries {
-			ch <- Result{Entry: e}
+			select {
+			case <-worker.Closing():
+				return
+			case out <- Result{Entry: e}:
 			}
-	}()
-
-	return DerivedResults(qr, ch)
+		}
+	})
 }
 
 func NaiveQueryApply(q Query, qr Results) Results {