/* (non-Javadoc)
@see com.ibm.ws.sib.processor.impl.interfaces.DownstreamControl#sendNotFlushedMessage(com.ibm.ws.sib.utils.SIBUuid12, long)
11776
public void sendNotFlushedMessage(SIBUuid8 ignore, SIBUuid12 streamID, long requestID) 
  throws SIResourceException 
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.entry(tc, "sendNotFlushedMessage", 
        new Object[]{ignore, streamID, new Long(requestID)});
        
    ControlNotFlushed notFlushed = createControlNotFlushed(_targetMEUuid, streamID, requestID);
    notFlushed = _internalOutputStreamManager.stampNotFlushed(notFlushed, streamID);
    
    // If the destination in a Link add Link specific properties to message
    if( _isLink )
    {
      notFlushed = (ControlNotFlushed)addLinkProps(notFlushed);
      notFlushed.setRoutingDestination(_routingDestination);
    }
    _mpio.sendToMe(_targetMEUuid, SIMPConstants.MSG_HIGH_PRIORITY, notFlushed);    
        
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "sendNotFlushedMessage");  
  }
682020
public void sendCancelledNotification(Jid to, String packetID) throws NotConnectedException, InterruptedException {
        // Create the message to send
        Message msg = new Message(to);
        // Create a MessageEvent Package and add it to the message
        MessageEvent messageEvent = new MessageEvent();
        messageEvent.setCancelled(true);
        messageEvent.setStanzaId(packetID);
        msg.addExtension(messageEvent);
        // Send the packet
        connection().sendStanza(msg);
    }
824890
private void sendMessages(final List<MessageEnvelope> msgEnvelopes) {
        try {
            boolean success=BackendOperation.execute(new BackendOperation.Transactional<Boolean>() {
                @Override
                public Boolean call(StoreTransaction txh) throws BackendException {
                    ListMultimap<StaticBuffer,Entry> mutations = ArrayListMultimap.create();
                    for (MessageEnvelope env : msgEnvelopes) {
                        mutations.put(env.key,env.entry);
                        long ts = env.entry.getColumn().getLong(0);
                        log.debug("Preparing to write {} to storage with column/timestamp {}", env, times.getTime(ts));
                    }

                    Map<StaticBuffer,KCVMutation> muts = new HashMap<StaticBuffer, KCVMutation>(mutations.keySet().size());
                    for (StaticBuffer key : mutations.keySet()) {
                        muts.put(key,new KCVMutation(mutations.get(key),KeyColumnValueStore.NO_DELETIONS));
                        log.debug("Built mutation on key {} with {} additions", key, mutations.get(key).size());
                    }
                    manager.storeManager.mutateMany(ImmutableMap.of(store.getName(),muts),txh);
                    log.debug("Wrote {} total envelopes with operation timestamp {}", msgEnvelopes.size(), txh.getConfiguration().getCommitTime());
                    return Boolean.TRUE;
                }
                @Override
                public String toString() {
                    return "messageSending";
                }
            },this, times, maxWriteTime);
            Preconditions.checkState(success);
            log.debug("Wrote {} messages to backend",msgEnvelopes.size());
            for (MessageEnvelope msgEnvelope : msgEnvelopes)
                msgEnvelope.message.delivered();
        } catch (TitanException e) {
            for (MessageEnvelope msgEnvelope : msgEnvelopes)
                msgEnvelope.message.failed(e);
            throw e;
        }
    }
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");
  }
596916
public boolean isFlushed(SIBUuid12 streamID)
  {
    //note that although it is true that this method is currently
    //equivalent to !hadStream(streamID), it may not always be true in
    //the future.
    if(streamSet == null) return true;

    return !streamID.equals(streamSet.getStreamID());
  }
766502
@GuardedBy("lock") protected void sendBundledMessages() {
        for(Map.Entry<Address,List<Message>> entry: msgs.entrySet()) {
            List<Message> list=entry.getValue();
            if(list.isEmpty())
                continue;

            output.position(0);
            if(list.size() == 1)
                sendSingleMessage(list.get(0));
            else {
                Address dst=entry.getKey();
                sendMessageList(dst, list.get(0).getSrc(), list);
                if(transport.statsEnabled())
                    transport.incrBatchesSent(1);
            }
        }
        clearMessages();
        count=0;
    }
1035342
private void sendHandshake(ChannelHandlerContext ctx) {
    boolean needToFlush = false;

    // Iterate until there is nothing left to write.
    while (true) {
      buffer = getOrCreateBuffer(ctx.alloc());
      try {
        handshaker.getBytesToSendToPeer(buffer);
      } catch (GeneralSecurityException e) {
        throw new RuntimeException(e);
      }
      if (!buffer.isReadable()) {
        break;
      }

      needToFlush = true;
      @SuppressWarnings("unused") // go/futurereturn-lsc
      Future<?> possiblyIgnoredError = ctx.write(buffer);
      buffer = null;
    }

    // If something was written, flush.
    if (needToFlush) {
      ctx.flush();
    }
  }
508901
Sink newMessageSink(int formatOpcode, long contentLength) {
    if (activeWriter) {
      throw new IllegalStateException("Another message writer is active. Did you call close()?");
    }
    activeWriter = true;

    // Reset FrameSink state for a new writer.
    frameSink.formatOpcode = formatOpcode;
    frameSink.contentLength = contentLength;
    frameSink.isFirstFrame = true;
    frameSink.closed = false;

    return frameSink;
  }
1160120
@Override
    public void sendNackMessage(
                                SIBUuid8 upstream,
                                SIBUuid12 destUuid,
                                SIBUuid8 busUuid,
                                long startTick,
                                long endTick,
                                int priority,
                                Reliability reliability,
                                SIBUuid12 stream) throws SIResourceException
    {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(
                        tc,
                        "sendNackMessage",
                        new Object[] { new Long(startTick), new Long(endTick) });

        ControlNack nackMsg = createControlNackMessage(priority, reliability, stream);

        nackMsg.setStartTick(startTick);
        nackMsg.setEndTick(endTick);

        if (upstream == null)
        {
            upstream = _originStreamMap.get(stream);
        }

        // Send this to MPIO
        // Send Nack messages at message priority +2
        _mpio.sendToMe(
                       upstream,
                       priority + 2,
                       nackMsg);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "sendNackMessage ");
    }
54083
@Override
    public void sendAckExpectedMessage(
                                       long ackExpStamp,
                                       int priority,
                                       Reliability reliability,
                                       SIBUuid12 stream)
                    throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "sendAckExpectedMessage",
                        new Object[] { new Long(ackExpStamp), new Integer(priority), reliability, stream });
        HashMap allPubSubOutputHandlers = _destination.getAllPubSubOutputHandlers();
        try
        {
            Iterator itr = allPubSubOutputHandlers.values().iterator();

            while (itr.hasNext())
            {
                PubSubOutputHandler handler = (PubSubOutputHandler) itr.next();

                // Send AckExpected to all OutputHandlers
                handler.processAckExpected(ackExpStamp, priority, reliability, stream);
            }
        } finally
        {
            // By calling the getAllPubSubOutputHandlers it will lock the 
            // handlers
            _destination.unlockPubsubOutputHandlers();
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "sendAckExpectedMessage");
    }
594257
protected void sendMessage(DataOutputStream outputStream, Message message) throws IOException {
        long startNs = System.nanoTime();
        ProtoUtils.writeMessage(outputStream, message);
        if(streamStats != null) {
            streamStats.reportNetworkTime(operation,
                                          Utils.elapsedTimeNs(startNs, System.nanoTime()));
        }
    }
580359
public void moveMessage(boolean discard)
    throws SIMPControllableNotFoundException,
           SIMPRuntimeOperationFailedException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "moveMessage", new Object[] { Boolean.valueOf(discard) });

    assertValidControllable();

    LocalTransaction tran = _txManager.createLocalTransaction(false);
    SIMPMessage msg = null;
    try
    {
      msg = getSIMPMessage();
    }
    catch (SIResourceException e)
    {
      // No FFDC code needed    
      SIMPRuntimeOperationFailedException finalE =
        new SIMPRuntimeOperationFailedException(
          nls.getFormattedMessage(
            "INTERNAL_MESSAGING_ERROR_CWSIP0003",
            new Object[] {"QueuedMessage.moveMessage",
                          "1:473:1.66",
                          e,
                          Long.toString(_messageID)},
            null), e,
            "INTERNAL_MESSAGING_ERROR_CWSIP0003", new Object[] {Long.toString(_messageID)});
  
      SibTr.exception(tc, finalE);
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
        SibTr.exit(tc, "moveMessage", finalE);
      throw finalE;
    } 
    if(!discard)
    { 
      try
      {
        copyMessageToExceptionDestination(tran);  
      }
      catch (Exception e)
      {
        // No FFDC code needed 
        SIMPRuntimeOperationFailedException finalE =
          new SIMPRuntimeOperationFailedException(e);
        
        SibTr.exception(tc, finalE);
        
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
          SibTr.exit(tc, "moveMessage", finalE);
        throw finalE; 
      }

      discard = true;
    }
    
    if (discard)
    {      
      try
      {
        if(msg.isInStore())
        {
          if(msg.getLockID()==AbstractItem.NO_LOCK_ID)
          {
            //lock the message
            // Cache a lock ID to lock the items with          
            long lockID = _messageProcessor.getMessageStore().getUniqueLockID(AbstractItem.STORE_NEVER);
            msg.lockItemIfAvailable(lockID);
          }
          Transaction msTran = _messageProcessor.resolveAndEnlistMsgStoreTransaction(tran);
          msg.remove(msTran, msg.getLockID());  
        }
      }
      catch (Exception e)
      {
        // No FFDC code needed
        
        SIMPRuntimeOperationFailedException finalE = null;
        
        boolean adding = false;
        boolean removing = false;      
        
        if (msg instanceof MessageItem)
        {
          adding = ((MessageItem)msg).isAdding();
          removing = ((MessageItem)msg).isRemoving();
        }
        else
        {
          adding = ((MessageItemReference)msg).isAdding();
          removing = ((MessageItemReference)msg).isRemoving();          
        }
        
        if (adding)
        {
          // If the message is adding it probably means that 
          // we have an indoubt transaction which hasn't been committed
          finalE =
            new SIMPRuntimeOperationFailedException(
              nls.getFormattedMessage(
                "MESSAGE_INDOUBT_WARNING_CWSIP0361",
                new Object[] {Long.toString(_messageID),
                              _destinationHandler.getName()},
                null), 
              e,
              "MESSAGE_INDOUBT_WARNING_CWSIP0361", new Object[] {Long.toString(_messageID),
                _destinationHandler.getName() });          
        }         
        else if (removing)
        {
          // If the message is deleting it probably means that 
          // we have delivered the message to a consumer
          finalE =
            new SIMPRuntimeOperationFailedException(
              nls.getFormattedMessage(
                "MESSAGE_INDOUBT_WARNING_CWSIP0362",
                new Object[] {Long.toString(_messageID),
                              _destinationHandler.getName()},
                null), 
              e,
              "MESSAGE_INDOUBT_WARNING_CWSIP0362", new Object[] { Long.toString(_messageID),
                  _destinationHandler.getName() });          
        }         
        else
        {
          // Unexpected exception
          FFDCFilter.processException(
            e,
            "com.ibm.ws.sib.processor.runtime.QueuedMessage.moveMessage",
            "1:579:1.66",
            this);
    
          finalE =
            new SIMPRuntimeOperationFailedException(
              nls.getFormattedMessage(
                "INTERNAL_MESSAGING_ERROR_CWSIP0003",
                new Object[] {"QueuedMessage.removeMessage",
                              "1:587:1.66",
                              e,
                              Long.toString(_messageID)},
                null), e,
                "INTERNAL_MESSAGING_ERROR_CWSIP0003", new Object[] { Long.toString(_messageID)});
        }
        
        try
        {
          tran.rollback();
        }
        catch (SIException ee)
        {
          FFDCFilter.processException(
            ee,
            "com.ibm.ws.sib.processor.runtime.QueuedMessage.moveMessage",
            "1:603:1.66",
            this);
          
          SibTr.exception(tc, ee);
        }
        

        SibTr.exception(tc, finalE);
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
          SibTr.exit(tc, "moveMessage", finalE);
        throw finalE;             
      }
    }
    try
    {
      tran.commit();
    }
    catch (SIException e)
    {
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.runtime.QueuedMessage.moveMessage",
        "1:625:1.66",
        this);

      SIMPRuntimeOperationFailedException finalE =
        new SIMPRuntimeOperationFailedException(
          nls.getFormattedMessage(
            "INTERNAL_MESSAGING_ERROR_CWSIP0003",
            new Object[] {"QueuedMessage.removeMessage",
                          "1:633:1.66",
                          e,
                          Long.toString(_messageID)},
            null), e,
            "INTERNAL_MESSAGING_ERROR_CWSIP0003", new Object[] {Long.toString(_messageID)});

      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"); 
  }
417649
public SIBusMessage nextLocked()
  throws SISessionUnavailableException, SISessionDroppedException,
         SIResourceException, SIConnectionLostException,
         SIErrorException, SIIncorrectCallException
  {
    if (TraceComponent.isAnyTracingEnabled() && CoreSPILockedMessageEnumeration.tc.isEntryEnabled())
      SibTr.entry(CoreSPILockedMessageEnumeration.tc, "nextLocked",
        new Object[] {new Integer(hashCode()), this});

    JsMessage jsMsg = null;
    JsMessageWrapper msg = null;
    LMEMessage dirtyMessage = null;
    boolean removeMsg = false;

    checkValidState("nextLocked");

    localConsumerPoint.checkNotClosed();

    // Make any list modifications under the lock
    synchronized(this)
    {
      // If we still have an unlocked message in the list remove it now
      if(currentUnlockedMessage != null)
      {
        removeMessage(currentUnlockedMessage);
        currentUnlockedMessage = null;
      }

      if(currentMsg != null)
        currentMsg = currentMsg.next;
      else if(!endReached)
        currentMsg = firstMsg;

      if(currentMsg != null)
      {
        msg = currentMsg.message;
        jsMsg = currentMsg.jsMessage;

        // Due to the slight dodgyness of setting jsMsg (see comment on dirtyMessage)
        // this tries to ensure our logic is correct
        if((jsMsg != null) && !currentMsg.wasRead)
        {
          SIErrorException e =
            new SIErrorException(nls.getFormattedMessage(
                            "INTERNAL_MESSAGING_ERROR_CWSIP0001",
                            new Object[] {
                              "com.ibm.ws.sib.processor.impl.AbstractLockedMessageEnumeration",
                              "1:1318:1.154.3.1" },
                              null));

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

          throw e;
        }

        // If the message was added to the message store but the consumer does
        // not require recovery we can delete the message before we return it.
        // We don't want to do it here under the lock, so we remember to do it
        // later but pretend we've already done it by the time we release the
        // lock - that way no-one else will think its in the message store when it
        // isn't.
        if(currentMsg.isStored && !currentMsg.isRecoverable)
        {
          removeMsg = true;
          currentMsg.isStored = false;
        }

        // We're about to release the list lock but we need to be able to set
        // the jsMessage back into the LMEMessage object later so we save a
        // pointer to it. BUT there is a slight possibility that by the time
        // we write the jsMessage into it the LMEMessage has been deleted from
        // the list (e.g. by an unlockAll() from the consumer), that's not a
        // problem in itself. BUT by deleting it from the list we actually can
        // put it into the set of pooled objects so you could be writing over
        // another message BUT we currently hold the AsynchConsumerLock and we
        // know no new messages can be added to the list unless they hold that
        // lock SO we may have moved into the pool but we couldn't have been
        // re-used SO we're safe!
        dirtyMessage = currentMsg;
        currentMsg.wasRead = true;

        //indicate that there is a message under the cursor
        messageAvailable = true;
      }
      else
      {
        messageAvailable = false;
        endReached = true;
      }
    } // synchronized

    //check it isn't null
    if((msg != null) && (jsMsg == null))
    {
      jsMsg = setPropertiesInMessage(currentMsg);


      if(removeMsg)
      {
        try
        {
          removeMessageFromStore(msg,
                                 null,
                                 false); // false = DON'T decrement active message count
                                         // (wait until the message is deleted/unlocked from the LME)
        }
        catch (SIMPMessageNotLockedException e)
        {
//        SIMPMessageNotLockedException shouldn't occur so FFDC.
          FFDCFilter.processException(
            e,
            "com.ibm.ws.sib.processor.impl.AbstractLockedMessageEnumeration.nextLocked",
            "1:1396:1.154.3.1",
            this);

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

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

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

      dirtyMessage.jsMessage = jsMsg; // Read earlier comment before touching this!
    }

    if (TraceComponent.isAnyTracingEnabled() && UserTrace.tc_mt.isDebugEnabled())
      UserTrace.trace_Receive(null,
                              jsMsg,
                              consumerSession.getDestinationAddress(),
                              consumerSession.getIdInternal());

    if (TraceComponent.isAnyTracingEnabled() && CoreSPILockedMessageEnumeration.tc.isEntryEnabled())
      SibTr.exit(CoreSPILockedMessageEnumeration.tc, "nextLocked", new Object[] {jsMsg});
    //return the message or null
    return jsMsg;
  }
796952
public void requestNextMessageBatch()
      throws SIConnectionDroppedException, SIConnectionLostException
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "requestNextMessageBatch");

      // This is done by simply requesting the ME end starts up its asynchronous consumer to deliver us another
      // batch of messages however since this is not a start from the application we ask for a restart so that
      // we can take into account whether the consumer session has been stopped by message processor or not.
      sendStart(true);                                                                                          //471642

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "requestNextMessageBatch");
        }
1155698
@Override
    public void sendFlushedMessage(SIBUuid8 ignore, SIBUuid12 streamID) throws SIResourceException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "sendFlushedMessage", new Object[] { streamID });

        // This flush should be broadcast to all downstream neighbors 
        // for this cell as it is the result of a startFlush()

        // This is a bit of a kludge since we may be sending flushes to cells
        // which have no information about a stream.
        HashMap allPubSubOutputHandlers = _destination.getAllPubSubOutputHandlers();
        try
        {
            Iterator itr = allPubSubOutputHandlers.values().iterator();

            while (itr.hasNext())
            {
                // Get the appropriate target cellule and forward the message
                PubSubOutputHandler handler = (PubSubOutputHandler) itr.next();

                // Note that the null Cellule we pass in here is ignored
                // as each OutputHandler knows its targetCellule 
                handler.sendFlushedMessage(null, streamID);

                // Also, tell the handler to remove any information for this
                // stream since it's just been flushed.
                handler.removeStream(streamID);
            }
        } finally
        {
            // By calling the getAllPubSubOutputHandlers it will lock the 
            // handlers
            _destination.unlockPubsubOutputHandlers();
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "sendFlushedMessage");
    }
583037
public void forceFlush(SIBUuid12 streamID) throws SIErrorException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "forceFlush", streamID);

        StreamSet streamSet = streamSets.get(streamID);
        streamSet.dereferenceControlAdapter();

        // Send out a flushed message.  If this fails, make sure we get
        // to at least invoke the callback.
        try
        {
            // The Cellule argument is null as it is ignored by our parent handler
            // this flush message should be broadcast downstream.
            // This will also implicitly remove the streamSet
            downControl.sendFlushedMessage(null, streamID);
        } catch (Exception e)
        {
            // FFDC
            FFDCFilter.processException(
                                        e,
                                        "com.ibm.ws.sib.processor.gd.InternalOutputStreamManager.forceFlush",
                                        "1:743:1.48.1.1",
                                        this);

            // Note that it doesn't make much sense to throw an exception here since
            // this is a callback from stream array map.
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                SibTr.exit(tc, "forceFlush", e);
            return;
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "forceFlush");
    }
1226865
public void requestMoreMessages(int receivedBytes, int requestedBytes)
      throws SIConnectionDroppedException, SIConnectionLostException
   {
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "requestMoreMessages",
                                           new Object[]{""+receivedBytes, ""+requestedBytes});

      if (sessionId == 0)
      {
         // If the session Id = 0, then no one called setSessionId(). As such we are unable to flow
         // to the server as we do not know which session to instruct the server to use.
         SIErrorException e = new SIErrorException(
            nls.getFormattedMessage("SESSION_ID_HAS_NOT_BEEN_SET_SICO1043", null, null)
         );

         FFDCFilter.processException(e, CLASS_NAME + ".requestMoreMessages",
                                     CommsConstants.CONVERSATIONHELPERIMPL_07, this);

         if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.debug(this, tc, e.getMessage(), e);

         throw e;
      }

      CommsByteBuffer request = getCommsByteBuffer();
      // Connection object id
      request.putShort(connectionObjectId);
      // Consumer session id
      request.putShort(sessionId);
      // Received bytes
      request.putInt(receivedBytes);
      // Requested bytes
      request.putInt(requestedBytes);

      // Pass on call to server
      jfapSend(request,
               JFapChannelConstants.SEG_REQUEST_MSGS,
               JFapChannelConstants.PRIORITY_MEDIUM,
               true,
               ThrottlingPolicy.BLOCK_THREAD);

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "requestMoreMessages");
   }