Fix Broadcaster interface mismatch to pubsubBroadcaster (#246)

Author: hsanjuan

crdt.go

@@ -60,9 +60,9 @@ var (
 // all replicas and to retrieve payloads broadcasted.
 type Broadcaster interface {
 	// Send payload to other replicas.
-	Broadcast([]byte) error
+	Broadcast(context.Context, []byte) error
 	// Obtain the next payload received from the network.
-	Next() ([]byte, error)
+	Next(context.Context) ([]byte, error)
 }
 
 // A SessionDAGService is a Sessions-enabled DAGService. This type of DAG-Service
@@ -351,7 +351,7 @@ func (store *Datastore) handleNext(ctx context.Context) {
 		default:
 		}
 
-		data, err := store.broadcaster.Next()
+		data, err := store.broadcaster.Next(ctx)
 		if err != nil {
 			if err == ErrNoMoreBroadcast || ctx.Err() != nil {
 				return
@@ -1301,7 +1301,7 @@ func (store *Datastore) broadcast(ctx context.Context, cids []cid.Cid) error {
 		return err
 	}
 
-	err = store.broadcaster.Broadcast(bcastBytes)
+	err = store.broadcaster.Broadcast(ctx, bcastBytes)
 	if err != nil {
 		return errors.Wrapf(err, "error broadcasting %s", cids)
 	}

crdt_test.go

@@ -120,7 +120,7 @@ func newBroadcasters(t testing.TB, n int) ([]*mockBroadcaster, context.CancelFun
 	return broadcasters, cancel
 }
 
-func (mb *mockBroadcaster) Broadcast(data []byte) error {
+func (mb *mockBroadcaster) Broadcast(ctx context.Context, data []byte) error {
 	var wg sync.WaitGroup
 
 	randg := rand.New(rand.NewSource(time.Now().UnixNano()))
@@ -155,10 +155,12 @@ func (mb *mockBroadcaster) Broadcast(data []byte) error {
 	return nil
 }
 
-func (mb *mockBroadcaster) Next() ([]byte, error) {
+func (mb *mockBroadcaster) Next(ctx context.Context) ([]byte, error) {
 	select {
 	case data := <-mb.myChan:
 		return data, nil
+	case <-ctx.Done():
+		return nil, ErrNoMoreBroadcast
 	case <-mb.ctx.Done():
 		return nil, ErrNoMoreBroadcast
 	}

pubsub_broadcaster.go

@@ -7,6 +7,8 @@ import (
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 )
 
+var _ Broadcaster = (*PubSubBroadcaster)(nil)
+
 // PubSubBroadcaster implements a Broadcaster using libp2p PubSub.
 type PubSubBroadcaster struct {
 	ctx   context.Context