/*
(non-Javadoc)

@see com.ibm.ws.sib.processor.impl.interfaces.ConsumerPoint#suspendConsumer()
763693
@Override
    public boolean suspendConsumer(int suspendFlag)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "suspendConsumer", Integer.valueOf(suspendFlag));

        boolean didSuspendConsumer;
        //  We can't hold the LCP lock before we hold the keyGroup lock and we don't want to
        // take the asynch lock as we'll block on a consumerMessage call but the keyGroup
        // is set inside the LCP and asynch locks and could therefore vanish between checking for it
        // and trying to lock it here. So we take a dirty read of the keyGroup and if we
        // got it we'll lock it. This does mean there are slight windows around setting
        // and unsetting of the keyGroup but the worst case is that we lock the group when
        // we didn't need to (as we're no longer a member of the group).
        JSKeyGroup cachedKeyGroup = _keyGroup;

        // We're a member of a group so we must lock it before we go for the LCP lock
        // and take the message
        if (cachedKeyGroup != null)
            cachedKeyGroup.lock();

        try
        {
            this.lock();
            try
            {
                if (!isConsumerSuspended(suspendFlag)) // Is the consumer already suspended
                {
                    _consumerSuspended = true;
                    _suspendFlags |= suspendFlag;
                    unsetReady();
                    didSuspendConsumer = true;
                }
                else
                {
                    didSuspendConsumer = false;
                }
            } finally
            {
                this.unlock();
            }
        } finally
        {
            if (cachedKeyGroup != null)
                cachedKeyGroup.unlock();
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "suspendConsumer", didSuspendConsumer);
        return didSuspendConsumer;
    }
545015
public void removeConsumer(MonitoredConsumer mc)
  {
    if (tc.isEntryEnabled()) 
      SibTr.entry(tc, 
                  "removeConsumer", 
                  new Object[]{mc});
      
    boolean selector = mc.isSelector();
    boolean isWildcarded = mc.isWildcarded();
    String topic = mc.getTopic();
    ArrayList consumerList = 
      _subscriptionRegistrar.getConsumerListForExpression(topic, selector, isWildcarded);
        
    //TODO: handle empty return
        
    if (tc.isDebugEnabled())
      SibTr.debug(tc, "Found existing entry with list: " + consumerList +
                      ", of size: " + consumerList.size());
        
    // Remove the consumer      
    boolean removed = consumerList.remove(mc);
    if(!removed)
    {
      // Couldn't find the consumer. This means we have an internal 
      // inconsistency in our tables, so throw an SIErrorException
    
      if (tc.isEntryEnabled()) 
        SibTr.exit(tc, "removeConsumer", "SIErrorException");
      
      SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0001",
        new Object[] {
          "com.ibm.ws.sib.processor.matching.ConsumerMonitoring",
          "1:248:1.7"
          });  
        
      throw new SIErrorException(
        nls.getFormattedMessage(
          "INTERNAL_MESSAGING_ERROR_CWSIP0001",
          new Object[] {
            "com.ibm.ws.sib.processor.matching.ConsumerMonitoring",
            "1:256:1.7"
          },
          null));      
    }          
                 
    // Only remove the row in the table from the subscription registrar
    // if this was the last consumer in the list
    if(consumerList.isEmpty())
    {
      _subscriptionRegistrar.removeConsumerListForExpression(topic,selector,isWildcarded);                  
    }
        
    // 1. Need to remove sub references from consumer monitor tables
    // 2. Drive callbacks where appropriate
    ArrayList exactMonitorList = (ArrayList)mc.getMatchingExactMonitorList();
    ArrayList wildcardMonitorList = (ArrayList)mc.getMatchingWildcardMonitorList();
        
    _consumerMonitorRegistrar.
      removeConsumerFromRegisteredMonitors(mc, exactMonitorList, wildcardMonitorList);
          
    if (tc.isEntryEnabled()) SibTr.exit(tc, "removeConsumer"); 
  }
381773
@Override
    public void resumeConsumer(int suspendFlag)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "resumeConsumer", this);

        // Use this flag to identify whether a new thread must be spun off to look for more
        // messages
        boolean spinOffNewThread = false;
        try
        {
            // Take the lock
            this.lock();

            if (_consumerSuspended)
            {
                boolean setStillSuspended = false;

                // If we're resuming due to the ConsumerSet being resumed then we need to
                // re-check the set's state due to us not holding the set's activeMessageCount
                // lock at this time (due to the hierarchy). And if we find that the set is
                // actually suspended (due to someone else in the set getting in and adding
                // another message then we just ignore this resume as there'll be another one
                // along when the count drops back below the limit again.
                if (suspendFlag == DispatchableConsumerPoint.SUSPEND_FLAG_SET_ACTIVE_MSGS)
                    setStillSuspended = _consumerKey.isConsumerSetSuspended();

                if (!setStillSuspended)
                {
                    // clear the bit provided in the _suspendFlags
                    _suspendFlags &= ~suspendFlag;

                    if (_suspendFlags == 0) // No flags set so resume the consumer
                    {
                        _consumerSuspended = false;

                        // If the consumer is still active (started) we need
                        // to kickstart the consumer back into life to check for more
                        // messages
                        if (!_stopped && !_closing)
                        {
                            if (_asynchConsumerRegistered)
                            {
                                if (!_asynchConsumer.isAsynchConsumerRunning())
                                {
                                    // We'll have to take the full lock hierarchy and
                                    // possibly spin off another thread to start looking for more messages
                                    // (just like start())
                                    spinOffNewThread = true;
                                }
                            }
                            else
                            {
                                // Wake up any synchronous receivers
                                _waiter.signal();
                            }
                        }
                    }
                }
            }
        } finally
        {
            this.unlock();
        }

        // start a new thread to run the callback if we determined that it was necessary   
        if (spinOffNewThread)
        {
            try
            {
                _messageProcessor.startNewThread(new AsynchThread(this, false));
            } catch (InterruptedException e)
            {
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.JSLocalConsumerPoint.resumeConsumer",
                                            "1:4978:1.22.5.1",
                                            this);

                SibTr.exception(tc, e);

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

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "resumeConsumer");
    }
1011788
public void stopConsumer() {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "stopConsumer");
        try {
            // lock the consumerSession to ensure visibility of update to started.
            synchronized (consumerSession) {
                consumerSession.getConsumerSession().stop();
                consumerSession.started = false;
            }
        } catch (Throwable t) {
            FFDCFilter.processException(t,
                                        CLASS_NAME + ".consumeMessages",
                                        CommsConstants.CATASYNCHRHREADER_CONSUME_MSGS_02,
                                        this);
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                SibTr.debug(this, tc, "Unable to stop consumer session due to Throwable: " + t);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "stopConsumer");
    }
894085
public void addConsumer(SnapshotConsumer consumer){
		if (consumers.contains(consumer))
			consumers.remove(consumer);
		consumers.add(consumer);
	}
673947
@SuppressWarnings("unchecked")
  void registerConsumer(Inspector<? super OutT> subInspector) {
    consumers.add((Inspector<OutT>) subInspector);
  }
935719
public void suspend() {
    if (suspension.compareAndSet(null, new CountDownLatch(1))) {
      try {
        suspendBarrier.await();
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      } catch (BrokenBarrierException e) {
        LOG.error("Exception during suspend: " + flowletContext, e);
      }
    }
  }
639082
protected void attachBifurcatedConsumer(BifurcatedConsumerSessionImpl consumer)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "attachBifurcatedConsumer", consumer);

        // Create a bifurcated list if required
        if (_bifurcatedConsumers == null)
        {
            synchronized (this)
            {
                if (_bifurcatedConsumers == null)
                    _bifurcatedConsumers = new LinkedList<BifurcatedConsumerSessionImpl>();
            }
        }

        synchronized (_bifurcatedConsumers)
        {
            _bifurcatedConsumers.add(consumer);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "attachBifurcatedConsumer");
    }
897491
@Override
    public final void closeConsumers() {
        final TaskRunner taskRunner = getTaskRunner();
        final List<RowProcessingConsumer> configurableConsumers = getConsumers();
        for (final RowProcessingConsumer consumer : configurableConsumers) {
            final TaskRunnable task = createCloseTask(consumer, null);
            taskRunner.run(task);
        }
    }
225563
private synchronized RealmOAuthConsumer getConsumer() {
        // could just inject these, but I kinda prefer pushing this out
        // to the properties file...
        if (consumer == null) {
            OAuthServiceProvider serviceProvider = new OAuthServiceProvider("", "", "");
            String realm =
                    propertyResolver.getProperty(
                            "org.jasig.rest.interceptor.oauth." + id + ".realm");
            String consumerKey =
                    propertyResolver.getProperty(
                            "org.jasig.rest.interceptor.oauth." + id + ".consumerKey");
            String secretKey =
                    propertyResolver.getProperty(
                            "org.jasig.rest.interceptor.oauth." + id + ".secretKey");

            Assert.notNull(
                    consumerKey,
                    "The property \"org.jasig.rest.interceptor.oauth."
                            + id
                            + ".consumerKey\" must be set.");
            Assert.notNull(
                    secretKey,
                    "The property \"org.jasig.rest.interceptor.oauth."
                            + id
                            + ".secretKey\" must be set.");

            consumer = new RealmOAuthConsumer(consumerKey, secretKey, realm, serviceProvider);
        }

        return consumer;
    }
1153180
@Override
    public void setConsumer(IDataSetConsumer consumer) {
        logger.debug("setConsumer(consumer) - start");

        if (this._columnSensing) {
            _consumer = new BufferedConsumer(consumer);
        } else {
            _consumer = consumer;
        }
    }
912210
private void pauseConsumers(SIMPMessage msg, long retryInterval)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "pauseConsumers", new Object[] { msg, Long.valueOf(retryInterval) });

        // We need to lock the consumerPoints to check for consumers to pause but
        // we can't actually suspend them under that lock or we'll deadlock (465125)
        // so we build a new consumer list and process them outside the lock
        LinkedList<DispatchableKey> cachedConsumerPoints = new LinkedList<DispatchableKey>();
        synchronized (consumerPoints)
        {
            Iterator<DispatchableKey> itr = consumerPoints.iterator();
            while (itr.hasNext())
            {
                DispatchableKey cKey = itr.next();

                // If consumer is interested in this message, suspend it.
                if (((LocalQPConsumerKey) cKey).filterMatches((AbstractItem) msg))
                {
                    cachedConsumerPoints.add(cKey);
                }
            }
        } // sync

        // Now pause the consumers outside the lock
        Iterator<DispatchableKey> itr2 = cachedConsumerPoints.iterator();
        while (itr2.hasNext())
        {
            DispatchableKey cKey = itr2.next();

            DispatchableConsumerPoint consumerPoint =
                            cKey.getConsumerPoint();
            // If this consumer is already suspended then don't sent up a new alarm and
            // don't resuspend the consumer
            BlockedConsumerRetryHandler blockedConsumerRetryHandler = new BlockedConsumerRetryHandler(consumerPoint);
            if (blockedConsumerRetryHandler.startSuspend())
            {
                // Spin of a new alarm that will suspend the consumer immediately
                //  then wake up and resume the consumer
                _messageProcessor.getAlarmManager().
                                create(retryInterval, blockedConsumerRetryHandler);
            }
        }

        // Spin of a new alarm that will mark the itemStream as blocked
        // and unblock it in time with the consumers. This allows us to
        // display the state via the MBeans/panels using the QueueMessage
        // objects
        _messageProcessor.getAlarmManager().create(retryInterval,
                                                   new BlockedRetryHandler());

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "pauseConsumers");
    }
706048
public ConsumerSession getConsumerSession() 
  throws SISessionUnavailableException, SISessionDroppedException,
         SIIncorrectCallException

  {
    if (TraceComponent.isAnyTracingEnabled() && CoreSPILockedMessageEnumeration.tc.isEntryEnabled())
      SibTr.entry(CoreSPILockedMessageEnumeration.tc, "getConsumerSession", this);
      
    checkValidState("getConsumerSession");
          
    _localConsumerPoint.checkNotClosed();
    
    if (TraceComponent.isAnyTracingEnabled() && CoreSPILockedMessageEnumeration.tc.isEntryEnabled())
      SibTr.exit(CoreSPILockedMessageEnumeration.tc, "getConsumerSession", 
        _localConsumerPoint.getConsumerSession());
    
    return _localConsumerPoint.getConsumerSession();
  }
249144
public Consumer createConsumer(Processor processor) throws Exception {
        LOG.debug(this.toString());

        ObjectHelper.notNull(this.mode, "mode");
        GpioPin pin = isAlreadyProvisioned();

        if (pin == null) {
            switch (this.mode) {
            case DIGITAL_INPUT:
                pin = getOrCreateController().provisionDigitalInputPin(getPin(), pullResistance);
                break;
            case ANALOG_INPUT:
                pin = getOrCreateController().provisionAnalogInputPin(getPin());
                break;
            case ANALOG_OUTPUT:
            case DIGITAL_OUTPUT: // PinMode.allOutput()
            case PWM_OUTPUT:
                LOG.error("Cannot create Consumer with OUTPUT Mode");
                return null;
            default:
                LOG.error("Cannot create Consumer w/o Mode");
                break;
            }

        } else {
            // enhancement we could manage several pins with one consumer
            throw new IllegalArgumentException("Cannot create twice same input pin [" + this.gpioId + "] for Consumer");
        }

        return new GPIOConsumer(this, processor, pin, state);
    }
624747
public void setConsumerDispatcher(ConsumerDispatcher consumerDispatcher)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "setConsumerDispatcher", consumerDispatcher);

        this.consumerDispatcher = consumerDispatcher;

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "setConsumerDispatcher");
    }
1094256
public void resumeConsumer(int suspendFlag)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "resumeConsumer", this);
    
    ArrayList<AORequestedTick> satisfiedTicks = null;

    synchronized(parent) // AOStream
    {
      try
      {
        // Take the lock
        this.lock();
        try
        {    
          if (_consumerSuspended)
          {
            // clear the bit provided in the _suspendFlags
            _suspendFlags &= ~suspendFlag;
      
            if (_suspendFlags == 0) // No flags set so resume the consumer
            {
              _consumerSuspended = false;
      
              // If the consumer is still active (started) we need
              // to kickstart the consumer back into life to check for more
              // messages
              if(!closed)
                satisfiedTicks = processQueuedMsgs(null);
            }
          }
        }
        finally
        {
          this.unlock();
        }   
      }
      catch(SINotPossibleInCurrentConfigurationException e)
      {
        // No FFDC code needed
        notifyException(e);
      }
      
      if (satisfiedTicks!=null)
      {
        // inform parent about satisfied ticks - outside lock
        int length = satisfiedTicks.size();
        for (int i = 0; i < length; i++)
        {
          AORequestedTick aotick = (AORequestedTick) satisfiedTicks.get(i);
          parent.satisfiedRequest(aotick.tick, aotick.getMessage());
        }
      }
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "resumeConsumer");
  }
1370325
public ConsumerSession getConsumerSession()
      throws SISessionUnavailableException, SISessionDroppedException,
             SIConnectionUnavailableException, SIConnectionDroppedException
	{
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc,"getConsumerSession");          // f173765.2
      checkValid();                                                            // f173765.2
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc,"getConsumerSession");           // f173765.2
      
      return consumerSession;
	}