enhancement/subscription (#60)

main.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	"github.com/status-im/mvds/node"
-	"github.com/status-im/mvds/protobuf"
 	"github.com/status-im/mvds/state"
 	"github.com/status-im/mvds/store"
 	"github.com/status-im/mvds/transport"
@@ -116,7 +115,6 @@ func createNode(transport transport.Transport, id state.PeerID, mode node.Mode)
 		0,
 		id,
 		mode,
-		make(chan protobuf.Message),
 	)
 }
 

node/node.go

@@ -60,7 +60,6 @@ func NewNode(
 	currentEpoch int64,
 	id state.PeerID,
 	mode Mode,
-	subscription chan<- protobuf.Message,
 ) *Node {
 	ctx, cancel := context.WithCancel(context.Background())
 
@@ -76,7 +75,6 @@ func NewNode(
 		ID:           id,
 		epoch:        currentEpoch,
 		mode:         mode,
-		subscription: subscription,
 	}
 }
 
@@ -117,6 +115,11 @@ func (n *Node) Stop() {
 	n.cancel()
 }
 
+// Subscribe subscribes to incoming messages.
+func (n *Node) Subscribe(sub chan <-protobuf.Message) {
+	n.subscription = sub
+}
+
 // AppendMessage sends a message to a given group.
 func (n *Node) AppendMessage(group state.GroupID, data []byte) (state.MessageID, error) {
 	m := protobuf.Message{
@@ -329,7 +332,9 @@ func (n *Node) onMessage(group state.GroupID, sender state.PeerID, msg protobuf.
 		}
 	}()
 
-	n.subscription <- msg
+	if n.subscription != nil {
+		n.subscription <- msg
+	}
 
 	err = n.store.Add(msg)
 	if err != nil {