[FAB-7049] Expose deliver service on peer

This CR exposes the deliver service on a peer, which will allow clients
to retrieve blocks using the same mechanism that is available on the
orderer.

Change-Id: Iafcbcf32c7a448680a6726ae4cca5fb94832e68b
Signed-off-by: Will Lahti <wtlahti@us.ibm.com>

Author: wlahti

bddtests/features/bootstrap.feature

@@ -96,7 +96,7 @@ Feature: Bootstrap
         |  peerOrg1     |
         |  peerOrg2     |
 
-    And user "configAdminOrdererOrg0" using cert alias "config-admin-cert" connects to deliver function on orderer "<orderer0>"
+    And user "configAdminOrdererOrg0" using cert alias "config-admin-cert" connects to deliver function on orderer "<orderer0>" using port "7050"
 
     And user "configAdminOrdererOrg0" retrieves the latest config update "latestOrdererConfig" from orderer "<orderer0>" for channel "{ordererSystemChannelId}"
 
@@ -156,7 +156,7 @@ Feature: Bootstrap
     # Requesting a deliver earlier may result in a SERVICE_UNAVAILABLE response and a connection drop
     And I wait "<ChannelJoinDelay>" seconds
 
-    When user "dev0Org0" using cert alias "consortium1-cert" connects to deliver function on orderer "<orderer0>"
+    When user "dev0Org0" using cert alias "consortium1-cert" connects to deliver function on orderer "<orderer0>" using port "7050"
     And user "dev0Org0" sends deliver a seek request on orderer "<orderer0>" with properties:
       | ChainId                               | Start |  End    |
       | com.acme.blockchain.jdoe.channel1     |   0   |  0      |
@@ -190,7 +190,7 @@ Feature: Bootstrap
         | User            | Peer     | Organization  |
         | peer0Signer     | peer0    | peerOrg0      |
 
-      And user "configAdminPeerOrg0" using cert alias "config-admin-cert" connects to deliver function on orderer "<orderer0>"
+      And user "configAdminPeerOrg0" using cert alias "config-admin-cert" connects to deliver function on orderer "<orderer0>" using port "7050"
 
       And user "configAdminPeerOrg0" retrieves the latest config update "latestChannelConfigUpdate" from orderer "<orderer0>" for channel "com.acme.blockchain.jdoe.channel1"
 
@@ -249,6 +249,19 @@ Feature: Bootstrap
         | User            | Peer     | Organization  |
         | peer2Signer     | peer2    | peerOrg1      |
 
+    When user "dev0Org0" using cert alias "consortium1-cert" connects to deliver function on orderer "peer0" using port "7051"
+         And user "dev0Org0" sends deliver a seek request on orderer "peer0" with properties:
+           | ChainId                               | Start |  End    |
+           | com.acme.blockchain.jdoe.channel1     |   0   |  0      |
+
+    Then user "dev0Org0" should get a delivery "genesisBlockForMyNewChannel" from "peer0" of "1" blocks with "1" messages within "1" seconds
+
+    When user "dev0Org0" using cert alias "consortium1-cert" connects to deliver function on orderer "peer2" using port "7051"
+         And user "dev0Org0" sends deliver a seek request on orderer "peer2" with properties:
+           | ChainId                               | Start |  End    |
+           | com.acme.blockchain.jdoe.channel1     |   0   |  0      |
+
+    Then user "dev0Org0" should get a delivery "genesisBlockForMyNewChannelFromOtherOrgsPeer" from "peer2" of "1" blocks with "1" messages within "1" seconds
 
     # Entry point for invoking on an existing channel
     When user "peer0Admin" creates a chaincode spec "ccSpec" with name "example02" of type "GOLANG" for chaincode "github.com/hyperledger/fabric/examples/chaincode/go/chaincode_example02" with args
@@ -378,9 +391,6 @@ Feature: Bootstrap
 
       Then user "dev0Org0" should get a delivery "deliveredInvokeTx1Block" from "<orderer0>" of "1" blocks with "1" messages within "1" seconds
 
-
-    # TODO: Once events are working, consider listen event listener as well.
-
     Examples: Orderer Options
       |          ComposeFile                                                                                                                       |  SystemUpWaitTime   | ConsensusType | ChannelJoinDelay | BroadcastWaitTime | orderer0 | orderer1 | orderer2 |Orderer Specific Info|
       |   dc-base.yml                                                                                                                              |        0            |     solo      |      2           |      2            | orderer0 | orderer0 | orderer0 |                     |

bddtests/steps/bootstrap_impl.py

@@ -268,12 +268,12 @@ def step_impl(context, userName, transactionAlias, orderer):
     bootstrap_util.broadcastCreateChannelConfigTx(context=context, certAlias=None, composeService=orderer, user=user, configTxEnvelope=transaction)
 
 
-@when(u'user "{userName}" using cert alias "{certAlias}" connects to deliver function on orderer "{composeService}"')
-@Given(u'user "{userName}" using cert alias "{certAlias}" connects to deliver function on orderer "{composeService}"')
-def step_impl(context, userName, certAlias, composeService):
+@when(u'user "{userName}" using cert alias "{certAlias}" connects to deliver function on orderer "{composeService}" using port "{port}"')
+@Given(u'user "{userName}" using cert alias "{certAlias}" connects to deliver function on orderer "{composeService}" using port "{port}"')
+def step_impl(context, userName, certAlias, composeService, port):
     directory = bootstrap_util.getDirectory(context)
     user = directory.getUser(userName=userName)
-    user.connectToDeliverFunction(context, composeService, nodeAdminTuple=user.tags[certAlias])
+    user.connectToDeliverFunction(context, composeService, nodeAdminTuple=user.tags[certAlias], port=port)
 
 @when(u'user "{userName}" sends deliver a seek request on orderer "{composeService}" with properties')
 def step_impl(context, userName, composeService):
@@ -390,4 +390,4 @@ def step_impl(context, certAlias):
 def step_impl(context, command, service_name):
     assert "composition" in context, "No composition found in context"
     composition = context.composition
-    composition.issueCommand([command], [service_name])
+    composition.issueCommand([command], [service_name])

bddtests/steps/orderer_util.py

@@ -190,12 +190,13 @@ def closeStreams(self):
             deliverStreamHelper.send(None)
         self.abDeliversStreamHelperDict.clear()
 
-    def connectToDeliverFunction(self, context, composeService, nodeAdminTuple, timeout=1):
+    def connectToDeliverFunction(self, context, composeService, nodeAdminTuple, timeout=1, port=7050):
         'Connect to the deliver function and drain messages to associated orderer queue'
         assert not composeService in self.abDeliversStreamHelperDict, "Already connected to deliver stream on {0}".format(composeService)
         streamHelper = DeliverStreamHelper(directory=self.directory,
                                            ordererStub=self.getABStubForComposeService(context=context,
-                                                                                       composeService=composeService),
+                                                                                       composeService=composeService,
+                                                                                       port=port),
                                            entity=self, nodeAdminTuple=nodeAdminTuple)
         self.abDeliversStreamHelperDict[composeService] = streamHelper
         return streamHelper
@@ -220,15 +221,16 @@ def broadcastMessages(self, context, numMsgsToBroadcast, composeService, dataFun
         print("Done")
         assert counter == int(numMsgsToBroadcast), "counter = {0}, expected {1}".format(counter, numMsgsToBroadcast)
 
-    def getABStubForComposeService(self, context, composeService):
+    def getABStubForComposeService(self, context, composeService,port=7050):
         'Return a Stub for the supplied composeService, will cache'
         if composeService in self.atomicBroadcastStubsDict:
             return self.atomicBroadcastStubsDict[composeService]
         # Get the IP address of the server that the user registered on
         root_certificates = self.directory.getTrustedRootsForOrdererNetworkAsPEM()
-        ipAddress, port = bdd_test_util.getPortHostMapping(context.compose_containers, composeService, 7050)
+        peer_root_certificates = self.directory.getTrustedRootsForPeerNetworkAsPEM()
+        ipAddress, port_to_use = bdd_test_util.getPortHostMapping(context.compose_containers, composeService, port)
         # print("ipAddress in getABStubForComposeService == {0}:{1}".format(ipAddress, port))
-        channel = bdd_grpc_util.getGRPCChannel(ipAddress=ipAddress, port=port, root_certificates=root_certificates, ssl_target_name_override=composeService)
+        channel = bdd_grpc_util.getGRPCChannel(ipAddress=ipAddress, port=port_to_use, root_certificates="".join([root_certificates,peer_root_certificates]), ssl_target_name_override=composeService)
         newABStub = ab_pb2_grpc.AtomicBroadcastStub(channel)
         self.atomicBroadcastStubsDict[composeService] = newABStub
         return newABStub

core/peer/atomicbroadcast.go

@@ -0,0 +1,70 @@
+/*
+Copyright IBM Corp. 2017 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 peer
+
+import (
+	"runtime/debug"
+
+	"github.com/hyperledger/fabric/common/deliver"
+	"github.com/hyperledger/fabric/common/flogging"
+	"github.com/hyperledger/fabric/common/policies"
+	"github.com/hyperledger/fabric/orderer/common/broadcast"
+	"github.com/hyperledger/fabric/protos/common"
+	ab "github.com/hyperledger/fabric/protos/orderer"
+	logging "github.com/op/go-logging"
+)
+
+const pkgLogID = "common/peer"
+
+var logger *logging.Logger
+
+func init() {
+	logger = flogging.MustGetLogger(pkgLogID)
+}
+
+type server struct {
+	bh broadcast.Handler
+	dh deliver.Handler
+}
+
+// Broadcast is not implemented/supported on a peer
+func (s *server) Broadcast(srv ab.AtomicBroadcast_BroadcastServer) error {
+	return srv.Send(&ab.BroadcastResponse{
+		Status: common.Status_NOT_IMPLEMENTED,
+	})
+}
+
+// Deliver sends a stream of blocks to a client after commitment
+func (s *server) Deliver(srv ab.AtomicBroadcast_DeliverServer) error {
+	logger.Debugf("Starting new Deliver handler")
+	defer func() {
+		if r := recover(); r != nil {
+			logger.Criticalf("Deliver client triggered panic: %s\n%s", r, debug.Stack())
+		}
+		logger.Debugf("Closing Deliver stream")
+	}()
+	return s.dh.Handle(srv)
+}
+
+// NewAtomicBroadcastServer creates an ab.AtomicBroadcastServer based on the
+// ledger Reader. Broadcast is not implemented/supported on the peer.
+func NewAtomicBroadcastServer() ab.AtomicBroadcastServer {
+	s := &server{
+		dh: deliver.NewHandlerImpl(DeliverSupportManager{}, policies.ChannelReaders),
+		bh: broadcast.NewHandlerImpl(nil),
+	}
+	return s
+}

core/peer/peer.go

@@ -15,7 +15,11 @@ import (
 	"github.com/hyperledger/fabric/common/channelconfig"
 	"github.com/hyperledger/fabric/common/configtx"
 	configtxtest "github.com/hyperledger/fabric/common/configtx/test"
+	"github.com/hyperledger/fabric/common/deliver"
 	"github.com/hyperledger/fabric/common/flogging"
+	commonledger "github.com/hyperledger/fabric/common/ledger"
+	"github.com/hyperledger/fabric/common/ledger/blockledger"
+	"github.com/hyperledger/fabric/common/ledger/blockledger/file"
 	mockchannelconfig "github.com/hyperledger/fabric/common/mocks/config"
 	mockconfigtx "github.com/hyperledger/fabric/common/mocks/configtx"
 	mockpolicies "github.com/hyperledger/fabric/common/mocks/policies"
@@ -58,7 +62,8 @@ type chainSupport struct {
 	bundleSource *resourcesconfig.BundleSource
 	channelconfig.Resources
 	channelconfig.Application
-	ledger ledger.PeerLedger
+	ledger     ledger.PeerLedger
+	fileLedger *fileledger.FileLedger
 }
 
 var transientStoreFactory = &storeProvider{}
@@ -132,6 +137,17 @@ func (cs *chainSupport) GetMSPIDs(cid string) []string {
 	return GetMSPIDs(cid)
 }
 
+// Sequence passes through to the underlying configtx.Validator
+func (cs *chainSupport) Sequence() uint64 {
+	return cs.ConfigtxValidator().Sequence()
+}
+func (cs *chainSupport) Reader() blockledger.Reader {
+	return cs.fileLedger
+}
+func (cs *chainSupport) Errored() <-chan struct{} {
+	return nil
+}
+
 // chain is a local struct to manage objects in a chain
 type chain struct {
 	cs        *chainSupport
@@ -300,6 +316,7 @@ func createChain(cid string, ledger ledger.PeerLedger, cb *common.Block) error {
 	cs := &chainSupport{
 		Application: ac, // TODO, refactor as this is accessible through Manager
 		ledger:      ledger,
+		fileLedger:  fileledger.NewFileLedger(fileLedgerBlockStore{ledger}),
 	}
 
 	peerSingletonCallback := func(bundle *resourcesconfig.Bundle) {
@@ -684,3 +701,30 @@ func CreatePeerServer(listenAddress string,
 func GetPeerServer() comm.GRPCServer {
 	return peerServer
 }
+
+//
+//  Deliver service support structs for the peer
+//
+
+// DeliverSupportManager provides access to a channel for performing deliver
+type DeliverSupportManager struct {
+}
+
+func (dsm DeliverSupportManager) GetChain(chainID string) (deliver.Support, bool) {
+	channel, ok := chains.list[chainID]
+	return channel.cs, ok
+}
+
+// fileLedgerBlockStore implements the interface expected by
+// common/ledger/blockledger/file to interact with a file ledger for deliver
+type fileLedgerBlockStore struct {
+	ledger.PeerLedger
+}
+
+func (flbs fileLedgerBlockStore) AddBlock(*common.Block) error {
+	return nil
+}
+
+func (flbs fileLedgerBlockStore) RetrieveBlocks(startBlockNumber uint64) (commonledger.ResultsIterator, error) {
+	return flbs.GetBlocksIterator(startBlockNumber)
+}

peer/node/start.go

@@ -43,6 +43,7 @@ import (
 	"github.com/hyperledger/fabric/peer/version"
 	cb "github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/ledger/rwset"
+	ab "github.com/hyperledger/fabric/protos/orderer"
 	pb "github.com/hyperledger/fabric/protos/peer"
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
@@ -157,6 +158,11 @@ func serve(args []string) error {
 		grpclog.Fatalf("Failed to create ehub server: %v", err)
 	}
 
+	// create the peer's AtomicBroadcastServer, which supports deliver but not
+	// broadcast
+	abServer := peer.NewAtomicBroadcastServer()
+	ab.RegisterAtomicBroadcastServer(peerServer.Server(), abServer)
+
 	// enable the cache of chaincode info
 	ccprovider.EnableCCInfoCache()