psUpdateMessage = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"));
psUpdateMessage = conn.prepareStatement(getSQLStatement("INC_CHANNELCOUNT"));
   protected void handleBeforePrepare(List refsToAdd, List refsToRemove, Transaction tx) throws Exception   {      //We only need to lock on the adds      List refs = new ArrayList(refsToAdd.size());            Iterator iter = refsToAdd.iterator();      while (iter.hasNext())      {         ChannelRefPair pair = (ChannelRefPair)iter.next();                  refs.add(pair.ref);      }            orderReferences(refs);            List addsToReverse = new ArrayList();            //We insert a tx record and      //a row for each ref with +      //and update the row for each delivery with "-"            PreparedStatement psReference = null;      PreparedStatement psInsertMessage = null;      PreparedStatement psUpdateMessage = null;      Connection conn = null;      TransactionWrapper wrap = new TransactionWrapper();            try      {         //get the locks         getLocks(refs);                  conn = ds.getConnection();                  //Insert the tx record         if (!refsToAdd.isEmpty() || !refsToRemove.isEmpty())         {            addTXRecord(conn, tx);         }                  iter = refsToAdd.iterator();                  boolean batch = usingBatchUpdates && refsToAdd.size() > 1;         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();                        if (!batch)            {               psReference = conn.prepareStatement(getSQLStatement("INSERT_MESSAGE_REF"));            }                        prepareToAddReference(pair.channelId, pair.ref, tx, psReference);                        if (batch)            {               psReference.addBatch();            }            else            {               int rows = psReference.executeUpdate();                              if (trace)               {                  log.trace("Inserted " + rows + " rows");               }                              psReference.close();               psReference = null;            }                        if (!batch)            {               psInsertMessage = conn.prepareStatement(getSQLStatement("INSERT_MESSAGE"));               psUpdateMessage = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_CHANNELCOUNT"));            }                        Message m = pair.ref.getMessage();                        m.incPersistentChannelCount();            addsToReverse.add(pair.ref);                                    boolean added;            if (m.getPersistentChannelCount() == 1)            {               //First time so persist the 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() > 1;         if (batch)         {            psReference = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_REF"));         }                  while (iter.hasNext())         {            ChannelRefPair pair = (ChannelRefPair) iter.next();                        if (!batch)            {               psReference = conn.prepareStatement(getSQLStatement("UPDATE_MESSAGE_REF"));            }                        prepareToRemoveReference(pair.channelId, pair.ref, tx, psReference);                        if (batch)            {               psReference.addBatch();            }            else            {               int rows = psReference.executeUpdate();                              if (trace)               {                  log.trace("updated " + rows + " rows");               }                              psReference.close();               psReference = null;            }         }                  if (batch)         {            int[] rows = psReference.executeBatch();                        if (trace)            {               logBatchUpdate(getSQLStatement("UPDATE_MESSAGE_REF"), rows, "updated");            }                        psReference.close();            psReference = 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 (conn != null)         {            try            {               conn.close();            }            catch (Throwable e)            {            }         }         try         {            wrap.end();                     }         finally         {            if (wrap.isFailed())            {               //reverse any incs               this.decPersistentCounts(addsToReverse);            }                        //release the locks                        this.releaseLocks(refs);         }      }   }