Invoking this method will cause the running message sweep to be
de-scheduled. Any currently present {@link Message}s will only be removed
from memory when this watch instance is removed from memory.
337249
@Override
    public synchronized boolean stop() {
        if (!this.isStarted()) {
            throw new IllegalStateException("Cannot terminate what was not started.");
        } else if (!this.isStopped.compareAndSet(false, true)) {
            return false;
        }
        DefaultLogWatch.LOGGER.info("Terminating {}.", this);
        this.tailing.stop();
        this.consumers.stop();
        this.handingDown.clear();
        this.storage.logWatchTerminated();
        DefaultLogWatch.LOGGER.info("Terminated {}.", this);
        return true;
    }
789585
@SuppressWarnings("unchecked")
    public <E> MessageHandler<? super E> removeReceivedMessageHandler(Class<E> type) {
        receivedMessageHandlerCache.clear();
        return (MessageHandler<? super E>) receivedMessageHandlers.remove(type);
    }
1082367
@Deprecated
    public void stopWatching(File file) {
        if (file == null) {
            throw new IllegalArgumentException("Null file argument");
        }
        synchronized (mutex) {
            stopWatchingInternal(file.toPath());
        }
    }
1358296
public void messagingEngineDestroyed(JsMessagingEngine messagingEngine)
    {
        final String methodName = "messagingEngineDestroyed";
        if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled())
        {
            SibTr.entry(this, TRACE, methodName, messagingEngine);
        }

        /*
         * If there are no longer any local messaging engines on the required
         * bus, switch to a remote messaging engine
         */

        final JsMessagingEngine[] localMessagingEngines = SibRaEngineComponent
                .getMessagingEngines(_endpointConfiguration.getBusName());

        if (0 == localMessagingEngines.length)
        {
            /*
             * The last local messaging engine for the required bus has been
             * destroyed; we may be able to now connect remotely so kick off
             * a check.
             */

            SibTr.info(TRACE, "ME_DESTROYED_CWSIV0779", new Object[] {
                    messagingEngine.getName(),
                    _endpointConfiguration.getBusName() });

          try
          {
                clearTimer ();
                timerLoop ();
            }
          catch (final ResourceException exception)
          {

                FFDCFilter.processException(exception, CLASS_NAME + "."
                        + methodName, FFDC_PROBE_1, this);
                SibTr.error(TRACE, "MESSAGING_ENGINE_STOPPING_CWSIV0765",
                        new Object[] { exception, messagingEngine.getName(),
                                messagingEngine.getBusName() });
            }
        }
        if (TraceComponent.isAnyTracingEnabled() && TRACE.isEntryEnabled())
        {
            SibTr.exit(this, TRACE, methodName);
        }
    }
363995
@Override
    public void dispatch(final KeyedMessage message, final boolean block) throws IllegalArgumentException, ContainerException {
        if(!isRunningLazy) {
            LOGGER.debug("Dispacth called on stopped container");
            statCollector.messageFailed(false);
        }

        if(message == null)
            return; // No. We didn't process the null message

        if(!inbound.doesMessageKeyBelongToNode(message.key)) {
            if(LOGGER.isDebugEnabled())
                LOGGER.debug("Message with key " + SafeString.objectDescription(message.key) + " sent to wrong container. ");
            statCollector.messageFailed(false);
            return;
        }

        boolean evictedAndBlocking;

        if(message == null || message.message == null)
            throw new IllegalArgumentException("the container for " + clusterId + " attempted to dispatch null message.");

        if(message.key == null)
            throw new ContainerException("Message " + objectDescription(message.message) + " contains no key.");

        try {
            numBeingWorked.incrementAndGet();

            do {
                evictedAndBlocking = false;

                final InstanceWrapper wrapper = getInstanceForKey(message.key);

                // wrapper will be null if the activate returns 'false'
                if(wrapper != null) {
                    final Object instance = wrapper.getExclusive(block);
                    if(instance != null) { // null indicates we didn't get the lock
                        final List<KeyedMessageWithType> response;
                        try {
                            if(wrapper.isEvicted()) {
                                response = null;
                                // if we're not blocking then we need to just return a failure. Otherwise we want to try again
                                // because eventually the current Mp will be passivated and removed from the container and
                                // a subsequent call to getInstanceForDispatch will create a new one.
                                if(block) {
                                    Thread.yield();
                                    evictedAndBlocking = true; // we're going to try again.
                                } else { // otherwise it's just like we couldn't get the lock. The Mp is busy being killed off.
                                    if(LOGGER.isTraceEnabled())
                                        LOGGER.trace("the container for " + clusterId + " failed handle message due to evicted Mp "
                                            + SafeString.valueOf(prototype));

                                    statCollector.messageCollision(message);
                                }
                            } else {
                                response = invokeOperation(wrapper.getInstance(), Operation.handle, message);
                            }
                        } finally {
                            wrapper.releaseLock();
                        }
                        if(response != null) {
                            try {
                                dispatcher.dispatch(response);
                            } catch(final Exception de) {
                                if(isRunning.get())
                                    LOGGER.warn("Failed on subsequent dispatch of " + response + ": " + de.getLocalizedMessage());
                            }
                        }
                    } else { // ... we didn't get the lock
                        if(LOGGER.isTraceEnabled())
                            LOGGER.trace("the container for " + clusterId + " failed to obtain lock on " + SafeString.valueOf(prototype));
                        statCollector.messageCollision(message);
                    }
                } else {
                    // if we got here then the activate on the Mp explicitly returned 'false'
                    if(LOGGER.isDebugEnabled())
                        LOGGER.debug("the container for " + clusterId + " failed to activate the Mp for " + SafeString.valueOf(prototype));
                    // we consider this "processed"
                    break; // leave the do/while loop
                }
            } while(evictedAndBlocking);

        } finally {
            numBeingWorked.decrementAndGet();
        }
    }
315813
public synchronized void stop() {
        if (scheduledFuture != null) {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) {
                Tr.event(this, tc, "Scan task cancelled");
            }
            scheduledFuture.cancel(false);
            scheduledFuture = null;
        }
    }
631068
public void moveMessage(boolean discard) throws SIMPControllableNotFoundException, 
                                                    SIMPRuntimeOperationFailedException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "moveMessage", Boolean.valueOf(discard));
      
    assertValidControllable();
    
    // This is a pubsub messages so we cannot remove the actual message from the ItemStream.
    // In fact, there is no need to remove the underlying message, we leave that to the stream
    // to call back to the PubSubInputHandler to remove the reference to the message if
    // possible (i.e. others are not also referencing it).
    try
    {
      Stream stream = getStream();
      if(stream != null)
        stream.writeSilenceForced(getStartTick());
    }
    catch (SIException e)
    {
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.runtime.LinkTransmitMessageControl.moveMessage",
        "1:184:1.5",
        this);
  
      SIMPRuntimeOperationFailedException finalE =
        new SIMPRuntimeOperationFailedException(
          nls.getFormattedMessage(
            "INTERNAL_MESSAGING_ERROR_CWSIP0003",
            new Object[] {"TransmitMessage.moveMessage",
                "1:192:1.5",
                          e,
                          new Long(getStartTick())},
            null), e);

      SibTr.exception(tc, finalE);
      
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "moveMessage", finalE);
      
      throw finalE;             
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "moveMessage");
  }
1066272
public void removeMessage(CmsNotificationMessage message) {

        m_messages.remove(message);
        if (hasWidget()) {
            m_widget.removeMessage(message);
        }
    }
745015
public void removeMessageListener(MessageListener<? extends Message> messageListener) {
        removeListener(messageListener.getChannelIdentifier(), messageListener.getMessageType(), messageListener);
    }
95944
public void deregisterForEvents(SIMPMessage msg)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "deregisterForEvents");
      
    msg.deregisterMessageEventListener(MessageEvents.POST_COMMIT_REMOVE, this);
    msg.deregisterMessageEventListener(MessageEvents.POST_ROLLBACK_REMOVE, this);
    msg.deregisterMessageEventListener(MessageEvents.POST_COMMIT_ADD, this);
    msg.deregisterMessageEventListener(MessageEvents.POST_ROLLBACK_ADD, this);
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "deregisterForEvents");
  }
196557
public void removeMessage(Message message) {
        synchronized (messages) {
            messages.removeIf(messageRef -> Objects.equals(messageRef.get(), message));
        }
    }
708080
public void moveMessage(boolean discard)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "removeMessage", new Boolean(discard));

    InvalidOperationException finalE =
      new InvalidOperationException(
        nls.getFormattedMessage(
          "INTERNAL_MESSAGING_ERROR_CWSIP0005",
          new Object[] {"RemoteMessageRequest.removeMessage",
                        "1:259:1.34",
                        this},
          null));

    SibTr.exception(tc, finalE);

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "removeMessage", finalE);
    throw finalE;
  }
1295960
private void ensureWatchThreadRunning() {
		if (!threadRunning) {
			thread = new Thread(watchThread);
			thread.setDaemon(true);
			thread.start();
			watchThread.setThread(thread);
			watchThread.updateName();
			threadRunning = true;
		}
	}
939577
private int lockMessages(boolean isolatedRun) throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "lockMessages", new Object[] { this, Boolean.valueOf(isolatedRun) });

        int numOfMsgs = 0;

        _transferredConsumer = null;
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "GatherMessages: " + Boolean.valueOf(_gatherMessages)
                            + ", instance isolatedRun: " + Boolean.valueOf(_isolatedRun)
                            + ", _hasLocalQP: " + Boolean.valueOf(_hasLocalQP));
        // If we're doing an isolated run then we don't want to become ready
        // but we do still want to drain the queue. Unfortunately, if we're remote
        // from the queue we probably won't have a message sat there waiting and
        // therefore we do need to make ourselves ready and issue a request back
        // to the DME. (unless we are gathering in which case we have to treat 
        // the consumer as if it is local)
        if (!isolatedRun || (!_hasLocalQP || _gatherMessages))
        {
            // If the LCP is stopped then do nothing (unless we're an isolated run)
            if (_stopped && !isolatedRun)
            {
                //we're not ready for more messages
                unsetReady();
                //don't try and get any more
                _drainQueue = false;
            }
            else
            {
                // we are ready for new messages, before we look for one on the ItemStream
                // we tell teh COnsumerDispatcher that we're available for messages just to
                // make sure we don't miss one coming in while we're looking on the ItemStream
                // (and not finding any). If we do find a suitable message on the ItemStream
                // we mark ourselves as unready. This obviously introduces the possibility of
                // the ConsumerDispatcher trying to give us a message, but better a false
                // possitive than a message falling down the gap.
                try
                {
                    setReady();
                } catch (SINotPossibleInCurrentConfigurationException e)
                {
                    // No FFDC code needed
                    //don't try and get any more
                    _drainQueue = false;
                }
            }

            // Remember we're in the middle of an isolated run (so that put() knows
            // what to do)
            if (isolatedRun)
                _isolatedRun = isolatedRun;
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(tc, "Instance isolatedRun: " + Boolean.valueOf(_isolatedRun));
        }

        if ((_keyGroup != null) && !_keyGroup.isStarted())
            _drainQueue = false;

        boolean consumerReady = true;

        // If the consumer has asked for batches of messages we'll look for that number
        // of available messages on the queue and give them to the asynch consumer all in
        // one go. If this is a stoppable consumer we'd like to keep the number of messages
        // being processed within the number of possible hidden messages (which equals the
        // maxSequentialFailuresThreshold). So we crop the batch size to the remaining
        // space we have left for hidden messages (or one) whichever is the larger.
        // However, that would limit the batch size when there are no problems which is a little
        // rough so we only do the cropping once we've started hiding messages (which can result
        // in a few more messages being processed/hidden to start with but that's life). 
        int currentMaxBatchSize = _maxBatchSize;
        if (_consumerStoppable && (currentMaxBatchSize > 1) && (_hiddenMessages.size() > 0))
        {
            int remainingHiddenSpace = _maxSequentialFailuresThreshold - _hiddenMessages.size();
            if (remainingHiddenSpace < 1)
                remainingHiddenSpace = 1;
            if (remainingHiddenSpace < currentMaxBatchSize)
                currentMaxBatchSize = remainingHiddenSpace;

            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(tc, "lockMessages", "Cropped maxBatchSize to " + currentMaxBatchSize);
        }

        //while we still want more messages
        while ((numOfMsgs < currentMaxBatchSize) && // filled the batch
               _drainQueue && // emptied the queue
               (!_lockRequested || (numOfMsgs == 0)) && // Asked to interupt
               (!_interruptConsumer || (numOfMsgs == 0)) && // Interrupted (ignore first time round)
               (_transferredConsumer == null)) // switched ordered consumer
        {
            SIMPMessage msg = null;

            if (_keyGroup != null)
            {
                // If we are part of a group, we want to remember which member of the group actually gets matched
                // when the filter is called.
                _keyGroup.setConsumerActive(true);
            }

            //try and get a locked eligible message
            msg = getEligibleMsgLocked(null);

            if (msg != null)
            {
                JSLocalConsumerPoint matchingConsumer = this;

                // We've found a message that matches. If we are a member of
                // an ordering group it is possible that this message matches
                // a different member of the group from us. If that is that is
                // the case we have to possibilities....
                // i) If this is the fist message in the batch we can pass the locked
                // message to the matching consumer for them to process.
                // ii) If we already have messages in our batch we will have to
                // unlock this message and process the batch first.
                if (_keyGroup != null)
                {
                    // find out which member of the group actually got matched
                    matchingConsumer = (JSLocalConsumerPoint) _keyGroup.getMatchingMember(_consumerKey).getConsumerPoint();
                    // and show that we're done with the filter for now
                    _keyGroup.setConsumerActive(false);
                }

                // The message is intended for this consumer
                if (matchingConsumer == this)
                {
                    // If we've got a message then we're no longer interested
                    // in messages straight from the ConsumerDispatcher but if
                    // looking for a batch of messages we may take a long time
                    // to exit this loop, so take ourselves out of the runnin
                    // as soon as possible.
                    if (consumerReady)
                    {
                        consumerReady = false;
                        unsetReady();
                    }

                    // Do we really need this message in the MS?
                    boolean isRecoverable = true;
                    if ((_unrecoverableOptions != Reliability.NONE) &&
                        (msg.getReliability().compareTo(_unrecoverableOptions) <= 0))
                        isRecoverable = false;

                    registerForEvents(msg);

                    //store a reference to the locked message in the LME
                    _allLockedMessages.addNewMessage(msg, true, isRecoverable);

                    //If we were successful, increment our counter
                    numOfMsgs++;

                    // We check for an interupt after every message is locked
                    if (_externalConsumerLock != null)
                        _interruptConsumer = _externalConsumerLock.isLockYieldRequested();
                }
                // The message is intended for another member of the key group
                else
                {
                    // If we already have messages locked to this consumer we must
                    // unlock the last message and process the others first. Then
                    // we can come back and pick this one up later (if it is still
                    // available)
                    if (numOfMsgs > 0)
                    {
                        try
                        {
                            msg.unlockMsg(msg.getLockID(), null, true);
                            msg = null;
                        } catch (MessageStoreException e)
                        {
                            // MessageStoreException shouldn't occur so FFDC.
                            FFDCFilter.processException(
                                                        e,
                                                        "com.ibm.ws.sib.processor.impl.JSLocalConsumerPoint.lockMessages",
                                                        "1:2995:1.22.5.1",
                                                        this);

                            SibTr.exception(tc, e);

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

                            throw new SIResourceException(
                                            nls.getFormattedMessage(
                                                                    "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                                    new Object[] {
                                                                                  "com.ibm.ws.sib.processor.impl.JSLocalConsumerPoint",
                                                                                  "1:3008:1.22.5.1",
                                                                                  e },
                                                                    null),
                                            e);
                        }
                    }
                    // This is the only message locked so we can just drive the
                    // correct consumer
                    else
                    {
                        numOfMsgs++;

                        // Mark the fact that we've transferred ownership of a locked message
                        // so that it can be processed by the calling method (outside of any
                        // LCP locks)
                        _transferredConsumer = matchingConsumer;

                        _transferredConsumer.lock();
                        try
                        {
                            _transferredConsumer.unsetReady();
                            _transferredConsumer.addLockedMessage(msg);
                        } finally
                        {
                            _transferredConsumer.unlock();
                        }

                        // We now want to drop out of the loop so we can process this message
                        // with the correct consumer

                        // We check for an interupt after every message is locked
                        if (_externalConsumerLock != null)
                            _interruptConsumer = _externalConsumerLock.isLockYieldRequested();
                    }
                }
            }
            if (msg == null)
            {
                // We have all the messages currently on the
                // queue so we are about to callack to
                // get them processed.

                //stop trying to get any more
                _drainQueue = false;
            }

            // We check for an interupt after every message is locked
            if (_externalConsumerLock != null)
                _interruptConsumer = _externalConsumerLock.isLockYieldRequested();
        } // end while

        // If we're an isolated run and we found a message (i.e. not ready) then
        // no-one will try to deliver a message to us and therefore we can forget
        // that we were ever here. On the other hand, if we couldn't find a message
        // and the queue is remote we need to signal that we're still waiting.
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "Instance isolatedRun: " + Boolean.valueOf(_isolatedRun) +
                            ", method isolatedRun: " + Boolean.valueOf(isolatedRun) +
                            ", ready: " + Boolean.valueOf(_ready));
        if (isolatedRun && !_ready)
            _isolatedRun = false;

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "lockMessages", Integer.valueOf(numOfMsgs));

        return numOfMsgs;
    }
598030
private Watcher wrapWatcher(final Watcher watcher) {
    if (watcher == null) {
      return null;
    }
    return new Watcher() {
      @Override
      public void process(final WatchedEvent event) {
        if (eventExecutor.isShutdown()) {
          LOG.debug("Already shutdown. Discarding event: {}", event);
          return;
        }
        eventExecutor.execute(new Runnable() {
          @Override
          public void run() {
            try {
              watcher.process(event);
            } catch (Throwable t) {
              LOG.error("Watcher throws exception.", t);
            }
          }
        });
      }
    };
  }
195032
protected void disconnectServiceUpIsRunning() {
        if (serviceProcessIsRunning != null) serviceProcessIsRunning.stop();
        // set null so the SERVICE_UP enricher runs (possibly removing it), then remove so
        // everything is removed
        // TODO race because the is-running check may be mid-task
        setAttribute(SERVICE_PROCESS_IS_RUNNING, null);
        removeAttribute(SERVICE_PROCESS_IS_RUNNING);
    }
870729
public void free()
    {
        if (m_messageMap != null)
        {
            for (BaseMessageQueue messageQueue : m_messageMap.values())
            {
                if (messageQueue != null)
                {       // Don't worry about removing these, since you are removing them all.
                    messageQueue.setMessageManager(null);   // Make sure it doesn't try to remove itself again
                    messageQueue.free();
                }                
            }
        }
        m_messageMap = null;
        super.free();
    }