Broadcast consumer connection event

@param consumer
Consumer that has connected
@param type
Event type
@param paramMap
Parameters passed with connection
328949
protected void fireConsumerConnectionEvent(IConsumer consumer, PipeConnectionEvent.EventType type, Map<String, Object> paramMap) {
        firePipeConnectionEvent(PipeConnectionEvent.build(this, type, consumer, paramMap));
    }
431254
private synchronized void createConsumerThreadPool() {
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.entry(tc, "createConsumerThreadPool");
        if (_consumerThreadPool == null) {
            // 214163 Add a threadpool listener to associate threads in trace
            // with the messaging engine.
            _consumerThreadPool = new ThreadPool(
                            "Consumer " + getMessagingEngineName(),
                            0,
                            _customProperties.get_max_consumer_threadpool_size(),
                            new ThreadPoolListener[] { new ThreadPoolListenerImpl(this) });
            _consumerThreadPool
                            .setRequestBufferExpansionLimit(Integer.MAX_VALUE);
        }
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.exit(tc, "createConsumerThreadPool");
    }
270327
public static void searchUnitMap(Consumer<Map<String, Unit>> consumer) {
        readWriteLock.readLock().lock();
        try {
            consumer.accept(Collections.unmodifiableMap(searchUnitMap));
        } finally {
            readWriteLock.readLock().unlock();
        }
    }
63825
public <T> void use(final Class<T> type, final Consumer<T> consumer) {
        if (uiThreadRunner.isOnUiThread()) {
            try {
                graph.use(type, Inject.class, consumer);
            } catch (PokeException e) {
                throw new MvcGraphException(e.getMessage(), e);
            }
        } else {
            uiThreadRunner.post(new Runnable() {
                @Override
                public void run() {
                    try {
                        graph.use(type, Inject.class, consumer);
                    } catch (PokeException e) {
                        throw new MvcGraphException(e.getMessage(), e);
                    }
                }
            });
        }
    }
106464
public static <T> void setIfPresent(final Environment env, final String key, final Class<T> type,
			final Consumer<T> function) {
		final T value = env.getProperty(key, type);
		if (value != null) {
			function.accept(value);
		}
	}
663034
@Override
  public <T> Consumer<T> injector(Class<T> type)
  {
    ArrayList<InjectProgram> injectList = new ArrayList<>();

    introspectInject(injectList, type);

    introspectInit(injectList, type);

    return new InjectProgramImpl<T>(injectList);
  }
40243
void addConsumer(ConsumerSessionImpl consumer) {
        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.entry(tc, "addConsumer", consumer);

        _consumerList.add(consumer);

        if (TraceComponent.isAnyTracingEnabled()
            && tc.isEntryEnabled())
            SibTr.exit(tc, "addConsumer");
    }
94504
void processMsgs(LockedMessageEnumeration msgEnumeration, ConsumerSession consumerSession)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "processMsgs", new Object[] { msgEnumeration, consumerSession });      

    // Remember that a callback is running
    asynchConsumerRunning = true;

    try
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        SibTr.debug(tc, "Entering asynchConsumerCallback.consumeMessages",
                     new Object[] {asynchConsumerCallback, msgEnumeration});      

      //Call the consumeMessages method on the registered
      //AsynchConsumerCallback object
      asynchConsumerCallback.consumeMessages(msgEnumeration);

      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        SibTr.debug(tc, "Exiting asynchConsumerCallback.consumeMessages");      
    }
    catch (Throwable e)
    {
      //Catch any exceptions thrown by consumeMessages.
      FFDCFilter.processException(
       e,
       "com.ibm.ws.sib.processor.impl.AsynchConsumer.processMsgs",
       "1:124:1.37",
       this);
       
      //Notify the consumer that this exception occurred.
      if(consumerSession!=null)
      {
        try
        {
          //Notify the consumer that this exception occurred.
          notifyExceptionListeners(e, consumerSession);
        }
        catch(Exception connectionClosed)
        {
          //No FFDC code needed
        }
            
      }
       
     //Trace the exception but otherwise swallow it since it was not
      //a failure in MP code or code upon which the MP is critically
      //dependent.
      if (e instanceof Exception)
        SibTr.exception(tc, (Exception)e);
      
      if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
        SibTr.debug(tc, "Exception occurred in consumeMessages " + e);
         
      // We're not allowed to swallow this exception - let it work its way back to
      // the threadpool
      if(e instanceof ThreadDeath)
      {     
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
          SibTr.exit(tc, "processMsgs", e); 
        throw (ThreadDeath)e;
      }
        
    }
    finally
    {
      asynchConsumerRunning = false;
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "processMsgs");
  }
601356
@Override
    public void addConnectionEventListener( ConnectionEventListener listener ) {
        if (listener == null) {
            throw new IllegalArgumentException("Listener is null");
        }
        listeners.add(listener);
    }
1262470
@SuppressWarnings({ "unchecked", "PMD.ShortVariable" })
    public <E extends EventBase<?>, C extends Channel> void forChannels(
            Class<C> type, BiConsumer<E, C> handler) {
        Arrays.stream(channels)
            .filter(c -> type.isAssignableFrom(c.getClass()))
            .forEach(c -> handler.accept((E) this, (C) c));
    }
199227
public void registerConsumer( LocalMessageConsumer consumer )
    {
    	consumersLock.writeLock().lock();
    	try
    	{
    		localConsumers.add(consumer);
    	}
    	finally
    	{
    		consumersLock.writeLock().unlock();
    	}
    }
235321
@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:
        }
    }
320817
@Beta
    public static <T> Consumer<T> sc(final Object mutex, final Consumer<T> consumer) {
        N.checkArgNotNull(mutex, "mutex");
        N.checkArgNotNull(consumer, "consumer");

        return new Consumer<T>() {
            @Override
            public void accept(T t) {
                synchronized (mutex) {
                    consumer.accept(t);
                }
            }
        };
    }
312058
@SuppressWarnings("unchecked")
    public void forEach(final Consumer<V> consumer)
    {
        for (final Object value : values)
        {
            if (null != value)
            {
                consumer.accept((V)value);
            }
        }
    }
981673
public void parseEvents(MarcXchangeEventConsumer consumer) throws XMLStreamException {
        Objects.requireNonNull(consumer);
        if (builder.getMarcListeners() != null) {
            for (Map.Entry<String, MarcListener> entry : builder.getMarcListeners().entrySet()) {
                consumer.setMarcListener(entry.getKey(), entry.getValue());
            }
        }
        XMLInputFactory inputFactory = XMLInputFactory.newInstance();
        XMLEventReader xmlEventReader = inputFactory.createXMLEventReader(builder.getInputStream());
        while (xmlEventReader.hasNext()) {
            consumer.add(xmlEventReader.nextEvent());
        }
        xmlEventReader.close();
    }
938896
public ConsumerKey attachConsumerPoint(
    ConsumerPoint consumerPoint,
    SelectionCriteria criteria,
    SIBUuid12 connectionUuid,
    boolean readAhead,
    boolean forwardScanning,
    JSConsumerSet consumerSet) throws SIDestinationLockedException, SISelectorSyntaxException, SIDiscriminatorSyntaxException, SINotPossibleInCurrentConfigurationException, SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this,tc, "attachConsumerPoint",
        new Object[]{consumerPoint,
                     criteria,
                     connectionUuid,
                     new Boolean(readAhead),
                     new Boolean(forwardScanning),
                     consumerSet});
    
    ConsumerKey returnKey = null;
    
    // There are a few times that we're unable to accept new consumers, e.g. forcing a
    // flush or changing cardinality. None of these events should last very long so
    // we just try again a few times, if it's still in this state then something else has
    // probably gone wrong so we bomb out. We obviously can't hold a lock while we're
    // retrying.
    int chances = 5;
    boolean repeat = true;
    while(repeat)
    {
      repeat = false;
      
      synchronized (consumerPoints)
      {
        if (!_currentReachability)
        { // don't allow an attachment if DME is not reachable or a CardinalityChange is in progress or a forced flush
          // is in progress
          // throw exception
          SIResourceException e = new SIResourceException(
            nls.getFormattedMessage("ANYCAST_STREAM_UNAVAILABLE_CWSIP0471",
                                    new Object[]{getDestName(), SIMPUtils.getMENameFromUuid(getLocalisationUuid().toString())},
                                    null));
          SibTr.exception(tc, e);
          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "attachConsumerPoint", e);
          throw e;
        }
        else if (_cardinalityOne)
        { // enforce cardinality
          if (consumerPoints.size() != 0)
          {
            // throw exception
            SIResourceException e = new SIResourceException(
              nls.getFormattedMessage("CONSUMERCARDINALITY_LIMIT_REACHED_CWSIP0472",
                                      new Object[]{getDestName(), getLocalisationUuid().toString()},
                                      null));
            SibTr.exception(tc, e);
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.exit(tc, "attachConsumerPoint", e);
            throw e;
          }
        }
        // The stream is in a funny state, try again in a few milliseconds
        else if (_cardinalityChange || _forceFlushInProgress)
        {
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "Unable to attach at this point " + _cardinalityChange + ":" +
                            _forceFlushInProgress);
          repeat = true;
        }
        
        // Work out how long we should wait for a response from the DME based on whether 
        // we are gathering or not
        long responseTimeout = 0;
        if (consumerPoint.isGatheringConsumer())
          responseTimeout = SIMPConstants.GATHERING_ANYCAST_RESPONSE_INTERVAL; // 10 seconds
        else
          responseTimeout = SIMPConstants.ANYCAST_RESPONSE_INTERVAL; // 5 minutes
        
        // do this in sync block to prevent two concurrent attachConsumerPoints from succeeding
        // when cardinalityOne, and to ensure that reachability changes are notified.
        if(_aih.consumerAttaching(responseTimeout))
        {
          returnKey = super.attachConsumerPoint(consumerPoint,
              criteria,
              connectionUuid,
              readAhead,
              forwardScanning,
              consumerSet);
        }
        else
        {
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "Unable to attach at this point, flush work in progress");
          repeat = true;
        }

        // If we're unable to attach at this exact point, but there's a probability that
        // we will in a split second, we sleep for a short time and try again (unless it's
        // time to give up).
        if(repeat)
        {
          if(chances > 0)
          {
            try
            {
              // Release the consumerPoints lock while we're waiting
              consumerPoints.wait(100);
            }
            catch (InterruptedException e)
            {
              // No FFDC code needed
            }
            chances--;
          }
          else
          {
            // throw exception, should be rare
            SIResourceException e =
              new SIResourceException(
                nls.getFormattedMessage(
                  "ANYCAST_STREAM_NOT_FLUSHED_CWSIP0512",
                  new Object[]{_messageProcessor.getMessagingEngineName(), _destName},
                  null));
            SibTr.exception(tc, e);
            
            // As we don't expect this to have taken so long, we FFDC just in case
            // someone is interested
            FFDCFilter.processException(e,
                                        "com.ibm.ws.sib.processor.impl.RemoteConsumerDispatcher.attachConsumerPoint",
                                        "1:445:1.97.2.21",
                                        this);
            
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.exit(tc, "attachConsumerPoint", e);
            
            throw e;
          }
        } // repeat
      } // end synchronized (consumerPoints)
    } // while
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "attachConsumerPoint", returnKey);
    return returnKey;
  }
1104578
public void addConnectionEventListener(ConnectionEventListener listener) {
    log.finest("addConnectionEventListener()");
    if (listener == null)
      throw new IllegalArgumentException("Listener is null");
    listeners.add(listener);
  }