public void handleNotFlushedMessage(ControlNotFlushed cMsg)

  throws SIIncorrectCallException, SIErrorException, SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this, tc, "handleNotFlushedMessage", new Object[] { cMsg });

    SIBUuid12 streamID = cMsg.getGuaranteedStreamUUID();
    //Synchronize to resolve racing messages.
    synchronized (flushMap)
    {
      FlushQueryRecord entry = flushMap.get(streamID);
      // make sure the request ID matches.
      long reqId = cMsg.getRequestID();
      //if there is no entry or the entry's request ID does not match then
      //it must have been a stale request and we ignore it.
      if ((entry != null) && (entry.requestID == reqId))
      {
        // ASSERT.isTrue(entry.cache != null)

        // The assertion basically means that we should only receive
        // NOTFLUSHED if the original query was AREYOUFLUSHED.  The
        // only legal reply to REQUESTFLUSH is, eventually, FLUSHED.

        // Our original query was for "are you flushed" and we now
        // have the reply, so:
        // 1) remove the entry from the flush map and disable the alarm
        // 2) handle the new prefixes and replay any cached data messages.
        flushMap.remove(streamID);
        entry.resend.cancel();

        StreamSet streamSet = getStreamSetForMessage(cMsg);

        if(streamSet == null)
        {
          //check that we don't already have a streamSet for this source, because there
          // should only be the one at any one time (except for links, see below...)
          SIBUuid8 sourceMEUuid = cMsg.getGuaranteedSourceMessagingEngineUUID();
          if(sourceMap.containsValue(sourceMEUuid))
          {
            //if we do, try to remove it
            Iterator<SIBUuid12> itr = sourceMap.keySet().iterator();
            while(itr.hasNext())
            {
              SIBUuid12 sid = itr.next();
              SIBUuid8 uuid = sourceMap.get(sid);
              if(sourceMEUuid.equals(uuid))
              {
                // 538529
                // If this is a link it is possible to have multiple streams of messages
                // coming in from the source ME concurrently. We have one for all point-to-point
                // messages and one per topicspace that is sending messages to us.
                // Ideally we need to check with the source before we flush any of our existing
                // streams, as they may still be valid. We would do this by asking the sender if
                // they are flushed (commented out code below). Unfortunately we can't do that because
                // the sender (the other end of the link) may not be the source of the messages
                // so it's not guaranteed to actually know about this stream even if it still
                // exists (for example the ME's just been restarted and lost the non-persistent
                // state for this stream). So, instead we simply keep all streams.
                // TODO This isn't ideal so maybe we can fix it in the future 
                if(destination.isLink())
                {
                  if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
                    SibTr.debug(tc, "WARNING: Existing Link stream, " + sid + ", not flushed");
                  
//                  StreamSet existingStreamSet = (StreamSet)streamSets.get(sid);
//
//                  // Create and store the new request record
//                  long reqID = messageProcessor.nextTick();
//                  FlushQueryRecord newEntry = new FlushQueryRecord(sourceMEUuid,
//                                               existingStreamSet.getDestUuid(),
//                                               existingStreamSet.getBusUuid(),
//                                               null,
//                                               reqID);
//                  flushMap.put(sid, newEntry);
//
//                  // Create and send the query
//                  upControl.sendAreYouFlushedMessage(sourceMEUuid,
//                                                     existingStreamSet.getDestUuid(),
//                                                     existingStreamSet.getBusUuid(),
//                                                     reqID,
//                                                     sid);
//
//                  // Start the alarm.  The context for the alarm is the stream id, which
//                  // we can use to look up the FlushQueryRecord if the alarm expires.
//                  newEntry.resend = am.create(GDConfig.FLUSH_QUERY_INTERVAL, this, sid);
                }
                // If we're not a link we only ever expect a single streamSet to be in use
                // by the source at any one time, so flush any that we already have.
                else
                {
                  flush(sid);
                  //there should only ever be at most one other streamSet found
                  //so stop looking if we found one.
                  break;
                }
              }
            }
          }
          
          // If this is a link, get the actual target destination 
          String linkTarget = SIMPConstants.PTOP_TARGET_STREAM;
          JsDestinationAddress routingDest = cMsg.getRoutingDestination();
          if (destination.isLink() &&  routingDest!= null)
          {
            // get the destination and check if it is a topicspace
            DestinationHandler dest = 
              destination.getDestinationManager().
                getDestinationInternal(routingDest.getDestinationName(),
                                       routingDest.getBusName(), true);
            if (dest!=null && dest.isPubSub())
              linkTarget = routingDest.getDestinationName();
          }           
          
          //create a new streamSet
          SIBUuid8 busId = cMsg.getGuaranteedCrossBusSourceBusUUID();
          if( busId == null)
          {
            busId = messageProcessor.getMessagingEngineBusUuid();
          }
          streamSet = addNewStreamSet(streamID,
                                      sourceMEUuid,
                                      cMsg.getGuaranteedTargetDestinationDefinitionUUID(),
                                      busId,
                                      linkTarget);
          //set all the completed prefixes
          long[] completedPrefixes = cMsg.getCompletedPrefixTicks();
          int[] priority  = cMsg.getCompletedPrefixPriority();
          int[] reliability = cMsg.getCompletedPrefixQOS();
          for(int i=0;i<completedPrefixes.length;i++)
          {
            streamSet.setPersistentData(priority[i],
                                                Reliability.getReliability(reliability[i]),
                                                completedPrefixes[i]);
          }
        }
        // Now replay any cached messages
        for(int i=0; i<entry.cacheIndex; i++)
        {
          handleMessage(entry.cache[i]);
        }
      }
    }
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "handleNotFlushedMessage");
  }
public void handleNotFlushedMessage ( ControlNotFlushed cMsg ) throws SIIncorrectCallException , SIErrorException , SIResourceException { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . entry ( this , tc , "handleNotFlushedMessage" , new Object [ ] { cMsg } ) ; SIBUuid12 streamID = cMsg . getGuaranteedStreamUUID ( ) ; synchronized ( flushMap ) { FlushQueryRecord entry = flushMap . get ( streamID ) ; long reqId = cMsg . getRequestID ( ) ; if ( ( entry != null ) && ( entry . requestID == reqId ) ) { flushMap . remove ( streamID ) ; entry . resend . cancel ( ) ; StreamSet streamSet = getStreamSetForMessage ( cMsg ) ; if ( streamSet == null ) { SIBUuid8 sourceMEUuid = cMsg . getGuaranteedSourceMessagingEngineUUID ( ) ; if ( sourceMap . containsValue ( sourceMEUuid ) ) { Iterator < SIBUuid12 > itr = sourceMap . keySet ( ) . iterator ( ) ; while ( itr . hasNext ( ) ) { SIBUuid12 sid = itr . next ( ) ; SIBUuid8 uuid = sourceMap . get ( sid ) ; if ( sourceMEUuid . equals ( uuid ) ) { if ( destination . isLink ( ) ) { if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isDebugEnabled ( ) ) SibTr . debug ( tc , "WARNING: Existing Link stream, " + sid + ", not flushed" ) ; } else { flush ( sid ) ; break ; } } } } String linkTarget = SIMPConstants . PTOP_TARGET_STREAM ; JsDestinationAddress routingDest = cMsg . getRoutingDestination ( ) ; if ( destination . isLink ( ) && routingDest != null ) { DestinationHandler dest = destination . getDestinationManager ( ) . getDestinationInternal ( routingDest . getDestinationName ( ) , routingDest . getBusName ( ) , true ) ; if ( dest != null && dest . isPubSub ( ) ) linkTarget = routingDest . getDestinationName ( ) ; } SIBUuid8 busId = cMsg . getGuaranteedCrossBusSourceBusUUID ( ) ; if ( busId == null ) { busId = messageProcessor . getMessagingEngineBusUuid ( ) ; } streamSet = addNewStreamSet ( streamID , sourceMEUuid , cMsg . getGuaranteedTargetDestinationDefinitionUUID ( ) , busId , linkTarget ) ; long [ ] completedPrefixes = cMsg . getCompletedPrefixTicks ( ) ; int [ ] priority = cMsg . getCompletedPrefixPriority ( ) ; int [ ] reliability = cMsg . getCompletedPrefixQOS ( ) ; for ( int i = 0 ; i < completedPrefixes . length ; i ++ ) { streamSet . setPersistentData ( priority [ i ] , Reliability . getReliability ( reliability [ i ] ) , completedPrefixes [ i ] ) ; } } for ( int i = 0 ; i < entry . cacheIndex ; i ++ ) { handleMessage ( entry . cache [ i ] ) ; } } } if ( TraceComponent . isAnyTracingEnabled ( ) && tc . isEntryEnabled ( ) ) SibTr . exit ( tc , "handleNotFlushedMessage" ) ; }
Handle a ControlNotFlushed message. Flush and streams from the same source
and create a new set of streams for the new (not flushed) stream ID.

@param cMsg
@throws SIResourceException