@@ -2842,15 +2842,16 @@ export function newFileKeyValStore(dir:string):KeyValStore {
2842
2842
/**
2843
2843
* The ChainCodeCBE is used internal to the EventHub to hold chaincode event registration callbacks.
2844
2844
*/
2845
- class ChainCodeCBE {
2845
+ export class ChainCodeCBE {
2846
+ // chaincode id
2846
2847
ccid : string ;
2847
- eventname : string ;
2848
- payload : Uint8Array ;
2848
+ // event name regex filter
2849
+ eventNameFilter : RegExp ;
2850
+ // callback function to invoke on successful filter match
2849
2851
cb : Function ;
2850
- constructor ( ccid : string , eventname : string , payload : Uint8Array , cb : Function ) {
2852
+ constructor ( ccid : string , eventNameFilter : string , cb : Function ) {
2851
2853
this . ccid = ccid ;
2852
- this . eventname = eventname ;
2853
- this . payload = payload ;
2854
+ this . eventNameFilter = new RegExp ( eventNameFilter ) ;
2854
2855
this . cb = cb ;
2855
2856
}
2856
2857
}
@@ -2879,7 +2880,7 @@ export class EventHub {
2879
2880
this . chaincodeRegistrants = new HashTable ( ) ;
2880
2881
this . blockRegistrants = new Set ( ) ;
2881
2882
this . txRegistrants = new HashTable ( ) ;
2882
- this . peeraddr = "localhost:7053" ;
2883
+ this . peeraddr = null ;
2883
2884
this . connected = false ;
2884
2885
}
2885
2886
@@ -2893,6 +2894,7 @@ export class EventHub {
2893
2894
2894
2895
public connect ( ) {
2895
2896
if ( this . connected ) return ;
2897
+ if ( ! this . peeraddr ) throw Error ( "Must set peer address before connecting." ) ;
2896
2898
this . events = grpc . load ( __dirname + "/protos/events.proto" ) . protos ;
2897
2899
this . client = new this . events . Events ( this . peeraddr , grpc . credentials . createInsecure ( ) ) ;
2898
2900
this . call = this . client . chat ( ) ;
@@ -2902,11 +2904,15 @@ export class EventHub {
2902
2904
let eh = this ; // for callback context
2903
2905
this . call . on ( 'data' , function ( event ) {
2904
2906
if ( event . Event == "chaincodeEvent" ) {
2905
- var cbe = eh . chaincodeRegistrants . get ( event . chaincodeEvent . chaincodeID + "/" + event . chaincodeEvent . eventName ) ;
2906
- if ( cbe ) {
2907
- cbe . payload = event . chaincodeEvent . payload ;
2908
- cbe . cb ( cbe ) ;
2909
- }
2907
+ var cbtable = eh . chaincodeRegistrants . get ( event . chaincodeEvent . chaincodeID ) ;
2908
+ if ( ! cbtable ) {
2909
+ return ;
2910
+ }
2911
+ cbtable . forEach ( function ( cbe ) {
2912
+ if ( cbe . eventNameFilter . test ( event . chaincodeEvent . eventName ) ) {
2913
+ cbe . cb ( event . chaincodeEvent ) ;
2914
+ }
2915
+ } ) ;
2910
2916
} else if ( event . Event == "block" ) {
2911
2917
eh . blockRegistrants . forEach ( function ( cb ) {
2912
2918
cb ( event . block ) ;
@@ -2928,19 +2934,35 @@ export class EventHub {
2928
2934
this . connected = false ;
2929
2935
}
2930
2936
2931
- public registerChaincodeEvent ( ccid : string , eventname : string , callback : Function ) {
2937
+ public registerChaincodeEvent ( ccid : string , eventname : string , callback : Function ) : ChainCodeCBE {
2932
2938
if ( ! this . connected ) return ;
2933
- let cb = new ChainCodeCBE ( ccid , eventname , null , callback ) ;
2934
- let register = { register : { events : [ { eventType : "CHAINCODE" , chaincodeRegInfo :{ chaincodeID : ccid , eventName : eventname } } ] } } ;
2935
- this . chaincodeRegistrants . put ( ccid + "/" + eventname , cb ) ;
2936
- this . call . write ( register ) ;
2939
+ let cb = new ChainCodeCBE ( ccid , eventname , callback ) ;
2940
+ let cbtable = this . chaincodeRegistrants . get ( ccid ) ;
2941
+ if ( ! cbtable ) {
2942
+ cbtable = new Set ( ) ;
2943
+ this . chaincodeRegistrants . put ( ccid , cbtable ) ;
2944
+ cbtable . add ( cb ) ;
2945
+ let register = { register : { events : [ { eventType : "CHAINCODE" , chaincodeRegInfo :{ chaincodeID : ccid , eventName : "" } } ] } } ;
2946
+ this . call . write ( register ) ;
2947
+ } else {
2948
+ cbtable . add ( cb ) ;
2949
+ }
2950
+ return cb ;
2937
2951
}
2938
2952
2939
- public unregisterChaincodeEvent ( ccid : string , eventname : string ) {
2953
+ public unregisterChaincodeEvent ( cbe : ChainCodeCBE ) {
2940
2954
if ( ! this . connected ) return ;
2941
- var unregister = { unregister : { events : [ { eventType : "CHAINCODE" , chaincodeRegInfo :{ chaincodeID : ccid , eventName : eventname } } ] } } ;
2942
- this . chaincodeRegistrants . remove ( ccid + "/" + eventname ) ;
2943
- this . call . write ( unregister ) ;
2955
+ let cbtable = this . chaincodeRegistrants . get ( cbe . ccid ) ;
2956
+ if ( ! cbtable ) {
2957
+ debug ( "No event registration for ccid %s " , cbe . ccid ) ;
2958
+ return ;
2959
+ }
2960
+ cbtable . delete ( cbe ) ;
2961
+ if ( cbtable . size <= 0 ) {
2962
+ var unregister = { unregister : { events : [ { eventType : "CHAINCODE" , chaincodeRegInfo :{ chaincodeID : cbe . ccid , eventName : "" } } ] } } ;
2963
+ this . chaincodeRegistrants . remove ( cbe . ccid ) ;
2964
+ this . call . write ( unregister ) ;
2965
+ }
2944
2966
}
2945
2967
2946
2968
public registerBlockEvent ( callback :Function ) {
0 commit comments