log.info("Removing ref " + pair.ref + " in tx");
   protected void handleBeforeCommit1PC(List refsToAdd, List refsToRemove, Transaction tx)      throws Exception   {      log.info("Before commit 1pc: I have " + refsToAdd.size() + " refs to add and " + refsToRemove.size() + " refs to remove");            //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 psIncMessage = null;      PreparedStatement psDecMessage = null;      PreparedStatement psDeleteMessage = null;      TransactionWrapper wrap = new TransactionWrapper();            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"));            psIncMessage = conn.prepareStatement(getSQLStatement("INC_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            log.info("Adding ref: " + ref + " in channel: " + pair.channelId);            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"));               psIncMessage = conn.prepareStatement(getSQLStatement("INC_CHANNELCOUNT"));            }                                     boolean added;            if (!m.isPersisted())            {               log.info("The message " + m.getMessageID() + " hasn't been persisted before so i'm gonna persist it");               //First time so add message               storeMessage(m, psInsertMessage);                              added = true;                              m.setPersisted(true);            }            else            {               log.info("The message " + m.getMessageID() + " has been peristed before so just updating it's channel count");               //Update message channel count               incrementChannelCount(m, psIncMessage);                              added = false;            }                        if (batch)            {               if (added)               {                  psInsertMessage.addBatch();                                    messageInsertsInBatch = true;               }               else               {                   psIncMessage.addBatch();                                    messageUpdatesInBatch = true;               }            }            else            {               if (added)               {                  int rows = psInsertMessage.executeUpdate();                                    if (trace) { log.trace("Inserted " + rows + " rows"); }               }               else               {                  int rows = psIncMessage.executeUpdate();                                    if (trace) { log.trace("Updated " + rows + " rows"); }               }               psInsertMessage.close();               psInsertMessage = null;               psIncMessage.close();               psIncMessage = 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 = psIncMessage.executeBatch();                              if (trace) { logBatchUpdate(getSQLStatement("INC_CHANNELCOUNT"), rowsMessage, "updated"); }            }                        psReference.close();            psReference = null;            psInsertMessage.close();            psInsertMessage = null;            psIncMessage.close();            psIncMessage = null;         }                  //Now the removes                  iter = refsToRemove.iterator();                  batch = usingBatchUpdates && refsToRemove.size() > 0;         psReference = null;         psDeleteMessage = null;                  if (batch)         {            psReference = conn.prepareStatement(getSQLStatement("DELETE_MESSAGE_REF"));            psDeleteMessage = conn.prepareStatement(getSQLStatement("DELETE_MESSAGE"));            psDecMessage = conn.prepareStatement(getSQLStatement("DEC_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"));               psDecMessage = conn.prepareStatement(getSQLStatement("DEC_CHANNELCOUNT"));            }                        Message m = pair.ref.getMessage();                                            //Update the channel count                        decrementChannelCount(m, psDecMessage);                        //Delete the message (if necessary)                        removeMessage(m, psDeleteMessage);                                   if (batch)            {               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"); }               psDeleteMessage.close();               psDeleteMessage = null;               psDecMessage.close();               psDecMessage = null;            }                     }                  if (batch)         {            int[] rows = psReference.executeBatch();                        if (trace) { logBatchUpdate(getSQLStatement("DELETE_MESSAGE_REF"), rows, "deleted"); }                        rows = psDecMessage.executeBatch();                        if (trace) { logBatchUpdate(getSQLStatement("DEC_CHANNELCOUNT"), rows, "updated"); }            rows = psDeleteMessage.executeBatch();                        if (trace) { logBatchUpdate(getSQLStatement("DELETE_MESSAGE"), rows, "deleted"); }            psReference.close();            psReference = null;            psDeleteMessage.close();            psDeleteMessage = null;            psDecMessage.close();            psDecMessage = 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 (psIncMessage != null)         {            try            {               psIncMessage.close();            }            catch (Throwable t)            {            }         }         if (psDecMessage != null)         {            try            {               psDecMessage.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         {              //Release the locks            this.releaseLocks(allRefs);         }      }   }