objectadapter.java

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

                if (vector.size() == 1) {
                    long ver;

                    if ((ver = ((PMDLongInteger) vector.get(0)).get())
                        != VERSIONNUM) {
                        _log.error(
                            "Incompatible Database version Schema\n" +
                            "Db Vesrsion = " + ver + "\tSchema Version = "
                            + VERSIONNUM + "\nExiting...");
                        System.exit(-1);
                    }
                } else {
                    _log.error("Corrupted Db schema version ID:\n " +
                        "Exiting...");
                    System.exit(-1);
                }
                session.getCurrentTransaction().abort();
            }
            try {
                PMDSessionManager.instance().destroySession();
            } catch (FailedToDestroySessionException exception) {
                _log.error("Failed to destroy session", exception);
            }
        } catch (Exception exception) {
            _log.error("Error verifying DB schema, exiting", exception);
            System.exit(-1);
        }

    }

    /**
     * Check to see if the Id root has been created.
     * If not create it and initialise it. This is used as a sequencer
     * to number all Messages that are delivered to OpenJMS server.
     *
     */
    private void createIdRoot()
        throws FailedToCreateSessionException, TransactionInProgressException,
        FailedToCreateCollectionException, DatabaseIOException,
        TransactionNotInProgressException, ObjectNameExistsException,
        TransactionException, FailedToCreateLockException,
        FailedToAcquireLockException, FailedToDestroySessionException {
        PMDVector vector;
        PMDHashMap map;
        SessionIfc session = getSession();

        session.getCurrentTransaction().begin();
        if ((vector = (PMDVector) session.lookup(IDSTORAGE)) == null) {
            _log.debug("Creating ID Storage root");
            vector =
                (PMDVector) session.getCollectionManager().createVector();
            session.createObject(vector);
            session.bind(IDSTORAGE, vector);
            session.getCurrentTransaction().commit();

            session.getCurrentTransaction().begin();
            vector = (PMDVector) session.lookup(IDSTORAGE);
            session.acquireLock(vector, MAX_WAIT_TIME);
            vector.addElement(new PMDLongInteger(0));
            session.updateObject(vector);
            session.getCurrentTransaction().commit();
        } else {
            session.getCurrentTransaction().abort();
        }

        session.getCurrentTransaction().begin();
        if ((map = (PMDHashMap) session.lookup(MESSAGES)) == null) {
            _log.debug("Creating Messages root");
            map =
                (PMDHashMap) session.getCollectionManager().createHashMap();
            session.createObject(map);
            session.bind(MESSAGES, map);
            session.getCurrentTransaction().commit();
        } else {
            session.getCurrentTransaction().abort();
        }

        session.getCurrentTransaction().begin();
        if ((vector = (PMDVector) session.lookup(DESTINATIONS)) == null) {
            _log.debug("Creating Destinations root");
            vector =
                (PMDVector) session.getCollectionManager().createVector();
            session.createObject(vector);
            session.bind(DESTINATIONS, vector);
            session.getCurrentTransaction().commit();
        } else {
            session.getCurrentTransaction().abort();
        }
        PMDSessionManager.instance().destroySession();
    }

    // implementation of PersistenceAdapter.getLastIds
    public long getLastId(Connection connection)
        throws PersistenceException {
        long lastId = -1;

        try {
            createIdRoot();
            PMDVector vector;
            SessionIfc session = getSession();

            session.getCurrentTransaction().begin();
            if ((vector = (PMDVector) session.lookup(IDSTORAGE)) != null) {
                if (vector.size() == 1) {
                    lastId = ((PMDLongInteger) vector.get(0)).get();
                }
            }
            session.getCurrentTransaction().abort();
        } catch (Exception err) {
            _log.error("Failed to get Id: " + err.getMessage());
        }
        try {
            PMDSessionManager.instance().destroySession();
        } catch (FailedToDestroySessionException sessErr) {
            _log.error("Failed to destroy session: " +
                sessErr.getMessage());
            sessErr.printStackTrace();
            // Session not destroyed.
        }
        return lastId;
    }

    // implementation of PersistenceAdapter.updateIds
    public void updateIds(Connection connection, long id)
        throws PersistenceException {
        try {
            PMDVector vector;
            SessionIfc session = getSession();

            session.getCurrentTransaction().begin();
            if ((vector = (PMDVector) session.lookup(IDSTORAGE)) != null) {
                session.acquireLock(vector, MAX_WAIT_TIME);
                if (vector.size() == 1) {
                    vector.set(0, new PMDLongInteger(id));
                    session.updateObject(vector);
                }

            }
            session.getCurrentTransaction().commit();
        } catch (Exception err) {
            throw new PersistenceException("Failed to get Id: " + err.getMessage());
        }
        try {
            PMDSessionManager.instance().destroySession();
        } catch (FailedToDestroySessionException sessErr) {
            _log.error("Failed to destroy session", sessErr);
        }
    }

    // implementation of PersistenceAdapter.addMessage
    public synchronized void addMessage(Connection connection,
                                        MessageImpl message)
        throws PersistenceException {

        try {
            DestinationImpl dest =
                (DestinationImpl) message.getJMSDestination();
            String name = dest.getDestination();
            PMDHashMap messages;
            SessionIfc session = getSession();

            session.getCurrentTransaction().begin();

            messages = messageTable(session);

            if (messages != null) {
                try {
                    session.acquireLock(messages, MAX_WAIT_TIME);

                    // check to see that the message does not exist
                    String id = message.getMessageId().getId();
                    if (messages.get(id) == null) {
                        PersistentMessage pm = new PersistentMessage(message);
                        session.createObject(pm);
                        // when we initially add a message to the database it
                        // must go in the unprocessed list
                        messages.put(id, session.createHandle(pm));
                        session.updateObject(messages);
                    } else {
                        throw new PersistenceException("Mesage with this id: " +
                            message.getJMSMessageID() +
                            " already exists: NOT ADDING");
                    }
                } catch (Exception err) {
                    throw new PersistenceException("Error in addMessage " +
                        err.toString());
                }
            } else {
                _log.error("Attempt to save message for non register" +
                    " queue/topic, name = " + name);
            }

            session.getCurrentTransaction().commit();
            PMDSessionManager.instance().destroySession();
        } catch (PersistenceException exception) {
            // rethrowe it
            throw exception;
        } catch (Exception err) {
            throw new PersistenceException("Error in addMessage " + err.toString());
        }
    }

    // implementation of PersistenceAdapter.updateMessage
    public synchronized void updateMessage(Connection connection,
                                           MessageImpl message)
        throws PersistenceException {

        try {
            DestinationImpl dest =
                (DestinationImpl) message.getJMSDestination();
            String name = dest.getDestination();
            PMDHashMap messages;
            SessionIfc session = getSession();

            session.getCurrentTransaction().begin();

            messages = messageTable(session);

            if (messages != null) {
                try {
                    session.acquireLock(messages, MAX_WAIT_TIME);

                    // check to see that the message does not exist
                    String id = message.getMessageId().getId();
                    if (messages.get(id) != null) {
                        // remove the old message and update it
                        PMDHandle handle = (PMDHandle) messages.get(id);
                        PersistentMessage pm = (PersistentMessage) handle.resolve();
                        pm.setMessage(message);
                        pm.setProcessed(message.getProcessed());
                        session.updateObject(pm);
                    } else {
                        throw new PersistenceException("Mesage with this id: " +
                            message.getJMSMessageID() +
                            " doesn't exists: NOT UPDATING");
                    }
                } catch (Exception err) {
                    throw new PersistenceException("Error in updateMessage " +
                        err.toString());
                }
            } else {
                throw new PersistenceException("Attempt to save message for non register" +
                    " queue/topic, name = " + name);
            }

            session.getCurrentTransaction().commit();
            PMDSessionManager.instance().destroySession();
        } catch (PersistenceException pe) {
            throw pe;
        } catch (Exception err) {
            throw new PersistenceException("Error in updateMessage " +
                err.toString());
        }
    }

    // implementation of PersistenceAdapter.removeMessage
    public synchronized void removeMessage(Connection connection, String id)
        throws PersistenceException {

        try {
            PMDHashMap map;
            SessionIfc session = getSession();

            session.getCurrentTransaction().begin();

            if ((map = messageTable(session)) != null) {
                try {
                    session.acquireLock(map, MAX_WAIT_TIME);
                    PMDHandle handle = (PMDHandle) map.remove(id);
                    if (handle != null) {
                        session.deleteObject
                            ((PersistentMessage) handle.resolve());
                    }

                    session.updateObject(map);
                } catch (Exception err) {
                    throw new PersistenceException("Error in removeMessage " +
                        err.toString());
                }
            } else {
                throw new PersistenceException("Error in removeMessage " +
                    "Cannot retrieve the message talbe ");
            }

            session.getCurrentTransaction().commit();
            PMDSessionManager.instance().destroySession();
        } catch (PersistenceException pe) {
            throw pe;
        } catch (Exception err) {
            throw new PersistenceException("Error in removeMessage " +
                err.toString());
        }
    }

    // implementation of PersistenceAdapter.getMessage
    public synchronized MessageImpl getMessage(Connection connection, String id)
        throws PersistenceException {
        MessageImpl message = null;

        try {
            PMDHashMap map;
            SessionIfc session = getSession();

            session.getCurrentTransaction().begin();

            if ((map = messageTable(session)) != null) {
                try {
                    session.acquireLock(map, MAX_WAIT_TIME);
                    PMDHandle handle = (PMDHandle) map.get(id);
                    PersistentMessage m = null;

                    if (handle != null) {
                        m = (PersistentMessage) handle.resolve();
                    }

                    if (m != null) {
                        message = m.getMessage();
                    }
                } catch (Exception err) {
                    throw new PersistenceException("Error in getMessage " +
                        err.toString());
                }
            } else {