dht: switch GetClosestPeers to return a slice of peers instead of a c…

…hannel

dht_test.go

@@ -1483,7 +1483,7 @@ func testFindPeerQuery(t *testing.T,
 	require.NoError(t, err)
 
 	var outpeers []peer.ID
-	for p := range out {
+	for _, p := range out {
 		outpeers = append(outpeers, p)
 	}
 
@@ -1521,7 +1521,7 @@ func TestFindClosestPeers(t *testing.T) {
 	}
 
 	var out []peer.ID
-	for p := range peers {
+	for _, p := range peers {
 		out = append(out, p)
 	}
 
@@ -2112,18 +2112,16 @@ func TestPreconnectedNodes(t *testing.T) {
 	}
 
 	// See if it works
-	peerCh, err := d2.GetClosestPeers(ctx, "testkey")
+	peers, err := d2.GetClosestPeers(ctx, "testkey")
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	select {
-	case p := <-peerCh:
-		if p == h1.ID() {
-			break
-		}
+	if len(peers) != 1 {
+		t.Fatal("why is there more than one peer?")
+	}
+
+	if peers[0] != h1.ID() {
 		t.Fatal("could not find peer")
-	case <-ctx.Done():
-		t.Fatal("test hung")
 	}
 }

lookup.go

@@ -16,7 +16,7 @@ import (
 //
 // If the context is canceled, this function will return the context error along
 // with the closest K peers it has found so far.
-func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key string) (<-chan peer.ID, error) {
+func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key string) ([]peer.ID, error) {
 	if key == "" {
 		return nil, fmt.Errorf("can't lookup empty key")
 	}
@@ -51,17 +51,10 @@ func (dht *IpfsDHT) GetClosestPeers(ctx context.Context, key string) (<-chan pee
 		return nil, err
 	}
 
-	out := make(chan peer.ID, dht.bucketSize)
-	defer close(out)
-
-	for _, p := range lookupRes.peers {
-		out <- p
-	}
-
 	if ctx.Err() == nil && lookupRes.completed {
 		// refresh the cpl for this key as the query was successful
 		dht.routingTable.ResetCplRefreshedAtForID(kb.ConvertKey(key), time.Now())
 	}
 
-	return out, ctx.Err()
+	return lookupRes.peers, ctx.Err()
 }

routing.go

@@ -65,13 +65,13 @@ func (dht *IpfsDHT) PutValue(ctx context.Context, key string, value []byte, opts
 		return err
 	}
 
-	pchan, err := dht.GetClosestPeers(ctx, key)
+	peers, err := dht.GetClosestPeers(ctx, key)
 	if err != nil {
 		return err
 	}
 
 	wg := sync.WaitGroup{}
-	for p := range pchan {
+	for _, p := range peers {
 		wg.Add(1)
 		go func(p peer.ID) {
 			ctx, cancel := context.WithCancel(ctx)
@@ -446,7 +446,7 @@ func (dht *IpfsDHT) Provide(ctx context.Context, key cid.Cid, brdcst bool) (err
 	}
 
 	wg := sync.WaitGroup{}
-	for p := range peers {
+	for _, p := range peers {
 		wg.Add(1)
 		go func(p peer.ID) {
 			defer wg.Done()