code: @SuppressWarnings("unused")
      public void onPipeConnectionEvent(PipeConnectionEvent event) {
          switch (event.getType()) {
              case PROVIDER_CONNECT_PUSH:
                  //log.debug("Provider connect");
                  if (event.getProvider() == this && event.getSource() != connMsgOut && (event.getParamMap() == null || !event.getParamMap().containsKey("record"))) {
                      livePipe = (IPipe) event.getSource();
                      //log.debug("Provider: {}", livePipe.getClass().getName());
                      for (IConsumer consumer : livePipe.getConsumers()) {
                          subscriberStats.increment();
                      }
                  }
                  break;
              case PROVIDER_DISCONNECT:
                  //log.debug("Provider disconnect");
                  //if (log.isDebugEnabled() && livePipe != null) {
                  //log.debug("Provider: {}", livePipe.getClass().getName());
                  //}
                  if (livePipe == event.getSource()) {
                      livePipe = null;
                  }
                  break;
              case CONSUMER_CONNECT_PUSH:
                  //log.debug("Consumer connect");
                  IPipe pipe = (IPipe) event.getSource();
                  //if (log.isDebugEnabled() && pipe != null) {
                  //log.debug("Consumer: {}", pipe.getClass().getName());
                  //}
                  if (livePipe == pipe) {
                      notifyChunkSize();
                  }
                  subscriberStats.increment();
                  break;
              case CONSUMER_DISCONNECT:
                  //log.debug("Consumer disconnect: {}", event.getSource().getClass().getName());
                  subscriberStats.decrement();
                  break;
              default:
          }
      }
docs: Pipe connection event handler
  
  @param event
  Pipe connection event