BFT Block Puller: block deliverer, connection source (#4258)

- Introduce an abstraction BlockDeliverer that can CFT/BFT implementations
- Expose ConnectionSource method Endpoints


Change-Id: I4c2d9b8f74a4004d55e779438e41e7a80b274504

Signed-off-by: Yoav Tock <[email protected]>

Author: tock-ibm

core/deliverservice/deliveryclient.go

@@ -41,12 +41,20 @@ type DeliverService interface {
 	Stop()
 }
 
+// BlockDeliverer communicates with orderers to obtain new blocks and send them to the committer service, for a
+// specific channel. It can be implemented using different protocols depending on the ordering service consensus type,
+// e.g CFT (etcdraft) or BFT (SmartBFT).
+type BlockDeliverer interface {
+	Stop()
+	DeliverBlocks()
+}
+
 // deliverServiceImpl the implementation of the delivery service
 // maintains connection to the ordering service and maps of
 // blocks providers
 type deliverServiceImpl struct {
 	conf           *Config
-	blockProviders map[string]*blocksprovider.Deliverer
+	blockProviders map[string]BlockDeliverer
 	lock           sync.RWMutex
 	stopping       bool
 }
@@ -78,7 +86,7 @@ type Config struct {
 func NewDeliverService(conf *Config) DeliverService {
 	ds := &deliverServiceImpl{
 		conf:           conf,
-		blockProviders: make(map[string]*blocksprovider.Deliverer),
+		blockProviders: make(map[string]BlockDeliverer),
 	}
 	return ds
 }

core/deliverservice/deliveryclient_test.go

@@ -77,8 +77,9 @@ eUCutqn1KYDMYh54i6p723cXbdDkmvL2UCciHyHdSWS9lmkKVdyNGIJ6
 		}
 
 		bp, ok := ds.blockProviders["channel-id"]
+		bpd := bp.(*blocksprovider.Deliverer)
 		require.True(t, ok, "map entry must exist")
-		require.Equal(t, "76f7a03f8dfdb0ef7c4b28b3901fe163c730e906c70e4cdf887054ad5f608bed", fmt.Sprintf("%x", bp.TLSCertHash))
+		require.Equal(t, "76f7a03f8dfdb0ef7c4b28b3901fe163c730e906c70e4cdf887054ad5f608bed", fmt.Sprintf("%x", bpd.TLSCertHash))
 	})
 
 	t.Run("Green Path without mutual TLS", func(t *testing.T) {
@@ -99,8 +100,9 @@ eUCutqn1KYDMYh54i6p723cXbdDkmvL2UCciHyHdSWS9lmkKVdyNGIJ6
 		}
 
 		bp, ok := ds.blockProviders["channel-id"]
+		bpd := bp.(*blocksprovider.Deliverer)
 		require.True(t, ok, "map entry must exist")
-		require.Nil(t, bp.TLSCertHash)
+		require.Nil(t, bpd.TLSCertHash)
 	})
 
 	t.Run("Exists", func(t *testing.T) {
@@ -131,11 +133,11 @@ func TestStopDeliverForChannel(t *testing.T) {
 	t.Run("Green path", func(t *testing.T) {
 		ds := NewDeliverService(&Config{}).(*deliverServiceImpl)
 		doneA := make(chan struct{})
-		ds.blockProviders = map[string]*blocksprovider.Deliverer{
-			"a": {
+		ds.blockProviders = map[string]BlockDeliverer{
+			"a": &blocksprovider.Deliverer{
 				DoneC: doneA,
 			},
-			"b": {
+			"b": &blocksprovider.Deliverer{
 				DoneC: make(chan struct{}),
 			},
 		}
@@ -153,11 +155,11 @@ func TestStopDeliverForChannel(t *testing.T) {
 
 	t.Run("Already stopping", func(t *testing.T) {
 		ds := NewDeliverService(&Config{}).(*deliverServiceImpl)
-		ds.blockProviders = map[string]*blocksprovider.Deliverer{
-			"a": {
+		ds.blockProviders = map[string]BlockDeliverer{
+			"a": &blocksprovider.Deliverer{
 				DoneC: make(chan struct{}),
 			},
-			"b": {
+			"b": &blocksprovider.Deliverer{
 				DoneC: make(chan struct{}),
 			},
 		}
@@ -169,11 +171,11 @@ func TestStopDeliverForChannel(t *testing.T) {
 
 	t.Run("Non-existent", func(t *testing.T) {
 		ds := NewDeliverService(&Config{}).(*deliverServiceImpl)
-		ds.blockProviders = map[string]*blocksprovider.Deliverer{
-			"a": {
+		ds.blockProviders = map[string]BlockDeliverer{
+			"a": &blocksprovider.Deliverer{
 				DoneC: make(chan struct{}),
 			},
-			"b": {
+			"b": &blocksprovider.Deliverer{
 				DoneC: make(chan struct{}),
 			},
 		}
@@ -185,18 +187,19 @@ func TestStopDeliverForChannel(t *testing.T) {
 
 func TestStop(t *testing.T) {
 	ds := NewDeliverService(&Config{}).(*deliverServiceImpl)
-	ds.blockProviders = map[string]*blocksprovider.Deliverer{
-		"a": {
+	ds.blockProviders = map[string]BlockDeliverer{
+		"a": &blocksprovider.Deliverer{
 			DoneC: make(chan struct{}),
 		},
-		"b": {
+		"b": &blocksprovider.Deliverer{
 			DoneC: make(chan struct{}),
 		},
 	}
 	require.False(t, ds.stopping)
 	for _, bp := range ds.blockProviders {
+		bpd := bp.(*blocksprovider.Deliverer)
 		select {
-		case <-bp.DoneC:
+		case <-bpd.DoneC:
 			require.Fail(t, "block providers should not be closed")
 		default:
 		}
@@ -206,8 +209,9 @@ func TestStop(t *testing.T) {
 	require.True(t, ds.stopping)
 	require.Len(t, ds.blockProviders, 2)
 	for _, bp := range ds.blockProviders {
+		bpd := bp.(*blocksprovider.Deliverer)
 		select {
-		case <-bp.DoneC:
+		case <-bpd.DoneC:
 		default:
 			require.Fail(t, "block providers should te closed")
 		}

internal/pkg/peer/blocksprovider/blocksprovider.go

@@ -76,6 +76,7 @@ type BlockVerifier interface {
 //go:generate counterfeiter -o fake/orderer_connection_source.go --fake-name OrdererConnectionSource . OrdererConnectionSource
 type OrdererConnectionSource interface {
 	RandomEndpoint() (*orderers.Endpoint, error)
+	Endpoints() []*orderers.Endpoint
 }
 
 //go:generate counterfeiter -o fake/dialer.go --fake-name Dialer . Dialer

internal/pkg/peer/blocksprovider/fake/orderer_connection_source.go

@@ -53,6 +53,13 @@ func (cs *ConnectionSource) RandomEndpoint() (*Endpoint, error) {
 	return cs.allEndpoints[rand.Intn(len(cs.allEndpoints))], nil
 }
 
+func (cs *ConnectionSource) Endpoints() []*Endpoint {
+	cs.mutex.RLock()
+	defer cs.mutex.RUnlock()
+
+	return cs.allEndpoints
+}
+
 func (cs *ConnectionSource) Update(globalAddrs []string, orgs map[string]OrdererOrg) {
 	cs.mutex.Lock()
 	defer cs.mutex.Unlock()