/*
(non-Javadoc)

@see com.ibm.wsspi.sib.core.ConsumerSession#registerAsynchConsumerCallback(com.ibm.wsspi.sib.core.AsynchConsumerCallback, int, boolean)
965173
@Override
    public void registerAsynchConsumerCallback(
                                               AsynchConsumerCallback callback,
                                               int maxActiveMessages,
                                               long messageLockExpiry,
                                               int maxBatchSize,
                                               OrderingContext context)
                    throws SISessionUnavailableException, SISessionDroppedException,
                    SIErrorException,
                    SIIncorrectCallException
    {
        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConsumerSession.tc.isEntryEnabled())
            SibTr.entry(CoreSPIConsumerSession.tc, "registerAsynchConsumerCallback",
                        new Object[] { this,
                                      callback,
                                      new Integer(maxActiveMessages),
                                      new Long(messageLockExpiry),
                                      new Integer(maxBatchSize),
                                      context });
        registerAsynchConsumerCallback(callback,
                                       maxActiveMessages,
                                       messageLockExpiry,
                                       maxBatchSize,
                                       null,
                                       false,
                                       context);

        if (TraceComponent.isAnyTracingEnabled() && CoreSPIConsumerSession.tc.isEntryEnabled())
            SibTr.exit(CoreSPIConsumerSession.tc, "registerAsynchConsumerCallback");
    }
1288570
@Override
    public boolean registerConsumerSetMonitor(
                                              SIDestinationAddress destinationAddress,
                                              String discriminatorExpression,
                                              ConsumerSetChangeCallback callback)
                    throws SIResourceException,
                    SINotPossibleInCurrentConfigurationException,
                    SIConnectionUnavailableException,
                    SIConnectionDroppedException,
                    SIIncorrectCallException,
                    SICommandInvocationFailedException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "registerConsumerSetMonitor",
                        new Object[] { destinationAddress,
                                      discriminatorExpression,
                                      callback });

        // Check that a listener has been specified
        if (callback == null)
        {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIIncorrectCallException");
            // Parameter is null, throw an excepiton
            throw new SIIncorrectCallException(nls.getFormattedMessage("NULL_CONSUMERSETCHANGECALLBACK_CWSIP0667", null, null));
        } else if (destinationAddress == null) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIIncorrectCallException");
            // Parameter is null, throw an excepiton
            throw new SIIncorrectCallException(nls.getFormattedMessage("NULL_DESTINATIONADDRESS_CWSIP0668", null, null));

        } else if (discriminatorExpression == null) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIIncorrectCallException");
            // Parameter is null, throw an excepiton
            throw new SIIncorrectCallException(nls.getFormattedMessage("NULL_DISCRIMINATOREXPRESSION_CWSIP0669", null, null));

        }
        boolean areConsumers = false;
        try
        {
            DestinationHandler topicSpace =
                            _destinationManager.getDestination((JsDestinationAddress) destinationAddress, false);

            areConsumers =
                            _messageProcessor.
                                            getMessageProcessorMatching().
                                            registerConsumerSetMonitor(topicSpace,
                                                                       discriminatorExpression,
                                                                       this,
                                                                       callback);
        } catch (Exception e)
        {
            // FFDC
            FFDCFilter
                            .processException(
                                              e,
                                              "com.ibm.ws.sib.processor.impl.ConnectionImpl.registerConsumerSetMonitor",
                                              "1:7378:1.347.1.25",
                                              this);

            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.exception(tc, e);

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "registerConsumerSetMonitor", "SIErrorException");
            //SIErrorException is RuntimeExcetion hence catching this exception in StaticCATConnection
            throw new SIErrorException(e);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "registerConsumerSetMonitor", Boolean.valueOf(areConsumers));
        return areConsumers;
    }
728336
public void deregisterConsumerSetMonitor(ConnectionImpl connection,
                                           ConsumerSetChangeCallback callback)
  throws SINotPossibleInCurrentConfigurationException   
  {
    if (tc.isEntryEnabled())
        SibTr.entry(
          tc,
          "deregisterConsumerSetMonitor",
          new Object[] { connection, callback}); 

     // Call the registrar to do the work
     _consumerMonitorRegistrar.deregisterMonitor(connection, callback);
    
    if (tc.isEntryEnabled())
      SibTr.exit(tc, "deregisterConsumerSetMonitor");
 
  }
764918
@Override
    public boolean addSync() throws ResourceException {

        final boolean isTracingEnabled = TraceComponent.isAnyTracingEnabled();

        if (isTracingEnabled && tc.isEntryEnabled()) {
            Tr.entry(this, tc, "addSync");
        }

        UOWCoordinator uowCoord = mcWrapper.getUOWCoordinator();

        if (uowCoord == null) {
            java.lang.IllegalStateException e = new java.lang.IllegalStateException("addSync: illegal state exception. uowCoord is null");
            Object[] parms = new Object[] { "addSync", e };
            Tr.error(tc, "ILLEGAL_STATE_EXCEPTION_J2CA0079", parms);
            if (isTracingEnabled && tc.isEntryEnabled())
                Tr.exit(this, tc, "addSync", e);
            throw e;
        }

        try {
            if (mcWrapper.isConnectionSynchronizationProvider()) {
                throw new NotSupportedException();
            }

            /*
             * This code will remain here just in case we run into this
             * case. We should not be isEnlistmentDisabled in this code.
             *
             * Log a message and return, if isEnlistmentDisabled is true.
             */
            if (mcWrapper.isEnlistmentDisabled()) {

                if (isTracingEnabled && tc.isDebugEnabled()) {
                    Tr.debug(this, tc, "Managed connection isEnlistmentDisabled is true.");
                    Tr.debug(this, tc, "Returning without registering.");
                }

                if (isTracingEnabled && tc.isEntryEnabled()) {
                    Tr.exit(this, tc, "addSync", false);
                }

                return false;

            }

            /*
             * Use our XATransactionWrapper
             */
            EmbeddableWebSphereTransactionManager tranMgr = mcWrapper.pm.connectorSvc.transactionManager;
            tranMgr.registerSynchronization(uowCoord, this);

        } catch (Exception e) {
            com.ibm.ws.ffdc.FFDCFilter.processException(
                                                        e,
                                                        "com.ibm.ejs.j2c.XATransactionWrapper.addSync",
                                                        "237",
                                                        this);
            Object[] parms = new Object[] { "addSync", e, "ResourceException" };
            Tr.error(tc, "REGISTER_WITH_SYNCHRONIZATION_EXCP_J2CA0026", parms);
            ResourceException re = new ResourceException("addSync: caught Exception");
            re.initCause(e);
            if (isTracingEnabled && tc.isEntryEnabled())
                Tr.exit(this, tc, "addSync", e);
            throw re;
        }

        if (isTracingEnabled && tc.isEntryEnabled())
            Tr.exit(this, tc, "addSync", true);
        return true;
    }
1128428
@Override
    public void eventPostRollbackAdd(Transaction transaction) throws SevereMessageStoreException
    {
        super.eventPostRollbackAdd(transaction);
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "eventPostRollbackAdd", transaction);

        if ((getMessageItem().getRegisterForPostEvents()) || failedInitInRestore)
        {
            stream = (MessageReferenceStream) getReferenceStream();
            //we can only register the listeners if we are fully initialized
            if ((stream instanceof DurableSubscriptionItemStream) || (stream instanceof ProxyReferenceStream))
            {
                stream.registerListeners(this);
            }
            failedInitInRestore = false;
            getMessageItem().setRegisterForPostEvents(failedInitInRestore);

        }

        try
        {
            if (POST_ROLLBACK_ADD_1 != null)
            {
                POST_ROLLBACK_ADD_1.messageEventOccurred(MessageEvents.POST_ROLLBACK_ADD, this, transaction);
            }
            if (POST_ROLLBACK_ADD_2 != null)
            {
                POST_ROLLBACK_ADD_2.messageEventOccurred(MessageEvents.POST_ROLLBACK_ADD, this, transaction);
            }
        } catch (SIException e)
        {
            FFDCFilter.processException(
                                        e,
                                        "com.ibm.ws.sib.processor.impl.store.items.MessageItemReference.eventPostRollbackAdd",
                                        "1:641:1.147",
                                        this);

            SibTr.exception(tc, e);

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "eventPostRollbackAdd", e);

            throw new SIErrorException(e);
        }

        resetEvents();

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "eventPostRollbackAdd");
    }
1029476
private void addConsumerToRegisteredMonitorMap(
    MonitoredConsumer mc,
    ArrayList monitorList,
    Map registeredMonitors)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.entry(
          tc,
          "addConsumerToRegisteredMonitorMap",
          new Object[] { registeredMonitors,
                         mc,
                         monitorList });
    // Finally, add this consumer to the appropriate places in the registered
    // consumer monitors tables
    Iterator i = monitorList.iterator();
    while (i.hasNext())
    {
      String tExpr = (String)i.next();

      // Get the next monitor
      RegisteredCallbacks rMonitor = (RegisteredCallbacks)registeredMonitors.get(tExpr);

      // Retrieve the current list of consumers
      ArrayList matchingConsumerList = rMonitor.getMatchingConsumers();

      // There is no need to drive any callbacks as we are placing subscriptions
      // in the registrations table that correspond to existing subscriptions.
      // So simply add the new subscription to the list.
      matchingConsumerList.add(mc);
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "addConsumerToRegisteredMonitorMap");

  }
759243
@Beta
    public static <T, U> BiConsumer<T, U> sc(final Object mutex, final BiConsumer<T, U> biConsumer) {
        N.checkArgNotNull(mutex, "mutex");
        N.checkArgNotNull(biConsumer, "biConsumer");

        return new BiConsumer<T, U>() {
            @Override
            public void accept(T t, U u) {
                synchronized (mutex) {
                    biConsumer.accept(t, u);
                }
            }
        };
    }
535463
@Override
    public long newReadyConsumer(JSConsumerKey consumerKey, boolean bSelector)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "newReadyConsumer",
                        new Object[] { consumerKey, Boolean.valueOf(bSelector) });
        // If this is a forward scanning consumer just remember we have one
        if (consumerKey.getForwardScanning())
        {
            //move the consumerPoint in to the ready list
            readyFwdScanningCPs.put((SimpleEntry) consumerKey);
        }
        else
        {
            // If the consumer has a selector...
            if (bSelector)
            {
                //increment the specific consumer count and the version number
                specificReadyConsumerCount++;

                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    SibTr.debug(tc, "sepcificReadyConsumerCount:" + specificReadyConsumerCount);
            }
            else
            {
                //move the consumerPoint in to the ready list
                nonSpecificReadyCPs.put((SimpleEntry) consumerKey);
            }
        }

        // Increment the ready version
        ++readyConsumerVersion;

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "newReadyConsumer", Long.valueOf(readyConsumerVersion));
        return readyConsumerVersion;
    }
724710
public boolean commitAsync(String topic) {
        KafkaConsumer<String, byte[]> consumer = _getConsumer(topic);
        synchronized (consumer) {
            Set<String> subscription = consumer.subscription();
            if (subscription == null || !subscription.contains(topic)) {
                // this consumer has not subscribed to the topic
                return false;
            }
            try {
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets,
                            Exception e) {
                        if (e != null) {
                            LOGGER.error(e.getMessage(), e);
                        }
                    }
                });
            } catch (WakeupException e) {
            } catch (Exception e) {
                throw new KafkaException(e);
            }
            return true;
        }
    }
844486
public void setConsumer(org.apache.kafka.clients.consumer.KafkaConsumer<Object, Object> consumer) {
        this.consumer = consumer;
    }
290961
@Override
  public void consume(SerializableConsumer<R> consumer) {
    checkNotNull(consumer, "consumer cannot be null");

    ConsumerStreamlet<R> consumerStreamlet = new ConsumerStreamlet<>(this, consumer);
    addChild(consumerStreamlet);
  }
199227
public void registerConsumer( LocalMessageConsumer consumer )
    {
    	consumersLock.writeLock().lock();
    	try
    	{
    		localConsumers.add(consumer);
    	}
    	finally
    	{
    		consumersLock.writeLock().unlock();
    	}
    }
1214420
public void setAsynchConsumerCallback(int requestNumber,
                                         int maxActiveMessages,
                                         long messageLockExpiry,
                                         int batchsize,
                                         OrderingContext orderContext,
                                         boolean stoppable,
                                         int maxSequentialFailures,
                                         long hiddenMessageDelay)
   {
     if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "setAsynchConsumerCallback",
                                          new Object[]
                                          {
                                             requestNumber,
                                             maxActiveMessages,
                                             messageLockExpiry,
                                             batchsize,
                                             orderContext,
                                             stoppable,
                                             maxSequentialFailures,
                                             hiddenMessageDelay});

     SIErrorException e = new SIErrorException(nls.getFormattedMessage("PROTOCOL_ERROR_SICO2003", null, null));

     FFDCFilter.processException(e, CLASS_NAME + ".setAsynchConsumerCallback",CommsConstants.CATCONSUMER_SETASYNCHCALLBACK_01,this);

     SibTr.error(tc, "PROTOCOL_ERROR_SICO2003", e);

     if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "setAsynchConsumerCallback");

     // Re-throw this exception so that the client will informed if required
     throw e;
   }
819779
@Override
    public boolean addSync() throws ResourceException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) { 
            Tr.entry(this, tc, "addSync");
        }

        UOWCoordinator uowCoord = mcWrapper.getUOWCoordinator();

        if (uowCoord == null) {
            // No transaction context.  Should never happen.
            // removed null check on mcWrapper since it is redundant from the above code (mcWrapper.getUOWCoordinator())
            String pmiName = mcWrapper.gConfigProps.cfName;

            Object[] parms = new Object[] { "addSync", pmiName };
            Tr.error(tc, "NO_VALID_TRANSACTION_CONTEXT_J2CA0040", parms); 
            throw new ResourceException("INTERNAL ERROR: No valid transaction context present");

        }

        try {

            //  No need to register for synchronization for unshared/LTC/res-control=Application.
            //  Instead of cleanup being done via after completion it will either happen at close or
            //  commit/rollback which ever happens last.  This will allow the connection to be returned
            //  to the pool prior to the end of the LTC scope allowing better utilization of the connection.
            boolean shareable = mcWrapper.getConnectionManager().shareable();
            if (!shareable && !J2CUtilityClass.isContainerAtBoundary(mcWrapper.pm.connectorSvc.transactionManager)) {
                //Register an instance of a no-transaction wrapper
                // so the TM will create an RRS NativeLocalTranasction.  The wrapper
                // provides "empty" TransactioWrapper and Synchronization behaviors,
                // but indicates to the TM that the LTC requires RRS support.  
                ((LocalTransactionCoordinator) uowCoord).enlistSynchronization(new RRSNoTransactionWrapper());

                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) { 
                    Tr.exit(this, tc, "addSync", "returning without registering.");
                }
                return false;
            } else { // Register a synchronization with the Transaction Manager
                ((LocalTransactionCoordinator) uowCoord).enlistSynchronization(this);
                mcWrapper.markRRSLocalTransactionWrapperInUse();
                registeredForSync = true;
            }
        } catch (ResourceException e) {
            com.ibm.ws.ffdc.FFDCFilter.processException(e, "com.ibm.ejs.j2c.RRSLocalTransactionWrapper.addSync", "594", this);
            Object[] parms = new Object[] { "addSync", e, "Exception" };
            Tr.error(tc, "REGISTER_WITH_SYNCHRONIZATION_EXCP_J2CA0026", parms);
            throw e;
        } catch (Exception e) {
            com.ibm.ws.ffdc.FFDCFilter.processException(e, "com.ibm.ejs.j2c.RRSLocalTransactionWrapper.addSync", "605", this);
            Object[] parms = new Object[] { "addSync", e, "Exception" };
            Tr.error(tc, "REGISTER_WITH_SYNCHRONIZATION_EXCP_J2CA0026", parms);
            ResourceException re = new ResourceException("addSync: caught Exception");
            re.initCause(e);
            throw re;
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) { 
            Tr.exit(this, tc, "addSync", registeredForSync);
        }

        return registeredForSync;

    }
274413
public void registerAsynchConsumerCallback(
            final AsynchConsumerCallback callback, final int maxActiveMessages,
            final long messageLockExpiry, final int maxBatchSize,
            final OrderingContext extendedMessageOrderingContext)
            throws SISessionUnavailableException, SISessionDroppedException,
            SIConnectionUnavailableException, SIConnectionDroppedException,
            SIIncorrectCallException {

        throw new SibRaNotSupportedException(NLS
                .getString("ASYNCHRONOUS_METHOD_CWSIV0250"));

    }
58108
public void unsetAsynchConsumerCallback(int requestNumber, boolean stoppable)                        //SIB0115d.comms
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "unsetAsynchConsumerCallback", "requestNumber="+requestNumber+",stoppable="+stoppable);

      checkNotBrowserSession();                                   // F171893

      if (subConsumer != null)
      {
         subConsumer.unsetAsynchConsumerCallback(requestNumber, stoppable);                             //SIB0115d.comms
      }

      subConsumer = null;

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "unsetAsynchConsumerCallback");
   }
1056700
@Override
    public boolean addSync() throws ResourceException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, hexId() + ":addSynch()");
        }
        return false;
    }