psUpdateMessage = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"));
psUpdateMessage = conn.prepareStatement(getSQLStatement("DEC_CHANNELCOUNT"));
   protected void handleBeforeCommit2PC(List refsToRemove, Transaction tx)      throws Exception   {                Connection conn = null;      PreparedStatement psUpdateMessage = null;      PreparedStatement psDeleteMessage = null;      TransactionWrapper wrap = new TransactionWrapper();            List refs = new ArrayList(refsToRemove.size());      Iterator iter = refsToRemove.iterator();      while (iter.hasNext())      {         ChannelRefPair pair = (ChannelRefPair)iter.next();         refs.add(pair.ref);      }                  orderReferences(refs);                  List removesToReverse = new ArrayList();            try      {         //get locks on all the refs         this.getLocks(refs);                  conn = ds.getConnection();                           //2PC commit                  //First we commit any refs in state "+" to "C" and delete any         //refs in state "-", then we         //remove any messages due to refs we just removed         //if they're not referenced elsewhere                  commitPreparedTransaction(tx, conn);                  boolean batch = usingBatchUpdates && refsToRemove.size() > 0;         boolean messageDeletionsInBatch = false;         boolean messageUpdatesInBatch = false;               if (batch)         {            psDeleteMessage = conn.prepareStatement(getSQLStatement("DELETE_MESSAGE"));            psUpdateMessage = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"));         }                           iter = refsToRemove.iterator();         while (iter.hasNext())         {            ChannelRefPair pair = (ChannelRefPair) iter.next();                        MessageReference ref = pair.ref;                        if (!batch)            {               psDeleteMessage = conn.prepareStatement(getSQLStatement("DELETE_MESSAGE"));               psUpdateMessage = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"));            }                        Message m = ref.getMessage();                                               //We may need to remove the message itself                        m.decPersistentChannelCount();            removesToReverse.add(ref);                        boolean removed;            if (m.getPersistentChannelCount() == 0)            {               //We can remove the message               removeMessage(m, psDeleteMessage);                              removed = true;            }            else            {               //Decrement channel count               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)         {            if (messageDeletionsInBatch)            {               int[] rows = psDeleteMessage.executeBatch();                              if (trace)               {                  logBatchUpdate(getSQLStatement("DELETE_MESSAGE"), rows, "deleted");               }                              psDeleteMessage.close();               psDeleteMessage = null;            }            if (messageUpdatesInBatch)            {               int[] rows = psUpdateMessage.executeBatch();                              if (trace)               {                  logBatchUpdate(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"), rows, "updated");               }                              psUpdateMessage.close();               psUpdateMessage = null;            }         }               }      catch (Exception e)      {         wrap.exceptionOccurred();         throw e;      }      finally      {         if (psDeleteMessage != null)         {            try            {               psDeleteMessage.close();            }            catch (Throwable t)            {            }         }         if (psUpdateMessage != null)         {            try            {               psUpdateMessage.close();            }            catch (Throwable t)            {            }         }         if (conn != null)         {            try            {               conn.close();            }            catch (Throwable e)            {            }         }         try         {            wrap.end();         }         finally         {            if (wrap.isFailed())            {               //reverse any decs               this.incPersistentCounts(removesToReverse);            }            //release the locks            this.releaseLocks(refs);         }      }   }