/*
(non-Javadoc)
@see net.timewalker.ffmq4.local.destination.AbstractLocalDestination#putLocked(net.timewalker.ffmq4.common.message.AbstractMessage, net.timewalker.ffmq4.local.session.LocalSession, net.timewalker.ffmq4.local.MessageLockSet)
438056
@Override
	public boolean putLocked( AbstractMessage srcMessage, LocalSession session , MessageLockSet locks ) throws JMSException
    {
    	checkNotClosed();
    	checkTransactionLock();
    
    	// Check delivery mode
    	if (!topicDef.supportDeliveryMode(srcMessage.getJMSDeliveryMode()))
    		throw new FFMQException("Topic does not support this delivery mode : "+
    				(srcMessage.getJMSDeliveryMode() == DeliveryMode.NON_PERSISTENT ? 
                    "DeliveryMode.NON_PERSISTENT" : "DeliveryMode.PERSISTENT"),
                    "INVALID_DELIVERY_MODE");
    	
    	sentToTopicCount.incrementAndGet();
    	
        boolean commitRequired = false;
        
        subscriptionsLock.readLock().lock();
        try
        {
        	// Pass 1 - Indexed subscriptions
        	if (!indexedSubscriptionMap.isEmpty())
        	{
        		srcMessage.ensureDeserializationLevel(MessageSerializationLevel.ALL_HEADERS);
        		for(Entry<String,Map<Object,List<LocalTopicSubscription>>> entry : indexedSubscriptionMap.entrySet())
        		{
        			String headerName = entry.getKey();
        			Object value;
        			if (headerName.equals("JMSCorrelationID")) // Intercept special headers
        				value = srcMessage.getJMSCorrelationID();
        			else
        				value = srcMessage.getObjectProperty(headerName);
        			
        			if (value != null)
        			{
        				value = normalizeLiteralValue(value);
        				List<LocalTopicSubscription> subscriptions = entry.getValue().get(value);
        				if (subscriptions != null)
        				{
        					if (pushToSubscriptions(srcMessage, session, locks, subscriptions))
        						commitRequired = true;
        				}
        			}
        		}
        	}
        	
        	// Pass 2 - Non indexed subscriptions
            if (!flatSubscriptions.isEmpty())
            	if (pushToSubscriptions(srcMessage, session, locks, flatSubscriptions))
            		commitRequired = true;
        }
        finally
        {
        	subscriptionsLock.readLock().unlock();
        }
        
        return commitRequired;
    }
1364935
public SIBusMessage nextLocked() throws SISessionUnavailableException, SISessionDroppedException, SIConnectionUnavailableException, SIConnectionDroppedException, SIResourceException, SIConnectionLostException, SIIncorrectCallException {
    
    return _cursor >= messages.size () ? null : (SIBusMessage) messages.get(_cursor++);
    
  }
658215
public final void ensureOrder(TotalOrderRemoteTransactionState state, Collection<?> keysModified) throws InterruptedException {
      //the retries due to state transfer re-uses the same state. we need that the keys previous locked to be release
      //in order to insert it again in the keys locked.
      //NOTE: this method does not need to be synchronized because it is invoked by a one thread at the time, namely
      //the thread that is delivering the messages in total order.
      state.awaitUntilReset();
      TotalOrderLatch transactionSynchronizedBlock = new TotalOrderLatchImpl(state.getGlobalTransaction().globalId());
      state.setTransactionSynchronizedBlock(transactionSynchronizedBlock);
         //this will collect all the count down latch corresponding to the previous transactions in the queue
         for (Object key : keysModified) {
            TotalOrderLatch prevTx = keysLocked.put(key, transactionSynchronizedBlock);
            if (prevTx != null) {
               state.addSynchronizedBlock(prevTx);
            }
            state.addLockedKey(key);

      }

      TotalOrderLatch stateTransfer = stateTransferInProgress.get();
      if (stateTransfer != null) {
         state.addSynchronizedBlock(stateTransfer);
      }

      if (trace) {
         log.tracef("Transaction [%s] will wait for %s and locked %s", state.getGlobalTransaction().globalId(),
                    state.getConflictingTransactionBlocks(), state.getLockedKeys() == null ? "[ClearCommand]" :
               state.getLockedKeys());
      }
   }
835724
private void lock() throws IOException {
		try {
			lock.lockInterruptibly();
		}
		catch (InterruptedException e) {
			Thread.currentThread().interrupt();
			throw new IOException("interrupted");
		}
	}
464893
@Override
    public void lock(Object entity, LockModeType lockMode)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            Tr.debug(tc, "em.lock(" + entity + ',' + lockMode + ");\n" + toString());

        getEMInvocationInfo(true).lock(entity, lockMode);
    }
789361
public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset local var
        // holding count negative to indicate failure unless set.
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            /*
             * Note that count is used in wait guard even though it is
             * not protected by lock. This works because count can
             * only decrease at this point (all other puts are shut
             * out by lock), and we (or some other waiting put) are
             * signalled if it ever changes from capacity. Similarly
             * for all other uses of count in other wait guards.
             */
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }
643728
public void put(E e) throws InterruptedException {
        Objects.requireNonNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }
1336496
public void unlockAndDeliver( MessageLock lockRef ) throws JMSException
    {
    	MessageStore targetStore;
    	if (lockRef.getDeliveryMode() == DeliveryMode.NON_PERSISTENT)
    	    targetStore = volatileStore;
    	else
    	    targetStore = persistentStore;
    	
    	int handle = lockRef.getHandle();
    	AbstractMessage message = lockRef.getMessage();
    	synchronized (storeLock)
        {
    		targetStore.unlock(handle);
        }
    	sentToQueueCount.incrementAndGet();
    	
    	sendAvailabilityNotification(message);
    }
1317482
public SIBusMessage nextLocked()
  throws SISessionUnavailableException, SISessionDroppedException,
         SIResourceException, SIConnectionLostException, 
         SIErrorException, SIIncorrectCallException
  {
    if (TraceComponent.isAnyTracingEnabled() && CoreSPILockedMessageEnumeration.tc.isEntryEnabled())
      SibTr.entry(CoreSPILockedMessageEnumeration.tc, "nextLocked", this);

    checkValidState("nextLocked");
          
    _localConsumerPoint.checkNotClosed();  

    if (_seenSingleMessage || _singleMessage == null)
    {
      _messageAvailable = false;
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "nextLocked", null);
      return null;
    }
    _seenSingleMessage = true;
    
    JsMessage msg = null;
    
    // Because this message is not in the message store there is no need to copy it
    // before giving it to a caller unless this is pubsub, in which case there may be
    // other subscriptions referencing the same message, in which case we must copy the
    // message (unless the caller indicates that they won't be altering it)
    if(_isPubsub &&
       ((ConnectionImpl)(_localConsumerPoint.getConsumerSession().getConnection())).getMessageCopiedWhenReceived())
    {
      try
      {
        msg = (_singleMessage.getMessage()).getReceived();
      }
      catch (MessageCopyFailedException e)
      {
        FFDCFilter.processException(
          e,
          "com.ibm.ws.sib.processor.impl.SingleLockedMessageEnumerationImpl.nextLocked",
          "1:172:1.44",
          this);
            
        SibTr.exception(tc, e);
        SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
          new Object[] {
            "com.ibm.ws.sib.processor.impl.SingleLockedMessageEnumerationImpl",
            "1:179:1.44",
            e });
          
        _seenSingleMessage = false;
        _messageAvailable = false;
  
        if (TraceComponent.isAnyTracingEnabled() && CoreSPILockedMessageEnumeration.tc.isEntryEnabled())
          SibTr.exit(CoreSPILockedMessageEnumeration.tc, "nextLocked", e);
  
        throw new SIResourceException(
          nls.getFormattedMessage(
            "INTERNAL_MESSAGING_ERROR_CWSIP0002",
            new Object[] {
              "com.ibm.ws.sib.processor.impl.SingleLockedMessageEnumerationImpl",
              "1:193:1.44",
              e },
            null),
          e);
      }
    }
    else
      msg = _singleMessage.getMessage();
          
    //indicate that there is a messageAvailable
    _messageAvailable = true;

    if (TraceComponent.isAnyTracingEnabled() && UserTrace.tc_mt.isDebugEnabled())                                   
      UserTrace.trace_Receive(null, 
                              msg, 
                              _localConsumerPoint.getConsumerSession().getDestinationAddress(), 
                              _localConsumerPoint.getConsumerSession().getIdInternal());
    
    if (TraceComponent.isAnyTracingEnabled() && CoreSPILockedMessageEnumeration.tc.isEntryEnabled())
          SibTr.exit(CoreSPILockedMessageEnumeration.tc, "nextLocked", msg);
    
    return msg;
  }
444349
@Override
    @SuppressWarnings("unchecked")
    public MessagePattern cloneAsThawed() {
        MessagePattern newMsg;
        try {
            newMsg=(MessagePattern)super.clone();
        } catch (CloneNotSupportedException e) {
            throw new ICUCloneNotSupportedException(e);
        }
        newMsg.parts=(ArrayList<Part>)parts.clone();
        if(numericValues!=null) {
            newMsg.numericValues=(ArrayList<Double>)numericValues.clone();
        }
        newMsg.frozen=false;
        return newMsg;
    }
1038864
public void writeUncommitted(SIMPMessage m) throws SIResourceException
  {

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this, tc, "writeUncommitted", new Object[] { m });

    TickRange tr = null;

    JsMessage jsMsg = m.getMessage();
   
    long stamp = jsMsg.getGuaranteedValueValueTick();
    long starts = jsMsg.getGuaranteedValueStartTick();
    long ends = jsMsg.getGuaranteedValueEndTick();

    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
    {
      SibTr.debug(tc, "writeUncommitted at: " + stamp + " on Stream " +  stream);
    }

    synchronized (this)
    {      
      if (m.getStreamIsGuess())
      {
        newGuessInStream(stamp);
      }
      msgAdded(stamp);

      // write into stream
      tr = TickRange.newUncommittedTick(stamp);
      tr.startstamp = starts;
      tr.endstamp = ends;
    
      // Save message in the stream while it is Uncommitted
      // It will be replaced by its index in the ItemStream once it becomes a Value
      tr.value = m;

      oststream.writeCombinedRange(tr);
 
      // Now update our lastMsgAdded with the tick of the
      // message we have just added
      lastMsgAdded = stamp;
      
      // SIB0105
      // If this is the first uncommitted messages to be added to the stream, start off the 
      // blocked message health state timer. This will check back in the future to see if the
      // message successfully sent.
      if (blockedStreamAlarm==null)
      {
        blockedStreamAlarm = new BlockedStreamAlarm(getCompletedPrefix());
        am.create(mp.getCustomProperties().getBlockedStreamHealthCheckInterval(), blockedStreamAlarm);
      }

    } // end synchronized

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "writeUncommitted");
  }
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();
        }
    }
600414
@Override
	public final void setMessageListener(MessageListener messageListener) throws JMSException
    {
        super.setMessageListener(messageListener);
        
        // If the connection was already started, wake up the new listener in case there were messages
        // waiting in the destination
        if (messageListener != null && session.getConnection().isStarted())
        	wakeUpMessageListener();
    }
559278
public void eventLocked()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "eventLocked");

    // modify prefetching info
    synchronized (this)
    {
      RemoteDispatchableKey dkey = key.getRemoteDispatchableKey();
      RemoteQPConsumerKey ck = null;
      if (dkey instanceof RemoteQPConsumerKey)
        ck = (RemoteQPConsumerKey) dkey;

      if (ck != null)
      {
        ck.messageLocked(key);
        informedConsumerKeyThatLocked = true;
      }
    }

    // call superclass
    super.eventLocked();

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "eventLocked");
  }
271838
public void setIsCloned(boolean isCloned)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "setIsCloned", Boolean.valueOf(isCloned));
        this.isCloned = isCloned;
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(this, tc, "setIsCloned");
    }
1279233
public void lock() throws IOException, InterruptedException {
        if (channel == null) {
            initializeChannel();
        }
        lockSem.acquire();
        lock = channel.lock();
    }
124923
public void blockTillTerminated() {
    while (!runState.compareAndSet(State.IDLE, State.REMOVED)) {
      // wait and retry
      LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(10));
      if (Thread.currentThread().isInterrupted()) {
        runState.set(State.REMOVED);
        return;
      }
    }
  }