re-enable sending of chaincode events

Chaincode events were originally stored in TransactionResults
in NonHashData(see fabric.proto). TransactionResults was removed as
part of removing error transactions from ledger which broke
sending of chaincode events. This is fixed in the patch which
stores chaincode events directly in NonHashData. Sample is also
enhanced.

Change-Id: Ifee8facb2e4760e878f15716af0e48346536c476
Signed-off-by: Srinivasan Muralidharan <[email protected]>

Author: Srinivasan Muralidharan

core/ledger/ledger.go

@@ -159,7 +159,31 @@ func (ledger *Ledger) CommitTxBatch(id interface{}, transactions []*protos.Trans
 	writeBatch := gorocksdb.NewWriteBatch()
 	defer writeBatch.Destroy()
 	block := protos.NewBlock(transactions, metadata)
-	block.NonHashData = &protos.NonHashData{}
+
+	ccEvents := []*protos.ChaincodeEvent{}
+
+	if transactionResults != nil {
+		ccEvents = make([]*protos.ChaincodeEvent, len(transactionResults))
+		for i := 0; i < len(transactionResults); i++ {
+			if transactionResults[i].ChaincodeEvent != nil {
+				ccEvents[i] = transactionResults[i].ChaincodeEvent
+			} else {
+				//We need the index so we can map the chaincode
+				//event to the transaction that generated it.
+				//Hence need an entry for cc event even if one
+				//wasn't generated for the transaction. We cannot
+				//use a nil cc event as protobuf does not like
+				//elements of a repeated array to be nil.
+				//
+				//We should discard empty events without chaincode
+				//ID when sending out events.
+				ccEvents[i] = &protos.ChaincodeEvent{}
+			}
+		}
+	}
+
+	//store chaincode events directly in NonHashData. This will likely change in New Consensus where we can move them to Transaction
+	block.NonHashData = &protos.NonHashData{ChaincodeEvents: ccEvents}
 	newBlockNumber, err := ledger.blockchain.addPersistenceChangesForNewBlock(context.TODO(), block, stateHash, writeBatch)
 	if err != nil {
 		ledger.resetForNextTxGroup(false)
@@ -180,6 +204,10 @@ func (ledger *Ledger) CommitTxBatch(id interface{}, transactions []*protos.Trans
 	ledger.blockchain.blockPersistenceStatus(true)
 
 	sendProducerBlockEvent(block)
+
+	//send chaincode events from transaction results
+	sendChaincodeEvents(transactionResults)
+
 	if len(transactionResults) != 0 {
 		ledgerLogger.Debug("There were some erroneous transactions. We need to send a 'TX rejected' message here.")
 	}
@@ -491,3 +519,16 @@ func sendProducerBlockEvent(block *protos.Block) {
 
 	producer.Send(producer.CreateBlockEvent(block))
 }
+
+//send chaincode events created by transactions
+func sendChaincodeEvents(trs []*protos.TransactionResult) {
+	if trs != nil {
+		for _, tr := range trs {
+			//we store empty chaincode events in the protobuf repeated array to make protobuf happy.
+			//when we replay off a block ignore empty events
+			if tr.ChaincodeEvent != nil && tr.ChaincodeEvent.ChaincodeID != "" {
+				producer.Send(producer.CreateChaincodeEvent(tr.ChaincodeEvent))
+			}
+		}
+	}
+}

events/producer/handler.go

@@ -64,7 +64,7 @@ func (d *handler) register(iMsg []*pb.Interest) error {
 	//if successfully done, continue....
 	for _, v := range iMsg {
 		if err := registerHandler(v, d); err != nil {
-			producerLogger.Errorf("could not register %s", v)
+			producerLogger.Errorf("could not register %s: %s", v, err)
 			continue
 		}
 		d.addInterest(v)

examples/events/block-listener/README.md

@@ -1,11 +1,11 @@
 # What is block-listener
-block-listener.go will connect to a peer and receive blocks. For every transaction in the block, it will test the ErrorCode field and print success/failure.
+block-listener.go will connect to a peer and receive blocks events, transaction rejection events and chaincode events (if a chaincode emits events).
 
 # To Run
 ```sh
 1. go build
 
-2. ./block-listener -events-address=< event address >
+2. ./block-listener -events-address=< event address > -listen-to-rejections=< true | false > -events-from-chaincode=< chaincode ID >
 ```
 
 # Example with PBFT
@@ -21,9 +21,11 @@ docker run --rm -it -e CORE_VM_ENDPOINT=http://172.17.0.1:2375 -e CORE_PEER_ID=v
 docker run --rm -it -e CORE_VM_ENDPOINT=http://172.17.0.1:2375 -e CORE_PEER_ID=vp3 -e CORE_PEER_ADDRESSAUTODETECT=true -e CORE_PEER_DISCOVERY_ROOTNODE=172.17.0.2:7051 -e CORE_PEER_VALIDATOR_CONSENSUS_PLUGIN=pbft hyperledger/fabric-peer peer node start
 
 ## Attach event client to a Peer
-./block-listener -events-address=172.17.0.2:7054
+```sh
+./block-listener -events-address=172.17.0.2:7053 -listen-to-rejections=true
+```
 
-Event client should output "Event Address: 172.17.0.2:7054" and wait for events.
+Event client should output "Event Address: 172.17.0.2:7053" and wait for events.
 
 ## Create a deploy transaction
 Submit a transaction to deploy chaincode_example02.
@@ -50,3 +52,31 @@ CORE_PEER_ADDRESS=172.17.0.2:7051 peer chaincode invoke -n 1edd7021ab71b766f4928
 ```
 
 Notice error transaction in events client.
+
+# Tesing chaincode events
+Chaincode github.com/hyperledger/fabric/examples/chaincode/go/eventsender can be used to test event sender.
+## Deploy eventsender chaincode
+Stop the event listener and restart it as follows  
+
+```
+CORE_PEER_ADDRESS=172.17.0.2:7051 ./peer chaincode deploy -p github.com/hyperledger/fabric/examples/chaincode/go/eventsender -c '{"Function":"init", "Args":[]}'
+```
+
+```
+Note the chaincode ID of the eventsender chaincode. This will be used in the commands below.
+```
+## Restart event listener
+Stop the event listener if running and restart it with `-events-from-chaincode` option
+
+```sh
+./block-listener -events-address=172.17.0.2:7053 -listen-to-rejections=true -events-from-chaincode=< event sender chaincode ID>
+```
+
+
+##Send an invoke request to event sender
+
+```sh
+CORE_PEER_ADDRESS=172.17.0.2:7051 ./peer chaincode invoke -n < eventsender chaincode ID > -c '{"Function":"greet", "Args":["hello","world"]}'
+```
+
+Note the output from the event listener terminal showing a chaincode event from the event sender chaincode in addition to the block event generated by the transaction.

examples/events/block-listener/block-listener.go

@@ -31,11 +31,23 @@ import (
 type adapter struct {
 	notfy              chan *pb.Event_Block
 	rejected           chan *pb.Event_Rejection
+	cEvent             chan *pb.Event_ChaincodeEvent
 	listenToRejections bool
+	chaincodeID        string
 }
 
 //GetInterestedEvents implements consumer.EventAdapter interface for registering interested events
 func (a *adapter) GetInterestedEvents() ([]*pb.Interest, error) {
+	if a.chaincodeID != "" {
+		return []*pb.Interest{
+			{EventType: pb.EventType_BLOCK},
+			{EventType: pb.EventType_REJECTION},
+			{EventType: pb.EventType_CHAINCODE,
+				RegInfo: &pb.Interest_ChaincodeRegInfo{
+					ChaincodeRegInfo: &pb.ChaincodeReg{
+						ChaincodeID: a.chaincodeID,
+						EventName:   ""}}}}, nil
+	}
 	return []*pb.Interest{{EventType: pb.EventType_BLOCK}, {EventType: pb.EventType_REJECTION}}, nil
 }
 
@@ -49,6 +61,10 @@ func (a *adapter) Recv(msg *pb.Event) (bool, error) {
 		a.rejected <- o
 		return true, nil
 	}
+	if o, e := msg.Event.(*pb.Event_ChaincodeEvent); e {
+		a.cEvent <- o
+		return true, nil
+	}
 	a.notfy <- nil
 	return false, nil
 }
@@ -59,12 +75,12 @@ func (a *adapter) Disconnected(err error) {
 	os.Exit(1)
 }
 
-func createEventClient(eventAddress string, listenToRejections bool) *adapter {
+func createEventClient(eventAddress string, listenToRejections bool, cid string) *adapter {
 	var obcEHClient *consumer.EventsClient
 
 	done := make(chan *pb.Event_Block)
 	reject := make(chan *pb.Event_Rejection)
-	adapter := &adapter{notfy: done, rejected: reject, listenToRejections: listenToRejections}
+	adapter := &adapter{notfy: done, rejected: reject, listenToRejections: listenToRejections, chaincodeID: cid, cEvent: make(chan *pb.Event_ChaincodeEvent)}
 	obcEHClient = consumer.NewEventsClient(eventAddress, adapter)
 	if err := obcEHClient.Start(); err != nil {
 		fmt.Printf("could not start chat %s\n", err)
@@ -78,13 +94,15 @@ func createEventClient(eventAddress string, listenToRejections bool) *adapter {
 func main() {
 	var eventAddress string
 	var listenToRejections bool
+	var chaincodeID string
 	flag.StringVar(&eventAddress, "events-address", "0.0.0.0:7053", "address of events server")
 	flag.BoolVar(&listenToRejections, "listen-to-rejections", false, "whether to listen to rejection events")
+	flag.StringVar(&chaincodeID, "events-from-chaincode", "", "listen to events from given chaincode")
 	flag.Parse()
 
 	fmt.Printf("Event Address: %s\n", eventAddress)
 
-	a := createEventClient(eventAddress, listenToRejections)
+	a := createEventClient(eventAddress, listenToRejections, chaincodeID)
 	if a == nil {
 		fmt.Printf("Error creating event client\n")
 		return
@@ -106,6 +124,12 @@ func main() {
 			fmt.Printf("Received rejected transaction\n")
 			fmt.Printf("--------------\n")
 			fmt.Printf("Transaction error:\n%s\t%s\n", r.Rejection.Tx.Uuid, r.Rejection.ErrorMsg)
+		case ce := <-a.cEvent:
+			fmt.Printf("\n")
+			fmt.Printf("\n")
+			fmt.Printf("Received chaincode event\n")
+			fmt.Printf("------------------------\n")
+			fmt.Printf("Chaincode Event:%v\n", ce)
 		}
 	}
 }

protos/fabric.pb.go

@@ -110,8 +110,11 @@ message BlockchainInfo {
 // the block hash when verifying the blockchain.
 // localLedgerCommitTimestamp - The time at which the block was added
 // to the ledger on the local peer.
+// chaincodeEvent - is an array ChaincodeEvents, one per transaction in the
+// block
 message NonHashData {
     google.protobuf.Timestamp localLedgerCommitTimestamp = 1;
+    repeated ChaincodeEvent chaincodeEvents = 2;
 }
 
 // Interface exported by the server.