func (eventHub *EventHub) notifyChaincodeRegistrants(channelID string, ccEvent *pb.ChaincodeEvent, patternMatch bool) {
	cbeArray := eventHub.getChaincodeRegistrants(ccEvent.ChaincodeId)
	if len(cbeArray) <= 0 {
		logger.Debugf("No event registration for ccid %s \n", ccEvent.ChaincodeId)
	}
	for _, v := range cbeArray {
		match := v.EventNameFilter == ccEvent.EventName
		if !match && patternMatch {
			match, _ = regexp.MatchString(v.EventNameFilter, ccEvent.EventName)
		}
		if match {
			callback := v.CallbackFunc
			if callback != nil {
				callback(&fab.ChaincodeEvent{
					ChaincodeID: ccEvent.ChaincodeId,
					TxID:        ccEvent.TxId,
					EventName:   ccEvent.EventName,
					Payload:     ccEvent.Payload,
					ChannelID:   channelID,
				})
			}
		}
	}
}
Utility function to fire callbacks for chaincode registrants
package events
import (
	"reflect"
	"regexp"
	"sync"
	"time"
	"github.com/golang/protobuf/proto"
	"github.com/spf13/cast"
	"google.golang.org/grpc/keepalive"
	"github.com/hyperledger/fabric-sdk-go/api/apiconfig"
	fab "github.com/hyperledger/fabric-sdk-go/api/apifabclient"
	common "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"
	cnsmr "github.com/hyperledger/fabric-sdk-go/internal/github.com/hyperledger/fabric/events/consumer"
	"github.com/hyperledger/fabric-sdk-go/pkg/config/urlutil"
	"github.com/hyperledger/fabric-sdk-go/pkg/errors/status"
	consumer "github.com/hyperledger/fabric-sdk-go/pkg/fabric-client/events/consumer"
	"github.com/hyperledger/fabric-sdk-go/pkg/logging"
	"github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/core/ledger/util"
	"github.com/hyperledger/fabric-sdk-go/third_party/github.com/hyperledger/fabric/protos/utils"
	"github.com/pkg/errors"
	"github.com/hyperledger/fabric-sdk-go/internal/github.com/tjfoc/gmsm/sm2"
)
var logger = logging.NewLogger("fabric_sdk_go")
type EventHub struct {
	mtx sync.RWMutex
	chaincodeRegistrants sync.Map
	blockRegistrants []func(*common.Block)
	txRegistrants sync.Map
	peerAddr string
	peerTLSCertificate *sm2.Certificate
	peerTLSServerHostOverride string
	grpcClient fab.EventsClient
	connected bool
	interestedEvents []*pb.Interest
	eventsClientFactory eventClientFactory
	provider fab.ProviderContext
	identity fab.IdentityContext
	kap      keepalive.ClientParameters
	failFast bool
}
type eventClientFactory interface {
	newEventsClient(provider fab.ProviderContext, identity fab.IdentityContext, peerAddress string, certificate *sm2.Certificate, serverHostOverride string, regTimeout time.Duration, adapter cnsmr.EventAdapter, kap keepalive.ClientParameters, failFast bool) (fab.EventsClient, error)
}
type consumerClientFactory struct{}
func (ccf *consumerClientFactory) newEventsClient(provider fab.ProviderContext, identity fab.IdentityContext, peerAddress string, certificate *sm2.Certificate, serverHostOverride string,
	regTimeout time.Duration, adapter cnsmr.EventAdapter, kap keepalive.ClientParameters, failFast bool) (fab.EventsClient, error) {
	return consumer.NewEventsClient(provider, identity, peerAddress, certificate, serverHostOverride, regTimeout, adapter, kap, failFast)
}
type Context struct {
	fab.ProviderContext
	fab.IdentityContext
}
func New(ctx Context) (*EventHub, error) {
	eventHub := EventHub{
		blockRegistrants:    nil,
		interestedEvents:    nil,
		eventsClientFactory: &consumerClientFactory{},
		provider:            ctx.ProviderContext,
		identity:            ctx.IdentityContext,
	}
	eventHub.RegisterBlockEvent(eventHub.txCallback)
	return &eventHub, nil
}
func FromConfig(ctx Context, peerCfg *apiconfig.PeerConfig) (*EventHub, error) {
	eventHub, err := New(ctx)
	if err != nil {
		return nil, err
	}
	serverHostOverride := ""
	if str, ok := peerCfg.GRPCOptions["ssl-target-name-override"].(string); ok {
		serverHostOverride = str
	}
	eventHub.peerAddr = peerCfg.EventURL
	if urlutil.IsTLSEnabled(eventHub.peerAddr) {
		eventHub.peerTLSCertificate, err = peerCfg.TLSCACerts.TLSCert()
		if err != nil {
			return nil, err
		}
	}
	eventHub.peerTLSServerHostOverride = serverHostOverride
	eventHub.kap = getKeepAliveOptions(peerCfg)
	eventHub.failFast = getFailFast(peerCfg)
	return eventHub, nil
}
func getFailFast(peerCfg *apiconfig.PeerConfig) bool {
	var failFast = true  
	if ff, ok := peerCfg.GRPCOptions["fail-fast"].(bool); ok {
		failFast = cast.ToBool(ff)
	}
	return failFast
}
func getKeepAliveOptions(peerCfg *apiconfig.PeerConfig) keepalive.ClientParameters {
	var kap keepalive.ClientParameters
	if kaTime, ok := peerCfg.GRPCOptions["keep-alive-time"]; ok {
		kap.Time = cast.ToDuration(kaTime)
	}
	if kaTimeout, ok := peerCfg.GRPCOptions["keep-alive-timeout"]; ok {
		kap.Timeout = cast.ToDuration(kaTimeout)
	}
	if kaPermit, ok := peerCfg.GRPCOptions["keep-alive-permit"]; ok {
		kap.PermitWithoutStream = cast.ToBool(kaPermit)
	}
	return kap
}
func (eventHub *EventHub) SetInterests(block bool) {
	eventHub.mtx.Lock()
	defer eventHub.mtx.Unlock()
	eventHub.interestedEvents = make([]*pb.Interest, 0)
	eventHub.blockRegistrants = make([]func(*common.Block), 0)
	if block {
		eventHub.blockRegistrants = append(eventHub.blockRegistrants, eventHub.txCallback)
		eventHub.interestedEvents = append(eventHub.interestedEvents, &pb.Interest{EventType: pb.EventType_BLOCK})
	}
}
func (eventHub *EventHub) Disconnect() error {
	eventHub.mtx.Lock()
	defer eventHub.mtx.Unlock()
	if !eventHub.connected {
		return nil
	}
	err := eventHub.grpcClient.UnregisterAsync(eventHub.interestedEvents)
	if err != nil {
		return errors.WithMessage(err, "event client UnregisterAsync failed")
	}
	err = eventHub.grpcClient.Stop()
	if err != nil {
		return errors.WithMessage(err, "event client stop failed")
	}
	eventHub.connected = false
	return nil
}
func (eventHub *EventHub) RegisterBlockEvent(callback func(*common.Block)) {
	eventHub.mtx.Lock()
	defer eventHub.mtx.Unlock()
	eventHub.blockRegistrants = append(eventHub.blockRegistrants, callback)
	if len(eventHub.blockRegistrants) == 1 {
		eventHub.interestedEvents = append(eventHub.interestedEvents, &pb.Interest{EventType: pb.EventType_BLOCK})
	}
}
func (eventHub *EventHub) UnregisterBlockEvent(callback func(*common.Block)) {
	eventHub.mtx.Lock()
	defer eventHub.mtx.Unlock()
	f1 := reflect.ValueOf(callback)
	for i := range eventHub.blockRegistrants {
		f2 := reflect.ValueOf(eventHub.blockRegistrants[i])
		if f1.Pointer() == f2.Pointer() {
			eventHub.blockRegistrants = append(eventHub.blockRegistrants[:i], eventHub.blockRegistrants[i+1:]...)
			break
		}
	}
	if len(eventHub.blockRegistrants) < 1 {
		blockEventInterest := pb.Interest{EventType: pb.EventType_BLOCK}
		eventHub.grpcClient.UnregisterAsync([]*pb.Interest{&blockEventInterest})
		for i, v := range eventHub.interestedEvents {
			if *v == blockEventInterest {
				eventHub.interestedEvents = append(eventHub.interestedEvents[:i], eventHub.interestedEvents[i+1:]...)
			}
		}
	}
}
func (eventHub *EventHub) addChaincodeInterest(ChaincodeID string, EventName string) {
	ccInterest := &pb.Interest{
		EventType: pb.EventType_CHAINCODE,
		RegInfo: &pb.Interest_ChaincodeRegInfo{
			ChaincodeRegInfo: &pb.ChaincodeReg{
				ChaincodeId: ChaincodeID,
				EventName:   EventName,
			},
		},
	}
	eventHub.interestedEvents = append(eventHub.interestedEvents, ccInterest)
	if eventHub.IsConnected() {
		eventHub.grpcClient.RegisterAsync([]*pb.Interest{ccInterest})
	}
}
func (eventHub *EventHub) removeChaincodeInterest(ChaincodeID string, EventName string) {
	ccInterest := &pb.Interest{
		EventType: pb.EventType_CHAINCODE,
		RegInfo: &pb.Interest_ChaincodeRegInfo{
			ChaincodeRegInfo: &pb.ChaincodeReg{
				ChaincodeId: ChaincodeID,
				EventName:   EventName,
			},
		},
	}
	for i, v := range eventHub.interestedEvents {
		if v.EventType == ccInterest.EventType && *(v.GetChaincodeRegInfo()) == *(ccInterest.GetChaincodeRegInfo()) {
			eventHub.interestedEvents = append(eventHub.interestedEvents[:i], eventHub.interestedEvents[i+1:]...)
		}
	}
	if eventHub.IsConnected() {
		eventHub.grpcClient.UnregisterAsync([]*pb.Interest{ccInterest})
	}
}
func (eventHub *EventHub) SetPeerAddr(peerURL string, peerTLSCertificate *sm2.Certificate, peerTLSServerHostOverride string) {
	eventHub.peerAddr = peerURL
	eventHub.peerTLSCertificate = peerTLSCertificate
	eventHub.peerTLSServerHostOverride = peerTLSServerHostOverride
}
func (eventHub *EventHub) IsConnected() bool {
	return eventHub.connected
}
func (eventHub *EventHub) Connect() error {
	eventHub.mtx.Lock()
	defer eventHub.mtx.Unlock()
	if eventHub.connected {
		logger.Debugf("Nothing to do - EventHub already connected")
		return nil
	}
	if eventHub.peerAddr == "" {
		return errors.New("peerAddr is required")
	}
	if eventHub.interestedEvents == nil || len(eventHub.interestedEvents) == 0 {
		return errors.New("at least one event must be registered")
	}
	if eventHub.grpcClient == nil {
		eventsClient, _ := eventHub.eventsClientFactory.newEventsClient(eventHub.provider, eventHub.identity,
			eventHub.peerAddr, eventHub.peerTLSCertificate, eventHub.peerTLSServerHostOverride,
			eventHub.provider.Config().TimeoutOrDefault(apiconfig.EventReg), eventHub, eventHub.kap, eventHub.failFast)
		eventHub.grpcClient = eventsClient
	}
	if err := eventHub.grpcClient.Start(); err != nil {
		eventHub.grpcClient.Stop()
		return errors.WithMessage(err, "event client start failed")
	}
	eventHub.connected = true
	return nil
}
func (eventHub *EventHub) GetInterestedEvents() ([]*pb.Interest, error) {
	return eventHub.interestedEvents, nil
}
func (eventHub *EventHub) Recv(msg *pb.Event) (bool, error) {
	go func() {
		switch msg.Event.(type) {
		case *pb.Event_Block:
			blockEvent := msg.Event.(*pb.Event_Block)
			logger.Debugf("Recv blockEvent for block number [%d]", blockEvent.Block.Header.Number)
			for _, v := range eventHub.getBlockRegistrants() {
				v(blockEvent.Block)
			}
			txFilter := util.TxValidationFlags(blockEvent.Block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
			for i, tdata := range blockEvent.Block.Data.Data {
				if txFilter.IsValid(i) {
					if ccEvent, channelID, err := getChainCodeEvent(tdata); err != nil {
						logger.Warnf("getChainCodeEvent return error: %v\n", err)
					} else if ccEvent != nil {
						eventHub.notifyChaincodeRegistrants(channelID, ccEvent, true)
					}
				} else {
					logger.Debugf("received invalid transaction")
				}
			}
			return
		case *pb.Event_ChaincodeEvent:
			ccEvent := msg.Event.(*pb.Event_ChaincodeEvent)
			logger.Debugf("Recv ccEvent for txID [%s]", ccEvent.ChaincodeEvent.TxId)
			if ccEvent != nil {
				eventHub.notifyChaincodeRegistrants("", ccEvent.ChaincodeEvent, false)
			}
			return
		default:
			return
		}
	}()
	return true, nil
}
func (eventHub *EventHub) Disconnected(err error) {
	if err != nil {
		logger.Warnf("EventHub was disconnected unexpectedly: %s", err)
	}
}
func (eventHub *EventHub) RegisterChaincodeEvent(ccid string, eventname string, callback func(*fab.ChaincodeEvent)) *fab.ChainCodeCBE {
	eventHub.mtx.Lock()
	defer eventHub.mtx.Unlock()
	eventHub.addChaincodeInterest(ccid, eventname)
	cbe := fab.ChainCodeCBE{CCID: ccid, EventNameFilter: eventname, CallbackFunc: callback}
	var cbeArray []*fab.ChainCodeCBE
	ccRegistrantArray, ok := eventHub.chaincodeRegistrants.Load(ccid)
	if !ok {
		cbeArray = make([]*fab.ChainCodeCBE, 0)
	} else {
		cbeArray = ccRegistrantArray.([]*fab.ChainCodeCBE)
	}
	cbeArray = append(cbeArray, &cbe)
	eventHub.chaincodeRegistrants.Store(ccid, cbeArray)
	return &cbe
}
func (eventHub *EventHub) UnregisterChaincodeEvent(cbe *fab.ChainCodeCBE) {
	eventHub.mtx.Lock()
	defer eventHub.mtx.Unlock()
	eventHub.removeChaincodeInterest(cbe.CCID, cbe.EventNameFilter)
	ccRegistrantArray, ok := eventHub.chaincodeRegistrants.Load(cbe.CCID)
	if ok {
		cbeArray := ccRegistrantArray.([]*fab.ChainCodeCBE)
		if len(cbeArray) <= 0 {
			logger.Debugf("No event registration for ccid %s \n", cbe.CCID)
			return
		}
		for i, v := range cbeArray {
			if v == cbe {
				newCbeArray := append(cbeArray[:i], cbeArray[i+1:]...)
				if len(newCbeArray) <= 0 {
					eventHub.chaincodeRegistrants.Delete(cbe.CCID)
				} else {
					eventHub.chaincodeRegistrants.Store(cbe.CCID, newCbeArray)
				}
				break
			}
		}
	}
}
func (eventHub *EventHub) RegisterTxEvent(txnID fab.TransactionID, callback func(string, pb.TxValidationCode, error)) {
	logger.Debugf("reg txid %s\n", txnID.ID)
	eventHub.txRegistrants.Store(txnID.ID, callback)
}
func (eventHub *EventHub) UnregisterTxEvent(txnID fab.TransactionID) {
	logger.Debugf("un-reg txid %s\n", txnID.ID)
	eventHub.txRegistrants.Delete(txnID.ID)
}
func (eventHub *EventHub) txCallback(block *common.Block) {
	txFilter := util.TxValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
	for i, v := range block.Data.Data {
		if env, err := utils.GetEnvelopeFromBlock(v); err != nil {
			logger.Debugf("error extracting Envelope from block: %v\n", err)
			return
		} else if env != nil {
			payload, err := utils.GetPayload(env)
			if err != nil {
				logger.Debugf("error extracting Payload from envelope: %v\n", err)
				return
			}
			channelHeaderBytes := payload.Header.ChannelHeader
			channelHeader := &common.ChannelHeader{}
			err = proto.Unmarshal(channelHeaderBytes, channelHeader)
			if err != nil {
				logger.Debugf("error extracting ChannelHeader from payload: %v\n", err)
				return
			}
			callback := eventHub.getTXRegistrant(channelHeader.TxId)
			if callback != nil {
				if txFilter.IsInvalid(i) {
					callback(channelHeader.TxId, txFilter.Flag(i),
						status.New(status.EventServerStatus, int32(txFilter.Flag(i)), "received invalid transaction", nil))
				} else {
					callback(channelHeader.TxId, txFilter.Flag(i), nil)
				}
			} else {
				logger.Debugf("No callback registered for TxID: %s\n", channelHeader.TxId)
			}
		}
	}
}
func (eventHub *EventHub) getBlockRegistrants() []func(*common.Block) {
	eventHub.mtx.RLock()
	defer eventHub.mtx.RUnlock()
	clone := make([]func(*common.Block), len(eventHub.blockRegistrants))
	copy(clone, eventHub.blockRegistrants)
	return clone
}
func (eventHub *EventHub) getChaincodeRegistrants(chaincodeID string) []*fab.ChainCodeCBE {
	eventHub.mtx.RLock()
	defer eventHub.mtx.RUnlock()
	registrants, ok := eventHub.chaincodeRegistrants.Load(chaincodeID)
	if !ok {
		return nil
	}
	cbeRegistrants := registrants.([]*fab.ChainCodeCBE)
	clone := make([]*fab.ChainCodeCBE, len(cbeRegistrants))
	copy(clone, cbeRegistrants)
	return clone
}
func (eventHub *EventHub) getTXRegistrant(txID string) func(string, pb.TxValidationCode, error) {
	v, ok := eventHub.txRegistrants.Load(txID)
	if !ok {
		return nil
	}
	return v.(func(string, pb.TxValidationCode, error))
}
func getChainCodeEvent(tdata []byte) (event *pb.ChaincodeEvent, channelID string, err error) {
	if tdata == nil {
		return nil, "", errors.New("Cannot extract payload from nil transaction")
	}
	if env, err := utils.GetEnvelopeFromBlock(tdata); err != nil {
		return nil, "", errors.Wrap(err, "tx from block failed")
	} else if env != nil {
		payload, err := utils.GetPayload(env)
		if err != nil {
			return nil, "", errors.Wrap(err, "extract payload from envelope failed")
		}
		channelHeaderBytes := payload.Header.ChannelHeader
		channelHeader := &common.ChannelHeader{}
		err = proto.Unmarshal(channelHeaderBytes, channelHeader)
		if err != nil {
			return nil, "", errors.Wrap(err, "unmarshal channel header failed")
		}
		channelID := channelHeader.ChannelId
		if common.HeaderType(channelHeader.Type) == common.HeaderType_ENDORSER_TRANSACTION {
			tx, err := utils.GetTransaction(payload.Data)
			if err != nil {
				return nil, "", errors.Wrap(err, "unmarshal transaction payload")
			}
			chaincodeActionPayload, err := utils.GetChaincodeActionPayload(tx.Actions[0].Payload)
			if err != nil {
				return nil, "", errors.Wrap(err, "chaincode action payload retrieval failed")
			}
			propRespPayload, err := utils.GetProposalResponsePayload(chaincodeActionPayload.Action.ProposalResponsePayload)
			if err != nil {
				return nil, "", errors.Wrap(err, "proposal response payload retrieval failed")
			}
			caPayload, err := utils.GetChaincodeAction(propRespPayload.Extension)
			if err != nil {
				return nil, "", errors.Wrap(err, "chaincode action retrieval failed")
			}
			ccEvent, err := utils.GetChaincodeEvents(caPayload.Events)
			if ccEvent != nil {
				return ccEvent, channelID, nil
			}
		}
	}
	return nil, "", nil
}