public boolean subscribe ( IConsumer consumer , Map < String , Object > paramMap ) { // pipe is possibly used by dozens of threads at once (like many subscribers for one server stream) boolean success = consumers . addIfAbsent ( consumer ) ; // if consumer is listener object register it as listener and consumer has just been added if ( success && consumer instanceof IPipeConnectionListener ) { listeners . addIfAbsent ( ( IPipeConnectionListener ) consumer ) ; } return success ; }
Connect consumer to this pipe . Doesn t allow to connect one consumer twice . Does register event listeners if instance of IPipeConnectionListener is given .