if (debug()) log.debug ("MessageResender: Resending " +MessageUtils.toString(msg));
rsndNum++;  if (debug()) log.debug ("MessageResender: Resending #"+rsndNum+" " +MessageUtils.toString(msg));
  public void run()   {    int len;    while (true)     {      //  Wait until we have some new messages or acks or we have timed out to       //  re-examine old messages.      synchronized (queue)       {        while (true)        {          //  If we have new data (new msgs and/or acks) and a non-empty msg queue,           //  don't wait.  New data can arrive during waiting or during queue processing.          if (haveNewData)          {            haveNewData = false;            if (queue.size() > 0) break;          }          //  Check how long to wait before we need to satisfy a resend deadline          long waitTime = 0;  // 0 = wait till notify (or interrupt)          if (queue.size() > 0)          {            if (minResendDeadline < Long.MAX_VALUE)            {              waitTime = minResendDeadline - now();              if (waitTime <= 0) break;            }          }          //  Wait for a specified time or until notify          try { queue.wait (waitTime); } catch (Exception e) {}        }        messages = (AttributedMessage[]) queue.toArray (messages);  // try array reuse        len = queue.size();        if (len > 1) Arrays.sort (messages, 0, len, deadlineSort);      }      //  Next we try to match the messages with the acks we've collected so far.      //  Possible many-to-many relationship between the messages and the acks.      if (len > 0 && debug())       {        log.debug ("MessageResender: reviewing queue (" +len+ " msg" +(len==1? ")" : "s)"));      }      for (int i=0; i<len; i++)      {        AttributedMessage msg = messages[i];                //  Avoid already successful sends (when a message is acked it is declared        //  and recorded as a successful send).        if (MessageAckingAspect.wasSuccessfulSend (msg))        {          if (debug()) log.debug ("MessageResender: Dropping already successful " +MessageUtils.toString(msg));          remove (msg);          messages[i] = null;          continue;        }        //  Search for match of message with current received ack data        int msgNum = MessageUtils.getMessageNumber (msg);        AckList ackList = AckList.findFirst (MessageAckingAspect.getReceivedAcks(msg), msgNum);        if (ackList != null)        {          //  We have an ack!!          if (debug()) log.debug ("MessageResender: Got ack for " +MessageUtils.toString(msg));          Ack ack = MessageUtils.getAck (msg);          String toNode = MessageUtils.getToAgentNode (msg);          MessageAckingAspect.addSuccessfulSend (msg);                          MessageAckingAspect.updateMessageHistory (ack, ackList);          MessageAckingAspect.removeAcksToSend (toNode, ack.getSpecificAcks()); // acks now acked          remove (msg);          messages[i] = null;        }      }            //  For any remaining un-acked messages we need to decide if      //  it is time to try to resend the message.      minResendDeadline = Long.MAX_VALUE;      for (int i=0; i<len; i++) if (messages[i] != null)      {        AttributedMessage msg = messages[i];        Ack ack = MessageUtils.getAck (msg);        //  See if time to resend message        int timeout = ack.getResendTimeout() + ack.getResendDelay();        long resendDeadline = ack.getSendTime() + timeout;        long timeLeft = resendDeadline - now();        if (debug()) log.debug ("MessageResender: Msg " +MessageUtils.getMessageNumber(msg)+          ": timeout=" +timeout+ "  timeLeft=" +timeLeft);        if (timeLeft <= 0)        {          //  Time to resend the message.  SendMessage re-inserts the message at the          //  front of the whole send pipeline: it'll come through the AckFrontend          //  again on its way out to an outgoing link.  The link selection policy has           //  been modified to try a different link with each message resend, as it          //  wants to try every possbile transport link in its efforts to get a           //  message through.          if (debug()) log.debug ("MessageResender: Resending " +MessageUtils.toString(msg));          ack.setSendTime (now());          SendMessage.sendMsg (msg);          //  Calculate the new message resend deadline.  Note that we do not know at          //  this point what new link will be chosen for the message we just resent,          //  so all we can do at this point is use the timeout we currently have.          //  At the end of link selection the ack is updated with the new link timing          //  information and so then the timeout will be correct, and it is expected          //  that this update will occur before a resend based on the old link info.          //  Note that with each resend we add some delay to the message resend timeout           //  if the message send count is getting high (num links + 1), so that will          //  also affect the time available for the link selection to occur.  Note that           //  the delay is limited by the max delay time set in the ack, with a current          //  default of 1 minute.          int highSendCount = ack.getNumberOfLinkChoices() + 1;          if (ack.getSendCount() > highSendCount) ack.addResendDelay (500);          resendDeadline = ack.getSendTime() + ack.getResendTimeout() + ack.getResendDelay();        }        if (resendDeadline < minResendDeadline) minResendDeadline = resendDeadline;      }      Arrays.fill (messages, null);  // release references    }  }