public JournalTransactionStore(JournalPersistenceAdapter adapter) 
{
        this.peristenceAdapter = adapter;
}

 public  void acknowledge(JournalTopicMessageStore store,
    JournalTopicAck ack,
    RecordLocation location) 
{
        Tx tx = getTx(ack.getTransactionId(), location);
        tx.add(store, ack);
}

  void addMessage(JournalMessageStore store,
    Message message,
    RecordLocation location) throws IOException 
{
        Tx tx = getTx(message.getTransactionId(), location);
        tx.add(store, message);
}

 public RecordLocation 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.
        RecordLocation rc = null;
        synchronized (inflightTransactions) {
            for (Iterator< Tx > iter = inflightTransactions.values().iterator(); iter.hasNext();) {
                Tx tx = iter.next();
                RecordLocation location = tx.location;
                if (rc == null || rc.compareTo(location) <  0) {
                    rc = location;
                }
            }
        }
        synchronized (preparedTransactions) {
            for (Iterator< Tx > iter = preparedTransactions.values().iterator(); iter.hasNext();) {
                Tx tx = iter.next();
                RecordLocation location = tx.location;
                if (rc == null || rc.compareTo(location) <  0) {
                    rc = location;
                }
            }
            return rc;
        }
}

 public  void commit(TransactionId txid,
    boolean wasPrepared) throws IOException 
{
        Tx 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);
        } else {
            peristenceAdapter.writeCommand(new JournalTransaction(JournalTransaction.LOCAL_COMMIT, txid,
                                                                  wasPrepared), true);
        }
}

 public Tx getTx(Object txid,
    RecordLocation location) 
{
        Tx tx = null;
        synchronized (inflightTransactions) {
            tx = inflightTransactions.get(txid);
        }
        if (tx == null) {
            tx = new Tx(location);
            inflightTransactions.put(txid, tx);
        }
        return tx;
}

 public boolean isDoingRecover() 
{
        return doingRecover;
}

 public  void prepare(TransactionId txid) throws IOException 
{
        Tx 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, Tx > txs = null;
            synchronized (preparedTransactions) {
                txs = new LinkedHashMap< TransactionId, Tx >(preparedTransactions);
            }
            for (Iterator< TransactionId > iter = txs.keySet().iterator(); iter.hasNext();) {
                Object txid = iter.next();
                Tx tx = txs.get(txid);
                listener.recover((XATransactionId)txid, tx.getMessages(), tx.getAcks());
            }
        } finally {
            this.doingRecover = false;
        }
}

 public  void removeMessage(JournalMessageStore store,
    MessageAck ack,
    RecordLocation location) throws IOException 
{
        Tx tx = getTx(ack.getTransactionId(), location);
        tx.add(store, ack);
}

 public Tx 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 
{
        Tx 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 
{
        Tx 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);
            } else {
                peristenceAdapter.writeCommand(new JournalTransaction(JournalTransaction.LOCAL_ROLLBACK,
                                                                      txid, false), true);
            }
        }
}

 public  void start() throws Exception 
{

}

 public  void stop() throws Exception 
{

}