public AMQTransactionStore(AMQPersistenceAdapter adapter) 
{
        this.peristenceAdapter = adapter;
}

 public  void acknowledge(AMQTopicMessageStore store,
    JournalTopicAck ack,
    Location location) 
{
        AMQTx tx = getTx(ack.getTransactionId(), location);
        tx.add(store, ack);
}

  void addMessage(AMQMessageStore store,
    Message message,
    Location location) throws IOException 
{
        AMQTx tx = getTx(message.getTransactionId(), location);
        tx.add(store, message, location);
}

 public Location checkpoint() throws IOException 
{
        // Nothing really to checkpoint.. since, we don't
        // checkpoint tx operations in to long term store until they are
        // committed.
        // But we keep track of the first location of an operation
        // that was associated with an active tx. The journal can not
        // roll over active tx records.
        Location minimumLocationInUse = null;
        synchronized (inflightTransactions) {
            for (Iterator< AMQTx > iter = inflightTransactions.values().iterator(); iter.hasNext();) {
                AMQTx tx = iter.next();
                Location location = tx.getLocation();
                if (minimumLocationInUse == null || location.compareTo(minimumLocationInUse) <  0) {
                    minimumLocationInUse = location;
                }
            }
        }
        synchronized (preparedTransactions) {
            for (Iterator< AMQTx > iter = preparedTransactions.values().iterator(); iter.hasNext();) {
                AMQTx tx = iter.next();
                Location location = tx.getLocation();
                if (minimumLocationInUse == null || location.compareTo(minimumLocationInUse) <  0) {
                    minimumLocationInUse = location;
                }
            }
            return minimumLocationInUse;
        }
}

 public  void commit(TransactionId txid,
    boolean wasPrepared) throws IOException 
{
        AMQTx tx;
        if (wasPrepared) {
            synchronized (preparedTransactions) {
                tx = preparedTransactions.remove(txid);
            }
        } else {
            synchronized (inflightTransactions) {
                tx = inflightTransactions.remove(txid);
            }
        }
        if (tx == null) {
            return;
        }
        if (txid.isXATransaction()) {
            peristenceAdapter.writeCommand(new JournalTransaction(JournalTransaction.XA_COMMIT, txid, wasPrepared), true,true);
        } else {
            peristenceAdapter.writeCommand(new JournalTransaction(JournalTransaction.LOCAL_COMMIT, txid, wasPrepared), true,true);
        }
}

 public Map<TransactionId, AMQTx> getPreparedTransactions() 
{
        return this.preparedTransactions;
}

 public AMQTx getTx(TransactionId txid,
    Location location) 
{
        AMQTx tx = null;
        synchronized (inflightTransactions) {
            tx = inflightTransactions.get(txid);
            if (tx == null) {
                tx = new AMQTx(location);
                inflightTransactions.put(txid, tx);
            }
        }
        return tx;
}

 public boolean isDoingRecover() 
{
        return doingRecover;
}

 public  void prepare(TransactionId txid) throws IOException 
{
        AMQTx tx = null;
        synchronized (inflightTransactions) {
            tx = inflightTransactions.remove(txid);
        }
        if (tx == null) {
            return;
        }
        peristenceAdapter.writeCommand(new JournalTransaction(JournalTransaction.XA_PREPARE, txid, false), true);
        synchronized (preparedTransactions) {
            preparedTransactions.put(txid, tx);
        }
}

 public synchronized  void recover(TransactionRecoveryListener listener) throws IOException 
{
        // All the in-flight transactions get rolled back..
        synchronized (inflightTransactions) {
            inflightTransactions.clear();
        }
        this.doingRecover = true;
        try {
            Map< TransactionId, AMQTx > txs = null;
            synchronized (preparedTransactions) {
                txs = new LinkedHashMap< TransactionId, AMQTx >(preparedTransactions);
            }
            for (Iterator< TransactionId > iter = txs.keySet().iterator(); iter.hasNext();) {
                Object txid = iter.next();
                AMQTx tx = txs.get(txid);
                listener.recover((XATransactionId)txid, tx.getMessages(), tx.getAcks());
            }
        } finally {
            this.doingRecover = false;
        }
}

 public  void removeMessage(AMQMessageStore store,
    MessageAck ack,
    Location location) throws IOException 
{
        AMQTx tx = getTx(ack.getTransactionId(), location);
        tx.add(store, ack);
}

 public AMQTx replayCommit(TransactionId txid,
    boolean wasPrepared) throws IOException 
{
        if (wasPrepared) {
            synchronized (preparedTransactions) {
                return preparedTransactions.remove(txid);
            }
        } else {
            synchronized (inflightTransactions) {
                return inflightTransactions.remove(txid);
            }
        }
}

 public  void replayPrepare(TransactionId txid) throws IOException 
{
        AMQTx tx = null;
        synchronized (inflightTransactions) {
            tx = inflightTransactions.remove(txid);
        }
        if (tx == null) {
            return;
        }
        synchronized (preparedTransactions) {
            preparedTransactions.put(txid, tx);
        }
}

 public  void replayRollback(TransactionId txid) throws IOException 
{
        boolean inflight = false;
        synchronized (inflightTransactions) {
            inflight = inflightTransactions.remove(txid) != null;
        }
        if (inflight) {
            synchronized (preparedTransactions) {
                preparedTransactions.remove(txid);
            }
        }
}

 public  void rollback(TransactionId txid) throws IOException 
{
        AMQTx tx = null;
        synchronized (inflightTransactions) {
            tx = inflightTransactions.remove(txid);
        }
        if (tx != null) {
            synchronized (preparedTransactions) {
                tx = preparedTransactions.remove(txid);
            }
        }
        if (tx != null) {
            if (txid.isXATransaction()) {
                peristenceAdapter.writeCommand(new JournalTransaction(JournalTransaction.XA_ROLLBACK, txid, false), true,true);
            } else {
                peristenceAdapter.writeCommand(new JournalTransaction(JournalTransaction.LOCAL_ROLLBACK, txid, false), true,true);
            }
        }
}

 public  void setPreparedTransactions(Map<TransactionId, AMQTx> preparedTransactions) 
{
        if (preparedTransactions != null) {
            this.preparedTransactions.clear();
            this.preparedTransactions.putAll(preparedTransactions);
        }
}

 public  void start() throws Exception 
{

}

 public  void stop() throws Exception 
{

}