private String registerTransactionListenerProcessor ( ) throws InvalidArgumentException { logger . debug ( format ( "Channel %s registerTransactionListenerProcessor starting" , name ) ) ; // Transaction listener is internal Block listener for transactions return registerBlockListener ( blockEvent -> { HFClient lclient = client ; if ( null == lclient || shutdown ) { //can happen if were not quite shutdown return ; } final String source = blockEvent . getPeer ( ) != null ? blockEvent . getPeer ( ) . toString ( ) : "not peer!" ; logger . debug ( format ( "is peer %b, is filtered: %b" , blockEvent . getPeer ( ) != null , blockEvent . isFiltered ( ) ) ) ; final Iterable < TransactionEvent > transactionEvents = blockEvent . getTransactionEvents ( ) ; if ( transactionEvents == null || ! transactionEvents . iterator ( ) . hasNext ( ) ) { // no transactions today we can assume it was a config or update block. if ( isLaterBlock ( blockEvent . getBlockNumber ( ) ) ) { ServiceDiscovery lserviceDiscovery = serviceDiscovery ; if ( null != lserviceDiscovery ) { client . getExecutorService ( ) . execute ( ( ) -> lserviceDiscovery . fullNetworkDiscovery ( true ) ) ; } } else { lclient . getExecutorService ( ) . execute ( ( ) -> { try { if ( ! shutdown ) { loadCACertificates ( true ) ; } } catch ( Exception e ) { logger . warn ( format ( "Channel %s failed to load certificates for an update" , name ) , e ) ; } } ) ; } return ; } if ( txListeners . isEmpty ( ) || shutdown ) { return ; } for ( TransactionEvent transactionEvent : blockEvent . getTransactionEvents ( ) ) { logger . debug ( format ( "Channel %s got event from %s for transaction %s in block number: %d" , name , source , transactionEvent . getTransactionID ( ) , blockEvent . getBlockNumber ( ) ) ) ; List < TL > txL = new ArrayList <> ( txListeners . size ( ) + 2 ) ; synchronized ( txListeners ) { LinkedList < TL > list = txListeners . get ( transactionEvent . getTransactionID ( ) ) ; if ( null != list ) { txL . addAll ( list ) ; } } for ( TL l : txL ) { try { // only if we get events from each eventhub on the channel fire the transactions event. // if (getEventHubs().containsAll(l.eventReceived(transactionEvent.getEventHub()))) { if ( shutdown ) { break ; } if ( l . eventReceived ( transactionEvent ) ) { l . fire ( transactionEvent ) ; } } catch ( Throwable e ) { logger . error ( e ) ; // Don't let one register stop rest. } } } } ) ; }
Own block listener to manage transactions .