[FAB-8203] Channel Client - EventHub not connected

Change-Id: Ia13d54bd08783629fe8e919330392d2d5a8ebd64 Signed-off-by: Sandra Vrtikapa <sandra.vrtikapa@securekey.com>
hyperledger · Feb 12, 2018 · 1fd4d4d · 1fd4d4d
1 parent d5a5917
commit 1fd4d4d
Show file tree

Hide file tree

Showing 5 changed files with 64 additions and 6 deletions.
diff --git a/api/apitxn/chclient/chclient.go b/api/apitxn/chclient/chclient.go
@@ -77,7 +77,7 @@ type ChannelClient interface {
 	// RegisterChaincodeEvent registers chain code event
 	// @param {chan bool} channel which receives event details when the event is complete
 	// @returns {object}  object handle that should be used to unregister
-	RegisterChaincodeEvent(notify chan<- *CCEvent, chainCodeID string, eventID string) Registration
+	RegisterChaincodeEvent(notify chan<- *CCEvent, chainCodeID string, eventID string) (Registration, error)
 
 	// UnregisterChaincodeEvent unregisters chain code event
 	UnregisterChaincodeEvent(registration Registration) error

diff --git a/pkg/fabric-txn/chclient/chclient.go b/pkg/fabric-txn/chclient/chclient.go
@@ -190,14 +190,20 @@ func (cc *ChannelClient) Close() error {
 // RegisterChaincodeEvent registers chain code event
 // @param {chan bool} channel which receives event details when the event is complete
 // @returns {object} object handle that should be used to unregister
-func (cc *ChannelClient) RegisterChaincodeEvent(notify chan<- *chclient.CCEvent, chainCodeID string, eventID string) chclient.Registration {
+func (cc *ChannelClient) RegisterChaincodeEvent(notify chan<- *chclient.CCEvent, chainCodeID string, eventID string) (chclient.Registration, error) {
+
+	if cc.eventHub.IsConnected() == false {
+		if err := cc.eventHub.Connect(); err != nil {
+			return nil, errors.WithMessage(err, "Event hub failed to connect")
+		}
+	}
 
 	// Register callback for CE
 	rce := cc.eventHub.RegisterChaincodeEvent(chainCodeID, eventID, func(ce *fab.ChaincodeEvent) {
 		notify <- &chclient.CCEvent{ChaincodeID: ce.ChaincodeID, EventName: ce.EventName, TxID: ce.TxID, Payload: ce.Payload}
 	})
 
-	return rce
+	return rce, nil
 }
 
 // UnregisterChaincodeEvent removes chain code event registration

diff --git a/test/integration/e2e/end_to_end.go b/test/integration/e2e/end_to_end.go
@@ -123,7 +123,10 @@ func Run(t *testing.T, configOpt apiconfig.ConfigProvider, sdkOpts ...fabsdk.Opt
 
 	// Register chaincode event (pass in channel which receives event details when the event is complete)
 	notifier := make(chan *chclient.CCEvent)
-	rce := chClient.RegisterChaincodeEvent(notifier, ccID, eventID)
+	rce, err := chClient.RegisterChaincodeEvent(notifier, ccID, eventID)
+	if err != nil {
+		t.Fatalf("Failed to register cc event: %s", err)
+	}
 
 	// Move funds
 	response, err = chClient.Execute(chclient.Request{ChaincodeID: ccID, Fcn: "invoke", Args: integration.ExampleCCTxArgs()})

diff --git a/test/integration/e2e/no_orderer_config.go b/test/integration/e2e/no_orderer_config.go
@@ -59,7 +59,10 @@ func runWithNoOrdererConfig(t *testing.T, configOpt apiconfig.ConfigProvider, sd
 
 	// Register chaincode event (pass in channel which receives event details when the event is complete)
 	notifier := make(chan *chclient.CCEvent)
-	rce := chClient.RegisterChaincodeEvent(notifier, ccID, eventID)
+	rce, err := chClient.RegisterChaincodeEvent(notifier, ccID, eventID)
+	if err != nil {
+		t.Fatalf("Failed to register cc event: %s", err)
+	}
 
 	// Move funds
 	response, err = chClient.Execute(chclient.Request{ChaincodeID: ccID, Fcn: "invoke", Args: integration.ExampleCCTxArgs()})

diff --git a/test/integration/sdk/channel_client_test.go b/test/integration/sdk/channel_client_test.go
@@ -100,6 +100,15 @@ func TestChannelClient(t *testing.T) {
 	// Test invocation of custom handler
 	testInvokeHandler(chainCodeID, chClient, t)
 
+	// Test receive event using separate client
+	listener, err := sdk.NewClient(fabsdk.WithUser("User1")).Channel(testSetup.ChannelID)
+	if err != nil {
+		t.Fatalf("Failed to create new channel client: %s", err)
+	}
+	defer listener.Close()
+
+	testChaincodeEventListener(chainCodeID, chClient, listener, t)
+
 	// Release channel client resources
 	err = chClient.Close()
 	if err != nil {
@@ -245,7 +254,10 @@ func testChaincodeEvent(ccID string, chClient chclient.ChannelClient, t *testing
 
 	// Register chaincode event (pass in channel which receives event details when the event is complete)
 	notifier := make(chan *chclient.CCEvent)
-	rce := chClient.RegisterChaincodeEvent(notifier, ccID, eventID)
+	rce, err := chClient.RegisterChaincodeEvent(notifier, ccID, eventID)
+	if err != nil {
+		t.Fatalf("Failed to register cc event: %s", err)
+	}
 
 	response, err := chClient.Execute(chclient.Request{ChaincodeID: ccID, Fcn: "invoke", Args: integration.ExampleCCTxArgs()})
 	if err != nil {
@@ -269,3 +281,37 @@ func testChaincodeEvent(ccID string, chClient chclient.ChannelClient, t *testing
 	}
 
 }
+
+func testChaincodeEventListener(ccID string, chClient chclient.ChannelClient, listener chclient.ChannelClient, t *testing.T) {
+
+	eventID := "test([a-zA-Z]+)"
+
+	// Register chaincode event (pass in channel which receives event details when the event is complete)
+	notifier := make(chan *chclient.CCEvent)
+	rce, err := listener.RegisterChaincodeEvent(notifier, ccID, eventID)
+	if err != nil {
+		t.Fatalf("Failed to register cc event: %s", err)
+	}
+
+	response, err := chClient.Execute(chclient.Request{ChaincodeID: ccID, Fcn: "invoke", Args: integration.ExampleCCTxArgs()})
+	if err != nil {
+		t.Fatalf("Failed to move funds: %s", err)
+	}
+
+	select {
+	case ccEvent := <-notifier:
+		t.Logf("Received cc event: %s", ccEvent)
+		if ccEvent.TxID != response.TransactionID.ID {
+			t.Fatalf("CCEvent(%s) and Execute(%s) transaction IDs don't match", ccEvent.TxID, response.TransactionID.ID)
+		}
+	case <-time.After(time.Second * 20):
+		t.Fatalf("Did NOT receive CC for eventId(%s)\n", eventID)
+	}
+
+	// Unregister chain code event using registration handle
+	err = listener.UnregisterChaincodeEvent(rce)
+	if err != nil {
+		t.Fatalf("Unregister cc event failed: %s", err)
+	}
+
+}