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"); } }
psDecMessage.addBatch();  psDeleteMessage.addBatch(); } else { int rows = psDecMessage.executeUpdate();  if (trace) { log.trace("Updated " + rows + " rows"); }  rows = psDeleteMessage.executeUpdate();  if (trace) { log.trace("Deleted " + rows + " rows"); }
   protected void handleBeforeCommit1PC(List refsToAdd, List refsToRemove, Transaction tx)      throws Exception   {      //TODO - A slight optimisation - it's possible we have refs referring to the same message      //so we will end up acquiring the lock more than once which is unnecessary      //If find unique set of messages can avoid this      List allRefs = new ArrayList(refsToAdd.size() + refsToRemove.size());      Iterator iter = refsToAdd.iterator();      while (iter.hasNext())      {         ChannelRefPair pair = (ChannelRefPair)iter.next();         allRefs.add(pair.ref);      }      iter = refsToRemove.iterator();      while (iter.hasNext())      {         ChannelRefPair pair = (ChannelRefPair)iter.next();         allRefs.add(pair.ref);      }                  orderReferences(allRefs);            //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();            List addsToReverse = new ArrayList();      List removesToReverse = new ArrayList();            try      {         conn = ds.getConnection();                  //obtain locks on all messages         getLocks(allRefs);                  //First the adds                  iter = refsToAdd.iterator();                  boolean batch = usingBatchUpdates && refsToAdd.size() > 0;                  boolean messageInsertsInBatch = false;         boolean messageUpdatesInBatch = false;                  if (batch)         {            psReference = conn.prepareStatement(getSQLStatement("INSERT_MESSAGE_REF"));            psInsertMessage = conn.prepareStatement(getSQLStatement("INSERT_MESSAGE"));            psUpdateMessage = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"));         }                  while (iter.hasNext())         {            ChannelRefPair pair = (ChannelRefPair) iter.next();                        MessageReference ref = pair.ref;                                                            if (!batch)            {               psReference = conn.prepareStatement(getSQLStatement("INSERT_MESSAGE_REF"));            }                        //Now store the reference            addReference(pair.channelId, ref, psReference, false);                          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(getSQLStatement("INSERT_MESSAGE"));               psUpdateMessage = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"));            }                                     m.incPersistentChannelCount();            addsToReverse.add(ref);                        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(getSQLStatement("INSERT_MESSAGE_REF"), rowsReference, "inserted");            }                        if (messageInsertsInBatch)            {               int[] rowsMessage = psInsertMessage.executeBatch();                              if (trace)               {                  logBatchUpdate(getSQLStatement("INSERT_MESSAGE"), rowsMessage, "inserted");               }            }            if (messageUpdatesInBatch)            {               int[] rowsMessage = psUpdateMessage.executeBatch();                              if (trace)               {                  logBatchUpdate(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"), 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(getSQLStatement("DELETE_MESSAGE_REF"));            psDeleteMessage = conn.prepareStatement(getSQLStatement("DELETE_MESSAGE"));            psUpdateMessage = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"));         }                  while (iter.hasNext())         {            ChannelRefPair pair = (ChannelRefPair) iter.next();                        if (!batch)            {               psReference = conn.prepareStatement(getSQLStatement("DELETE_MESSAGE_REF"));            }                        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(getSQLStatement("DELETE_MESSAGE"));               psUpdateMessage = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"));            }                        Message m = pair.ref.getMessage();                                            m.decPersistentChannelCount();                        removesToReverse.add(pair.ref);                        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(getSQLStatement("DELETE_MESSAGE_REF"), rowsReference, "deleted");            }                        if (messageDeletionsInBatch)            {               int[] rowsMessage = psDeleteMessage.executeBatch();                              if (trace)               {                  logBatchUpdate(getSQLStatement("DELETE_MESSAGE"), rowsMessage, "deleted");               }            }            if (messageUpdatesInBatch)            {               int[] rowsMessage = psUpdateMessage.executeBatch();                              if (trace)               {                  logBatchUpdate(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"), 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)            {            }         }         try         {            wrap.end();                                 }         finally         {            if (wrap.isFailed())            {               //Reverse any incs and decs we made               this.decPersistentCounts(addsToReverse);               this.incPersistentCounts(removesToReverse);            }                        //Release the locks            this.releaseLocks(allRefs);         }      }   }