FAB-1292 Gossip pull refactoring: Mediator

This commit introduces:
1) A pull.Mediator, an object that offloads
handling of messages and implementation of algo.PullAdapter from
the rest of the gossip code, so that it can be used for any type of
message we want to sync using the pull mechanism.

2) A certStore object, that holds certificates that are going
   to be replicated using the pull mechanism.
3) An addition of an identity message which is a certificate that's
   replicated among peers.

Change-Id: Ia7682c3c9874ee40fa3588706239f914f1e334fb
Signed-off-by: Yacov Manevich <[email protected]>

Author: yacovm

gossip/comm/comm_impl.go

@@ -254,7 +254,7 @@ func (c *commImpl) isStopping() bool {
 
 func (c *commImpl) Probe(peer *RemotePeer) error {
 	if c.isStopping() {
-		return fmt.Errorf("Stopping!")
+		return fmt.Errorf("Stopping")
 	}
 	c.logger.Debug("Entering, endpoint:", peer.Endpoint, "PKIID:", peer.PKIID)
 	var err error

gossip/comm/comm_test.go

@@ -173,7 +173,7 @@ func TestBasic(t *testing.T) {
 	m2 := comm2.Accept(acceptAll)
 	out := make(chan uint64, 2)
 	reader := func(ch <-chan ReceivedMessage) {
-		m := <- ch
+		m := <-ch
 		out <- m.GetGossipMessage().Nonce
 	}
 	go reader(m1)

gossip/gossip/certstore.go

@@ -0,0 +1,104 @@
+/*
+Copyright IBM Corp. 2016 All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+		 http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package gossip
+
+import (
+	"sync"
+
+	prot "github.com/golang/protobuf/proto"
+	"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"
+	"github.com/hyperledger/fabric/gossip/proto"
+	"github.com/hyperledger/fabric/gossip/util"
+)
+
+// certStore supports pull dissemination of identity messages
+type certStore struct {
+	sync.RWMutex
+	selfIdentity api.PeerIdentityType
+	idMapper     identity.Mapper
+	pull         pull.Mediator
+	logger       *util.Logger
+}
+
+func newCertStore(mcs api.MessageCryptoService, selfIdentity api.PeerIdentityType, pullMed pull.Mediator) *certStore {
+	certStore := &certStore{
+		idMapper:     identity.NewIdentityMapper(mcs),
+		selfIdentity: selfIdentity,
+		pull:         pullMed,
+	}
+
+	selfPKIID := certStore.idMapper.GetPKIidOfCert(selfIdentity)
+
+	certStore.logger = util.GetLogger("certStore", string(selfPKIID))
+
+	if err := certStore.idMapper.Put(selfPKIID, selfIdentity); err != nil {
+		certStore.logger.Panic("Failed associating self PKIID to cert:", err)
+	}
+
+	pullMed.Add(certStore.createIdentityMessage())
+
+	pullMed.RegisterMsgHook(pull.ResponseMsgType, func(_ []string, msgs []*proto.GossipMessage, _ comm.ReceivedMessage) {
+		for _, msg := range msgs {
+			pkiID := common.PKIidType(msg.GetPeerIdentity().PkiID)
+			cert := api.PeerIdentityType(msg.GetPeerIdentity().Cert)
+			if err := certStore.idMapper.Put(pkiID, cert); err != nil {
+				certStore.logger.Warning("Failed adding identity", cert, ", reason:", err)
+			}
+		}
+	})
+
+	return certStore
+}
+
+func (cs *certStore) createIdentityMessage() *proto.GossipMessage {
+	identity := &proto.PeerIdentity{
+		Cert:     cs.selfIdentity,
+		Metadata: nil,
+		PkiID:    cs.idMapper.GetPKIidOfCert(cs.selfIdentity),
+		Sig:      nil,
+	}
+
+	b, err := prot.Marshal(identity)
+	if err != nil {
+		cs.logger.Warning("Failed marshalling identity message:", err)
+		return nil
+	}
+
+	sig, err := cs.idMapper.Sign(b)
+	if err != nil {
+		cs.logger.Warning("Failed signing identity message:", err)
+		return nil
+	}
+	identity.Sig = sig
+
+	return &proto.GossipMessage{
+		Channel: nil,
+		Nonce:   0,
+		Tag:     proto.GossipMessage_EMPTY,
+		Content: &proto.GossipMessage_PeerIdentity{
+			PeerIdentity: identity,
+		},
+	}
+}
+
+func (cs *certStore) stop() {
+	cs.pull.Stop()
+}