@Override
    public void writeValue(MessageItem msgItem)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(this, tc, "writeValue", new Object[] { msgItem });

        TickRange tr = null;
        List msgList = null;
        JsMessage m = msgItem.getMessage();

        long valueTick = m.getGuaranteedValueValueTick();
        timeLastMsgReceived = System.currentTimeMillis();

        // Take lock on target stream and hold it until messages have been
        // added to batch by deliverOrderedMessages 
        synchronized (this)
        {
            synchronized (oststream)
            {
                //see if this stream can accept new messages BEFORE
                //we write messages into the stream in the 'value' state
                boolean canProcessMessage = false;
                try
                {
                    canProcessMessage = streamCanAcceptNewMessage(msgItem, valueTick);
                } catch (SIException e)
                {
                    //FFDC
                    FFDCFilter.processException(
                                                e,
                                                "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream.writeValue",
                                                "1:452:1.110",
                                                this);

                    SibTr.exception(tc, e);
                    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                        SibTr.exit(tc, "writeValue", e);
                    return;
                }
                if (canProcessMessage)
                {
                    boolean repeatedValue = false;
                    long completedPrefix = oststream.getCompletedPrefix();

                    // Check if this is a repeated Value message which we have
                    // already Acked and process accordingly
                    if (valueTick > completedPrefix)
                    {
                        // Message has not already been acked
                        tr = TickRange.newValueTick(valueTick, msgItem, -1);
                        tr.startstamp = m.getGuaranteedValueStartTick();
                        tr.endstamp = m.getGuaranteedValueEndTick();

                        // write this into the state stream
                        boolean streamChanged = false;
                        streamChanged = oststream.writeCombinedRange(tr);

                        // Check that we didn't already have this Value in the stream
                        if (streamChanged)
                        {
                            if (tr.valuestamp > valueHorizon)
                            {
                                valueHorizon = tr.valuestamp;
                            }

                            boolean gapcreated = false;

                            // determine the range of tick in the Completed state
                            // either side of this Value tick
                            // Note only the startstamp and endstamp will have been modified
                            tr = oststream.findCompletedRange(tr);

                            if (tr.endstamp > unknownHorizon)
                            {
                                // check if gap created
                                if (tr.startstamp > (unknownHorizon + 1))
                                {
                                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                                        SibTr.debug(
                                                    tc,
                                                    "created gap ["
                                                                    + (unknownHorizon + 1)
                                                                    + ", "
                                                                    + (tr.startstamp - 1)
                                                                    + "]");

                                    gapcreated = true;

                                    handleNewGap(unknownHorizon + 1, tr.startstamp - 1);
                                }
                                unknownHorizon = tr.endstamp;
                            }

                            if (!gapcreated)
                            {
                                // Only if gapcreated == false, can this message have the possibility
                                // of advancing the doubtHorizon and being delivered.

                                if (tr.startstamp <= doubtHorizon)
                                {
                                    // Deliver this message
                                    // We use a list to deliver messages
                                    if (msgList == null)
                                        msgList = new ArrayList();
                                    msgList.add(msgItem);

                                    // Set doubtHorizon to tr.endstamp + 1
                                    doubtHorizon = tr.endstamp + 1;
                                    msgList = advanceDoubtHorizon(msgList);
                                }
                                else
                                {
                                    // This case is only used if the previous batch of 
                                    // messages we delivered was rolled back and we need
                                    // to deliver them again
                                    msgList = advanceDoubtHorizon(msgList);
                                }
                            } // end if (!gapCreated)

                            // 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;
                                }
                            }
                        } // end if ( streamChanged )
                        else
                        {
                            // Stream was not changed so this was a repeated Value which 
                            // we have not yet Acked
                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                                SibTr.debug(
                                            tc,
                                            "Repeated V tick value - not yet acked "
                                                            + valueTick
                                                            + " : "
                                                            + completedPrefix);

                            repeatedValue = true;
                        }
                    }
                    else
                    { // valueTick <= completedPrefix

                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                            SibTr.debug(
                                        tc,
                                        "Repeated V tick value - already acked "
                                                        + valueTick
                                                        + " : "
                                                        + completedPrefix);

                        repeatedValue = true;
                    }

                    // It's acceptable to recieve the odd repeated value, for example a message
                    // is sent from the source slightly out of order with subsequent messages
                    // due to it's transaction taking longer to complete than others). So we (the
                    // target) see a gap appear, nack the gap (after a few milliseconds) then the
                    // message arrives anyway so we add it to the stream. Then the source receives
                    // the nack and re-sends the message. When that arrives at us we just ignore it
                    // becuase we already have it.
                    // You'd expect this to happen occasionally but if it happens too often it
                    // will add unnecessary traffic, reducing performance.
                    // There are a couple of reasons this can occur:
                    //  1) the initial nack is sent too fast to allow for slightly out of order
                    //     streams. This is set by GDConfig.GD_GAP_CURIOSITY_THRESHOLD (to 200 msecs)
                    //  2) The end-to-end transmission time of messages is excessively long, so
                    //     once a gap is introduced (by the sender overloading the socket) the gap
                    //     is nacked multiple times. This is due to the nack repetition timer (NRT)
                    //     popping multiple times before the re-sent message (from the first nack) is
                    //     actually recieved and the NRT cancelled. This end-to-end delay has been seen due
                    //     to excessive buffering of inbound messages in the Comms layer before being
                    //     processed by us. This buffer in Comms can be configured using
                    //     com.ibm.ws.sib.jfapchanel.RL_DISPATCH_MAXQUEUESIZE. The excessive buffering
                    //     is still only a symptom of the socket being overloaded with work, the thing
                    //     sending all the messages needs to back off.
                    // We would expect to see this high repetition after a restart of an ME as a whole
                    // 'sendWindow' of messages will flood into the system, this defaults to 1000 and could
                    // be turned down using sib.processor.sendWindow
                    // (510343)

                    // If we're monitoring the number of repeated values that we see, check
                    // to see if we've seen too many...
                    int repeatedPercent = mp.getCustomProperties().getRepeatedValuePercentage();
                    int repeatedInterval = mp.getCustomProperties().getRepeatedValueInterval();
                    if ((repeatedPercent > 0) &&
                        (repeatedInterval > 0))
                    {
                        // Repeated or not, this counts towards the interval that we count
                        // repeats in.
                        valueCounter++;

                        // Calculate how many repeated messages we allow per REPEATED_VALUE_INTERVAL
                        int repeatedValueThreshold =
                                        (repeatedPercent * repeatedInterval) / 100;

                        // If this was a repeated value (a message tick that we've already seen)
                        // add it to the counter and check to see if we've breached the threshold.
                        if (repeatedValue)
                        {
                            repeatedValueCounter++;

                            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                                SibTr.debug(tc, "valueCounter:" + valueCounter +
                                                " repeatedValueThreshold:" + repeatedValueThreshold +
                                                " repeatedValueCounter:" + repeatedValueCounter +
                                                " lastRepeatedValueWarningTime:" + lastRepeatedValueWarningTime);

                            if (repeatedValueCounter >= repeatedValueThreshold)
                            {
                                // Issue a warning message (if it's not too soon after the last message issued,
                                // to prevent flooding the log)
                                long currentTime = System.currentTimeMillis();
                                if ((currentTime - lastRepeatedValueWarningTime) > SIMPConstants.REPEATED_VALUE_WARNING_INTERVAL)
                                {
                                    // Issue a message
                                    int percent = (repeatedValueCounter * 100) / valueCounter;
                                    deliverer.reportRepeatedMessages(remoteEngineUUID.toString(), percent);

                                    lastRepeatedValueWarningTime = currentTime;

                                    // Reset the counters
                                    valueCounter = 0;
                                    repeatedValueCounter = 0;
                                }
                            }
                        }

                        // If we've completed a whole interval, reset the counters.
                        if (valueCounter == repeatedInterval)
                        {
                            // Reset the counters as we've entered a new interval
                            valueCounter = 0;
                            repeatedValueCounter = 0;
                        }
                    }
                }
                else
                {
                    //the stream is blocked - cannot accept new messages
                    if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                        SibTr.debug(
                                    tc,
                                    "The stream is blocked: msg discarded for resend.");
                }
            } // 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 ostsstream lock to update the completedPrefix. If we call
            // the BatchHandler when we hold the oststream lock it could cause
            // a deadlock
            if (msgList != null)
            {
                // Call the Input or Output Handler to deliver the messages
                try
                {
                    messagesInBatch += msgList.size();
                    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, "writeValue", "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.writeValue",
                                                "1:709:1.110",
                                                this);

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

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

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

        } // end sync - release lock on target stream

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

    }
@ Override public void writeValue ( MessageItem msgItem ) { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( this , tc , "writeValue" , new Object [ ] { msgItem } ) ; TickRange tr = null ; List msgList = null ; JsMessage m = msgItem . getMessage ( ) ; long valueTick = m . getGuaranteedValueValueTick ( ) ; timeLastMsgReceived = System . currentTimeMillis ( ) ; synchronized ( this ) { synchronized ( oststream ) { boolean canProcessMessage = false ; try { canProcessMessage = streamCanAcceptNewMessage ( msgItem , valueTick ) ; } catch ( SIException e ) { FFDCFilter . processException ( e , "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream.writeValue" , "1:452:1.110" , this ) ; SibTr . exception ( tc , e ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "writeValue" , e ) ; return ; } if ( canProcessMessage ) { boolean repeatedValue = false ; long completedPrefix = oststream . getCompletedPrefix ( ) ; if ( valueTick > completedPrefix ) { tr = TickRange . newValueTick ( valueTick , msgItem , - 1 ) ; tr . startstamp = m . getGuaranteedValueStartTick ( ) ; tr . endstamp = m . getGuaranteedValueEndTick ( ) ; boolean streamChanged = false ; streamChanged = oststream . writeCombinedRange ( tr ) ; if ( streamChanged ) { if ( tr . valuestamp > valueHorizon ) { valueHorizon = tr . valuestamp ; } boolean gapcreated = false ; tr = oststream . findCompletedRange ( tr ) ; if ( tr . endstamp > unknownHorizon ) { if ( tr . startstamp > ( unknownHorizon + 1 ) ) { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "created gap [" + ( unknownHorizon + 1 ) + ", " + ( tr . startstamp - 1 ) + "]" ) ; gapcreated = true ; handleNewGap ( unknownHorizon + 1 , tr . startstamp - 1 ) ; } unknownHorizon = tr . endstamp ; } if ( ! gapcreated ) { if ( tr . startstamp <= doubtHorizon ) { if ( msgList == null ) msgList = new ArrayList ( ) ; msgList . add ( msgItem ) ; doubtHorizon = tr . endstamp + 1 ; msgList = advanceDoubtHorizon ( msgList ) ; } else { msgList = advanceDoubtHorizon ( msgList ) ; } } if ( lastNackTick >= 0 ) { if ( tr . startstamp <= lastNackTick && tr . endstamp >= lastNackTick ) { getControlAdapter ( ) . getHealthState ( ) . updateHealth ( HealthStateListener . MSG_LOST_ERROR_STATE , HealthState . GREEN ) ; lastNackTick = - 1 ; } } } else { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "Repeated V tick value - not yet acked " + valueTick + " : " + completedPrefix ) ; repeatedValue = true ; } } else { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "Repeated V tick value - already acked " + valueTick + " : " + completedPrefix ) ; repeatedValue = true ; } int repeatedPercent = mp . getCustomProperties ( ) . getRepeatedValuePercentage ( ) ; int repeatedInterval = mp . getCustomProperties ( ) . getRepeatedValueInterval ( ) ; if ( ( repeatedPercent > 0 ) && ( repeatedInterval > 0 ) ) { valueCounter ++ ; int repeatedValueThreshold = ( repeatedPercent * repeatedInterval ) / 100 ; if ( repeatedValue ) { repeatedValueCounter ++ ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "valueCounter:" + valueCounter + " repeatedValueThreshold:" + repeatedValueThreshold + " repeatedValueCounter:" + repeatedValueCounter + " lastRepeatedValueWarningTime:" + lastRepeatedValueWarningTime ) ; if ( repeatedValueCounter >= repeatedValueThreshold ) { long currentTime = System . currentTimeMillis ( ) ; if ( ( currentTime - lastRepeatedValueWarningTime ) > SIMPConstants . REPEATED_VALUE_WARNING_INTERVAL ) { int percent = ( repeatedValueCounter * 100 ) / valueCounter ; deliverer . reportRepeatedMessages ( remoteEngineUUID . toString ( ) , percent ) ; lastRepeatedValueWarningTime = currentTime ; valueCounter = 0 ; repeatedValueCounter = 0 ; } } } if ( valueCounter == repeatedInterval ) { valueCounter = 0 ; repeatedValueCounter = 0 ; } } } else { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "The stream is blocked: msg discarded for resend." ) ; } } if ( msgList != null ) { try { messagesInBatch += msgList . size ( ) ; deliverer . deliverOrderedMessages ( msgList , this , priority , reliability ) ; } catch ( SINotPossibleInCurrentConfigurationException e ) { SibTr . exception ( tc , e ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "writeValue" , "GDException" ) ; } catch ( SIException e ) { FFDCFilter . processException ( e , "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream.writeValue" , "1:709:1.110" , this ) ; SibTr . exception ( tc , e ) ; SibTr . error ( tc , "INTERNAL_MESSAGING_ERROR_CWSIP0002" , new Object [ ] { "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream" , "1:716:1.110" , e } ) ; if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "writeValue" , "GDException" ) ; throw new SIErrorException ( nls . getFormattedMessage ( "INTERNAL_MESSAGING_ERROR_CWSIP0002" , new Object [ ] { "com.ibm.ws.sib.processor.gd.GuaranteedTargetStream" , "1:726:1.110" , e } , null ) , e ) ; } } } if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "writeValue" ) ; }
@exception GDException thrown from either initialiseStream or checkForWindowAdvanceGaps