code: 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;
      }
docs: Connect consumer to this pipe. Doesn't allow to connect one consumer twice. Does register event listeners if instance of IPipeConnectionListener is given.
  
  @param consumer
  Consumer
  @param paramMap
  Parameters passed with connection, used in concrete pipe implementations
  @return true if consumer was added, false otherwise