sendSilenceMessage may be called from InternalOutputStream
when a Nack is recevied
506037
public void sendSilenceMessage(
     long startStamp,
     long endStamp,
     long completedPrefix,
     boolean requestedOnly,
     int priority,
     Reliability reliability,
     SIBUuid12 stream)
     throws SIResourceException
   {

     if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
       SibTr.entry(tc, "sendSilenceMessage",
         new Object[] {
           new Long(startStamp),
           new Long(endStamp),
           new Long(completedPrefix),
           new Integer(priority),
           reliability });

     ControlSilence sMsg;
     try
     {
       // Create new Silence message
       sMsg = _cmf.createNewControlSilence();
     }
     catch (Exception e)
     {
       // FFDC
       FFDCFilter.processException(
         e,
         "com.ibm.ws.sib.processor.impl.PubSubOutputHandler.sendSilenceMessage",
         "1:787:1.164.1.5",
         this);

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

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

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

     // As we are using the Guaranteed Header - set all the attributes as 
     // well as the ones we want.
     SIMPUtils.setGuaranteedDeliveryProperties(sMsg,
         _messageProcessor.getMessagingEngineUuid(), 
         null,
         stream,
         null,
         _destinationHandler.getUuid(),
         ProtocolType.PUBSUBINPUT,
         GDConfig.PROTOCOL_VERSION);
    
     sMsg.setStartTick(startStamp);
     sMsg.setEndTick(endStamp);
     sMsg.setPriority(priority);
     sMsg.setReliability(reliability);
     sMsg.setCompletedPrefix(completedPrefix);
     sMsg.setRequestedOnly(requestedOnly);
     
     // If the destination in a Link add Link specific properties to message
     if( _isLink )
     {
       sMsg = (ControlSilence)addLinkProps(sMsg);
     }
    
     // Send message to destination
     // Using MPIO

     // Send the message to the MessageTransmitter
     //add a target cellule to the array for sending
     SIBUuid8[] fromTo = new SIBUuid8[1];
     fromTo[0] = _targetMEUuid;
      
     // Send at priority+1 if this is a response to a Nack
     if( requestedOnly )
       _mpio.sendDownTree(fromTo, priority+1, sMsg);
     else
       _mpio.sendDownTree(fromTo, priority, sMsg);

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

   }
492962
@Override
    protected void sendMessage(@NonNull INDArray message, int iterationNumber, int epochNumber) {
        // here we'll send our stuff to other executores over the wire
        // and let's pray for udp broadcast availability

        // Send this message away
        // FIXME: do something with unsafe duplication, which is bad and used ONLY for local spark
        try (MemoryWorkspace wsO = Nd4j.getMemoryManager().scopeOutOfWorkspaces()) {
            long updateId = updatesCounter.getAndIncrement();

            val m = message.unsafeDuplication();
            ModelParameterServer.getInstance().sendUpdate(m, iterationNumber, epochNumber);
        }


        // heere we update local queue
        super.sendMessage(message, iterationNumber, epochNumber);
    }
733738
@Override
  public void sendPushProfileAnswer(JPushProfileAnswer answer) throws InternalException, IllegalDiameterStateException, RouteException, OverloadException {
    send(Event.Type.SEND_MESSAGE, null, answer);
  }
502548
private void sendNSFailed(IConnection conn, String errorCode, String description, String name, Number streamId) {
        StreamService.sendNetStreamStatus(conn, errorCode, description, name, Status.ERROR, streamId);
    }
1278584
@Override
    public void onMessage(StreamMessage received)
    {
        if (this.reloadConfig && this.watcher != null)
        {
            Map<String, Object> reloadedConfig = null;

            try
            {
                reloadedConfig = this.watcher.readIfUpdated();
            }
            catch (IOException ex)
            {
                String logFormat = "Config file reload failed. Skip reload config.";
                logger.warn(logFormat, ex);
            }

            if (reloadedConfig != null)
            {
                onUpdate(reloadedConfig);
            }
        }

        this.executingKeyHistory = received.getHeader().getHistory();
        this.responsed = false;

        // Execute message processing.
        // This class not handles exceptions.
        // Because if throws exception, this class not need ack.
        try
        {
            onExecute(received);
        }
        finally
        {
            clearExecuteStatus();
        }

        // If not responsed, auto ack
        if (this.responsed == false)
        {
            super.ack();
        }
    }
1043489
public void sendNackMessage(SIBUuid8 meUuid, 
                              SIBUuid12 destUuid,
                              SIBUuid8  busUuid,  
                              long startTick, 
                              long endTick, 
                              int priority, 
                              Reliability reliability, 
                              SIBUuid12 streamID)
  throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "sendNackMessage", 
        new Object[] { meUuid,
                       new Long(startTick),
                       new Long(endTick),
                       new Integer(priority),
                       reliability,
                       streamID });

    ControlNack newNackMsg =
      createControlNackMessage(
        priority,
        reliability,
        streamID);
    
    newNackMsg.setStartTick(startTick);
    newNackMsg.setEndTick(endTick);
        
    try
    {
      _parentInputHandler.handleControlMessage(null, newNackMsg);
    }
    catch (SIException e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.impl.PubSubOutputHandler.sendNackMessage",
        "1:1639:1.164.1.5",
        this);
        
      SibTr.exception(tc, e);
      
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      {
        SibTr.exception(tc, e);
        SibTr.exit(tc, "sendNackMessage", e);
      }
    }
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "sendNackMessage");
  }
953001
@Override
    public void sendNackMessage(SIBUuid8 sourceMEUuid,
                                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();

        // As we are using the Guaranteed Header - set all the attributes as
        // well as the ones we want.
        SIMPUtils.setGuaranteedDeliveryProperties(nackMsg,
                                                  _messageProcessor.getMessagingEngineUuid(),
                                                  sourceMEUuid,
                                                  stream,
                                                  null,
                                                  destUuid,
                                                  ProtocolType.UNICASTOUTPUT,
                                                  GDConfig.PROTOCOL_VERSION);

        nackMsg.setPriority(priority);
        nackMsg.setReliability(reliability);

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

        //  Send Nack messages at original message priority +2
        sendToME(nackMsg, sourceMEUuid, busUuid, priority + 2);

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "sendNackMessage ");
    }
1351882
@Override
    public void messageReceived(IoSession session, Object buf) {
        final IoSessionInputStream in = (IoSessionInputStream) session
                .getAttribute(KEY_IN);
        in.write((IoBuffer) buf);
    }
745241
public void sendSilenceMessage(
    long startStamp,
    long endStamp,
    long completedPrefix,
    boolean requestedOnly,      
    int priority,
    Reliability reliability,
    SIBUuid12 stream)
    throws SIResourceException
  {

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "sendSilenceMessage");

    ControlSilence sMsg;
    try
    {
      // Create new Silence message
      sMsg = cmf.createNewControlSilence();
    }
    catch (Exception e)
    {
      // FFDC
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.impl.PtoPOutputHandler.sendSilenceMessage",
        "1:849:1.241",
        this);

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

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

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

    // As we are using the Guaranteed Header - set all the attributes as
    // well as the ones we want.
    SIMPUtils.setGuaranteedDeliveryProperties(sMsg,
        messageProcessor.getMessagingEngineUuid(), 
        targetMEUuid,
        stream,
        null,
        destinationHandler.getUuid(),
        ProtocolType.UNICASTINPUT,
        GDConfig.PROTOCOL_VERSION);
  
    sMsg.setStartTick(startStamp);
    sMsg.setEndTick(endStamp);
    sMsg.setPriority(priority);
    sMsg.setReliability(reliability);
    sMsg.setCompletedPrefix(completedPrefix);

    // If the destination in a Link add Link specific properties to message
    if( isLink )
    {
      sMsg = (ControlSilence)addLinkProps(sMsg);
    }
    
    // If the destination is system or temporary then  the 
    // routingDestination into th message
    if( this.isSystemOrTemp )
    {
      sMsg.setRoutingDestination(routingDestination); 
    }
 
    // Send message to destination
    // Using MPIO
    // If requestedOnly then this is a response to a Nack so resend at priority+1 
    if( requestedOnly )
      mpio.sendToMe(routingMEUuid, priority+1, sMsg);
    else
      mpio.sendToMe(routingMEUuid, priority, sMsg);
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "sendSilenceMessage");

  }
350729
protected boolean invokeMessageRouters(RoutedMessage routedMessage) {

        MessageRouter externalMsgRouter = externalMessageRouter.get();
        WsMessageRouter internalMsgRouter = internalMessageRouter.get();

        boolean retMe = true;

        try {
            /*
             * Need a recursion Counter for invokingMessageRouters
             * If an osgi framework debugging options is enabled then a message
             * is sent to sysout/syserr from wsMessageRouterImpl during the
             * Reentrant Read Write lock and will cause an infinite loop -- crash.
             *
             * Setting it to 2 to allows the "osgi" debug statement to occur once.
             * i.e Original message goes through and then osgi statement is emitted and is looped back once
             * and recursion counter will prevent any further statements.
             */
            if (!(counterForMessageRouter.incrementCount() > 2)) {
                if (externalMsgRouter != null) {
                    retMe &= externalMsgRouter.route(routedMessage.getFormattedMsg(), routedMessage.getLogRecord());
                }
                if (internalMsgRouter != null) {
                    retMe &= internalMsgRouter.route(routedMessage);
                } else {
                    String message = formatter.messageLogFormat(routedMessage.getLogRecord(), routedMessage.getFormattedVerboseMsg());
                    RoutedMessage specialRoutedMessage = new RoutedMessageImpl(routedMessage.getFormattedMsg(), routedMessage.getFormattedVerboseMsg(), message, routedMessage.getLogRecord());
                    synchronized (this) {
                        if (earlierMessages != null) {
                            earlierMessages.add(specialRoutedMessage);
                        }
                    }
                }
            }
        } finally {
            counterForMessageRouter.decrementCount();
        }
        return retMe;
    }
550217
private void send(AuditEventMessage msg, Destination<S> destination) throws Exception {
        if (EventUtils.isEmptyOrNull(msg)) {
            return;
        }
        // Serialize and format event message for syslog
        byte[] msgBytes = getTransportPayload(msg);
        if (EventUtils.isEmptyOrNull(msgBytes)) {
            return;
        }
        destination.write(msgBytes);
    }
452550
private void handleSendDataRequest(SerialMessage incomingMessage) {
		logger.trace("Handle Message Send Data Request");
		
		int callbackId = incomingMessage.getMessagePayloadByte(0);
		TransmissionState status = TransmissionState.getTransmissionState(incomingMessage.getMessagePayloadByte(1));
		SerialMessage originalMessage = this.lastSentMessage;
		
		if (status == null) {
			logger.warn("Transmission state not found, ignoring.");
			return;
		}

		logger.debug("CallBack ID = {}", callbackId);
		logger.debug(String.format("Status = %s (0x%02x)", status.getLabel(), status.getKey()));
		
		if (originalMessage == null || originalMessage.getCallbackId() != callbackId) {
			logger.warn("Already processed another send data request for this callback Id, ignoring.");
			return;
		}
		
		switch (status) {
			case COMPLETE_OK:
				ZWaveNode node = this.getNode(originalMessage.getMessageNode());
				node.resetResendCount();
				
				// in case we received a ping response and the node is alive, we proceed with the next node stage for this node.
				if (node != null && node.getNodeStage() == NodeStage.NODEBUILDINFO_PING) {
					node.advanceNodeStage();
				}
				if (incomingMessage.getMessageClass() == originalMessage.getExpectedReply() && !incomingMessage.isTransActionCanceled()) {
					notifyEventListeners(new ZWaveEvent(ZWaveEventType.TRANSACTION_COMPLETED_EVENT, this.lastSentMessage.getMessageNode(), 1, this.lastSentMessage));
					transactionCompleted.release();
					logger.trace("Released. Transaction completed permit count -> {}", transactionCompleted.availablePermits());
				}
				return;
			case COMPLETE_NO_ACK:
			case COMPLETE_FAIL:
			case COMPLETE_NOT_IDLE:
			case COMPLETE_NOROUTE:
				try {
					handleFailedSendDataRequest(originalMessage);
				} finally {
					transactionCompleted.release();
					logger.trace("Released. Transaction completed permit count -> {}", transactionCompleted.availablePermits());
				}
			default:
		}
	}
1200519
@Override
    public void sendInternal(NotificationType type, String... messages) throws NotificationException {
        if (producer == null) {
            createProducer();
        }
        sendInternalToProducer(producer, type, messages);
    }
540736
private void ack() {
    // Send a message to the other side of the connection indicating the
    // last message that we received in order. This will allow it to
    // purge messages we've already received from its queue.
    if (open && connected) {
      if (log.isDebugEnabled()) {
        log.debug(String.format("%s - Acking messages up to: %d", this, lastReceived));
      }
      eventBus.send(outAddress, new JsonObject().putString("action", "ack").putNumber("id", lastReceived));
      lastFeedbackTime = System.currentTimeMillis();
    }
  }
1078795
@Override
    public void processAckExpected(long stamp)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "processAckExpected", new Object[] { Long.valueOf(stamp) });

        List nackList = null;
        List msgList = null;
        boolean processNack = true;

        // Take lock on target stream and hold it until messages have been
        // added to batch by deliverOrderedMessages 
        synchronized (this)
        {
            synchronized (oststream) //see defect 289889
            {
                long completedPrefix = oststream.getCompletedPrefix();
                long blockingTick = -1;
                // Send Ack message for ticks up to completedPrefix
                sendAck();
                if (isStreamBlocked())
                {
                    // If the link was implicitly blocked due to an 'unexpected' problem when we tried
                    // to deliver a message (the most obvious reason is due to not being authorized)
                    // we can't keep the stream blocked as we need to allow messages in again to re-try
                    // the deliverery of the message to be able to tell if the problem still exists.
                    // However, rather than nack all messages currently blocked on the stream (which is
                    // what the source end will be ack-expecting) we only nack the blocking message,
                    // if that succeeds the stream will be left un-blocked, but if it fails the stream
                    // will be re-locked. If the stream is left un-blocked, the next time an ack-expected
                    // comes in we'll nack everything.
                    if (isStreamBlockedUnexpectedly())
                    {
                        stamp = linkBlockingTick;
                        blockingTick = linkBlockingTick;

                        // We need to unblock the stream, otherwise the failing message won't be re-delivered
                        // when it arrives (it'll get bounced in writeValue), and that's the only way to tell
                        // is the failure is still valid (as it's not detected by streamCanAcceptNewMessage).
                        setStreamIsBlocked(false, DestinationHandler.OUTPUT_HANDLER_FOUND, null, null);
                    }
                    else
                    {
                        // The stream has been marked as blocked. We should only send 
                        // NACKs if we are able to accept the resent message once it arrives.
                        // Because ACKs are for the stream as a whole, if this is a link
                        // then we need to check whether the link's exception destination is
                        // able to accept messages. If no exception destination is defined on the
                        // link then we check the link blocking destination.
                        int blockingReason = deliverer.checkStillBlocked();
                        if (blockingReason == DestinationHandler.OUTPUT_HANDLER_FOUND)
                        {
                            //the destination is no longer blocked
                            setStreamIsBlocked(false, DestinationHandler.OUTPUT_HANDLER_FOUND, null, null);
                        }
                        else
                        {
                            // Its possible that the reason for the blockage has changed. This code allows
                            // the link health state to be altered to reflect that change (but keeps the
                            // same blocking tick as that hasn't changed)
                            setStreamIsBlocked(true, blockingReason, null, streamBlockingAddress);

                            // It's possible that the blockage has been cleared on the sending side without
                            // us knowing about it (e.g. the offending message was deleted but the silence never
                            // made it to us). So, to err on the side of caution, we occasionally forget about
                            // the blockage and allow the blocking tick to be nack'd (just that one, not all
                            // of them, we don't want them all to be re-sent until we know the problem doesn't
                            // still exist). This will either cause the same problematic message to be re-sent
                            // and the link will stay blocked or a silence will arrive indicating the blockage
                            // has been cleared, allowing the subsequent ackExcepted to result in a full Nack
                            if (blockingCount++ > 3)
                            {
                                stamp = linkBlockingTick;
                                blockingCount = 0;
                                blockingTick = linkBlockingTick;

                                // Note: In this case (unlike the unexpected block case above) we can leave the
                                // stream blocked as the original problem was detected by streamCanAcceptNewMessage,
                                // if the mesasge is re-sent writeValue will have a chance of processing it if
                                // the call to streamCanAcceptNewMessage thinks the problem has been resolved.

                                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                                    SibTr.debug(tc, "Nacking the first tick (" + stamp + ") of a blocked link");
                            }
                            // The stream is blocked (for now) so we don't want the blocking message (and all
                            // the ones built up behind it) to be resent because we nack them!
                            else
                                processNack = false;
                        }
                    }
                }
                //We're allowed to send Nacks
                if (processNack)
                {
                    if (stamp <= completedPrefix)
                    {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                            SibTr.exit(tc, "processAckExpected");
                        return;
                    }

                    // walk the oststream from the completedPrefix to endstamp and Nack
                    // everything that is Unknown or Requested
                    // Change everything which is Unknown to Requested.
                    nackList = new ArrayList();
                    long startstamp = completedPrefix + 1;
                    long endstamp = stamp;
                    oststream.setCursor(startstamp);
                    TickRange tr2 = null;
                    TickRange tr1 = oststream.getNext();
                    do
                    {
                        if ((tr1.type == TickRange.Unknown)
                            || (tr1.type == TickRange.Requested))
                        {
                            long ss, es;

                            // Handle case where start or endstamps fall within a range
                            ss = tr1.startstamp > startstamp ? tr1.startstamp : startstamp;
                            es = endstamp > tr1.endstamp ? tr1.endstamp : endstamp;

                            TickRange tr = new TickRange(TickRange.Requested, ss, es);

                            // Only need to update stream if this is currently Unknown
                            if (tr1.type == TickRange.Unknown)
                            {
                                oststream.writeRange(tr);
                            }

                            // SIB0105
                            // Set the head msg health state to AMBER if neccessary
                            synchronized (pendingAlarms)
                            {
                                // If we have no current gaps then we must be missing msg at the head of the stream
                                if (!getAlarms(this).hasNext())
                                {
                                    getControlAdapter().getHealthState().updateHealth(HealthStateListener.MSG_LOST_ERROR_STATE,
                                                                                      HealthState.AMBER);
                                    lastNackTick = endstamp;
                                }
                            }

                            nackList.add(tr);
                        }
                        // If the stream is blocked but we want to force the message (or a silence if it's
                        // been deleted on the source side) to be sent again then we need to add the blocking
                        // range to the nack list, even if it's in value state (which is possible if we tried
                        // to deliver it before but failed (so it's already added to the stream, but not
                        // delivered yet).
                        else if ((tr1.type == TickRange.Value) && (tr1.endstamp == blockingTick))
                        {
                            nackList.add(tr1);
                        }
                        tr2 = tr1;
                        tr1 = oststream.getNext();
                    } while ((tr1.startstamp <= endstamp) && (tr1 != tr2));
                }//end processNack
                else
                {
                    //we did not process the AckExpected message for any nacks
                    //as we did not have room for the replies
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                        SibTr.debug(tc, "did not process AckExpected message for nacks.");

                }

                // PK57736 We should should this stimulus from the sending side
                // to advance the doubt horizon, as it is possible they are not
                // sending messages any longer (which is the other stimulus).
                msgList = advanceDoubtHorizon(null);

            } // end synchronized

            // PK57736 Deliver messages outside of synchronise 
            // We do this because deliverOrderedMessages takes the 
            // BatchHandler lock and the BatchHandler callbacks require
            // the stream lock to update the completedPrefix. If we call
            // the BatchHandler when we hold the stream lock it could cause
            // a deadlock
            if (msgList != null)
            {
                // Call the Input or Output Handler to deliver the messages
                try
                {
                    deliverer.deliverOrderedMessages(msgList, this, priority, reliability);
                } catch (SINotPossibleInCurrentConfigurationException e)
                {
                    // No FFDC code needed
                    SibTr.exception(tc, e);

                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                        SibTr.debug(tc, "processAckExpected", "GDException");

                    // Dont rethrow the exception. The GD protocols will handle the resend of
                    // the message
                } catch (SIException e)
                {
                    // FFDC
                    FFDCFilter.processException(
                                                e,
                                                "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream.processAckExpected",
                                                "1:1201:1.110",
                                                this);

                    SibTr.exception(tc, e);
                    SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                new Object[] {
                                              "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream",
                                              "1:1208:1.110",
                                              e });

                    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                        SibTr.exit(tc, "processAckExpected", "GDException");

                    throw new SIErrorException(
                                    nls.getFormattedMessage(
                                                            "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                            new Object[] {
                                                                          "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream",
                                                                          "1:1218:1.110",
                                                                          e },
                                                            null),
                                    e);
                }
            }
        } // end sync - release lock on target stream

        // send nacks
        for (int j = 0; processNack && j < nackList.size(); j++)
        {
            TickRange temptr = (TickRange) nackList.get(j);

            try
            {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    SibTr.debug(tc, "processAckExpected sending Nack from: " + temptr.startstamp + " to " + temptr.endstamp);

                upControl.sendNackMessage(
                                          streamSet.getRemoteMEUuid(),
                                          streamSet.getDestUuid(),
                                          streamSet.getBusUuid(),
                                          temptr.startstamp,
                                          temptr.endstamp,
                                          priority,
                                          reliability,
                                          streamSet.getStreamID());
            } catch (SIResourceException e)
            {
                // FFDC
                FFDCFilter.processException(
                                            e,
                                            "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream.processAckExpected",
                                            "1:1252:1.110",
                                            this);

                SibTr.exception(tc, e);
                SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                            new Object[] {
                                          "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream",
                                          "1:1259:1.110",
                                          e });

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

                throw new SIErrorException(
                                nls.getFormattedMessage(
                                                        "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                        new Object[] {
                                                                      "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream",
                                                                      "1:1270:1.110",
                                                                      e },
                                                        null),
                                e);
            }
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "processAckExpected");
    }
627083
private void writeSilenceInternal(TickRange tr, boolean forced)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "writeSilenceInternal", new Object[] { tr, Boolean.valueOf(forced) });

        List msgList = null;

        // Take lock on target stream and hold it until messages have been
        // added to batch by deliverOrderedMessages 
        synchronized (this)
        {
            // We are only allowed to remove a message from the stream and replace 
            // it with Silence if it has not yet been added to a Batch to be delivered
            // We know that this can't happen asynchronously once we hold the targetStream lock
            // and the nextCompletedPrefix tells us the last message in the current batch 
            if (!forced || tr.valuestamp > this.nextCompletedPrefix)
            {

                synchronized (oststream)
                {
                    // Get the completedPrefix 
                    long completedPrefix = oststream.getCompletedPrefix();

                    // check if all ticks in Silence msg are already acked as if
                    // so they will be changed to Completed anyway and we don't need
                    // to do it now
                    if (tr.endstamp > completedPrefix)
                    {
                        if (forced == false)
                            oststream.writeRange(tr);
                        else
                            oststream.writeCompletedRangeForced(tr);

                        // Get updated completedPrefix
                        completedPrefix = oststream.getCompletedPrefix();

                    }

                    // The completedPrefix may have advanced when the Silence message
                    // was writen to the stream or both
                    if ((completedPrefix + 1) > doubtHorizon)
                    {
                        // advance the doubt horizon
                        doubtHorizon = completedPrefix + 1;

                        // now call the method which trys to advance the doubtHorizon
                        // from it's current setting by moving over any ticks in Completed 
                        // state and sending Values messages until it reaches a tick in
                        // Unknown or Requested
                        msgList = advanceDoubtHorizon(null);
                    }

                    if ((doubtHorizon - 1) > unknownHorizon)
                        unknownHorizon = doubtHorizon - 1;

                    // the message was writen to the stream
                    // see if this message created a gap
                    if (tr.endstamp > unknownHorizon)
                    {
                        // check if gap created
                        if (tr.startstamp > (unknownHorizon + 1))
                        {
                            handleNewGap(unknownHorizon + 1, tr.startstamp - 1);
                        }
                        unknownHorizon = tr.endstamp;
                    }

                    // Reset the health state
                    if (lastNackTick >= 0)
                    {
                        if (tr.startstamp <= lastNackTick && tr.endstamp >= lastNackTick)
                        {
                            getControlAdapter().getHealthState().updateHealth(HealthStateListener.MSG_LOST_ERROR_STATE,
                                                                              HealthState.GREEN);
                            lastNackTick = -1;
                        }
                    }

                    // If the stream is blocked, see if we have a silence for the blocking tick
                    if (isStreamBlocked())
                    {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                            SibTr.debug(tc, "Stream is blocked on tick: " + linkBlockingTick + ", see if we can unblock it");
                        if (tr.endstamp >= linkBlockingTick)
                        {
                            if (tr.startstamp <= linkBlockingTick)
                            {
                                // The stream is no longer blocked
                                setStreamIsBlocked(false, DestinationHandler.OUTPUT_HANDLER_FOUND, null, null);
                            }
                        }
                    } // eof isStreamBlocked()
                } // end sync ( release the oststream lock )

                // Deliver messages outside of synchronise 
                // We do this because deliverOrderedMessages takes the 
                // BatchHandler lock and the BatchHandler callbacks require
                // the stream lock to update the completedPrefix. If we call
                // the BatchHandler when we hold the stream lock it could cause
                // a deadlock
                if (msgList != null)
                {
                    // Call the Input or Output Handler to deliver the messages
                    try
                    {
                        deliverer.deliverOrderedMessages(msgList, this, priority, reliability);
                    } catch (SINotPossibleInCurrentConfigurationException e)
                    {
                        // No FFDC code needed
                        SibTr.exception(tc, e);

                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                            SibTr.debug(tc, "writeSilenceInternal", "GDException");

                        // Dont rethrow the exception. The GD protocols will handle the resend of
                        // the message
                    } catch (SIException e)
                    {
                        // FFDC
                        FFDCFilter.processException(
                                                    e,
                                                    "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream.writeSilenceInternal",
                                                    "1:960:1.110",
                                                    this);

                        SibTr.exception(tc, e);
                        SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                    new Object[] {
                                                  "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream",
                                                  "1:967:1.110",
                                                  e });

                        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                            SibTr.exit(tc, "writeSilenceInternal", "GDException");

                        throw new SIErrorException(
                                        nls.getFormattedMessage(
                                                                "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                                new Object[] {
                                                                              "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream",
                                                                              "1:977:1.110",
                                                                              e },
                                                                null),
                                        e);
                    }
                }

            } // else can't remove as already in batch 
        } // end sync - release lock on target stream

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "writeSilenceInternal");
    }
989065
@Override
    public List sendValueMessages(
                                  List msgList,
                                  long completedPrefix,
                                  boolean requestedOnly,
                                  int priority,
                                  Reliability reliability,
                                  SIBUuid12 stream)
    {
        // NOP, PubSubOutputHandlers are responsible for pushing value
        // messages downstream.
        return null;
    }