transactionlog.java

来自「一个java方面的消息订阅发送的源码」· Java 代码 · 共 441 行 · 第 1/2 页

    /**
     * Close the transaction log
     *
     * @throws TransactionLogException - if it fails to close the log
     */
    public void close()
        throws TransactionLogException {
        try {
            if (_dos != null) {
                _dos.close();
            }
        } catch (IOException exception) {
            throw new TransactionLogException("Error in close " +
                exception.toString());
        }
    }

    /**
     * Return the size of the transaction log file.
     *
     * @return long - the length of the file
     */
    public long size() {
        return _size;
    }

    /**
     * Force a recovery of this log file. This will close the output file stream
     * if one is opened and then read each entry from the log file and send it to
     * the specified listener, if one is allocated.
     * <p>
     * The returned data structure is a HashMap, where the key is a
     * {@link ExternalXid} and the entries are LinkedList of {@link
     * BaseTransactionLogEntry} objects
     *
     * @return HashMap - a list of open transactions
     * @throws TransactionLogException - if there is a prob recovering
     */
    public synchronized HashMap recover()
        throws TransactionLogException {
        return getOpenTransactionList();
    }

    /**
     * Check if we can garbage collect this transaction log. It will go through
     * the log file and check to see whether there are any open transaction. If
     * there are no open transactions the it is a candidate for garage collection
     *
     * @return boolean - true if we can garbage collect; false otherwise
     */
    public synchronized boolean canGarbageCollect() {
        boolean result = false;

        try {
            HashMap records = getOpenTransactionList();
            if (records.size() == 0) {
                result = true;
            }
        } catch (Exception ignore) {
            ignore.printStackTrace();
        }

        return result;
    }

    /**
     * Destroy this transaction log, which basically removes it from the
     * file system
     *
     * @throws TransactionLogException
     */
    public synchronized void destroy()
        throws TransactionLogException {
        try {
            close();
            if (!(new File(_name)).delete()) {
                _log.error("Failed to destroy " + _name);
            }
        } catch (Exception exception) {
            throw new TransactionLogException("Error in destroy " +
                exception.toString());

        }
    }

    // override Object.equals
    public boolean equals(Object obj) {
        boolean result = false;

        if ((obj instanceof TransactionLog) &&
            (((TransactionLog) obj)._name.equals(_name))) {
            result = true;
        }

        return result;
    }

    /**
     * Return an instance of the output stream. If one does not exist then
     * create it.
     *
     * @return DataOutputStream - the output stream
     */
    private DataOutputStream getOutputStream()
        throws IOException, FileNotFoundException {
        if (_dos == null) {
            _dos = new DataOutputStream(
                new BufferedOutputStream(
                    new FileOutputStream(_name, true)));
        }

        return _dos;
    }

    /**
     * Return a list of open transactions in a HashMap. The key is the transaction
     * id and the data is a vector of associated data records in a LinkedList
     *
     * @return HashMap
     * @throws TransactionLogException - if there is a prob recovering
     */
    private HashMap getOpenTransactionList()
        throws TransactionLogException {

        HashMap records = new HashMap();

        // if the output stream is opened then close it
        try {
            if (_dos != null) {
                _dos.close();
                _dos = null;
            }
        } catch (Exception exception) {
            throw new TransactionLogException("Error in recover " +
                exception.toString());
        }


        FileInputStream fis = null;
        try {
            fis = new FileInputStream(_name);
            DataInputStream dis = new DataInputStream(new BufferedInputStream(fis));

            while (dis.available() > 0) {
                byte[] blob = new byte[(int) dis.readLong()];
                dis.readFully(blob);
                Object object = SerializationHelper.deserialize(blob);
                if (object instanceof StateTransactionLogEntry) {
                    StateTransactionLogEntry state = (StateTransactionLogEntry) object;
                    LinkedList list = null;
                    switch (state.getState().getOrd()) {
                        case TransactionState.OPENED_ORD:
                            if (records.containsKey(state.getExternalXid())) {
                                _log.error("OPENED_ORD : Transaction log is inconsistent");
                                continue;
                            }

                            list = new LinkedList();
                            records.put(state.getExternalXid(), list);
                            list.add(state);
                            break;

                        case TransactionState.PREPARED_ORD:
                            list = (LinkedList) records.get(state.getExternalXid());
                            if (list == null) {
                                _log.error("PREPARED_ORD : Transaction log is inconsistent");
                                continue;
                            }

                            list.add(state);
                            break;

                        case TransactionState.CLOSED_ORD:
                            if (records.get(state.getExternalXid()) == null) {
                                _log.error("CLOSED_ORD : Transaction log is inconsistent");
                                continue;
                            }

                            records.remove(state.getExternalXid());
                            break;

                        default:
                            break;
                    }
                } else if (object instanceof DataTransactionLogEntry) {
                    DataTransactionLogEntry data = (DataTransactionLogEntry) object;
                    LinkedList list = (LinkedList) records.get(data.getExternalXid());
                    if (list == null) {
                        _log.error("DATA : Transaction log is inconsistent");
                        continue;
                    }

                    list.add(data);
                } else {
                    System.err.println("There is no support for log entry " +
                        "records of type " + object.getClass().getName());
                }

            }
        } catch (Exception exception) {
            throw new TransactionLogException("Error in recover " +
                exception.toString());
        } finally {
            if (fis != null) {
                try {
                    fis.close();
                } catch (Exception exception) {
                    throw new TransactionLogException("Error in recover " +
                        exception.toString());
                }
            }
        }

        return records;

    }

} //-- TransactionLog