Merge "[FAB-2198] Introduce envelopes to gossip message"

gossip/comm/comm.go

@@ -38,7 +38,7 @@ type Comm interface {
 
 	// Accept returns a dedicated read-only channel for messages sent by other nodes that match a certain predicate.
 	// Each message from the channel can be used to send a reply back to the sender
-	Accept(common.MessageAcceptor) <-chan ReceivedMessage
+	Accept(common.MessageAcceptor) <-chan proto.ReceivedMessage
 
 	// PresumedDead returns a read-only channel for node endpoints that are suspected to be offline
 	PresumedDead() <-chan common.PKIidType
@@ -63,20 +63,3 @@ type RemotePeer struct {
 func (p *RemotePeer) String() string {
 	return fmt.Sprintf("%s, PKIid:%v", p.Endpoint, p.PKIID)
 }
-
-// ReceivedMessage is a GossipMessage wrapper that
-// enables the user to send a message to the origin from which
-// the ReceivedMessage was sent from.
-// It also allows to know the identity of the sender
-type ReceivedMessage interface {
-
-	// Respond sends a GossipMessage to the origin from which this ReceivedMessage was sent from
-	Respond(msg *proto.GossipMessage)
-
-	// GetGossipMessage returns the underlying GossipMessage
-	GetGossipMessage() *proto.GossipMessage
-
-	// GetPKIID returns the PKI-ID of the remote peer
-	// that sent the message
-	GetPKIID() common.PKIidType
-}

gossip/comm/comm_impl.go

@@ -98,7 +98,7 @@ func NewCommInstanceWithServer(port int, idMapper identity.Mapper, peerIdentity
 		deadEndpoints:     make(chan common.PKIidType, 100),
 		stopping:          int32(0),
 		exitChan:          make(chan struct{}, 1),
-		subscriptions:     make([]chan ReceivedMessage, 0),
+		subscriptions:     make([]chan proto.ReceivedMessage, 0),
 		blackListedPKIIDs: make([]common.PKIidType, 0),
 	}
 	commInst.connStore = newConnStore(commInst, commInst.logger)
@@ -154,7 +154,7 @@ type commImpl struct {
 	exitChan          chan struct{}
 	stopping          int32
 	stopWG            sync.WaitGroup
-	subscriptions     []chan ReceivedMessage
+	subscriptions     []chan proto.ReceivedMessage
 	blackListedPKIIDs []common.PKIidType
 }
 
@@ -290,9 +290,9 @@ func (c *commImpl) Probe(remotePeer *RemotePeer) error {
 	return err
 }
 
-func (c *commImpl) Accept(acceptor common.MessageAcceptor) <-chan ReceivedMessage {
+func (c *commImpl) Accept(acceptor common.MessageAcceptor) <-chan proto.ReceivedMessage {
 	genericChan := c.msgPublisher.AddChannel(acceptor)
-	specificChan := make(chan ReceivedMessage, 10)
+	specificChan := make(chan proto.ReceivedMessage, 10)
 
 	if c.isStopping() {
 		c.logger.Warning("Accept() called but comm module is stopping, returning empty channel")

gossip/comm/comm_test.go

@@ -94,7 +94,7 @@ func newCommInstance(port int, sec api.MessageCryptoService) (Comm, error) {
 	return inst, err
 }
 
-func handshaker(endpoint string, comm Comm, t *testing.T, sigMutator func([]byte) []byte, pkiIDmutator func([]byte) []byte) <-chan ReceivedMessage {
+func handshaker(endpoint string, comm Comm, t *testing.T, sigMutator func([]byte) []byte, pkiIDmutator func([]byte) []byte) <-chan proto.ReceivedMessage {
 	c := &commImpl{}
 	err := generateCertificates("key.pem", "cert.pem")
 	defer os.Remove("cert.pem")
@@ -189,7 +189,7 @@ func TestBasic(t *testing.T) {
 	m1 := comm1.Accept(acceptAll)
 	m2 := comm2.Accept(acceptAll)
 	out := make(chan uint64, 2)
-	reader := func(ch <-chan ReceivedMessage) {
+	reader := func(ch <-chan proto.ReceivedMessage) {
 		m := <-ch
 		out <- m.GetGossipMessage().Nonce
 	}
@@ -228,7 +228,7 @@ func TestBlackListPKIid(t *testing.T) {
 	defer comm3.Stop()
 	defer comm4.Stop()
 
-	reader := func(instance string, out chan uint64, in <-chan ReceivedMessage) {
+	reader := func(instance string, out chan uint64, in <-chan proto.ReceivedMessage) {
 		for {
 			msg := <-in
 			if msg == nil {
@@ -328,7 +328,7 @@ func TestResponses(t *testing.T) {
 	defer comm1.Stop()
 	defer comm2.Stop()
 
-	nonceIncrememter := func(msg ReceivedMessage) ReceivedMessage {
+	nonceIncrememter := func(msg proto.ReceivedMessage) proto.ReceivedMessage {
 		msg.GetGossipMessage().Nonce++
 		return msg
 	}
@@ -365,11 +365,11 @@ func TestAccept(t *testing.T) {
 	comm2, _ := newCommInstance(7612, naiveSec)
 
 	evenNONCESelector := func(m interface{}) bool {
-		return m.(ReceivedMessage).GetGossipMessage().Nonce%2 == 0
+		return m.(proto.ReceivedMessage).GetGossipMessage().Nonce%2 == 0
 	}
 
 	oddNONCESelector := func(m interface{}) bool {
-		return m.(ReceivedMessage).GetGossipMessage().Nonce%2 != 0
+		return m.(proto.ReceivedMessage).GetGossipMessage().Nonce%2 != 0
 	}
 
 	evenNONCES := comm1.Accept(evenNONCESelector)
@@ -381,7 +381,7 @@ func TestAccept(t *testing.T) {
 	out := make(chan uint64, defRecvBuffSize)
 	sem := make(chan struct{}, 0)
 
-	readIntoSlice := func(a *[]uint64, ch <-chan ReceivedMessage) {
+	readIntoSlice := func(a *[]uint64, ch <-chan proto.ReceivedMessage) {
 		for m := range ch {
 			*a = append(*a, m.GetGossipMessage().Nonce)
 			out <- m.GetGossipMessage().Nonce
@@ -422,7 +422,7 @@ func TestReConnections(t *testing.T) {
 	comm1, _ := newCommInstance(3611, naiveSec)
 	comm2, _ := newCommInstance(3612, naiveSec)
 
-	reader := func(out chan uint64, in <-chan ReceivedMessage) {
+	reader := func(out chan uint64, in <-chan proto.ReceivedMessage) {
 		for {
 			msg := <-in
 			if msg == nil {

gossip/comm/mock/mock_comm.go

@@ -46,7 +46,7 @@ type packetMock struct {
 type channelMock struct {
 	accept common.MessageAcceptor
 
-	channel chan comm.ReceivedMessage
+	channel chan proto.ReceivedMessage
 }
 
 type commMock struct {
@@ -91,11 +91,19 @@ func (packet *packetMock) Respond(msg *proto.GossipMessage) {
 	}
 }
 
+// GetSourceMessage Returns the SignedGossipMessage the ReceivedMessage was
+// constructed with
+func (packet *packetMock) GetSourceMessage() *proto.SignedGossipMessage {
+	return nil
+}
+
 // GetGossipMessage returns the underlying GossipMessage
 func (packet *packetMock) GetGossipMessage() *proto.GossipMessage {
 	return packet.msg.(*proto.GossipMessage)
 }
 
+// GetPKIID returns the PKI-ID of the remote peer
+// that sent the message
 func (packet *packetMock) GetPKIID() common.PKIidType {
 	return nil
 }
@@ -151,8 +159,8 @@ func (mock *commMock) Probe(peer *comm.RemotePeer) error {
 
 // Accept returns a dedicated read-only channel for messages sent by other nodes that match a certain predicate.
 // Each message from the channel can be used to send a reply back to the sender
-func (mock *commMock) Accept(accept common.MessageAcceptor) <-chan comm.ReceivedMessage {
-	ch := make(chan comm.ReceivedMessage)
+func (mock *commMock) Accept(accept common.MessageAcceptor) <-chan proto.ReceivedMessage {
+	ch := make(chan proto.ReceivedMessage)
 	mock.acceptors = append(mock.acceptors, &channelMock{accept, ch})
 	return ch
 }

gossip/comm/mock/mock_comm_test.go

@@ -37,8 +37,8 @@ func TestMockComm(t *testing.T) {
 	defer comm1.Stop()
 
 	msgCh := comm1.Accept(func(message interface{}) bool {
-		return message.(comm.ReceivedMessage).GetGossipMessage().GetStateRequest() != nil ||
-			message.(comm.ReceivedMessage).GetGossipMessage().GetStateResponse() != nil
+		return message.(proto.ReceivedMessage).GetGossipMessage().GetStateRequest() != nil ||
+			message.(proto.ReceivedMessage).GetGossipMessage().GetStateResponse() != nil
 	})
 
 	comm2 := NewCommMock(second.endpoint, members)

gossip/comm/msg.go

@@ -30,6 +30,12 @@ type ReceivedMessageImpl struct {
 	conn *connection
 }
 
+// GetSourceMessage Returns the SignedGossipMessage the ReceivedMessage was
+// constructed with
+func (m *ReceivedMessageImpl) GetSourceMessage() *proto.SignedGossipMessage {
+	return nil
+}
+
 // Respond sends a msg to the source that sent the ReceivedMessageImpl
 func (m *ReceivedMessageImpl) Respond(msg *proto.GossipMessage) {
 	m.conn.send(msg, func(e error) {})

gossip/election/adapter.go

@@ -22,7 +22,6 @@ import (
 	"time"
 
 	"github.com/hyperledger/fabric/gossip/api"
-	"github.com/hyperledger/fabric/gossip/comm"
 	"github.com/hyperledger/fabric/gossip/common"
 	"github.com/hyperledger/fabric/gossip/discovery"
 	proto "github.com/hyperledger/fabric/protos/gossip"
@@ -61,7 +60,7 @@ type gossip interface {
 	// If passThrough is false, the messages are processed by the gossip layer beforehand.
 	// If passThrough is true, the gossip layer doesn't intervene and the messages
 	// can be used to send a reply back to the sender
-	Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan comm.ReceivedMessage)
+	Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan proto.ReceivedMessage)
 
 	// Gossip sends a message to other peers to the network
 	Gossip(msg *proto.GossipMessage)

gossip/election/adapter_test.go

@@ -24,7 +24,6 @@ import (
 	"time"
 
 	"github.com/hyperledger/fabric/gossip/api"
-	"github.com/hyperledger/fabric/gossip/comm"
 	"github.com/hyperledger/fabric/gossip/common"
 	"github.com/hyperledger/fabric/gossip/discovery"
 	proto "github.com/hyperledger/fabric/protos/gossip"
@@ -216,7 +215,7 @@ func (g *peerMockGossip) Peers() []discovery.NetworkMember {
 	return res
 }
 
-func (g *peerMockGossip) Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan comm.ReceivedMessage) {
+func (g *peerMockGossip) Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan proto.ReceivedMessage) {
 	ch := make(chan *proto.GossipMessage, 100)
 	g.acceptorLock.Lock()
 	g.acceptors = append(g.acceptors, &mockAcceptor{

gossip/gossip/certstore.go

@@ -22,7 +22,6 @@ import (
 	"sync"
 
 	"github.com/hyperledger/fabric/gossip/api"
-	"github.com/hyperledger/fabric/gossip/comm"
 	"github.com/hyperledger/fabric/gossip/common"
 	"github.com/hyperledger/fabric/gossip/gossip/pull"
 	"github.com/hyperledger/fabric/gossip/identity"
@@ -63,7 +62,7 @@ func newCertStore(puller pull.Mediator, idMapper identity.Mapper, selfIdentity a
 
 	puller.Add(certStore.createIdentityMessage())
 
-	puller.RegisterMsgHook(pull.ResponseMsgType, func(_ []string, msgs []*proto.GossipMessage, _ comm.ReceivedMessage) {
+	puller.RegisterMsgHook(pull.ResponseMsgType, func(_ []string, msgs []*proto.GossipMessage, _ proto.ReceivedMessage) {
 		for _, msg := range msgs {
 			pkiID := common.PKIidType(msg.GetPeerIdentity().PkiID)
 			cert := api.PeerIdentityType(msg.GetPeerIdentity().Cert)
@@ -78,7 +77,7 @@ func newCertStore(puller pull.Mediator, idMapper identity.Mapper, selfIdentity a
 	return certStore
 }
 
-func (cs *certStore) handleMessage(msg comm.ReceivedMessage) {
+func (cs *certStore) handleMessage(msg proto.ReceivedMessage) {
 	if update := msg.GetGossipMessage().GetDataUpdate(); update != nil {
 		for _, m := range update.Data {
 			if !m.IsIdentityMsg() {

gossip/gossip/certstore_test.go

@@ -50,6 +50,12 @@ type sentMsg struct {
 	mock.Mock
 }
 
+// GetSourceMessage Returns the SignedGossipMessage the ReceivedMessage was
+// constructed with
+func (s *sentMsg) GetSourceMessage() *proto.SignedGossipMessage {
+	return nil
+}
+
 func (s *sentMsg) Respond(msg *proto.GossipMessage) {
 	s.Called(msg)
 }
@@ -80,30 +86,30 @@ func (m *membershipSvcMock) GetMembership() []discovery.NetworkMember {
 }
 
 func TestCertStoreBadSignature(t *testing.T) {
-	badSignature := func(nonce uint64) comm.ReceivedMessage {
+	badSignature := func(nonce uint64) proto.ReceivedMessage {
 		return createUpdateMessage(nonce, createBadlySignedUpdateMessage())
 	}
 
 	testCertificateUpdate(t, badSignature, false)
 }
 
 func TestCertStoreMismatchedIdentity(t *testing.T) {
-	mismatchedIdentity := func(nonce uint64) comm.ReceivedMessage {
+	mismatchedIdentity := func(nonce uint64) proto.ReceivedMessage {
 		return createUpdateMessage(nonce, createMismatchedUpdateMessage())
 	}
 
 	testCertificateUpdate(t, mismatchedIdentity, false)
 }
 
 func TestCertStoreShouldSucceed(t *testing.T) {
-	totallyFineIdentity := func(nonce uint64) comm.ReceivedMessage {
+	totallyFineIdentity := func(nonce uint64) proto.ReceivedMessage {
 		return createUpdateMessage(nonce, createValidUpdateMessage())
 	}
 
 	testCertificateUpdate(t, totallyFineIdentity, true)
 }
 
-func testCertificateUpdate(t *testing.T, updateFactory func(uint64) comm.ReceivedMessage, shouldSucceed bool) {
+func testCertificateUpdate(t *testing.T, updateFactory func(uint64) proto.ReceivedMessage, shouldSucceed bool) {
 	config := pull.PullConfig{
 		MsgType:           proto.PullMsgType_IdentityMsg,
 		PeerCountToSelect: 1,
@@ -257,7 +263,7 @@ func createValidUpdateMessage() *proto.GossipMessage {
 	return m
 }
 
-func createUpdateMessage(nonce uint64, idMsg *proto.GossipMessage) comm.ReceivedMessage {
+func createUpdateMessage(nonce uint64, idMsg *proto.GossipMessage) proto.ReceivedMessage {
 	update := &proto.GossipMessage{
 		Tag: proto.GossipMessage_EMPTY,
 		Content: &proto.GossipMessage_DataUpdate{
@@ -271,7 +277,7 @@ func createUpdateMessage(nonce uint64, idMsg *proto.GossipMessage) comm.Received
 	return &sentMsg{msg: update}
 }
 
-func createDigest(nonce uint64) comm.ReceivedMessage {
+func createDigest(nonce uint64) proto.ReceivedMessage {
 	digest := &proto.GossipMessage{
 		Tag: proto.GossipMessage_EMPTY,
 		Content: &proto.GossipMessage_DataDig{

gossip/gossip/channel/channel.go

@@ -67,7 +67,7 @@ type GossipChannel interface {
 	IsSubscribed(member discovery.NetworkMember) bool
 
 	// HandleMessage processes a message sent by a remote peer
-	HandleMessage(comm.ReceivedMessage)
+	HandleMessage(proto.ReceivedMessage)
 
 	// AddToMsgStore adds a given GossipMessage to the message store
 	AddToMsgStore(msg *proto.GossipMessage)
@@ -354,7 +354,7 @@ func (gc *gossipChannel) ConfigureChannel(joinMsg api.JoinChannelMessage) {
 }
 
 // HandleMessage processes a message sent by a remote peer
-func (gc *gossipChannel) HandleMessage(msg comm.ReceivedMessage) {
+func (gc *gossipChannel) HandleMessage(msg proto.ReceivedMessage) {
 	if !gc.verifyMsg(msg) {
 		return
 	}
@@ -500,7 +500,7 @@ func (gc *gossipChannel) createStateInfoSnapshot() *proto.GossipMessage {
 	}
 }
 
-func (gc *gossipChannel) verifyMsg(msg comm.ReceivedMessage) bool {
+func (gc *gossipChannel) verifyMsg(msg proto.ReceivedMessage) bool {
 	if msg == nil {
 		gc.logger.Warning("Messsage is nil")
 		return false

gossip/gossip/channel/channel_test.go

@@ -124,6 +124,12 @@ type receivedMsg struct {
 	mock.Mock
 }
 
+// GetSourceMessage Returns the SignedGossipMessage the ReceivedMessage was
+// constructed with
+func (m *receivedMsg) GetSourceMessage() *proto.SignedGossipMessage {
+	return nil
+}
+
 func (m *receivedMsg) GetGossipMessage() *proto.GossipMessage {
 	return m.msg
 }

gossip/gossip/gossip.go

@@ -56,7 +56,7 @@ type Gossip interface {
 	// If passThrough is false, the messages are processed by the gossip layer beforehand.
 	// If passThrough is true, the gossip layer doesn't intervene and the messages
 	// can be used to send a reply back to the sender
-	Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan comm.ReceivedMessage)
+	Accept(acceptor common.MessageAcceptor, passThrough bool) (<-chan *proto.GossipMessage, <-chan proto.ReceivedMessage)
 
 	// JoinChan makes the Gossip instance join a channel
 	JoinChan(joinMsg api.JoinChannelMessage, chainID common.ChainID)