Implementation of delegated routing based on the Edelweiss compiler and Reframe spec (#11)

Author: petar

client/client.go

@@ -0,0 +1,56 @@
+package client
+
+import (
+	"context"
+
+	"github.com/ipfs/go-cid"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"github.com/libp2p/go-libp2p-core/routing"
+)
+
+type ContentRoutingClient struct {
+	client DelegatedRoutingClient
+}
+
+var _ routing.ContentRouting = (*ContentRoutingClient)(nil)
+
+func NewContentRoutingClient(c DelegatedRoutingClient) *ContentRoutingClient {
+	return &ContentRoutingClient{client: c}
+}
+
+func (c *ContentRoutingClient) Provide(context.Context, cid.Cid, bool) error {
+	return routing.ErrNotSupported
+}
+
+func (c *ContentRoutingClient) FindProvidersAsync(ctx context.Context, key cid.Cid, numResults int) <-chan peer.AddrInfo {
+	addrInfoCh := make(chan peer.AddrInfo)
+	resultCh, err := c.client.FindProvidersAsync(ctx, key)
+	if err != nil {
+		close(addrInfoCh)
+		return addrInfoCh
+	}
+	go func() {
+		numProcessed := 0
+		closed := false
+		for asyncResult := range resultCh {
+			if asyncResult.Err != nil {
+				logger.Infof("find providers async emitted a transient error (%v)", asyncResult.Err)
+			} else {
+				for _, peerAddr := range asyncResult.AddrInfo {
+					if numResults <= 0 || numProcessed < numResults {
+						addrInfoCh <- peerAddr
+					}
+					numProcessed++
+					if numProcessed == numResults {
+						close(addrInfoCh)
+						closed = true
+					}
+				}
+			}
+		}
+		if !closed {
+			close(addrInfoCh)
+		}
+	}()
+	return addrInfoCh
+}

client/client_test.go

@@ -0,0 +1,101 @@
+package client
+
+import (
+	"context"
+	"testing"
+
+	"github.com/ipfs/go-cid"
+	"github.com/libp2p/go-libp2p-core/peer"
+)
+
+type TestDelegatedRoutingClient struct {
+	NumResults int
+}
+
+func (t TestDelegatedRoutingClient) FindProviders(ctx context.Context, key cid.Cid) ([]peer.AddrInfo, error) {
+	panic("not supported")
+}
+
+func (t TestDelegatedRoutingClient) FindProvidersAsync(ctx context.Context, key cid.Cid) (<-chan FindProvidersAsyncResult, error) {
+	ch := make(chan FindProvidersAsyncResult)
+	go func() {
+		defer close(ch)
+		for i := 0; i < t.NumResults; i++ {
+			ch <- FindProvidersAsyncResult{
+				AddrInfo: []peer.AddrInfo{{}},
+			}
+		}
+	}()
+	return ch, nil
+}
+
+func (t TestDelegatedRoutingClient) GetIPNS(ctx context.Context, id []byte) ([]byte, error) {
+	panic("not supported")
+}
+
+func (t TestDelegatedRoutingClient) GetIPNSAsync(ctx context.Context, id []byte) (<-chan GetIPNSAsyncResult, error) {
+	panic("not supported")
+}
+
+func (t TestDelegatedRoutingClient) PutIPNS(ctx context.Context, id []byte, record []byte) error {
+	panic("not supported")
+}
+
+func (t TestDelegatedRoutingClient) PutIPNSAsync(ctx context.Context, id []byte, record []byte) (<-chan PutIPNSAsyncResult, error) {
+	panic("not supported")
+}
+
+// TestContentRoutingFindProvidersUnlimitedResults is testing that ContentRoutingClient.FindProvidersAsync
+// correctly wraps DelegatedRoutingClient.FindProvidersAsync in the regime when the former allows for unlimited results.
+// This is a test of async semantics only. This is why values are not checked for validity.
+// Non-test implementations of DelegatedRoutingClient are responsible for returning valid values.
+func TestContentRoutingFindProvidersUnlimitedResults(t *testing.T) {
+	providedResults := 5
+	c := NewContentRoutingClient(TestDelegatedRoutingClient{providedResults})
+	ch := c.FindProvidersAsync(context.Background(), cid.Cid{}, 0)
+	num := 0
+	for range ch {
+		num++
+	}
+	if num != providedResults {
+		t.Errorf("expecting %v results, got %v", providedResults, num)
+	}
+}
+
+// TestContentRoutingFindProvidersFewerResults is testing that ContentRoutingClient.FindProvidersAsync
+// correctly wraps DelegatedRoutingClient.FindProvidersAsync in the regime when the former allows for
+// fewer results than are available.
+// This is a test of async semantics only. This is why values are not checked for validity.
+// Non-test implementations of DelegatedRoutingClient are responsible for returning valid values.
+func TestContentRoutingFindProvidersFewerResults(t *testing.T) {
+	providedResults := 5
+	wantResults := 3
+	c := NewContentRoutingClient(TestDelegatedRoutingClient{providedResults})
+	ch := c.FindProvidersAsync(context.Background(), cid.Cid{}, wantResults)
+	num := 0
+	for range ch {
+		num++
+	}
+	if num != wantResults {
+		t.Errorf("expecting %v results, got %v", wantResults, num)
+	}
+}
+
+// TestContentRoutingFindProvidersMoreResults is testing that ContentRoutingClient.FindProvidersAsync
+// correctly wraps DelegatedRoutingClient.FindProvidersAsync in the regime when the former allows for
+// more results than are available.
+// This is a test of async semantics only. This is why values are not checked for validity.
+// Non-test implementations of DelegatedRoutingClient are responsible for returning valid values.
+func TestContentRoutingFindProvidersMoreResults(t *testing.T) {
+	providedResults := 5
+	wantResults := 7
+	c := NewContentRoutingClient(TestDelegatedRoutingClient{providedResults})
+	ch := c.FindProvidersAsync(context.Background(), cid.Cid{}, wantResults)
+	num := 0
+	for range ch {
+		num++
+	}
+	if num != providedResults {
+		t.Errorf("expecting %v results, got %v", providedResults, num)
+	}
+}