gt:     i = _registeredWildcardConsumerMonitors.keySet().iterator();
context: /*******************************************************************************
   * Copyright (c) 2012 IBM Corporation and others.
   * All rights reserved. This program and the accompanying materials
   * are made available under the terms of the Eclipse Public License v1.0
   * which accompanies this distribution, and is available at
   * http://www.eclipse.org/legal/epl-v10.html
   *
   * Contributors:
   *     IBM Corporation - initial API and implementation
   *******************************************************************************/
  package com.ibm.ws.sib.processor.matching;
  
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.Map;
  
  import com.ibm.websphere.ras.TraceComponent;
  import com.ibm.ejs.ras.TraceNLS;
  import com.ibm.websphere.sib.exception.SIErrorException;
  import com.ibm.websphere.sib.exception.SINotPossibleInCurrentConfigurationException;
  import com.ibm.wsspi.sib.core.ConsumerSetChangeCallback;
  import com.ibm.ws.sib.processor.SIMPConstants;
  import com.ibm.ws.sib.processor.impl.ConnectionImpl;
  import com.ibm.ws.sib.processor.impl.MessageProcessor;
  import com.ibm.ws.sib.utils.ras.SibTr;
  import com.ibm.wsspi.sib.core.exception.SIDiscriminatorSyntaxException;
  
  /**
   * This class manages the set of consumer monitors that have been registered
   * with the system.
   *
   * The class was provided to support WSN Registered publishers.
   */
  public class ConsumerMonitorRegistrar
  {
    private static final TraceComponent tc =
      SibTr.register(
        ConsumerMonitorRegistrar.class,
        SIMPConstants.MP_TRACE_GROUP,
        SIMPConstants.RESOURCE_BUNDLE);
  
    // NLS for component
    private static final TraceNLS nls =
      TraceNLS.getTraceNLS(SIMPConstants.RESOURCE_BUNDLE);
  
    // Back references to the messageprocessor and mpm
    private MessageProcessor _messageProcessor;
    private MessageProcessorMatching _mpm;
  
    // Maps of registered consumer monitors
    private Map _registeredExactConsumerMonitors = new HashMap();
    private Map _registeredWildcardConsumerMonitors = new HashMap();
  
    // The callback index is keyed on callback object and allows us to
    // retrieve the topicexpression against which it was registered (specifically
    // when deregistering consumer set monitors)
    private Map _callbackIndex = new HashMap();
  
    private class RegisteredCallbacks
    {
      private ArrayList wrappedCallbacks;
      private ArrayList matchingConsumers;
  
      RegisteredCallbacks(ArrayList wrappedCallbacks,
                         ArrayList matchingConsumers)
      {
        this.wrappedCallbacks = wrappedCallbacks;
        this.matchingConsumers = matchingConsumers;
      }
  
      public ArrayList getWrappedCallbacks()
      {
        return wrappedCallbacks;
      }
  
      public ArrayList getMatchingConsumers()
      {
        return matchingConsumers;
      }
    }
  
    /**
     * TopicRecords are stored in the _callbackIndex and are used to help locate
     * appropriate registrar table entries using a key of callback.
     */
    private class TopicRecord
    {
      public String topicExpression;
      public boolean isWildcarded;
  
      TopicRecord(String topicExpression, boolean isWildcarded)
      {
        this.topicExpression = topicExpression;
        this.isWildcarded = isWildcarded;
      }
    }
  
    //------------------------------------------------------------------------------
    // Constructors for ConsumerMonitorRegistrar
    //------------------------------------------------------------------------------
  
    public ConsumerMonitorRegistrar(MessageProcessor messageProcessor,
                                    MessageProcessorMatching mpm)
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.entry(tc, "ConsumerMonitorRegistrar", new Object[]{ messageProcessor, mpm});
  
      // Set ref to messageProcessor and mpm
      _messageProcessor = messageProcessor;
      _mpm = mpm;
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "ConsumerMonitorRegistrar", this);
    }
  
    /**
     * Method registerCallbackOnNewExpression
     *
     * Adds a list of matching consumers in a new entry to the appropriate wildcard
     * or exact expression table.
     *
     * @param topicExpression
     * @param isWildcarded
     * @param callback
     * @param matchingConsumers
     * @return
     */
    public boolean registerCallbackOnNewExpression(
      ConnectionImpl connection,
      String topicExpression,
      boolean isWildcarded,
      ConsumerSetChangeCallback callback,
      ArrayList matchingConsumers)
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
          SibTr.entry(
            tc,
            "registerCallbackOnNewExpression",
            new Object[] { connection,
                           topicExpression,
                           new Boolean(isWildcarded),
                           callback,
                           matchingConsumers});
  
      boolean areConsumers = false;
  
      // Add the new callback to the index, so we can find it when we deregister
      addCallbackToConnectionIndex(connection, topicExpression, isWildcarded, callback);
  
      // Set up the callback list for this registration
      ArrayList callbackList = new ArrayList();
      WrappedConsumerSetChangeCallback wcb = new WrappedConsumerSetChangeCallback(callback);
      callbackList.add(wcb);
  
      // Register the monitor in the appropriate table
      RegisteredCallbacks regMonitor =
        new RegisteredCallbacks(callbackList, matchingConsumers);
  
      if(isWildcarded)
        _registeredWildcardConsumerMonitors.put(topicExpression, regMonitor);
      else
        _registeredExactConsumerMonitors.put(topicExpression, regMonitor);
  
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "registerCallbackOnNewExpression", new Boolean(areConsumers));
      return areConsumers;
    }
  
    /**
     * Method deregisterMonitor
     *
     * Deregisters a previously registered callback.
     *
     * @param callback
     * @throws SINotPossibleInCurrentConfigurationException
     */
    public void deregisterMonitor(ConnectionImpl connection,
                                  ConsumerSetChangeCallback callback)
    throws SINotPossibleInCurrentConfigurationException
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
          SibTr.entry(
            tc,
            "deregisterMonitor",
            new Object[] { connection, callback});
  
      // Retrieve the callback from the index
      TopicRecord tRecord = retrieveCallbackFromConnectionIndex(connection, callback);
  
      if(tRecord != null)
      {
        // Have found the callback in the index, now remove the callback from
        // the appropriate registered monitors table.
        if(tRecord.isWildcarded)
        {
          removeCallbackFromRegisteredMonitors(callback,
                                               tRecord.topicExpression,
                                               true, // isWildcarded
                                               _registeredWildcardConsumerMonitors);
        }
        else
        {
          // Handle the non-wildcard case
          removeCallbackFromRegisteredMonitors(callback,
                                               tRecord.topicExpression,
                                               false, // isWildcarded
                                               _registeredExactConsumerMonitors);
        }
      }
      else
      {
        // Failed to find any reference to callback
        // Build the message for the Exception
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
          SibTr.exit(tc, "deregisterMonitor", callback);
  
        SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0005",
          new Object[] {
            "com.ibm.ws.sib.processor.matching.ConsumerMonitorRegistrar",
            "1:237:1.10",
            callback });
        throw new SINotPossibleInCurrentConfigurationException(
          nls.getFormattedMessage(
            "INTERNAL_MESSAGING_ERROR_CWSIP0005",
            new Object[] {
              "com.ibm.ws.sib.processor.matching.ConsumerMonitorRegistrar",
              "1:244:1.10",
              callback },
            null));
  
      }
  
      // Now remove the callback from the index
      removeCallbackFromConnectionIndex(connection, callback);
  
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "deregisterMonitor");
    }
  
    /**
     * Method removeConsumerSetMonitors
     *
     * Removes all callbacks associated with a connection.
     *
     * @param connection
     * @throws SINotPossibleInCurrentConfigurationException
     */
    public void removeConsumerSetMonitors(ConnectionImpl connection)
    throws SINotPossibleInCurrentConfigurationException
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.entry(tc, "removeConsumerSetMonitors", new Object[] { connection });
  
      // Locate the connection in the callback index
      if(_callbackIndex.containsKey(connection))
      {
        // Have found registered callbacks for this connection
        Map connMap = (HashMap)_callbackIndex.get(connection);
  
        // Iterate through the map removing individual callbacks
        Iterator i = connMap.entrySet().iterator();
        while (i.hasNext())
        {
          Map.Entry entry = (Map.Entry)i.next();
          ConsumerSetChangeCallback callback = (ConsumerSetChangeCallback)entry.getKey();
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "Found registered callback: " + callback);
  
          // Locate the TopicRecord for the callback
          TopicRecord tRecord = (TopicRecord)entry.getValue();
          if(tRecord != null)
          {
            // Have found the callback in the index, now remove the callback from
            // the appropriate registered monitors table.
            if(tRecord.isWildcarded)
            {
              removeCallbackFromRegisteredMonitors(callback,
                                                   tRecord.topicExpression,
                                                   true, // isWildcarded
                                                   _registeredWildcardConsumerMonitors);
            }
            else
            {
              // Handle the non-wildcard case
              removeCallbackFromRegisteredMonitors(callback,
                                                   tRecord.topicExpression,
                                                   false, // isWildcarded
                                                   _registeredExactConsumerMonitors);
            }
          }
        }
      }
  
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "removeConsumerSetMonitors");
    }
  
    /**
     * Method removeCallbackFromRegisteredMonitors
     *
     * Removes a previously registered callback from the appropriate map.
     *
     * Also walks through the list of matching consumers removing references to this
     * registration.
     *
     * @param callback
     * @param topicExpression
     * @param isWildcarded
     * @param registeredMonitorsMap
     * @throws SINotPossibleInCurrentConfigurationException
     * @throws SIErrorException
     */
    public void removeCallbackFromRegisteredMonitors(
      ConsumerSetChangeCallback callback,
      String topicExpression,
      boolean isWildcarded,
      Map registeredMonitorsMap)
    throws SINotPossibleInCurrentConfigurationException,
           SIErrorException
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
          SibTr.entry(
            tc,
            "removeCallbackFromRegisteredMonitors",
            new Object[] { callback,
                           topicExpression,
                           new Boolean(isWildcarded)});
  
      ArrayList matchingConsumerList = null;
  
      RegisteredCallbacks rMonitor = (RegisteredCallbacks)registeredMonitorsMap.get(topicExpression);
  
      //Handle the case where we cannot find the monitor
      if(rMonitor != null)
      {
        // Have located the monitor record
        // Get the list of callbacks prepare to remove this one
        ArrayList callbackList = rMonitor.getWrappedCallbacks();
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          SibTr.debug(tc, "Found existing entry with callbacks: " + callbackList);
        WrappedConsumerSetChangeCallback wcb = new WrappedConsumerSetChangeCallback(callback);
        boolean cbRemoved = callbackList.remove(wcb);
        if(!cbRemoved)
        {
          //Couldn't find the callback. This means we have an internal inconsistency
          // in our tables, so throw an SIErrorException
  
          // Build the message for the Exception
          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "removeCallbackFromRegisteredMonitors", callback);
  
          SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0005",
            new Object[] {
              "com.ibm.ws.sib.processor.matching.ConsumerMonitorRegistrar",
              "1:372:1.10",
              callback });
          throw new SIErrorException(
            nls.getFormattedMessage(
              "INTERNAL_MESSAGING_ERROR_CWSIP0005",
              new Object[] {
                "com.ibm.ws.sib.processor.matching.ConsumerMonitorRegistrar",
                "1:379:1.10",
                callback },
              null));
  
        }
  
        // If the callbacklist is now empty, then we need to walk the matching
        // consumers, removing references and delete the record in the table
        if(callbackList.isEmpty())
        {
          // Get the list of matching consumers and remove the monitor references from
          // them.
          matchingConsumerList = rMonitor.getMatchingConsumers();
          // Are there consumers for this expression
          boolean areConsumers = !matchingConsumerList.isEmpty();
  
          if(areConsumers)
          {
            // Remove reference to this registration from each consumer
            Iterator i = matchingConsumerList.iterator();
            while (i.hasNext())
            {
              MonitoredConsumer mc = (MonitoredConsumer)i.next();
  
              if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(tc, "Remove registration from consumer: " + mc);
  
              // Remove the reference from the wildcarded list in the consumer
              boolean removed;
              if(isWildcarded)
                removed = mc.removeMatchingWildcardMonitor(topicExpression);
              else
                removed = mc.removeMatchingExactMonitor(topicExpression);
  
              if(!removed)
              {
                // Couldn't find the registration. This means we have an internal
                // inconsistency in our tables, so throw an SIErrorException
  
                // Build the message for the Exception
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                  SibTr.exit(tc, "removeCallbackFromRegisteredMonitors", mc + ":" + topicExpression);
  
                SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0005",
                  new Object[] {
                    "com.ibm.ws.sib.processor.matching.ConsumerMonitorRegistrar",
                    "1:425:1.10",
                    mc + ":" + topicExpression });
                throw new SIErrorException(
                  nls.getFormattedMessage(
                    "INTERNAL_MESSAGING_ERROR_CWSIP0005",
                    new Object[] {
                      "com.ibm.ws.sib.processor.matching.ConsumerMonitorRegistrar",
                      "1:432:1.10",
                      mc + ":" + topicExpression },
                    null));
              }
            }
          }
  
          // Now remove the record from the table
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "No more callbacks for this topicexpression, delete record");
  
          registeredMonitorsMap.remove(topicExpression);
        }
      }
      else
      {
        // Failed to find the callback
        // Build the message for the Exception
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
          SibTr.exit(tc, "removeCallbackFromRegisteredMonitors", callback);
  
        SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0005",
          new Object[] {
            "com.ibm.ws.sib.processor.matching.ConsumerMonitorRegistrar",
            "1:456:1.10",
            callback });
        throw new SINotPossibleInCurrentConfigurationException(
          nls.getFormattedMessage(
            "INTERNAL_MESSAGING_ERROR_CWSIP0005",
            new Object[] {
              "com.ibm.ws.sib.processor.matching.ConsumerMonitorRegistrar",
              "1:463:1.10",
              callback },
            null));
      }
  
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "removeCallbackFromRegisteredMonitors");
    }
  
    /**
     * Method matchNewExactConsumerToMonitors
     *
     * This method searches through the consumer monitor tables looking for
     * monitor topic expressions that match the supplied consumer topic expression.
     *
     * This method is driven where the consumer's topic expression is fully qualified.
     * The style of matching differs between monitors themselves registered
     * n fully qualified topics and those that were registered with a wildcard.
     * We use string matching to match against the fully qualified monitors
     * and direct MatchSpace evaluation to match against wildcarded monitors.
     *
     * Where matches are found we check to see whether a callback needs to be
     * driven because there were previously no matches.
     *
     * @param topicExpression of the consumer
     * @param mc the consumer
     * @throws SIDiscriminatorSyntaxException
     */
    public void matchNewExactConsumerToMonitors(
      String topicExpression,
      MonitoredConsumer mc)
    throws SIDiscriminatorSyntaxException
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
          SibTr.entry(
            tc,
            "matchNewExactConsumerToMonitors",
            new Object[] { topicExpression,
                           mc });
      // We'll set any monitors that we find into the following lists which
      // will be added to the consumer itself. This will ease consumer removal
      // processing
      ArrayList exactMonitorList = new ArrayList();
      ArrayList wildcardMonitorList = new ArrayList();
  
      // Iterate through the map of monitors defined on exact topic expressions,
      // Note we can use string matching in this case
      Iterator i = _registeredExactConsumerMonitors.keySet().iterator();
      while (i.hasNext())
      {
        String registeredTopic = (String)i.next();
  
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          SibTr.debug(tc, "Found registered topic: " + registeredTopic);
  
        // Process those registered topics that might match by virtue
        // of a common topic stem
        if(topicExpression.equals(registeredTopic))
        {
  
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "topics match, Add consumer to list");
  
          addConsumerToMonitorList(registeredTopic,
                                   _registeredExactConsumerMonitors,
                                   mc,
                                   false); // exact
  
          // Finally add reference to the monitor to the list for the new consumer
          exactMonitorList.add(registeredTopic);
        }
      }
  
      // Iterate through the map of monitors defined on wildcard topic expressions
      i = _registeredWildcardConsumerMonitors.keySet().iterator();
      while (i.hasNext())
      {
        String registeredTopic = (String)i.next();
        // Retrieve the non-wildcarded stem
        String registeredTopicStem = _mpm.retrieveNonWildcardStem(registeredTopic);
  
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          SibTr.debug(tc, "Found registered topic: " + registeredTopic + "with stem: " + registeredTopicStem);
  
        // Process those registered topics that might match by virtue
        // of a common topic stem
        if(topicExpression.startsWith(registeredTopicStem))
        {
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "Drive direct evaluation for topic: " + registeredTopic);
  
          if(_mpm.evaluateDiscriminator(topicExpression,registeredTopic))
          {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
              SibTr.debug(tc, "Add consumer to list");
  
            addConsumerToMonitorList(registeredTopic,
                                     _registeredWildcardConsumerMonitors,
                                     mc,
                                     true); // wildcarded
  
            // Finally add reference to the monitor to the list for the new consumer
            wildcardMonitorList.add(registeredTopic);
          }
        }
      }
  
      // Finally add lists of references to the monitors into the new consumer
      mc.setMatchingExactMonitorList(exactMonitorList);
      mc.setMatchingWildcardMonitorList(wildcardMonitorList);
  
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "matchNewExactConsumerToMonitors");
  
    }
  
    /**
     * Method matchNewWildcardConsumerToMonitors
     *
     * This method searches through the consumer monitor tables looking for
     * monitor topic expressions that match the supplied consumer topic expression.
     *
     * This method is driven where the consumer's topic expression is wildcarded.
     * The style of matching differs between monitors themselves registered with
     * a wildcard and those that were registered on fully qualified topics.
     * We can use direct MatchSpace evaluation to match against the fully qualified
     * monitors but the best that we can do with wildcarded monitor is to match
     * common non-wildcarded stems and conclude that the consumer "might" match
     * the monitor.
     *
     * Where matches are found we check to see whether a callback needs to be
     * driven because there were previously no matches.
     *
     * @param topicExpression of the consumer
     * @param mc the consumer
     * @throws SIDiscriminatorSyntaxException
     */
    public void matchNewWildcardConsumerToMonitors(
      String topicExpression,
      MonitoredConsumer mc)
    throws SIDiscriminatorSyntaxException
    {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
          SibTr.entry(
            tc,
            "matchNewWildcardConsumerToMonitors",
            new Object[] { topicExpression,
                           mc });
      // We'll set any monitors that we find into the following lists which
      // will be added to the consumer itself. This will ease consumer removal
      // processing
      ArrayList exactMonitorList = new ArrayList();
      ArrayList wildcardMonitorList = new ArrayList();
  
      // Use MatchSpace direct evaluation code once we've isolated candidate
      // expressions through string matching
  
      // Retrieve the non-wildcarded stem from the subscribers topic expression
      String stem = _mpm.retrieveNonWildcardStem(topicExpression);
  
      // Iterate through the map of monitors defined on exact topic expressions
      Iterator i = _registeredExactConsumerMonitors.keySet().iterator();
      while (i.hasNext())
      {
        String registeredTopic = (String)i.next();
  
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
          SibTr.debug(tc, "Found registered topic: " + registeredTopic);
  
        // Process those registered topics that might match by virtue
        // of a common topic stem
        if(registeredTopic.startsWith(stem))
        {
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "Drive direct evaluation for topic: " + registeredTopic);
  
          if(_mpm.evaluateDiscriminator(registeredTopic,topicExpression))
          {
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
              SibTr.debug(tc, "Add consumer to list");
  
            addConsumerToMonitorList(registeredTopic,
                                     _registeredExactConsumerMonitors,
                                     mc,
                                     false); // exact
  
            // Finally add reference to the monitor to the list for the new consumer
            exactMonitorList.add(registeredTopic);
          }
        }
      }
  
      // Iterate through the map of monitors defined on wildcard topic expressions
  