docs: @Override
      public void registerAsynchConsumer(
                                         AsynchConsumerCallback callback,
                                         int maxActiveMessages,
                                         long messageLockExpiry,
                                         int maxBatchSize,
                                         Reliability unrecoverableReliability,
                                         boolean inLine,
                                         OrderingContextImpl orderingGroup,
                                         ExternalConsumerLock optionalCallbackBusyLock)
                      throws SISessionUnavailableException, SISessionDroppedException,
                      SIErrorException,
                      SIIncorrectCallException
      {
          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.entry(
                          tc,
                          "registerAsynchConsumer",
                          new Object[] {
                                        this,
                                        callback,
                                        Integer.valueOf(maxActiveMessages),
                                        Long.valueOf(messageLockExpiry),
                                        Integer.valueOf(maxBatchSize),
                                        unrecoverableReliability,
                                        Boolean.valueOf(inLine),
                                        orderingGroup,
                                        optionalCallbackBusyLock });
  
          // Check that the values are valid
          checkParams(maxActiveMessages, messageLockExpiry, maxBatchSize);
  
          if (optionalCallbackBusyLock != null)
          {
              // The callback wants to block our asynch consumer thread sometimes.
              // It wants to know that when it has this lock, the asynch consumer
              // thread is not currently busy.
              // As of 14May2004, the only consumer to use this function is the
              // mediation point. Doing so prevents deadlocks between the mediation
              // state machine lock, and a separate async consumer lock, by making
              // them one and the same lock.
              // Thus, when an action is being processed within the state machine,
              // the async consumer is blocked, and similarly, if the async
              // consumer is busy passing a message to the mediation, no new events
              // can enter the mediation point state machine.
  
              // Use this optional lock instead of the default lock we wouls use.
              // ie: The "this" reference on the async consumer itself.
              _asynchConsumerBusyLock = optionalCallbackBusyLock;
          }
  
          // Synchronized against any other asyncConsumer work on this LCP or keyGroup
          synchronized (_asynchConsumerBusyLock)
          {
              // Synchronize on the consumer state
              this.lock();
              try
              {
                  // Only valid if the consumer session is still open
                  checkNotClosed();
  
                  // we can't register a callback if the LCP is not in a stopped state
                  // so throw an exception
                  if (!_stopped)
                  {
                      SIIncorrectCallException e =
                                      new SIIncorrectCallException(
                                                      nls.getFormattedMessage(
                                                                              "ASYNCH_CONSUMER_RUN_ERROR_CWSIP0176",
                                                                              new Object[] { _consumerDispatcher.getDestination().getName(),
                                                                                            _consumerDispatcher.getMessageProcessor().getMessagingEngineName() },
                                                                              null));
  
                      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                          SibTr.exit(tc, "registerAsynchConsumer", e);
                      throw e;
                  }
                  if (_waiting) // We're in a synchronous receive!
                  {
                      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                          SibTr.exit(tc, "registerAsynchConsumer", "waiting == true");
                      throw new SIIncorrectCallException(
                                      nls.getFormattedMessage("RECEIVE_USAGE_ERROR_CWSIP0174",
                                                              new Object[] { _consumerDispatcher.getDestination().getName(),
                                                                            _messageProcessor.getMessagingEngineName() },
                                                              null));
                  }
  
                  // set the recoverability options (they will only be overridden by
                  // an internal Core SPI user (Comms).
                  if (unrecoverableReliability != null)
                      setUnrecoverability(unrecoverableReliability);
  
                  // Remember the callback object (could be null)
                  _asynchConsumer.registerCallback(callback);
                  if (callback != null)
                  {
                      // If ordered, override maxbatchsize to be 1
                      // defect 114998 
                      if (_destinationAttachedTo.isOrdered() && maxBatchSize > 1)
                      {
                          _maxBatchSize = 1;
                          // Override and warn if ordered destination
                          SibTr.warning(tc, SibTr.Suppressor.ALL_AFTER_FIRST_SIMILAR_INSERTS, "MAX_BATCH_SIZE_OVERRIDE_WARNING_CWSIP0181",
                                        new Object[] { _destinationAttachedTo.getName() });
                      }
                      else
                          _maxBatchSize = maxBatchSize;
  
                      _inLine = inLine;
                      _maxActiveMessages = maxActiveMessages;
                      if (_maxActiveMessages > 0)
                      {
                          // (F001731)
                          // If the destination has the "MsgBlockWarningInterval" context property set (probably WPS event sequencing)
                          // then we need to warn if this consumer ever stays stuck for too long with its maximum active messages.
                          _activeMsgBlockInterval = -1;
                          Object value = _destinationAttachedTo.getContextValue("MsgBlockWarningInterval");
                          if ((value != null) && (value instanceof Integer))
                          {
                              _activeMsgBlockInterval = ((Integer) value).intValue();
                              if (_activeMsgBlockInterval < 0)
                                  _activeMsgBlockInterval = -1;
                          }
                      }
  
                      // Reset the active messages under this new asynch consumer
                      _currentActiveMessages = 0;
  
                      // Tell the LME the currentmessageLockExpiry value
                      _allLockedMessages.setMessageLockExpiry(messageLockExpiry);
  
                      // This flag is checked by synchronous consumers
                      _asynchConsumerRegistered = true;
  
                      // Asynch consumers are always transacted.
                      _transacted = true;
  
                      // Create an ordering group member if we need one.
                      // Ordering in conjunction with pubsub or reading ahead means nothing
                      // so we just ignore the request to order
                      if ((orderingGroup != null) && !_destinationAttachedTo.isPubSub() && !_consumerSession.getReadAhead())
                      {
                          // If we're not just re-registering with the same orderingGroup we have some work
                          // to do.
                          if (_orderingGroup != orderingGroup)
                          {
                              // If we're already in a different group remove ourselves from it first
                              if (_orderingGroup != null)
                                  _consumerKey.leaveKeyGroup();
  
                              _orderingGroup = orderingGroup;
  
                              // Move this consumer's key into a key group that represents this
                              // ordering group on this consumerDispatcher
                              try
                              {
                                  _keyGroup = (LocalQPConsumerKeyGroup) _consumerDispatcher.joinKeyGroup(_consumerKey, orderingGroup);
                              } catch (SIResourceException e)
                              {
                                  // FFDC
                                  FFDCFilter.processException(
                                                              e,
                                                              "com.ibm.ws.sib.processor.impl.JSLocalConsumerPoint.registerAsynchConsumer",
                                                              "1:2367:1.22.5.1",
                                                              this);
  
                                  if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                                      SibTr.exit(tc, "registerAsynchConsumer", "SIIncorrectCallException");
                                  throw new SIIncorrectCallException(
                                                  nls.getFormattedMessage(
                                                                          "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                                          new Object[] {
                                                                                        "com.ibm.ws.sib.processor.impl.ConsumerDispatcher",
                                                                                        "1:2376:1.22.5.1",
                                                                                        SIMPUtils.getStackTrace(e) },
                                                                          null));
                              }
  
                              if (_keyGroup != null)
                              {
                                  // Where we would lock our asynchConsumer we now need to lock the
                                  // entire keyGroup.
                                  // NOTE: if a callback busy lock was supplied this overrides the group
                                  //       lock, we assume this user knows what they are doing (all the ordered
                                  //       members have the same callback lock passed in).
                                  // From this point on the lock we currently hold (asynchConsumer) is next
                                  // to useless as anyone else will lock the asynchGroupLock instead so we
                                  // must not do ANYTHING else important after this point!
                                  if (optionalCallbackBusyLock == null)
                                      _asynchConsumerBusyLock = _keyGroup.getAsynchGroupLock();
                              }
                          }
                      }
                  }
                  else
                  {
                      // The callback is actually being deregistered
                      _asynchConsumerRegistered = false;
                      // Restore the recoverability of the session
                      resetBaseUnrecoverability();
                      _inLine = false;
                      if (_orderingGroup != null)
                      {
                          // Remove us from the keyGroup
                          _consumerKey.leaveKeyGroup();
                          _keyGroup = null;
                          _asynchConsumerBusyLock = _asynchConsumer;
                          _orderingGroup = null;
                      }
                  }
  
                  // We keep a handle to the external lock object as we need to check
                  // with it periodically to see if there is anyone wanting to take
                  // the lock that we hold.
                  _externalConsumerLock = optionalCallbackBusyLock;
              } // synchronized
              finally
              {
                  this.unlock();
              }
          } // synchronized
  
          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.exit(tc, "registerAsynchConsumer");
      }