feat: contentrouter implements routing.PeerRouting

Author: hacdias

routing/http/contentrouter/contentrouter.go

@@ -19,13 +19,15 @@ var logger = logging.Logger("routing/http/contentrouter")
 
 type Client interface {
 	GetProviders(ctx context.Context, key cid.Cid) (iter.ResultIter[types.Record], error)
+	GetPeers(ctx context.Context, pid peer.ID) (peers iter.ResultIter[types.Record], err error)
 }
 
 type contentRouter struct {
 	client Client
 }
 
 var _ routing.ContentRouting = (*contentRouter)(nil)
+var _ routing.PeerRouting = (*contentRouter)(nil)
 var _ routinghelpers.ProvideManyRouter = (*contentRouter)(nil)
 
 type option func(c *contentRouter)
@@ -100,3 +102,43 @@ func (c *contentRouter) FindProvidersAsync(ctx context.Context, key cid.Cid, num
 	go readProviderResponses(resultsIter, ch)
 	return ch
 }
+
+func (c *contentRouter) FindPeer(ctx context.Context, pid peer.ID) (peer.AddrInfo, error) {
+	iter, err := c.client.GetPeers(ctx, pid)
+	if err != nil {
+		return peer.AddrInfo{}, err
+	}
+	defer iter.Close()
+
+	for iter.Next() {
+		res := iter.Val()
+		if res.Err != nil {
+			logger.Warnw("error iterating provider responses: %s", res.Err)
+			continue
+		}
+		v := res.Val
+		if v.GetSchema() == types.SchemaPeer {
+			result, ok := v.(*types.PeerRecord)
+			if !ok {
+				logger.Errorw(
+					"problem casting find providers result",
+					"Schema", v.GetSchema(),
+					"Type", reflect.TypeOf(v).String(),
+				)
+				continue
+			}
+
+			var addrs []multiaddr.Multiaddr
+			for _, a := range result.Addrs {
+				addrs = append(addrs, a.Multiaddr)
+			}
+
+			return peer.AddrInfo{
+				ID:    *result.ID,
+				Addrs: addrs,
+			}, nil
+		}
+	}
+
+	return peer.AddrInfo{}, err
+}

routing/http/contentrouter/contentrouter_test.go

@@ -20,6 +20,10 @@ func (m *mockClient) GetProviders(ctx context.Context, key cid.Cid) (iter.Result
 	args := m.Called(ctx, key)
 	return args.Get(0).(iter.ResultIter[types.Record]), args.Error(1)
 }
+func (m *mockClient) GetPeers(ctx context.Context, pid peer.ID) (iter.ResultIter[types.Record], error) {
+	args := m.Called(ctx, pid)
+	return args.Get(0).(iter.ResultIter[types.Record]), args.Error(1)
+}
 func (m *mockClient) Ready(ctx context.Context) (bool, error) {
 	args := m.Called(ctx)
 	return args.Bool(0), args.Error(1)
@@ -38,7 +42,7 @@ func makeCID() cid.Cid {
 	return c
 }
 
-func TestGetProvidersAsync(t *testing.T) {
+func TestFindProvidersAsync(t *testing.T) {
 	key := makeCID()
 	ctx := context.Background()
 	client := &mockClient{}
@@ -79,3 +83,28 @@ func TestGetProvidersAsync(t *testing.T) {
 
 	require.Equal(t, expected, actualAIs)
 }
+
+func TestFindPeer(t *testing.T) {
+	ctx := context.Background()
+	client := &mockClient{}
+	crc := NewContentRoutingClient(client)
+
+	p1 := peer.ID("peer1")
+	ais := []types.Record{
+		&types.UnknownRecord{
+			Schema: "Unknown",
+		},
+		&types.PeerRecord{
+			Schema:    types.SchemaPeer,
+			ID:        &p1,
+			Protocols: []string{"transport-bitswap"},
+		},
+	}
+	aisIter := iter.ToResultIter[types.Record](iter.FromSlice(ais))
+
+	client.On("GetPeers", ctx, p1).Return(aisIter, nil)
+
+	peer, err := crc.FindPeer(ctx, p1)
+	require.NoError(t, err)
+	require.Equal(t, peer.ID, p1)
+}