Feat/add discovery client

client.go

@@ -9,10 +9,10 @@ import (
 	pb "github.com/libp2p/go-libp2p-rendezvous/pb"
 
 	ggio "github.com/gogo/protobuf/io"
-	host "github.com/libp2p/go-libp2p-host"
-	inet "github.com/libp2p/go-libp2p-net"
-	peer "github.com/libp2p/go-libp2p-peer"
-	pstore "github.com/libp2p/go-libp2p-peerstore"
+
+	"github.com/libp2p/go-libp2p-core/host"
+	inet "github.com/libp2p/go-libp2p-core/network"
+	"github.com/libp2p/go-libp2p-core/peer"
 )
 
 var (
@@ -27,16 +27,16 @@ type RendezvousPoint interface {
 }
 
 type Registration struct {
-	Peer pstore.PeerInfo
+	Peer peer.AddrInfo
 	Ns   string
 	Ttl  int
 }
 
 type RendezvousClient interface {
 	Register(ctx context.Context, ns string, ttl int) error
 	Unregister(ctx context.Context, ns string) error
-	Discover(ctx context.Context, ns string, limit int, cookie []byte) ([]pstore.PeerInfo, []byte, error)
-	DiscoverAsync(ctx context.Context, ns string) (<-chan pstore.PeerInfo, error)
+	Discover(ctx context.Context, ns string, limit int, cookie []byte) ([]peer.AddrInfo, []byte, error)
+	DiscoverAsync(ctx context.Context, ns string) (<-chan peer.AddrInfo, error)
 }
 
 func NewRendezvousPoint(host host.Host, p peer.ID) RendezvousPoint {
@@ -73,7 +73,7 @@ func (rp *rendezvousPoint) Register(ctx context.Context, ns string, ttl int) err
 	r := ggio.NewDelimitedReader(s, inet.MessageSizeMax)
 	w := ggio.NewDelimitedWriter(s)
 
-	req := newRegisterMessage(ns, pstore.PeerInfo{ID: rp.host.ID(), Addrs: rp.host.Addrs()}, ttl)
+	req := newRegisterMessage(ns, peer.AddrInfo{ID: rp.host.ID(), Addrs: rp.host.Addrs()}, ttl)
 	err = w.WriteMsg(req)
 	if err != nil {
 		return err
@@ -264,32 +264,32 @@ func discoverAsync(ctx context.Context, ns string, s inet.Stream, ch chan Regist
 	}
 }
 
-func (rc *rendezvousClient) Discover(ctx context.Context, ns string, limit int, cookie []byte) ([]pstore.PeerInfo, []byte, error) {
+func (rc *rendezvousClient) Discover(ctx context.Context, ns string, limit int, cookie []byte) ([]peer.AddrInfo, []byte, error) {
 	regs, cookie, err := rc.rp.Discover(ctx, ns, limit, cookie)
 	if err != nil {
 		return nil, nil, err
 	}
 
-	pinfos := make([]pstore.PeerInfo, len(regs))
+	pinfos := make([]peer.AddrInfo, len(regs))
 	for i, reg := range regs {
 		pinfos[i] = reg.Peer
 	}
 
 	return pinfos, cookie, nil
 }
 
-func (rc *rendezvousClient) DiscoverAsync(ctx context.Context, ns string) (<-chan pstore.PeerInfo, error) {
+func (rc *rendezvousClient) DiscoverAsync(ctx context.Context, ns string) (<-chan peer.AddrInfo, error) {
 	rch, err := rc.rp.DiscoverAsync(ctx, ns)
 	if err != nil {
 		return nil, err
 	}
 
-	ch := make(chan pstore.PeerInfo)
+	ch := make(chan peer.AddrInfo)
 	go discoverPeersAsync(ctx, rch, ch)
 	return ch, nil
 }
 
-func discoverPeersAsync(ctx context.Context, rch <-chan Registration, ch chan pstore.PeerInfo) {
+func discoverPeersAsync(ctx context.Context, rch <-chan Registration, ch chan peer.AddrInfo) {
 	defer close(ch)
 	for {
 		select {

client_test.go

@@ -5,8 +5,8 @@ import (
 	"testing"
 	"time"
 
-	host "github.com/libp2p/go-libp2p-host"
-	pstore "github.com/libp2p/go-libp2p-peerstore"
+	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/peer"
 )
 
 func getRendezvousClients(t *testing.T, hosts []host.Host) []RendezvousClient {
@@ -110,7 +110,7 @@ func TestClientRegistrationAndDiscoveryAsync(t *testing.T) {
 	DiscoverAsyncInterval = 2 * time.Minute
 }
 
-func checkPeerInfo(t *testing.T, pi pstore.PeerInfo, host host.Host) {
+func checkPeerInfo(t *testing.T, pi peer.AddrInfo, host host.Host) {
 	if pi.ID != host.ID() {
 		t.Fatal("bad registration: peer ID doesn't match host ID")
 	}

db/dbi.go

@@ -1,7 +1,7 @@
 package dbi
 
 import (
-	peer "github.com/libp2p/go-libp2p-peer"
+	"github.com/libp2p/go-libp2p-core/peer"
 )
 
 type RegistrationRecord struct {

db/sqlite/db.go

@@ -16,7 +16,7 @@ import (
 	_ "github.com/mattn/go-sqlite3"
 
 	logging "github.com/ipfs/go-log"
-	peer "github.com/libp2p/go-libp2p-peer"
+	"github.com/libp2p/go-libp2p-core/peer"
 )
 
 var log = logging.Logger("rendezvous/db")

db/sqlite/db_test.go

@@ -7,7 +7,7 @@ import (
 	"testing"
 	"time"
 
-	peer "github.com/libp2p/go-libp2p-peer"
+	"github.com/libp2p/go-libp2p-core/peer"
 	ma "github.com/multiformats/go-multiaddr"
 )
 

discovery_client.go

@@ -0,0 +1,134 @@
+package rendezvous
+
+import (
+	"context"
+	"github.com/libp2p/go-libp2p-core/discovery"
+	"github.com/libp2p/go-libp2p-core/host"
+	"github.com/libp2p/go-libp2p-core/peer"
+	"math"
+	"math/rand"
+	"sync"
+	"time"
+)
+
+type rendezvousDiscoveryClient struct {
+	rp        RendezvousPoint
+	peerCache sync.Map //is a map[string]discoveredPeerCache
+	rng       *rand.Rand
+	rngMux    sync.Mutex
+}
+
+type discoveredPeerCache struct {
+	cachedRegs []Registration
+	mux        sync.Mutex
+}
+
+func NewRendezvousDiscoveryClient(host host.Host, rendezvousPeer peer.ID) discovery.Discovery {
+	rp := NewRendezvousPoint(host, rendezvousPeer)
+	return &rendezvousDiscoveryClient{rp, sync.Map{}, rand.New(rand.NewSource(rand.Int63())), sync.Mutex{}}
+}
+
+func (c *rendezvousDiscoveryClient) Advertise(ctx context.Context, ns string, opts ...discovery.Option) (time.Duration, error) {
+	// Get options
+	var options discovery.Options
+	err := options.Apply(opts...)
+	if err != nil {
+		return 0, err
+	}
+
+	ttl := options.Ttl
+	var ttlSeconds int
+
+	// Default is minimum duration
+	if ttl == 0 {
+		ttlSeconds = 120
+	} else {
+		ttlSeconds = int(math.Round(ttl.Seconds()))
+	}
+
+	if err := c.rp.Register(ctx, ns, ttlSeconds); err != nil {
+		return 0, err
+	}
+
+	actualTTL := time.Duration(ttlSeconds) * time.Second
+	return actualTTL, nil
+}
+
+func (c *rendezvousDiscoveryClient) FindPeers(ctx context.Context, ns string, opts ...discovery.Option) (<-chan peer.AddrInfo, error) {
+	// Get options
+	var options discovery.Options
+	err := options.Apply(opts...)
+	if err != nil {
+		return nil, err
+	}
+
+	const maxLimit = 1000
+	limit := options.Limit
+	if limit == 0 || limit > maxLimit {
+		limit = maxLimit
+	}
+
+	// Get cached peers
+	var cache *discoveredPeerCache
+
+	genericCache, _ := c.peerCache.LoadOrStore(ns, &discoveredPeerCache{})
+	cache = genericCache.(*discoveredPeerCache)
+
+	cache.mux.Lock()
+	cachedRegs := cache.cachedRegs
+
+	// Remove all expired entries from cache
+	currentTime := int(time.Now().Unix())
+	newCacheSize := len(cachedRegs)
+
+	for i := 0; i < newCacheSize; i++ {
+		reg := cachedRegs[i]
+		if reg.Ttl < currentTime {
+			newCacheSize--
+			if i != newCacheSize {
+				cachedRegs[i] = cachedRegs[newCacheSize]
+				i--
+			}
+		}
+	}
+	cache.cachedRegs = cachedRegs[:newCacheSize]
+	cache.mux.Unlock()
+
+	// Discover new records if we don't have enough
+	var discoveryErr error
+	if newCacheSize < limit {
+		if discoveryRecords, _, err := c.rp.Discover(ctx, ns, limit, nil); err == nil {
+			for i := range discoveryRecords {
+				discoveryRecords[i].Ttl += currentTime
+			}
+			cache.mux.Lock()
+			cache.cachedRegs = discoveryRecords
+			cache.mux.Unlock()
+		} else {
+			// TODO: Should we return error even if we have valid cached results?
+			discoveryErr = err
+		}
+	}
+
+	// Randomize and fill channel with available records
+	cache.mux.Lock()
+	cachedRegs = cache.cachedRegs
+	sendQuantity := len(cachedRegs)
+	if limit < sendQuantity {
+		sendQuantity = limit
+	}
+
+	chPeer := make(chan peer.AddrInfo, sendQuantity)
+
+	c.rngMux.Lock()
+	perm := c.rng.Perm(len(cachedRegs))[0:sendQuantity]
+	c.rngMux.Unlock()
+
+	for _, i := range perm {
+		chPeer <- cachedRegs[i].Peer
+	}
+
+	cache.mux.Unlock()
+	close(chPeer)
+	return chPeer, discoveryErr
+}