psInsertMessage.addBatch(); messageInsertsInBatch = true;
log.trace("Updated " + rows + " rows");
   protected void handleBeforeCommit1PC(List refsToAdd, List refsToRemove, Transaction tx)      throws Exception   {      //For one phase we simply add rows corresponding to the refs      //and remove rows corresponding to the deliveries in one jdbc tx      //We also need to store or remove messages as necessary, depending      //on whether they've already been stored or still referenced by other      //channels               Connection conn = null;      PreparedStatement psReference = null;      PreparedStatement psInsertMessage = null;      PreparedStatement psUpdateMessage = null;      PreparedStatement psDeleteMessage = null;      TransactionWrapper wrap = new TransactionWrapper();            try      {         conn = ds.getConnection();                  //First the adds                  Iterator iter = refsToAdd.iterator();                  boolean batch = usingBatchUpdates && refsToAdd.size() > 0;                  boolean messageInsertsInBatch = false;         boolean messageUpdatesInBatch = false;                  if (batch)         {            psReference = conn.prepareStatement(insertMessageRef);            psInsertMessage = conn.prepareStatement(insertMessage);            psUpdateMessage = conn.prepareStatement(updateMessageChannelCount);         }                  while (iter.hasNext())         {            ChannelRefPair pair = (ChannelRefPair) iter.next();                        MessageReference ref = pair.ref;                                                            if (!batch)            {               psReference = conn.prepareStatement(insertMessageRef);            }                        //Now store the reference            addReference(pair.channelId, ref, psReference, true);                                    if (batch)            {               psReference.addBatch();            }            else            {               int rows = psReference.executeUpdate();                              if (trace)               {                  log.trace("Inserted " + rows + " rows");               }                              psReference.close();               psReference = null;            }                        Message m = ref.getMessage();                                if (!batch)            {               psInsertMessage = conn.prepareStatement(insertMessage);               psUpdateMessage = conn.prepareStatement(updateMessageChannelCount);            }                         synchronized (m)            {                                    m.incPersistentChannelCount();                                                                           boolean added;               if (m.getPersistentChannelCount() == 1)               {                  //First time so add message                  storeMessage(m, psInsertMessage);                                    added = true;               }               else               {                  //Update message channel count                  updateMessageChannelCount(m, psUpdateMessage);                                    added = false;               }                              if (batch)               {                  if (added)                  {                     psInsertMessage.addBatch();                     messageInsertsInBatch = true;                  }                  else                  {                     psUpdateMessage.addBatch();                     messageUpdatesInBatch = true;                  }               }               else               {                  if (added)                  {                     int rows = psInsertMessage.executeUpdate();                                          if (trace)                     {                        log.trace("Inserted " + rows + " rows");                     }                  }                  else                  {                     int rows = psUpdateMessage.executeUpdate();                                          if (trace)                     {                        log.trace("Updated " + rows + " rows");                     }                  }                  psInsertMessage.close();                  psInsertMessage = null;                  psUpdateMessage.close();                  psUpdateMessage = null;               }            }         }                  if (batch)         {            int[] rowsReference = psReference.executeBatch();                        if (trace)            {               logBatchUpdate(insertMessageRef, rowsReference, "inserted");            }                        if (messageInsertsInBatch)            {               int[] rowsMessage = psInsertMessage.executeBatch();                              if (trace)               {                  logBatchUpdate(insertMessage, rowsMessage, "inserted");               }            }            if (messageUpdatesInBatch)            {               int[] rowsMessage = psUpdateMessage.executeBatch();                              if (trace)               {                  logBatchUpdate(updateMessageChannelCount, rowsMessage, "updated");               }            }                        psReference.close();            psReference = null;            psInsertMessage.close();            psInsertMessage = null;            psUpdateMessage.close();            psUpdateMessage = null;         }                  //Now the removes                  iter = refsToRemove.iterator();                  batch = usingBatchUpdates && refsToRemove.size() > 0;         boolean messageDeletionsInBatch = false;         messageUpdatesInBatch = false;         psReference = null;         psUpdateMessage = null;         psDeleteMessage = null;                  if (batch)         {            psReference = conn.prepareStatement(deleteMessageRef);            psDeleteMessage = conn.prepareStatement(deleteMessage);            psUpdateMessage = conn.prepareStatement(updateMessageChannelCount);         }                  while (iter.hasNext())         {            ChannelRefPair pair = (ChannelRefPair) iter.next();                        if (!batch)            {               psReference = conn.prepareStatement(deleteMessageRef);            }                        removeReference(pair.channelId, pair.ref, psReference);                        if (batch)            {               psReference.addBatch();            }            else            {               int rows = psReference.executeUpdate();                              if (trace)               {                  log.trace("Deleted " + rows + " rows");               }                              psReference.close();               psReference = null;            }                        if (!batch)            {               psDeleteMessage = conn.prepareStatement(deleteMessage);               psUpdateMessage = conn.prepareStatement(updateMessageChannelCount);            }                        Message m = pair.ref.getMessage();                        synchronized (m)            {                                 m.decPersistentChannelCount();                                              boolean removed;                              if (m.getPersistentChannelCount() == 0)               {                  //No more refs - message can be deleted                  removeMessage(m, psDeleteMessage);                                       removed = true;               }               else               {                  //Update channel count for message                  updateMessageChannelCount(m, psUpdateMessage);                                    removed = false;               }                              if (batch)               {                  if (removed)                  {                     psDeleteMessage.addBatch();                     messageDeletionsInBatch = true;                  }                  else                  {                     psUpdateMessage.addBatch();                     messageUpdatesInBatch = true;                                          }               }               else               {                  if (removed)                  {                     int rows = psDeleteMessage.executeUpdate();                                          if (trace)                     {                        log.trace("Deleted " + rows + " rows");                     }                  }                  else                  {                     int rows = psUpdateMessage.executeUpdate();                                          if (trace)                     {                        log.trace("Updated " + rows + " rows");                     }                  }                  psDeleteMessage.close();                  psDeleteMessage = null;                  psUpdateMessage.close();                  psUpdateMessage = null;               }            }         }                  if (batch)         {            int[] rowsReference = psReference.executeBatch();                        if (trace)            {               logBatchUpdate(deleteMessageRef, rowsReference, "deleted");            }                        if (messageDeletionsInBatch)            {               int[] rowsMessage = psDeleteMessage.executeBatch();                              if (trace)               {                  logBatchUpdate(deleteMessage, rowsMessage, "deleted");               }            }            if (messageUpdatesInBatch)            {               int[] rowsMessage = psUpdateMessage.executeBatch();                              if (trace)               {                  logBatchUpdate(updateMessageChannelCount, rowsMessage, "updated");               }            }                        psReference.close();            psReference = null;            psDeleteMessage.close();            psDeleteMessage = null;            psUpdateMessage.close();            psUpdateMessage = null;         }               }      catch (Exception e)      {         wrap.exceptionOccurred();         throw e;      }      finally      {         if (psReference != null)         {            try            {               psReference.close();            }            catch (Throwable t)            {            }         }         if (psInsertMessage != null)         {            try            {               psInsertMessage.close();            }            catch (Throwable t)            {            }         }         if (psUpdateMessage != null)         {            try            {               psUpdateMessage.close();            }            catch (Throwable t)            {            }         }         if (psDeleteMessage != null)         {            try            {               psDeleteMessage.close();            }            catch (Throwable t)            {            }         }         if (conn != null)         {            try            {               conn.close();            }            catch (Throwable e)            {            }         }         wrap.end();      }   }