feat: plumb through datastore contexts

fetch.go

@@ -24,7 +24,7 @@ type fetchProtocol struct {
 	host host.Host
 }
 
-type getValue func(key string) ([]byte, error)
+type getValue func(ctx context.Context, key string) ([]byte, error)
 
 func newFetchProtocol(ctx context.Context, host host.Host, getData getValue) *fetchProtocol {
 	p := &fetchProtocol{ctx, host}
@@ -46,7 +46,7 @@ func (p *fetchProtocol) receive(s network.Stream, getData getValue) {
 		return
 	}
 
-	response, err := getData(msg.Identifier)
+	response, err := getData(p.ctx, msg.Identifier)
 	var respProto pb.FetchResponse
 
 	if err != nil {

fetch_test.go

@@ -22,7 +22,7 @@ type datastore struct {
 	data map[string][]byte
 }
 
-func (d *datastore) Lookup(key string) ([]byte, error) {
+func (d *datastore) Lookup(_ context.Context, key string) ([]byte, error) {
 	v, ok := d.data[key]
 	if !ok {
 		return nil, errors.New("key not found")

go.mod

@@ -3,15 +3,15 @@ module github.com/libp2p/go-libp2p-pubsub-router
 go 1.16
 
 require (
-	github.com/gogo/protobuf v1.3.1
-	github.com/ipfs/go-datastore v0.4.4
+	github.com/gogo/protobuf v1.3.2
+	github.com/ipfs/go-datastore v0.5.0
 	github.com/ipfs/go-ipfs-ds-help v0.1.1
-	github.com/ipfs/go-log/v2 v2.1.1
+	github.com/ipfs/go-log/v2 v2.3.0
 	github.com/libp2p/go-libp2p-blankhost v0.2.0
-	github.com/libp2p/go-libp2p-core v0.7.0
-	github.com/libp2p/go-libp2p-pubsub v0.4.0
+	github.com/libp2p/go-libp2p-core v0.11.0
+	github.com/libp2p/go-libp2p-pubsub v0.6.0
 	github.com/libp2p/go-libp2p-record v0.1.3
-	github.com/libp2p/go-libp2p-swarm v0.3.1
+	github.com/libp2p/go-libp2p-swarm v0.8.0
 	github.com/libp2p/go-msgio v0.0.6
-	golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 )

pubsub.go

@@ -127,7 +127,7 @@ func (p *PubsubValueStore) PutValue(ctx context.Context, key string, value []byt
 
 	ti.dbWriteMx.Lock()
 	defer ti.dbWriteMx.Unlock()
-	recCmp, err := p.putLocal(ti, key, value)
+	recCmp, err := p.putLocal(ctx, ti, key, value)
 	if err != nil {
 		return err
 	}
@@ -147,12 +147,12 @@ func (p *PubsubValueStore) PutValue(ctx context.Context, key string, value []byt
 // First return value is 0 if equal, greater than 0 if better, less than 0 if worse.
 // Second return value is true if valid.
 //
-func (p *PubsubValueStore) compare(key string, val []byte) (int, bool) {
+func (p *PubsubValueStore) compare(ctx context.Context, key string, val []byte) (int, bool) {
 	if p.Validator.Validate(key, val) != nil {
 		return -1, false
 	}
 
-	old, err := p.getLocal(key)
+	old, err := p.getLocal(ctx, key)
 	if err != nil {
 		// If the old one is invalid, the new one is *always* better.
 		return 1, true
@@ -192,7 +192,7 @@ func (p *PubsubValueStore) Subscribe(key string) error {
 		src peer.ID,
 		msg *pubsub.Message,
 	) pubsub.ValidationResult {
-		cmp, valid := p.compare(key, msg.GetData())
+		cmp, valid := p.compare(ctx, key, msg.GetData())
 		if !valid {
 			return pubsub.ValidationReject
 		}
@@ -271,7 +271,7 @@ func (p *PubsubValueStore) rebroadcast(ctx context.Context) {
 			p.mx.Unlock()
 			if len(topics) > 0 {
 				for i, k := range keys {
-					val, err := p.getLocal(k)
+					val, err := p.getLocal(ctx, k)
 					if err == nil {
 						topic := topics[i].topic
 						select {
@@ -300,16 +300,16 @@ func (p *PubsubValueStore) psPublishChannel(ctx context.Context, topic *pubsub.T
 // Requires that the ti.dbWriteMx is held when called
 // Returns true if the value is better then what is currently in the datastore
 // Returns any errors from putting the data in the datastore
-func (p *PubsubValueStore) putLocal(ti *topicInfo, key string, value []byte) (int, error) {
-	cmp, valid := p.compare(key, value)
+func (p *PubsubValueStore) putLocal(ctx context.Context, ti *topicInfo, key string, value []byte) (int, error) {
+	cmp, valid := p.compare(ctx, key, value)
 	if valid && cmp > 0 {
-		return cmp, p.ds.Put(dshelp.NewKeyFromBinary([]byte(key)), value)
+		return cmp, p.ds.Put(ctx, dshelp.NewKeyFromBinary([]byte(key)), value)
 	}
 	return cmp, nil
 }
 
-func (p *PubsubValueStore) getLocal(key string) ([]byte, error) {
-	val, err := p.ds.Get(dshelp.NewKeyFromBinary([]byte(key)))
+func (p *PubsubValueStore) getLocal(ctx context.Context, key string) ([]byte, error) {
+	val, err := p.ds.Get(ctx, dshelp.NewKeyFromBinary([]byte(key)))
 	if err != nil {
 		// Don't invalidate due to ds errors.
 		if err == ds.ErrNotFound {
@@ -330,7 +330,7 @@ func (p *PubsubValueStore) GetValue(ctx context.Context, key string, opts ...rou
 		return nil, err
 	}
 
-	return p.getLocal(key)
+	return p.getLocal(ctx, key)
 }
 
 func (p *PubsubValueStore) SearchValue(ctx context.Context, key string, opts ...routing.Option) (<-chan []byte, error) {
@@ -342,7 +342,7 @@ func (p *PubsubValueStore) SearchValue(ctx context.Context, key string, opts ...
 	defer p.watchLk.Unlock()
 
 	out := make(chan []byte, 1)
-	lv, err := p.getLocal(key)
+	lv, err := p.getLocal(ctx, key)
 	if err == nil {
 		out <- lv
 		close(out)
@@ -513,7 +513,7 @@ func (p *PubsubValueStore) handleSubscription(ctx context.Context, ti *topicInfo
 		}
 
 		ti.dbWriteMx.Lock()
-		recCmp, err := p.putLocal(ti, key, data)
+		recCmp, err := p.putLocal(ctx, ti, key, data)
 		ti.dbWriteMx.Unlock()
 		if recCmp > 0 {
 			if err != nil {

pubsub_test.go

@@ -20,8 +20,15 @@ import (
 )
 
 func newNetHost(ctx context.Context, t *testing.T) host.Host {
-	netw := swarmt.GenSwarm(t, ctx)
-	return bhost.NewBlankHost(netw)
+	netw := swarmt.GenSwarm(t)
+	bh := bhost.NewBlankHost(netw)
+	t.Cleanup(func() {
+		err := bh.Close()
+		if err != nil {
+			panic(err)
+		}
+	})
+	return bh
 }
 
 func newNetHosts(ctx context.Context, t *testing.T, n int) []host.Host {
@@ -32,6 +39,7 @@ func newNetHosts(ctx context.Context, t *testing.T, n int) []host.Host {
 		out = append(out, h)
 	}
 
+	time.Sleep(3 * time.Second)
 	return out
 }
 
@@ -159,6 +167,7 @@ func TestPubsubPublishSubscribe(t *testing.T) {
 	defer cancel()
 
 	pub, vss := setupTest(ctx, t)
+	defer pub.host.Close()
 
 	key := "/namespace/key"
 	key2 := "/namespace/key2"
@@ -268,6 +277,7 @@ func TestWatch(t *testing.T) {
 	defer cancel()
 
 	pub, vss := setupTest(ctx, t)
+	defer pub.host.Close()
 
 	key := "/namespace/key"
 	key2 := "/namespace/key2"

version.json

@@ -1,3 +1,3 @@
 {
-  "version": "v0.4.0"
+  "version": "v0.5.0"
 }