go-libp2p-kad-dht: implement GetValuesAsync

routing.go

@@ -143,33 +143,85 @@ func (dht *IpfsDHT) GetValue(ctx context.Context, key string) ([]byte, error) {
 }
 
 func (dht *IpfsDHT) GetValues(ctx context.Context, key string, nvals int) ([]routing.RecvdVal, error) {
+	var resChan chan *routing.RecvdVal = make(chan *routing.RecvdVal, 0)
+	go dht.getValuesAsyncRoutine(ctx, key, nvals, resChan, nil)
+
 	var vals []routing.RecvdVal
+loop:
+	for {
+		select {
+		case res := <-resChan:
+			if res == nil {
+				break loop
+			} else if res.Error != nil {
+				return nil, res.Error
+			} else {
+				vals = append(vals, *res)
+			}
+		}
+	}
+
+	return vals, nil
+}
+
+func (dht *IpfsDHT) GetValuesAsync(ctx context.Context, key string, nvals int) <-chan *routing.RecvdVal {
+	var resChan chan *routing.RecvdVal = make(chan *routing.RecvdVal, 0)
+	var errChan chan error = make(chan error, 0)
+	go dht.getValuesAsyncRoutine(ctx, key, nvals, resChan, errChan)
+	go func() {
+		for err := range errChan {
+			log.Debugf("Query error: %s", err)
+			notif.PublishQueryEvent(ctx, &notif.QueryEvent{
+				Type:  notif.QueryError,
+				Extra: err.Error(),
+			})
+		}
+	}()
+	return resChan
+}
+
+func (dht *IpfsDHT) getValuesAsyncRoutine(ctx context.Context, key string, nvals int, resChan chan<- *routing.RecvdVal, errChan chan<- error) {
 	var valslock sync.Mutex
+	var sentRes int
+
+	defer close(resChan)
+	if errChan != nil {
+		defer close(errChan)
+	}
 
 	// If we have it local, dont bother doing an RPC!
 	lrec, err := dht.getLocal(key)
 	if err == nil {
 		// TODO: this is tricky, we dont always want to trust our own value
 		// what if the authoritative source updated it?
 		log.Debug("have it locally")
-		vals = append(vals, routing.RecvdVal{
+		sentRes = sentRes + 1
+		resChan <- &routing.RecvdVal{
 			Val:  lrec.GetValue(),
 			From: dht.self,
-		})
+		}
 
-		if nvals <= 1 {
-			return vals, nil
+		if nvals == 0 || nvals == 1 {
+			return
 		}
 	} else if nvals == 0 {
-		return nil, err
+		if errChan != nil {
+			errChan <- err
+		}
+		resChan <- &routing.RecvdVal{Error: err}
+		return
 	}
 
 	// get closest peers in the routing table
 	rtp := dht.routingTable.NearestPeers(kb.ConvertKey(key), KValue)
 	log.Debugf("peers in rt: %d %s", len(rtp), rtp)
 	if len(rtp) == 0 {
 		log.Warning("No peers from routing table!")
-		return nil, kb.ErrLookupFailure
+		if errChan != nil {
+			errChan <- err
+		}
+		resChan <- &routing.RecvdVal{Error: kb.ErrLookupFailure}
+		return
 	}
 
 	// setup the Query
@@ -205,11 +257,12 @@ func (dht *IpfsDHT) GetValues(ctx context.Context, key string, nvals int) ([]rou
 				Val:  rec.GetValue(),
 				From: p,
 			}
+			resChan <- &rv
 			valslock.Lock()
-			vals = append(vals, rv)
+			sentRes += 1
 
 			// If weve collected enough records, we're done
-			if len(vals) >= nvals {
+			if sentRes >= nvals || nvals >= 0 {
 				res.success = true
 			}
 			valslock.Unlock()
@@ -226,14 +279,15 @@ func (dht *IpfsDHT) GetValues(ctx context.Context, key string, nvals int) ([]rou
 
 	// run it!
 	_, err = query.Run(ctx, rtp)
-	if len(vals) == 0 {
+	if sentRes == 0 {
 		if err != nil {
-			return nil, err
+			if errChan != nil {
+				errChan <- err
+			}
+			resChan <- &routing.RecvdVal{Error: err}
 		}
 	}
 
-	return vals, nil
-
 }
 
 // Value provider layer of indirection.