Feat/add discovery client

client.go

@@ -9,34 +9,34 @@ 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 (
 	DiscoverAsyncInterval = 2 * time.Minute
 )
 
 type RendezvousPoint interface {
-	Register(ctx context.Context, ns string, ttl int) error
+	Register(ctx context.Context, ns string, ttl int) (time.Duration, error)
 	Unregister(ctx context.Context, ns string) error
 	Discover(ctx context.Context, ns string, limit int, cookie []byte) ([]Registration, []byte, error)
 	DiscoverAsync(ctx context.Context, ns string) (<-chan Registration, error)
 }
 
 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
+	Register(ctx context.Context, ns string, ttl int) (time.Duration, 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 {
@@ -63,7 +63,7 @@ type rendezvousClient struct {
 	rp RendezvousPoint
 }
 
-func (rp *rendezvousPoint) Register(ctx context.Context, ns string, ttl int) error {
+func (rp *rendezvousPoint) Register(ctx context.Context, ns string, ttl int) (time.Duration, error) {
 	s, err := rp.host.NewStream(ctx, rp.p, RendezvousProto)
 	if err != nil {
 		return err
@@ -73,42 +73,43 @@ 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
+		return 0, err
 	}
 
 	var res pb.Message
 	err = r.ReadMsg(&res)
 	if err != nil {
-		return err
+		return 0, err
 	}
 
 	if res.GetType() != pb.Message_REGISTER_RESPONSE {
-		return fmt.Errorf("Unexpected response: %s", res.GetType().String())
+		return 0, fmt.Errorf("Unexpected response: %s", res.GetType().String())
 	}
 
-	status := res.GetRegisterResponse().GetStatus()
+	response := res.GetRegisterResponse()
+	status := response.GetStatus()
 	if status != pb.Message_OK {
-		return RendezvousError{Status: status, Text: res.GetRegisterResponse().GetStatusText()}
+		return 0, RendezvousError{Status: status, Text: res.GetRegisterResponse().GetStatusText()}
 	}
 
-	return nil
+	return time.Duration(*response.Ttl) * time.Second, nil
 }
 
-func (rc *rendezvousClient) Register(ctx context.Context, ns string, ttl int) error {
+func (rc *rendezvousClient) Register(ctx context.Context, ns string, ttl int) (time.Duration, error) {
 	if ttl < 120 {
-		return fmt.Errorf("registration TTL is too short")
+		return 0, fmt.Errorf("registration TTL is too short")
 	}
 
-	err := rc.rp.Register(ctx, ns, ttl)
+	returnedTTL, err := rc.rp.Register(ctx, ns, ttl)
 	if err != nil {
-		return err
+		return 0, err
 	}
 
 	go registerRefresh(ctx, rc.rp, ns, ttl)
-	return nil
+	return returnedTTL, nil
 }
 
 func registerRefresh(ctx context.Context, rz RendezvousPoint, ns string, ttl int) {
@@ -264,32 +265,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,148 @@
+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 map[peer.ID]*Registration
+	cookie     []byte
+	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
+
+	if ttl == 0 {
+		ttlSeconds = 7200
+	} else {
+		ttlSeconds = int(math.Round(ttl.Seconds()))
+	}
+
+	if returnedTTL, err := c.rp.Register(ctx, ns, ttlSeconds); err != nil {
+		return 0, err
+	} else {
+		return returnedTTL, 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()
+
+	// Remove all expired entries from cache
+	currentTime := int(time.Now().Unix())
+	newCacheSize := len(cache.cachedRegs)
+
+	for p := range cache.cachedRegs {
+		reg := cache.cachedRegs[p]
+		if reg.Ttl < currentTime {
+			newCacheSize--
+			delete(cache.cachedRegs, p)
+		}
+	}
+
+	cookie := cache.cookie
+	cache.mux.Unlock()
+
+	// Discover new records if we don't have enough
+	var discoveryErr error
+	if newCacheSize < limit {
+		if discoveryRecords, newCookie, err := c.rp.Discover(ctx, ns, limit, cookie); err == nil {
+			cache.mux.Lock()
+			if cache.cachedRegs == nil {
+				cache.cachedRegs = make(map[peer.ID]*Registration)
+			}
+			for i := range discoveryRecords {
+				rec := &discoveryRecords[i]
+				rec.Ttl += currentTime
+				cache.cachedRegs[rec.Peer.ID] = rec
+			}
+			cache.cookie = newCookie
+			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()
+	sendQuantity := len(cache.cachedRegs)
+	if limit < sendQuantity {
+		sendQuantity = limit
+	}
+
+	chPeer := make(chan peer.AddrInfo, sendQuantity)
+
+	c.rngMux.Lock()
+	perm := c.rng.Perm(len(cache.cachedRegs))[0:sendQuantity]
+	c.rngMux.Unlock()
+
+	permSet := make(map[int]int)
+	for i, v := range perm {
+		permSet[v] = i
+	}
+
+	sendLst := make([]*peer.AddrInfo, sendQuantity)
+	iter := 0
+	for k := range cache.cachedRegs {
+		if sendIndex, ok := permSet[iter]; ok {
+			sendLst[sendIndex] = &cache.cachedRegs[k].Peer
+		}
+		iter++
+	}
+
+	for _, send := range sendLst {
+		chPeer <- *send
+	}
+
+	cache.mux.Unlock()
+	close(chPeer)
+	return chPeer, discoveryErr
+}