/*
(non-Javadoc)

@seecom.ibm.ws.sib.processor.runtime.SIMPLocalTopicSpaceControllable#
getPubSubDeliveryStreamSetReceiverIterator
515950
public SIMPIterator getPubSubDeliveryStreamSetReceiverIterator() {
		if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
			SibTr.entry(tc, "getPubSubDeliveryStreamSetReceiverIterator");

		SIMPIterator iterator = this.getInputHandler().getTargetStreamManager()
				.getTargetStreamSetControlIterator();

		if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
			SibTr.entry(tc, "getPubSubDeliveryStreamSetReceiverIterator");
		return iterator;
	}
428950
public SIMPIterator getAttatchedRemoteSubscribers()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "getAttatchedRemoteSubscribers");

    // TODO : This method needs to throw runtime invalid to admin
    SIMPIterator itr = null;
    try
    {
      ProtocolItemStream protocolItemStream =
        ((ProtocolItemStream)_streamSet.getItemStream());
      DestinationHandler destHandler = protocolItemStream.getDestinationHandler();

      itr =
        new BasicSIMPIterator(destHandler.getAOControlAdapterIterator());
    }
    catch (MessageStoreException e)
    {
      FFDCFilter.processException(
          e,
          "com.ibm.ws.sib.processor.runtime.InternalOutputStreamSetControl.getAttatchedRemoteSubscribers",
          "1:613:1.30",
          this);
      SibTr.exception(tc, e);
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "getAttatchedRemoteSubscribers", itr);
    return itr;
  }
883313
public PubSubMessageItemStream getPublishPoint()
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        {
            SibTr.entry(tc, "getPublishPoint");
            SibTr.exit(tc, "getPublishPoint", _pubsubMessageItemStream);
        }

        return _pubsubMessageItemStream;
    }
991229
public SIMPIterator getQueuedMessageIterator() throws SIMPRuntimeOperationFailedException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "getQueuedMessageIterator");

    List<ControlAdapter> messages = new ArrayList<ControlAdapter>();
    NonLockingCursor cursor = null;
    try
    {
      cursor = _aoStream.itemStream.newNonLockingItemCursor(null);
      cursor.allowUnavailableItems();
      AOValue item = (AOValue)cursor.next();
      while(item != null)
      {
        try
        {
          // It's possible to get a null adapter back 
          ControlAdapter cAdapter = item.getControlAdapter();
          if(cAdapter != null)
            messages.add(cAdapter);
          
          item = (AOValue)cursor.next();
        }
        catch (NotInMessageStore ef)
        {
          // No FFDC code needed
          // The message has already been consumed. Trace the exception but allow processing to continue.
          SibTr.exception(tc, ef);
        }
      }      
    }
    catch (MessageStoreException e)
    {
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.runtime.RemoteConsumerTransmit.getQueuedMessageIterator",
        "1:467:1.44", 
        this);
        
      SIMPRuntimeOperationFailedException e1 =
        new SIMPRuntimeOperationFailedException(
          nls.getFormattedMessage(
            "INTERNAL_MESSAGING_ERROR_CWSIP0003",
            new Object[] {"RemoteConsumerTransmit.getQueuedMessageIterator",
                          "1:475:1.44", 
                          e,
                          _aoStream.streamId},
            null), e);
            
      SibTr.exception(tc, e1);
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "getQueuedMessageIterator", e1);
      throw e1;
    }
    finally
    {
      if (cursor != null)
        cursor.finished();               
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "getQueuedMessageIterator");
      
    return new BasicSIMPIterator(messages.iterator());    
  }
241410
static Message readMessageFromDelivery(Receiver receiveLink, Delivery delivery)
	{
	    int msgSize = delivery.pending();
        byte[] buffer = new byte[msgSize];
        
        int read = receiveLink.recv(buffer, 0, msgSize);
        
        Message message = Proton.message();
        message.decode(buffer, 0, read);
        return message;	    
	}
1339510
public SIMPIterator getReceivedMessageIterator(int maxMsgs)
  {
    List receivedMessages = new LinkedList();
    Iterator<MessageItem> streamMessages = 
      _targetStream.getAllMessagesOnStream().iterator();
    boolean allMsgs = (maxMsgs == SIMPConstants.SIMPCONTROL_RETURN_ALL_MESSAGES);
    int index = 0;
    while((allMsgs || (index < maxMsgs)) && streamMessages.hasNext())
    {
      MessageItem msgItem = (MessageItem)streamMessages.next();

      SIMPReceivedMessageControllable receivedMessage =
        new LinkReceivedMessageControl(msgItem, _tsm.getDestinationHandler().getMessageProcessor(), _tsm.getDestinationHandler());
      receivedMessages.add(receivedMessage);
      index++;
    }
    return new BasicSIMPIterator(receivedMessages.iterator());
  }
1071409
public Short getAvailableRefIndex(boolean consumerToProvider) {
        if (headerCache == null) {
            synchronized (this) {
                if (headerCache == null) {
                    headerCache = new TwoWayMap<Short, String>();
                }
            }
        }
        if (consumerToProvider) { // from consumer to provider : 0~32766
            for (short i = 0; i < Short.MAX_VALUE; i++) {
                if (!headerCache.containsKey(i)) {
                    return i;
                }
            }
        } else { // from provider to consumer : -1~-32767
            for (short i = -1; i > Short.MIN_VALUE; i--) {
                if (!headerCache.containsKey(i)) {
                    return i;
                }
            }
        }
        return null;
    }
101890
public SIMPIterator getStreams() throws SIMPControllableNotFoundException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "getStreams");

    assertValidControllable();

    Iterator it = null;
    try
    {
      it = _streamSet.iterator();
    }
    catch (SIResourceException e)
    {
      FFDCFilter.processException(
          e,
          "com.ibm.ws.sib.processor.runtime.SourceStreamSetControl.getStreams",
          "1:713:1.39",
          this);

        SIMPRuntimeOperationFailedException finalE =
          new SIMPRuntimeOperationFailedException(
            nls.getFormattedMessage(
              "INTERNAL_MESSAGING_ERROR_CWSIP0002",
              new Object[] {"SourceStreamSetControl.getStreams",
                  "1:721:1.39",
                            e},
              null), e);

        SibTr.exception(tc, finalE);
//        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "getStreams", finalE);
//        throw finalE;
    }
    SIMPIterator returnIterator=new SourceStreamControllableIterator(it);
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "getStreams", returnIterator);
    return returnIterator;
  }
786111
protected UDPBufferImpl getUDPBufferImpl() {
        UDPBufferImpl ret = (UDPBufferImpl) udpBufferObjectPool.get();
        if (ret == null) {
            ret = new UDPBufferImpl(this);
        }
        return ret;
    }
817447
public void reconstitutePubSubSourceStreams(StreamSet streamSet, int startMode)
                    throws
                    SIErrorException, SIDiscriminatorSyntaxException, SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "reconstitutePubSubSourceStreams", new Object[] { streamSet, new Integer(startMode) });

        _sourceStreamManager.reconstituteStreamSet(streamSet);

        // Don't do flush if we are asked to start in recovery mode 
        if (((startMode & JsConstants.ME_START_FLUSH) == JsConstants.ME_START_FLUSH)
            && ((startMode & JsConstants.ME_START_RECOVERY) == 0))
        {
            this.sendFlushedMessage(null, streamSet.getStreamID());
            // Now change streamID of streamSet
            streamSet.setStreamID(new SIBUuid12());
            // This calls requestUpdate on the StreamSet Item which will
            // cause a callback to the streamSet.getPersistentData() by msgstore
            Transaction tran = _txManager.createAutoCommitTransaction();
            try
            {
                streamSet.requestUpdate(tran);
            } catch (MessageStoreException e)
            {
                // MessageStoreException shouldn't occur so FFDC.
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.impl.PubSubInputHandler.reconstitutePubSubSourceStreams",
                                            "1:3145:1.329.1.1",
                                            this);

                SibTr.exception(tc, e);
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "reconstitutePubSubSourceStreams", "SIStoreException");
                throw new SIResourceException(e);
            }
        }

        try
        {
            NonLockingCursor cursor = _proxyReferenceStream.newNonLockingCursor(
                            new ClassEqualsFilter(MessageItemReference.class));
            cursor.allowUnavailableItems();
            MessageItemReference ref = (MessageItemReference) cursor.next();

            // Rebuild internalOutputStreams by running refs through matchspace 
            LocalTransaction siTran = null;
            Transaction msTran = null;
            int batchCount = 0;
            while (ref != null)
            {
                // If the count of reconstituted messages has reached 50, commit the transaction
                if (batchCount >= 50) {
                    try
                    {
                        siTran.commit();
                        // Null out the transaction object so that a new one is created next
                        siTran = null;
                    } catch (SIIncorrectCallException e)
                    {
                        // No FFDC code needed
                        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                            SibTr.exit(tc, "reconstitutePubSubSourceStreams", "SIResourceException");
                        throw new SIResourceException(e);
                    }
                }

                // If the transaction object is null then we've either never created it or have just committed it.
                // Create a new transaction 
                if (siTran == null) {
                    siTran = _txManager.createLocalTransaction(true);
                    msTran = (Transaction) siTran;
                    batchCount = 0;
                }

                boolean commitInsert = false;

                // Change streamID in message to streamID of StreamSet
                // If we are restoring from a stale backup this will
                // ensure that the restoreFanOut method puts it on a new InteralOutputStream
                if (ref.getGuaranteedStreamUuid() != streamSet.getStreamID())
                {
                    ref.setGuaranteedStreamUuid(streamSet.getStreamID());
                }

                if (!(ref.isAdding() || ref.isRemoving()))
                    commitInsert = true;
                // If there are no matching neighbours, remove reference
                if (!restoreFanOut(ref, commitInsert))
                {
                    if (!(ref.isAdding() || ref.isRemoving()))
                    {
                        ref.remove(msTran, ref.getLockID());
                        batchCount++;
                    }
                    else
                    // TODO: ref is indoubt, need to mark it for later deletion on commit
                    {
                    }
                }
                ref = (MessageItemReference) cursor.next();
            }
            try
            {
                if (siTran != null) {
                    siTran.commit();
                }
            } catch (SIIncorrectCallException e)
            {
                // No FFDC code needed
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                    SibTr.exit(tc, "reconstitutePubSubSourceStreams", "SIResourceException");
                throw new SIResourceException(e);
            }

            // Ditch the cursor & create again to get back to head of list
            cursor.finished();
            cursor = _proxyReferenceStream.newNonLockingCursor(
                            new ClassEqualsFilter(MessageItemReference.class));
            cursor.allowUnavailableItems();
            ref = (MessageItemReference) cursor.next();

            // Add all messages back in to the SourceStreams
            while (ref != null)
            {
                // Change streamID in message to streamID of StreamSet
                // If we are restoring from a stale backup this will
                // and the restoreMessage method puts it on a new SourceStream
                if (ref.getGuaranteedStreamUuid() != streamSet.getStreamID())
                {
                    ref.setGuaranteedStreamUuid(streamSet.getStreamID());
                }

                //add all messages back in to the streams
                if (!(ref.isAdding() || ref.isRemoving()))
                {
                    // commit those which are not in doubt
                    _sourceStreamManager.restoreMessage(ref, true);
                }
                else
                {
                    // add to stream in uncommitted state
                    _sourceStreamManager.restoreMessage(ref, false);
                }

                ref = (MessageItemReference) cursor.next();
            }
            // Consolidate all streams which may have been reconstituted
            _sourceStreamManager.consolidateStreams(startMode);

            cursor.finished();

        } catch (MessageStoreException e)
        {
            // MessageStoreException shouldn't occur so FFDC.
            FFDCFilter.processException(
                                        e,
                                        "com.ibm.ws.sib.processor.impl.PubSubInputHandler.reconstitutePubSubSourceStreams",
                                        "1:3277:1.329.1.1",
                                        this);

            SibTr.exception(tc, e);
            SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                        new Object[] {
                                      "com.ibm.ws.sib.processor.impl.PubSubInputHandler",
                                      "1:3284:1.329.1.1",
                                      e });

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

            throw new SIResourceException(
                            nls.getFormattedMessage(
                                                    "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                    new Object[] {
                                                                  "com.ibm.ws.sib.processor.impl.PubSubInputHandler",
                                                                  "1:3295:1.329.1.1",
                                                                  e },
                                                    null),
                            e);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "reconstitutePubSubSourceStreams");
    }
223018
@Override
    public long getDeliveryDelay()
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "getDeliveryDelay");
        long deliveryDelay = 0;

        try {
            deliveryDelay = getJSMessage(true).getDeliveryDelay();
            return deliveryDelay;
        } finally {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(this, tc, "getDeliveryDelay", deliveryDelay);

        }

    }
1229901
@Override
	public BufferProvider getBufferProvider(JobID jobID, ChannelID sourceChannelID) throws IOException {
		EnvelopeReceiverList receiverList = getReceiverList(jobID, sourceChannelID, false);
		
		// check if the receiver is already gone
		if (receiverList == null) {
			return this.discardBufferPool;
		}

		if (!receiverList.hasLocalReceiver() || receiverList.hasRemoteReceiver()) {
			throw new IOException("The destination to be looked up is not a single local endpoint.");
		}
		

		ChannelID localReceiver = receiverList.getLocalReceiver();
		Channel channel = this.channels.get(localReceiver);
		
		if (channel == null) {
			// receiver is already canceled
			return this.discardBufferPool;
		}

		if (!channel.isInputChannel()) {
			throw new IOException("Channel context for local receiver " + localReceiver + " is not an input channel context");
		}

		return (InputChannel<?>) channel;
	}
379708
public SIMPIterator getRemoteSubscriptionIterator()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.entry(tc, "getRemoteSubscriptionIterator");
      
    DestinationTypeFilter filter = new DestinationTypeFilter();
    filter.QUEUE = Boolean.FALSE;
    
    Collection mpAIHCollection = new LinkedList();
    for (SIMPIterator iter = destinationIndex.iterator(filter); iter.hasNext();)
    {
      BaseDestinationHandler destination = (BaseDestinationHandler) iter.next();
      Map destAIHMap = destination.getPseudoDurableAIHMap();
      mpAIHCollection.add(destAIHMap.values());
    }
    
    AttachedRemoteSubscriberIterator remoteSubscriptionItr 
      = new AttachedRemoteSubscriberIterator(mpAIHCollection);
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.exit(tc, "getRemoteSubscriptionIterator", remoteSubscriptionItr);
    return remoteSubscriptionItr;
  }
869723
@Override
    public ConsumerDispatcher getDurableSubscriptionConsumerDispatcher(ConsumerDispatcherState subState)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "getDurableSubscriptionConsumerDispatcher", subState);

        ConsumerDispatcher consumerDispatcher = _pubSubRealization.
                        getDurableSubscriptionConsumerDispatcher(subState);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(
                       tc,
                       "getDurableSubscriptionConsumerDispatcher",
                       consumerDispatcher);

        return consumerDispatcher;
    }
1005058
@Override
    public E peek() {
        // Copy field to avoid re-reading after volatile load
        final AtomicReferenceArray<E> buffer = this.buffer;

        final long consumerIndex = lvConsumerIndex(); // LoadLoad
        final int offset = calcElementOffset(consumerIndex);
        E e = lvElement(buffer, offset);
        if (null == e) {
            /*
             * NOTE: Queue may not actually be empty in the case of a producer (P1) being interrupted after
             * winning the CAS on offer but before storing the element in the queue. Other producers may go on
             * to fill up the queue after this element.
             */
            if (consumerIndex != lvProducerIndex()) {
                do {
                    e = lvElement(buffer, offset);
                } while (e == null);
            } else {
                return null;
            }
        }
        return e;
    }
925822
public SIMPIterator getQueuedMessageIterator()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "getQueuedMessageIterator");

    List<ControlAdapter> messages = new ArrayList<ControlAdapter>();
    
    NonLockingCursor cursor = null;

    MessageItem referredItem = null;
    try
    {
      cursor =
        consumerDispatcher.getReferenceStream().newNonLockingCursor(null);
      cursor.allowUnavailableItems();
      MessageItemReference item = (MessageItemReference) cursor.next();
      while(item != null)
      {
        try
        {
          referredItem = (MessageItem)item.getReferredItem();
          if(referredItem != null)
          {
            // force the arrival time to be written to the jsMessage  
            referredItem.forceCurrentMEArrivalTimeToJsMessage();
          }
          
          // It's possible to get a null adapter back 
          ControlAdapter cAdapter = item.getControlAdapter();
          if(cAdapter != null)
            messages.add(cAdapter);
          
          item = (MessageItemReference) cursor.next();
        }
        catch (NotInMessageStore ef)
        {
          // No FFDC code needed
          // The message has already been consumed. Trace the exception but allow processing to continue.
          SibTr.exception(tc, ef);
        }
        catch (MessageStoreException e)
        {
          // FFDC
          FFDCFilter.processException(
            e,
            "com.ibm.ws.sib.processor.runtime.LocalSubscriptionControl.getQueuedMessageIterator",
            "1:275:1.36",
            this);
            

          SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
            new Object[] { "com.ibm.ws.sib.processor.runtime.LocalSubscriptionControl.getQueuedMessageIterator", 
                           "1:281:1.36", 
                           SIMPUtils.getStackTrace(e) });
        }        
      }      
    }
    catch (MessageStoreException e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.runtime.LocalSubscriptionControl.getQueuedMessageIterator",
        "1:292:1.36",
        this);
        

      SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
        new Object[] { "com.ibm.ws.sib.processor.runtime.LocalSubscriptionControl.getQueuedMessageIterator", 
                       "1:298:1.36", 
                       SIMPUtils.getStackTrace(e) });
    }
    finally
    {
      if (cursor != null)
        cursor.finished();
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "getQueuedMessageIterator");
      
    return new BasicSIMPIterator(messages.iterator());    
  }
1366917
public String getTargetBus() {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.entry(tc, "getTargetBus");
    
    String bus = null;
    if(parent instanceof LinkPublicationPointControl)
    {
      LinkPublicationPointControl lppc = (LinkPublicationPointControl)parent;
      bus = lppc.getTargetBusName();
    }
    else
    {
      // parent is not an instance of LinkPublicationPointControl   
      SIErrorException finalE =
        new SIErrorException(
          nls.getFormattedMessage(
            "INTERNAL_MESSAGING_ERROR_CWSIP0005",
            new Object[] {"LinkPubSubTransmitMessageControl.getTargetBus",
                "1:149:1.7",
                parent},
            null));
      
      // FFDC
      FFDCFilter.processException(
          finalE,
          "com.ibm.ws.sib.processor.runtime.LinkPubSubTransmitMessageControl.getTargetBus",
          "1:157:1.7",
          this);
      
      SibTr.exception(tc, finalE);
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
        SibTr.exit(tc, "getTargetBus", finalE);
      throw finalE;      
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "getTargetBus", bus);
    return bus;
  }