resourcemanager.java

实现了Jms的服务器源码,支持多种适配器,DB,FTP,支持多种数据库

     * @return TransactionLog
     * @throws TransactionLogException - if there is tx log exception
     * @throws ResourceManagerException - if there is a resource problem.
     */
    private TransactionLog getTransactionLog(ExternalXid txid)
        throws TransactionLogException, ResourceManagerException {
        TransactionLog log = (TransactionLog) _tridToLogCache.get(txid);
        if (log == null) {
            log = getCurrentTransactionLog();
            addTridLogEntry(txid, log);
        }

        return log;
    }

    /**
     * Get the current transaction log. It will check the last transaction
     * log opened by the resource manager and determine whether there is
     * space enough to process another transaction.
     * <p>
     * If there is space enough then it will return that transaction,
     * otherwise it will create a new transaction log for the resource
     *
     * @return TransactionLog - the transaction log to use
     * @throws ResourceManagerException
     * @throws TransactionLogException
     */
    private TransactionLog getCurrentTransactionLog()
        throws TransactionLogException, ResourceManagerException {
        TransactionLog log = null;

        synchronized (_logs) {
            if (_logs.size() > 0) {
                log = (TransactionLog) _logs.last();
            }

            if ((log == null) ||
                (log.size() > _logFileSize)) {
                log = createNextTransactionLog();
            }
        }

        return log;
    }

    /**
     * Add an entry to the trid log cache table for the specified trid and
     * transaction log mapping.
     *
     * @param trid - the transaction identifier
     * @param log - the transaction log
     */
    private void addTridLogEntry(ExternalXid trid, TransactionLog log) {
        synchronized (_cacheLock) {
            // one to one relationship
            _tridToLogCache.put(trid, log);

            // one to many relationship
            Vector trids = (Vector) _logToTridCache.get(log);
            if (trids == null) {
                trids = new Vector();
                _logToTridCache.put(log, trids);
            }
            trids.addElement(trid);
        }
    }

    /**
     * Check whether the specified log is also the current log
     *
     * @param log - the log to check
     * @return boolean - true if it is
     */
    private boolean isCurrentTransactionLog(TransactionLog log) {
        boolean result = false;

        if (_logs.size() > 0) {
            result = log.equals(_logs.last());
        }

        return result;
    }

    /**
     * Remove an entry to the trid log cache table for the specified trid and
     * transaction log mapping.
     *
     * @param trid - the transaction identifier
     * @param log - the transaction log
     */
    private void removeTridLogEntry(ExternalXid trid, TransactionLog log) {
        synchronized (_cacheLock) {

            // one to one relationship
            _tridToLogCache.remove(trid);

            // one to many relationship
            Vector trids = (Vector) _logToTridCache.get(log);
            if (trids != null) {
                trids.remove(trid);
                if (trids.size() == 0) {
                    _logToTridCache.remove(log);
                }
            }
        }
    }

    /**
     * Return an arrya of records, both state and date, for the specified
     * global transaction
     *
     * @param xid - the global transaction id
     * @param rid - the resource id
     * @return Object[] - array of records
     */
    protected Object[] getTransactionRecords(ExternalXid xid, String rid) {
        Object[] records;

        // we also want to add this to the transaction data for that
        // txid
        LinkedList list = (LinkedList) _activeTransactions.get(xid);
        if (list != null) {
            records = list.toArray();
        } else {
            records = new Object[0];
        }

        return records;
    }


    /**
     * Return the sequence number of the file
     * files are associated with a unique number
     *
     * @param name - the file name to investigate
     * @return long - the transaction log number
     * @throws ResourceManagerException
     */
    protected long getSequenceNumber(String name)
        throws ResourceManagerException {
        int start = name.indexOf(RM_LOGFILE_PREFIX) +
            RM_LOGFILE_PREFIX.length();
        int end = name.indexOf(RM_LOGFILE_EXTENSION);

        // the number must be between the start and end positions
        try {
            return Long.parseLong(name.substring(start, end));
        } catch (NumberFormatException exception) {
            throw new ResourceManagerException(
                "Invalid name assigned to resource manager file " + name);
        }
    }

    /**
     * Return true if the specified transaction is active
     *
     * @param xid - the gobal transaction identifier
     */
    private synchronized boolean isTransactionActive(ExternalXid xid) {
        return _activeTransactions.containsKey(xid);
    }

    /**
     * Dump the specified records to the screen
     */
    private void dumpRecovered(HashMap records) {
        Iterator iter = records.keySet().iterator();
        while (iter.hasNext()) {
            ExternalXid txid = (ExternalXid) iter.next();
            LinkedList list = (LinkedList) records.get(txid);
            Iterator oiter = list.iterator();
            while (oiter.hasNext()) {
                Object object = oiter.next();
                if (object instanceof StateTransactionLogEntry) {
                    System.err.println("Recovered [" + txid + "] Class " +
                        object.getClass().getName() + " [" +
                        ((StateTransactionLogEntry) object).getState().toString() + "]");
                } else {
                    System.err.println("Recovered [" + txid + "] Class " +
                        object.getClass().getName());
                }
            }
        }
    }


    /**
     * Helper and type-safe method for creating a wrapper object for published
     * messages
     *
     * @param message - the message published
     * @return PublishedMessageWrapper
     */
    private PublishedMessageWrapper createPublishedMessageWrapper(
        MessageImpl message) {
        return new PublishedMessageWrapper(message);
    }

    /**
     * Helper and type-safe method for creating a wrapper object for received
     * messages
     *
     * @param id - the identity of the consumer receiving the message
     * @param handle - the handle of the message received
     * @return ReceivedMessageWrapper
     */
    private ReceivedMessageWrapper createReceivedMessageWrapper(
        String id, MessageHandle handle) {
        return new ReceivedMessageWrapper(id, handle);
    }

    /**
     * This functor is used by various collections to order the transaction log
     * files created by this resource manager. The resource manager will create
     * log files with sequentially increasing numbers (i.e xxx01.log, xxx2.log
     */
    private class TranLogFileComparator
        implements Comparator {

        // implementation of Comparator.comapre
        public int compare(Object o1, Object o2) {
            int result = -1;

            try {
                if ((o1 instanceof TransactionLog) &&
                    (o2 instanceof TransactionLog)) {
                    long seq1 = getSequenceNumber(((TransactionLog) o1).getName());
                    long seq2 = getSequenceNumber(((TransactionLog) o2).getName());

                    if (seq1 > seq2) {
                        result = 1;
                    } else if (seq1 < seq2) {
                        result = -1;
                    } else {
                        result = 0;
                    }
                } else {
                    throw new ClassCastException("o1 = " +
                        o1.getClass().getName() + " and o2 = " +
                        o2.getClass().getName());
                }
            } catch (Exception exception) {
                throw new RuntimeException("Error in ResourceManager.compare " +
                    exception.toString());
            }

            return result;
        }

        // implementation of Comparator.equals
        public boolean equals(Object obj) {
            if (obj instanceof TranLogFileComparator) {
                return true;
            }

            return false;
        }
    }


    /**
     * This private member class is used to wrap the transactional object,
     * which for this particular resource manager is a published message or
     * a received message handle.
     */
    abstract private class TransactionalObjectWrapper {

        /**
         * The transactional object instance
         */
        private Object _object;

        /**
         * Create an instance of the wrapper using the type and the object
         *
         * @param object - the associated object
         */
        public TransactionalObjectWrapper(Object object) {
            _object = object;
        }

        /**
         * Check whether the wrapper contains a published message. Note that a
         * published message has a {@link MessageImpl} a the transactional
         * object.
         *
         * @return boolean - true if it is
         */
        public boolean isPublishedMessage() {
            return this instanceof PublishedMessageWrapper;
        }

        /**
         * Check whether the wrapper contains a received message handle. Note
         * that a received message contains a {@link MessageHandle} as the
         * transactional object.
         *
         * @return boolean - true if it does
         */
        public boolean isReceivedMessage() {
            return this instanceof ReceivedMessageWrapper;
        }

        /**
         * Return the transaction object
         *
         * @return Object
         */
        public Object getObject() {
            return _object;
        }

    }


    /**
     * This private member class is used to wrap a published message
     */
    private class PublishedMessageWrapper extends TransactionalObjectWrapper {

        /**
         * Create an instance of the wrapper using the specified message
         *
         * @param message - the message to wrap
         */
        public PublishedMessageWrapper(MessageImpl message) {
            super(message);
        }

        /**
         * Return an instance of the message object
         *
         * @return MessageImpl
         */
        public MessageImpl getMessage() {
            return (MessageImpl) super.getObject();
        }
    }


    /**
     * This private member class is used to wrap a received message
     */
    private class ReceivedMessageWrapper extends TransactionalObjectWrapper {

        /**
         * Caches the id of the {@link ConsumerEndpoint} that is processed
         * this handle
         */
        private String _consumerId;

        /**
         * Create an instance of the wrapper using the specified message
         *
         * @param id - the identity of the consumer endpoint
         * @param handle - the handle to the message
         */
        public ReceivedMessageWrapper(String id, MessageHandle handle) {
            super(handle);
            _consumerId = id;
        }

        /**
         * Return a reference to the  consumer identity
         *
         * @return String
         */
        public String getConsumerId() {
            return _consumerId;
        }

        /**
         * Return an instance of the message handle
         *
         * @return MessageHandle
         */
        public MessageHandle getMessageHandle() {
            return (MessageHandle) super.getObject();
        }
    }

} //-- ResourceManager