WIP- Fabric gossip component

This is a commit that contains only APIs and protobuff
The API between the ledger and the gossip component is in gossip/api/api.go

Change-Id: I9f6aef85f3b03e2d3a6b9850148e9cf4d1a93ce3
Signed-off-by: Yacov Manevich <yacovm@il.ibm.com>

gossip/api/api.go

@@ -0,0 +1,93 @@
+/*
+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 api
+
+import (
+	"github.com/hyperledger/fabric/gossip/discovery"
+	"google.golang.org/grpc"
+)
+
+type GossipEmitterFactory interface {
+	NewGossipEmitter(id string, discSvc discovery.DiscoveryService) GossipService
+}
+
+// GossipService is used to publish new blocks to the gossip network
+type GossipService interface {
+	// payload: Holds the block's content, hash and seqNum
+	Publish(payload Payload) error
+}
+
+type BindAddress struct {
+	Host string
+	Port int16
+}
+
+// Payload defines an object that contains a ledger block
+type Payload struct {
+	Data   []byte // The content of the message, possibly encrypted or signed
+	Hash   string // The message hash
+	SeqNum uint64 // The message sequence number
+}
+
+type GossipMemberFactory interface {
+	NewGossipMember(discovery.DiscoveryService, ReplicationProvider, MessageCryptoService, MessagePolicyVerifier, *grpc.Server) GossipMember
+
+	NewGossipMemberWithRPCServer(discovery.DiscoveryService, ReplicationProvider, MessageCryptoService, MessagePolicyVerifier, BindAddress) (GossipMember, error)
+}
+
+// GossipMember is used to obtain new blocks from the gossip network
+type GossipMember interface {
+	// RegisterCallback registers a callback that is invoked on messages
+	// from startSeq to endSeq and invokes the callback when they arrive
+	RegisterCallback(startSeq uint64, endSeq uint64, callback func([]Payload))
+}
+
+// ReplicationProvider used by the GossipMember in order to obtain Blocks of
+// certain seqNum range to be sent to the requester
+type ReplicationProvider interface {
+	// GetData used by the gossip component to obtain certain blocks from the ledger.
+	// Returns the blocks requested with the given sequence numbers, or an error if
+	// some block requested is not available.
+	GetData(startSeqNum uint64, endSeqNum uint64) ([]Payload, error)
+
+	// LastBlockSeq used by the gossip component to obtain the last sequence of a block the ledger has.
+	LastBlockSeq() uint64
+}
+
+// MessageCryptoVerifier verifies the message's authenticity,
+// if messages are cryptographically signed
+type MessageCryptoService interface {
+	// Verify returns nil whether the message and its identifier are authentic,
+	// otherwise returns an error
+	Verify(seqNum uint64, sender string, payload Payload) error
+
+	// Sign signs the payload
+	Sign(sender string, Payload Payload) Payload
+
+	// SignBlob signs a blob
+	SignBlob([]byte) []byte
+
+	// VerifyBlob verifies a blob, returns error on failure
+	// and nil if the blob is correctly signed
+	VerifyBlob(sender string, blob []byte) error
+}
+
+// MessagePolicyVerifier verifies whether the message conforms to all required policies,
+// and can be safely delivered to the user.
+type MessagePolicyVerifier interface {
+	Verify(seqNum uint64, sender string, payload Payload) error
+}

gossip/comm/comm.go

@@ -0,0 +1,53 @@
+/*
+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 comm
+
+import (
+	"github.com/hyperledger/fabric/gossip/proto"
+	"sync"
+)
+
+type CommModule interface {
+	// Send sends a message to endpoints
+	Send(msg *proto.GossipMessage, endpoints ...string)
+
+	// Probe probes a remote node and returns nil if its responsive
+	Probe(endpoint string) 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
+	Accept(MessageAcceptor) <-chan *ReceivedMessage
+
+	// PresumedDead returns a read-only channel for node endpoints that are suspected to be offline
+	PresumedDead() <-chan string
+
+	// CloseConn closes a connection to a certain endpoint
+	CloseConn(endpoint string)
+
+	// Stop stops the module
+	Stop()
+}
+
+
+type MessageAcceptor func(*proto.GossipMessage) bool
+
+type ReceivedMessage struct {
+	*proto.GossipMessage
+	lock      *sync.Mutex
+	srvStream proto.Gossip_GossipStreamServer
+	clStream  proto.Gossip_GossipStreamClient
+}

gossip/discovery/discovery.go

@@ -0,0 +1,78 @@
+/*
+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 discovery
+
+import "github.com/hyperledger/fabric/gossip/proto"
+
+// CryptoService is an interface that the discovery expects to be implemented and passed on creation
+type CryptoService interface {
+	// validateAliveMsg validates that an Alive message is authentic
+	ValidateAliveMsg(*proto.AliveMessage) bool
+
+	// SignMessage signs an AliveMessage and updates its signature field
+	SignMessage(*proto.AliveMessage) *proto.AliveMessage
+}
+
+// CommService is an interface that the discovery expects to be implemented and passed on creation
+type CommService interface {
+	// Gossip gossips a message
+	Gossip(msg *proto.GossipMessage)
+
+	// SendToPeer sends to a given peer a message.
+	// The nonce can be anything since the communication module handles the nonce itself
+	SendToPeer(peer *NetworkMember, msg *proto.GossipMessage)
+
+	// Ping probes a remote peer and returns if it's responsive or not
+	Ping(peer *NetworkMember) bool
+
+	// Accept returns a read-only channel for membership messages sent from remote peers
+	Accept() <-chan GossipMsg
+
+	// PresumedDead returns a read-only channel for peers that are presumed to be dead
+	PresumedDead() <-chan string
+
+	// CloseConn orders to close the connection with a certain peer
+	CloseConn(id string)
+}
+
+type GossipMsg interface {
+	GetGossipMessage() *proto.GossipMessage
+}
+
+type NetworkMember struct {
+	Id       string
+	Endpoint string
+	Metadata []byte
+}
+
+type DiscoveryService interface {
+
+	// Self returns this instance's membership information
+	Self() NetworkMember
+
+	// UpdateMetadata updates this instance's metadata
+	UpdateMetadata([]byte)
+
+	// UpdateEndpoint updates this instance's endpoint
+	UpdateEndpoint(string)
+
+	// Stops this instance
+	Stop()
+
+	// GetMembership returns the alive members in the view
+	GetMembership() []NetworkMember
+}

gossip/gossip/gossip.go

@@ -0,0 +1,60 @@
+/*
+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 (
+	"github.com/hyperledger/fabric/gossip/discovery"
+	"github.com/hyperledger/fabric/gossip/proto"
+	"time"
+)
+
+type GossipService interface {
+
+	// GetPeersMetadata returns a mapping of endpoint --> metadata
+	GetPeersMetadata() map[string][]byte
+
+	// UpdateMetadata updates the self metadata of the discovery layer
+	UpdateMetadata([]byte)
+
+	// Gossip sends a message to other peers to the network
+	Gossip(msg *proto.GossipMessage)
+
+	// Accept returns a channel that outputs messages from other peers
+	Accept(MessageAcceptor) <-chan *proto.GossipMessage
+
+	// Stop stops the gossip component
+	Stop()
+}
+
+type MessageAcceptor func(*proto.GossipMessage) bool
+
+type GossipConfig struct {
+	BindPort            int
+	Id                  string
+	SelfEndpoint        string
+	BootstrapPeers      []*discovery.NetworkMember
+	PropagateIterations int
+	PropagatePeerNum    int
+
+	MaxMessageCountToStore int
+
+	MaxPropagationBurstSize    int
+	MaxPropagationBurstLatency time.Duration
+
+	PullInterval time.Duration
+	PullPeerNum  int
+}