docs: public ConsumableKey attachToRemoteDurableSubscription(
                                                             LocalConsumerPoint consumerPoint,
                                                             ConsumerDispatcherState subState,
                                                             SIBUuid8 durableMEUuid)
                      throws
                      SIDurableSubscriptionNotFoundException,
                      SIDestinationLockedException,
                      SISelectorSyntaxException,
                      SIDiscriminatorSyntaxException,
                      SINotPossibleInCurrentConfigurationException,
                      SIResourceException,
                      SIDurableSubscriptionMismatchException
      {
          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.entry(
                          tc,
                          "attachToRemoteDurableSubscription",
                          new Object[] { consumerPoint, subState, durableMEUuid });
  
          // We'll send the request to the remote ME, from which one of four replies may result:
          // 1) success
          // 2) failure because someone else is already attached
          // 3) failure because the subscription does not exist
          // 4) some other failure (storage problems, bugs, etc).
  
          //get the durable ME uuid
          String remSubName =
                          _baseDestinationHandler.constructPseudoDurableDestName(
                                                                                 durableMEUuid.toString(),
                                                                                 subState.getSubscriberID());
  
          //defect 259036: we do not need to create a consumer dispatcher
          //if one already exists and we are a cloned subscription
          boolean createConsumerDispatcher = true;
  
          ConsumableKey data = null;
  
          synchronized (_consumerDispatchersDurable)
          {
              // An existing ConsumerDispatcher implies an existing connected client which
              // means the connect attempt fails...unless we're waiting on a durable
              // subscription which is in the process of being deleted.  In which case we'll
              // see if the flush is expected (DME is up) and if so kick it a bit to give it
              // another chance (or ten)
              Object current = null;
              boolean tryAgain = true;
              int remainingAttempts = 10;
  
              // Keep trying this until it works, fails, or we get bored
              while (tryAgain)
              {
                  tryAgain = false;
  
                  // See if we already have an RCD for this subscription
                  current = _consumerDispatchersDurable.get(remSubName);
  
                  // If there is a RCD for this subscription but it's pending a delete try to
                  // get in and cancel the delete, if we succeed we can re-use the RCD, otherwise
                  // we'll drop out and try again in a bit, to give it a chance to complete
                  // the delete.
                  if (current != null)
                  {
                      // We don't just store RCDs in this map - we stick in Strings to represent RCDs
                      // currently being created (!)
                      if (current instanceof RemoteConsumerDispatcher)
                      {
                          RemoteConsumerDispatcher rcd = (RemoteConsumerDispatcher) current;
                          // If the RCD is in pending delete state there's a chance we can get in and
                          // cancel the delete to re-use it. Or, at the very least we can get it to
                          // resend the flush if the DME is up but wasn't before.
                          if (rcd.getPendingDelete())
                          {
                              current = rcd.getResolvedDurableCD(rcd);
  
                              // A null returned indicates the RCD is not in a usable state (cleaning up)
                              // Otherwise, we've rescued the old RCD for further use 
                              if (current != null)
                                  ((RemoteConsumerDispatcher) current).setPendingDelete(false);
                          }
                          // If it's not in pending delete, we'll try to use it
                      }
                      // A String represents an RCD/AIH that's being created. If the subscription is
                      // clonable (supports multiple consumers) we'll wait for the completion as we'll
                      // be able to use it. Otherwise, we couldn't use it even if it existed (receive
                      // exclusive) so bomb out now.
                      else if ((current instanceof String) && subState.isCloned())
                      {
                          // Set to null so we go round the loop again (after a delay)
                          current = null;
                      }
  
                      // If we don't have an RCD anymore (we did when we came in) we wait for a bit and
                      // try again (unless we've bored of waiting)
                      if (current == null)
                      {
                          remainingAttempts--;
  
                          // If we've already tried a few times, give up - something else has
                          // maybe gone wrong
                          if (remainingAttempts == 0)
                          {
                              if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                                  SibTr.exit(
                                             tc,
                                             "attachToRemoteDurableSubscription",
                                             "Consumers already attached");
  
                              throw new SIDestinationLockedException(
                                              nls.getFormattedMessage(
                                                                      "SUBSCRIPTION_IN_USE_ERROR_CWSIP0152",
                                                                      new Object[] {
                                                                                    subState.getSubscriberID(),
                                                                                    _messageProcessor.getMessagingEngineName() },
                                                                      null));
                          }
  
                          // Otherwise, sleep for a little but (releasing the lock) and try again later.
                          // If the cleanup of the RCD/AIH completes while we're waiting we'll be notified.
                          try
                          {
                              if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                                  SibTr.debug(tc, "Waiting on " + _consumerDispatchersDurable);
                              _consumerDispatchersDurable.wait(100);
                          } catch (InterruptedException e)
                          {
                              // No FFDC code needed
                          }
                          tryAgain = true;
                      }
                  }
                  else if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                      SibTr.debug(tc, "No existing consumer dispatcher found");
              }
  
              // By this point we should either have something in our hands (an RCD or a String) for
              // us to use. Or we need to create a new one
  
              if (current != null)
              {
                  createConsumerDispatcher = false;
  
                  // If we have an RCD we try to attach to it, if we're not clonable and someone is already
                  // attached we'll fail.
                  if (current instanceof RemoteConsumerDispatcher)
                  {
                      try
                      {
                          data =
                                          (ConsumableKey) ((RemoteConsumerDispatcher) current).attachConsumerPoint(
                                                                                                                   consumerPoint,
                                                                                                                   null,
                                                                                                                   consumerPoint.getConsumerSession().getConnectionUuid(),
                                                                                                                   consumerPoint.getConsumerSession().getReadAhead(),
                                                                                                                   consumerPoint.getConsumerSession().getForwardScanning(),
                                                                                                                   null);
                      } catch (SIResourceException e)
                      {
                          // No FFDC code needed
  
                          // See if there is a consumer already attached locally, to give a better exception
                          // text
                          if (((RemoteConsumerDispatcher) current).hasConsumersAttached())
                          {
                              if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                                  SibTr.exit(
                                             tc,
                                             "attachToRemoteDurableSubscription",
                                             new Object[] { "Consumers already attached locally", e });
  
                              throw new SIDestinationLockedException(
                                              nls.getFormattedMessage(
                                                                      "SUBSCRIPTION_IN_USE_ERROR_CWSIP0152",
                                                                      new Object[] {
                                                                                    subState.getSubscriberID(),
                                                                                    _messageProcessor.getMessagingEngineName() },
                                                                      null));
                          }
                          // Otherwise, check to see if the stream was created, again for a better exception text
                          else if (((RemoteConsumerDispatcher) current).getAnycastInputHandler().
                                          testStreamStatus(AnycastInputHandler.AIStreamStatus.STREAM_NON_EXISTENT))
                          {
                              if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                                  SibTr.exit(
                                             tc,
                                             "attachToRemoteDurableSubscription",
                                             new Object[] { "Consumers already attached remotely", e });
  
                              throw new SIDestinationLockedException(
                                              nls.getFormattedMessage(
                                                                      "SUBSCRIPTION_IN_USE_ERROR_CWSIP0152",
                                                                      new Object[] {
                                                                                    subState.getSubscriberID(),
                                                                                    _messageProcessor.getMessagingEngineName() },
                                                                      null));
                          }
                          // Otherwise, just throw what we were given
                          else
                          {
                              if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                                  SibTr.exit(
                                             tc,
                                             "attachToRemoteDurableSubscription",
                                             new Object[] { "Failed to attach remotely", e });
                              throw e;
                          }
                      }
                  }
                  else if (!subState.isCloned() && (current instanceof String))
                  {
                      // This is a String so must be a placeholder for a creating one, which means
                      // someone else is about to attach so we would fail anyway (as we're not cloned)
                      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                          SibTr.exit(
                                     tc,
                                     "attachToRemoteDurableSubscription",
                                     "Consumer in the process of attaching locally");
  
                      throw new SIDestinationLockedException(
                                      nls.getFormattedMessage(
                                                              "SUBSCRIPTION_IN_USE_ERROR_CWSIP0152",
                                                              new Object[] {
                                                                            subState.getSubscriberID(),
                                                                            _messageProcessor.getMessagingEngineName() },
                                                              null));
                  }
                  else
                  {
                      //The durable home tells us that this subscription
                      //is remote, but now we find that there is a non-remote
                      //ConsumerDispatcher for the subscription.
                      //This is inconsistent so we throw an exception
                      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                          SibTr.exit(
                                     tc,
                                     "attachToRemoteDurableSubscription",
                                     "A ConsumerDispatcher for this subscription exists locally.");
  
                      throw new SIDestinationLockedException(
                                      nls.getFormattedMessage(
                                                              "SUBSCRIPTION_ATTACH_FAILED_CWSIP0035",
                                                              new Object[] { subState.getSubscriberID(), durableMEUuid },
                                                              null));
                  }
              }
  
              if (createConsumerDispatcher)
              {
                  // Ok, no CD yet, which means we'll have to attempt to create a ConsumerDispatcher.
                  // If we do all this while we're holding the lock then we may hold up all other durable
                  // subs on this topic.  So create a bogus durSub entry and release the lock.  This will
                  // only block out durSubs attempting to connect to exactly the same durable subscription.
                  // If the create succeeds, then we'll fix the entry.  Otherwise we'll just remove it.
                  _consumerDispatchersDurable.put(remSubName, remSubName);
              }
          } //end sync
  
          if (createConsumerDispatcher)
          {
              //we are now in a position to create the RCD
              // RCD we'll use to attach our ConsumerPoint below
              RemoteConsumerDispatcher rcd = null;
  
              try
              {
                  rcd = _pubSubRemoteSupport.
                                  createRemoteConsumerDispatcher(remSubName,
                                                                 subState,
                                                                 durableMEUuid);
              } catch (Exception e)
              {
                  // No FFDC code needed
                  // Cleanup before exiting
                  synchronized (_consumerDispatchersDurable)
                  {
                      _consumerDispatchersDurable.remove(remSubName);
                  }
  
                  // rethrow not founds
                  if (e instanceof SIDurableSubscriptionNotFoundException)
                  {
                      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                          SibTr.exit(
                                     tc,
                                     "attachToRemoteDurableSubscription",
                                     "SIDurableSubscriptionNotFoundException");
                      throw (SIDurableSubscriptionNotFoundException) e;
                  }
  
                  // rethrow locked
                  if (e instanceof SIDestinationLockedException)
                  {
                      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                          SibTr.exit(
                                     tc,
                                     "attachToRemoteDurableSubscription",
                                     "SIDestinationLockedException");
                      throw (SIDestinationLockedException) e;
                  }
  
                  // rethrow resource exceptions
                  if (e instanceof SIResourceException)
                  {
                      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                          SibTr.exit(
                                     tc,
                                     "attachToRemoteDurableSubscription",
                                     "SIResourceException");
                      throw (SIResourceException) e;
                  }
  
                  // rethrow durable mismatch exception
                  if (e instanceof SIDurableSubscriptionMismatchException)
                  {
                      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                          SibTr.exit(
                                     tc,
                                     "attachToRemoteDurableSubscription",
                                     "SIDurableSubscriptionMismatchException");
                      throw (SIDurableSubscriptionMismatchException) e;
                  }
  
                  // There shouldn't be anything else, but if so FFDC the
                  // original and rethrow an SIErrorException
  
                  // Otherwise, exception shouldn't occur so FFDC.
                  FFDCFilter.processException(
                                              e,
                                              "com.ibm.ws.sib.processor.impl.destination.PubSubRealization.attachToRemoteDurableSubscription",
                                              "1:3044:1.35.2.4",
                                              this);
  
                  SibTr.exception(tc, e);
  
                  if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                      SibTr.exit(
                                 tc,
                                 "attachToRemoteDurableSubscription",
                                 "SIErrorException");
                  throw new SIErrorException(
                                  nls.getFormattedMessage(
                                                          "INTERNAL_MESSAGING_ERROR_CWSIP0001",
                                                          new Object[] {
                                                                        "com.ibm.ws.sib.processor.impl.destination.PubSubRealization",
                                                                        "1:3059:1.35.2.4" },
                                                          null));
              }
  
              // Register the "real" consumer dispatcher and attach our consumer point
              synchronized (_consumerDispatchersDurable)
              {
                  _consumerDispatchersDurable.put(remSubName, rcd);
  
                  //we do the attach in the synschronized
                  data =
                                  (ConsumableKey) rcd.attachConsumerPoint(
                                                                          consumerPoint,
                                                                          null,
                                                                          consumerPoint.getConsumerSession().getConnectionUuid(),
                                                                          consumerPoint.getConsumerSession().getReadAhead(),
                                                                          consumerPoint.getConsumerSession().getForwardScanning(),
                                                                          null);
  
                  // Wake up anyone who may be waiting for this
                  if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                      SibTr.debug(tc, "Notifying all on " + _consumerDispatchersDurable);
                  _consumerDispatchersDurable.notifyAll();
              }
  
          } //end create RCD
  
          // All done, return
          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.exit(tc, "attachToRemoteDurableSubscription", data);
  
          return data;
      }