[FAB-8756] Remove Event Hub

- Removed all references to the old EventHub
- Added an EventClient cache to InfraProvider
- Added reusable caching constructs:
  - LazyReference
  - LazyCache
  - FutureValue

Change-Id: I63b630bd4c339a88a263bc65b638b4b440b9db78
Signed-off-by: Bob Stasyszyn <Bob.Stasyszyn@securekey.com>

Author: bstasyszyn

pkg/client/channel/api.go

@@ -15,6 +15,7 @@ import (
 )
 
 // CCEvent contains the data for a chaincocde event
+// Deprecated since EventHub is replaced with EventService
 type CCEvent struct {
 	TxID        string
 	ChaincodeID string
@@ -24,6 +25,7 @@ type CCEvent struct {
 
 // Registration is a handle that is returned from a successful Register Chaincode Event.
 // This handle should be used in Unregister in order to unregister the event.
+// Deprecated since EventHub is replaced with EventService
 type Registration interface {
 }
 

pkg/client/channel/chclient.go

@@ -8,7 +8,6 @@ SPDX-License-Identifier: Apache-2.0
 package channel
 
 import (
-	"reflect"
 	"time"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/channel/invoke"
@@ -39,7 +38,7 @@ type Client struct {
 	context         context.Channel
 	membership      fab.ChannelMembership
 	transactor      fab.Transactor
-	eventHub        fab.EventHub
+	eventService    fab.EventService
 	greylist        *greylist.Filter
 	discoveryFilter fab.TargetFilter
 }
@@ -78,9 +77,9 @@ func New(channelProvider context.ChannelProvider, opts ...ClientOption) (*Client
 		return nil, errors.New("channel service not initialized")
 	}
 
-	eventHub, err := channelContext.ChannelService().EventHub()
+	eventService, err := channelContext.ChannelService().EventService()
 	if err != nil {
-		return nil, errors.WithMessage(err, "event hub creation failed")
+		return nil, errors.WithMessage(err, "event service creation failed")
 	}
 
 	transactor, err := channelContext.ChannelService().Transactor()
@@ -94,10 +93,10 @@ func New(channelProvider context.ChannelProvider, opts ...ClientOption) (*Client
 	}
 
 	channelClient := Client{
-		membership: membership,
-		transactor: transactor,
-		eventHub:   eventHub,
-		greylist:   greylistProvider,
+		membership:   membership,
+		transactor:   transactor,
+		eventService: eventService,
+		greylist:     greylistProvider,
 	}
 
 	for _, param := range opts {
@@ -189,11 +188,11 @@ func (cc *Client) prepareHandlerContexts(request Request, o opts) (*invoke.Reque
 	}
 
 	clientContext := &invoke.ClientContext{
-		Selection:  cc.context.SelectionService(),
-		Discovery:  cc.context.DiscoveryService(),
-		Membership: cc.membership,
-		Transactor: cc.transactor,
-		EventHub:   cc.eventHub,
+		Selection:    cc.context.SelectionService(),
+		Discovery:    cc.context.DiscoveryService(),
+		Membership:   cc.membership,
+		Transactor:   cc.transactor,
+		EventService: cc.eventService,
 	}
 
 	requestContext := &invoke.RequestContext{
@@ -240,45 +239,23 @@ func (cc *Client) addDefaultTimeout(timeOutType core.TimeoutType, options ...Opt
 	return options
 }
 
-// Close releases channel client resources (disconnects event hub etc.)
+// Close ...
+// TODO: This function should probably be deprecated since all
+// resources (including caches) are on the providers and will
+// be freed when Close() is called on the SDK.
 func (cc *Client) Close() error {
-	if cc.eventHub.IsConnected() == true {
-		return cc.eventHub.Disconnect()
-	}
-
 	return nil
 }
 
 // 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 *Client) RegisterChaincodeEvent(notify chan<- *CCEvent, chainCodeID string, eventID string) (Registration, error) {
-
-	if cc.eventHub.IsConnected() == false {
-		if err := cc.eventHub.Connect(); err != nil {
-			return nil, errors.WithMessage(err, "Event hub failed to connect")
-		}
-	}
-
+func (cc *Client) RegisterChaincodeEvent(chainCodeID string, eventFilter string) (fab.Registration, <-chan *fab.CCEvent, error) {
 	// Register callback for CE
-	rce := cc.eventHub.RegisterChaincodeEvent(chainCodeID, eventID, func(ce *fab.ChaincodeEvent) {
-		notify <- &CCEvent{ChaincodeID: ce.ChaincodeID, EventName: ce.EventName, TxID: ce.TxID, Payload: ce.Payload}
-	})
-
-	return rce, nil
+	return cc.eventService.RegisterChaincodeEvent(chainCodeID, eventFilter)
 }
 
 // UnregisterChaincodeEvent removes chain code event registration
-func (cc *Client) UnregisterChaincodeEvent(registration Registration) error {
-
-	switch regType := registration.(type) {
-
-	case *fab.ChainCodeCBE:
-		cc.eventHub.UnregisterChaincodeEvent(regType)
-	default:
-		return errors.Errorf("Unsupported registration type: %v", reflect.TypeOf(registration))
-	}
-
-	return nil
-
+func (cc *Client) UnregisterChaincodeEvent(registration fab.Registration) {
+	cc.eventService.Unregister(registration)
 }

pkg/client/channel/chclient_test.go

@@ -353,7 +353,7 @@ func TestOrdererStatusError(t *testing.T) {
 	testOrderer1 := fcmocks.NewMockOrderer("", make(chan *fab.SignedEnvelope))
 	orderers := []fab.Orderer{testOrderer1}
 	chClient := setupChannelClientWithNodes(peers, orderers, t)
-	chClient.eventHub = fcmocks.NewMockEventHub()
+	chClient.eventService = fcmocks.NewMockEventService()
 
 	testOrderer1.EnqueueSendBroadcastError(status.New(status.OrdererClientStatus,
 		status.ConnectionFailed.ToInt32(), testErrorMessage, nil))
@@ -371,22 +371,21 @@ func TestOrdererStatusError(t *testing.T) {
 
 func TestTransactionValidationError(t *testing.T) {
 	validationCode := pb.TxValidationCode_BAD_RWSET
-	mockEventHub := fcmocks.NewMockEventHub()
+	mockEventService := fcmocks.NewMockEventService()
 	testPeer1 := fcmocks.NewMockPeer("Peer1", "http://peer1.com")
 	peers := []fab.Peer{testPeer1}
 
 	go func() {
 		select {
-		case callback := <-mockEventHub.RegisteredTxCallbacks:
-			callback("txid", validationCode,
-				status.New(status.EventServerStatus, int32(validationCode), "test", nil))
+		case txStatusReg := <-mockEventService.TxStatusRegCh:
+			txStatusReg.Eventch <- &fab.TxStatusEvent{TxID: txStatusReg.TxID, TxValidationCode: validationCode}
 		case <-time.After(time.Second * 5):
 			t.Fatal("Timed out waiting for execute Tx to register event callback")
 		}
 	}()
 
 	chClient := setupChannelClient(peers, t)
-	chClient.eventHub = mockEventHub
+	chClient.eventService = mockEventService
 	response, err := chClient.Execute(Request{ChaincodeID: "test", Fcn: "invoke",
 		Args: [][]byte{[]byte("move"), []byte("a"), []byte("b"), []byte("1")}})
 	assert.Nil(t, response.Payload, "Expected nil result on failed execute operation")

pkg/client/channel/invoke/api.go

@@ -47,12 +47,12 @@ type Handler interface {
 
 //ClientContext contains context parameters for handler execution
 type ClientContext struct {
-	CryptoSuite core.CryptoSuite
-	Discovery   fab.DiscoveryService
-	Selection   fab.SelectionService
-	Membership  fab.ChannelMembership
-	Transactor  fab.Transactor
-	EventHub    fab.EventHub
+	CryptoSuite  core.CryptoSuite
+	Discovery    fab.DiscoveryService
+	Selection    fab.SelectionService
+	Membership   fab.ChannelMembership
+	Transactor   fab.Transactor
+	EventService fab.EventService
 }
 
 //RequestContext contains request, opts, response parameters for handler execution

pkg/client/channel/invoke/txnhandler.go

@@ -10,14 +10,15 @@ import (
 	"bytes"
 	"time"
 
+	"github.com/hyperledger/fabric-sdk-go/pkg/errors/status"
 	"github.com/pkg/errors"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/context/api/fab"
-	"github.com/hyperledger/fabric-sdk-go/pkg/errors/status"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/peer"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/txn"
 	"github.com/hyperledger/fabric-sdk-go/pkg/logging"
 	"github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/common"
+	pb "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/peer"
 )
 
 var logger = logging.NewLogger("fabsdk/client")
@@ -138,32 +139,28 @@ type CommitTxHandler struct {
 
 //Handle handles commit tx
 func (c *CommitTxHandler) Handle(requestContext *RequestContext, clientContext *ClientContext) {
-
-	//Connect to Event hub if not yet connected
-	if clientContext.EventHub.IsConnected() == false {
-		err := clientContext.EventHub.Connect()
-		if err != nil {
-			requestContext.Error = err
-			return
-		}
-	}
-
 	txnID := requestContext.Response.TransactionID
 
 	//Register Tx event
-	statusNotifier := txn.RegisterStatus(txnID, clientContext.EventHub)
-	_, err := createAndSendTransaction(clientContext.Transactor, requestContext.Response.Proposal, requestContext.Response.Responses)
+	reg, statusNotifier, err := clientContext.EventService.RegisterTxStatusEvent(string(txnID)) // TODO: Change func to use TransactionID instead of string
+	if err != nil {
+		requestContext.Error = errors.Wrap(err, "error registering for TxStatus event")
+		return
+	}
+	defer clientContext.EventService.Unregister(reg)
+
+	_, err = createAndSendTransaction(clientContext.Transactor, requestContext.Response.Proposal, requestContext.Response.Responses)
 	if err != nil {
 		requestContext.Error = errors.Wrap(err, "CreateAndSendTransaction failed")
 		return
 	}
 
 	select {
-	case result := <-statusNotifier:
-		requestContext.Response.TxValidationCode = result.Code
+	case txStatus := <-statusNotifier:
+		requestContext.Response.TxValidationCode = txStatus.TxValidationCode
 
-		if result.Error != nil {
-			requestContext.Error = result.Error
+		if txStatus.TxValidationCode != pb.TxValidationCode_VALID {
+			requestContext.Error = status.New(status.EventServerStatus, int32(txStatus.TxValidationCode), "received invalid transaction", nil)
 			return
 		}
 	case <-time.After(requestContext.Opts.Timeout):

pkg/client/channel/invoke/txnhandler_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/context"
 	"github.com/hyperledger/fabric-sdk-go/pkg/context/api/fab"
 	fcmocks "github.com/hyperledger/fabric-sdk-go/pkg/fab/mocks"
+	pb "github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/peer"
 )
 
 const (
@@ -65,13 +66,13 @@ func TestExecuteTxHandlerSuccess(t *testing.T) {
 	clientContext := setupChannelClientContext(nil, nil, []fab.Peer{mockPeer1, mockPeer2}, t)
 
 	//Prepare mock eventhub
-	mockEventHub := fcmocks.NewMockEventHub()
-	clientContext.EventHub = mockEventHub
+	mockEventService := fcmocks.NewMockEventService()
+	clientContext.EventService = mockEventService
 
 	go func() {
 		select {
-		case callback := <-mockEventHub.RegisteredTxCallbacks:
-			callback("txid", 0, nil)
+		case txStatusReg := <-mockEventService.TxStatusRegCh:
+			txStatusReg.Eventch <- &fab.TxStatusEvent{TxID: txStatusReg.TxID, TxValidationCode: pb.TxValidationCode_VALID}
 		case <-time.After(requestContext.Opts.Timeout):
 			t.Fatal("Execute handler : time out not expected")
 		}

pkg/client/resmgmt/resmgmt.go

@@ -21,6 +21,7 @@ import (
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/context"
 	contextImpl "github.com/hyperledger/fabric-sdk-go/pkg/context"
 	"github.com/hyperledger/fabric-sdk-go/pkg/errors/multi"
+	"github.com/hyperledger/fabric-sdk-go/pkg/errors/status"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/orderer"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/peer"
 	"github.com/hyperledger/fabric-sdk-go/pkg/fab/resource"
@@ -506,20 +507,17 @@ func (rc *Client) sendCCProposal(ccProposalType chaincodeProposalType, channelID
 		return errors.WithMessage(err, "sending deploy transaction proposal failed")
 	}
 
-	eventHub, err := channelService.EventHub()
+	eventService, err := channelService.EventService()
 	if err != nil {
-		return errors.WithMessage(err, "Unable to get EventHub")
-	}
-	if eventHub.IsConnected() == false {
-		err := eventHub.Connect()
-		if err != nil {
-			return err
-		}
-		defer eventHub.Disconnect()
+		return errors.WithMessage(err, "unable to get event service")
 	}
 
 	// Register for commit event
-	statusNotifier := txn.RegisterStatus(tp.TxnID, eventHub)
+	reg, statusNotifier, err := eventService.RegisterTxStatusEvent(string(tp.TxnID))
+	if err != nil {
+		return errors.WithMessage(err, "error registering for TxStatus event")
+	}
+	defer eventService.Unregister(reg)
 
 	transactionRequest := fab.TransactionRequest{
 		Proposal:          tp,
@@ -535,11 +533,11 @@ func (rc *Client) sendCCProposal(ccProposalType chaincodeProposalType, channelID
 	}
 
 	select {
-	case result := <-statusNotifier:
-		if result.Error == nil {
+	case txStatus := <-statusNotifier:
+		if txStatus.TxValidationCode == pb.TxValidationCode_VALID {
 			return nil
 		}
-		return errors.WithMessage(result.Error, "instantiateOrUpgradeCC failed")
+		return status.New(status.EventServerStatus, int32(txStatus.TxValidationCode), "instantiateOrUpgradeCC failed", nil)
 	case <-time.After(timeout):
 		return errors.New("instantiateOrUpgradeCC timeout")
 	}