objectadapter.java

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

                throw new PersistenceException("Error in getMessage " +
                    "Failed to retrieve the message table.");
            }
            session.getCurrentTransaction().abort();
            PMDSessionManager.instance().destroySession();
        } catch (PersistenceException pe) {
            throw pe;
        } catch (Exception err) {
            throw new PersistenceException("Error in getMessage " +
                err.toString());
        }
        return message;
    }

    // implementation of PersistenceAdapter.getUnprocessedMessages
    public synchronized Vector getUnprocessedMessages(Connection connection)
        throws PersistenceException {
        Vector result = new Vector();

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

            session.getCurrentTransaction().begin();

            if ((map = messageTable(session)) != null) {
                try {
                    session.acquireLock(map, MAX_WAIT_TIME);
                    Enumeration iter = map.elements();
                    while (iter.hasMoreElements()) {
                        PMDHandle handle = (PMDHandle) iter.nextElement();
                        PersistentMessage m =
                            (PersistentMessage) handle.resolve();

                        if (!m.getProcessed()) {
                            result.add(m.getMessage());
                        }
                    }
                } catch (Exception err) {
                    throw new PersistenceException(
                        "Error in getUnprocessedMessages " + err.toString());
                }
            } else {
                throw new PersistenceException(
                    "Error in getUnprocessedMessage. Failed to get message table");
            }
            session.getCurrentTransaction().abort();
            PMDSessionManager.instance().destroySession();
        } catch (PersistenceException pe) {
            throw pe;
        } catch (Exception err) {
            throw new PersistenceException("Error in getUnprocessedMessages "
                + err.toString());
        }
        return result;
    }

    // implementation of PersistenceAdapter.getMessages
    public synchronized Vector getMessages(Connection connection,
                                           PersistentMessageHandle handle)
        throws PersistenceException {
        Vector messages = new Vector();

        // for the jdbm only ever retrieve a single message. We could improve
        // this at a later date.
        MessageImpl message = getMessage(connection,
            handle.getMessageId().getId());
        if (message != null) {
            messages.add(message);
        }

        return messages;
    }

    // implementation of PersistenceAdapter.addMessageHandle
    public synchronized void addMessageHandle(Connection connection,
                                              PersistentMessageHandle handle)
        throws PersistenceException {

        try {
            PMDVector vector;
            SessionIfc session = getSession();
            String key = getHandlesRootName(handle.getDestination(),
                handle.getConsumerName());


            session.getCurrentTransaction().begin();

            if ((vector = handleTable(key, session)) != null) {
                try {
                    session.acquireLock(vector, MAX_WAIT_TIME);
                    vector.addElement(handle);
                    session.updateObject(vector);
                } catch (Exception err) {
                    throw new PersistenceException("Error in addMessageHandle " +
                        err.toString());
                }
            } else {
                throw new PersistenceException("Error in addMessageHandle " +
                    "Cannot get handle table for " + key);
            }

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

    // implementation of PersistenceAdapter.updateMessageHandle
    public synchronized void updateMessageHandle(Connection connection,
                                                 PersistentMessageHandle handle)
        throws PersistenceException {

        try {
            PMDVector vector;
            SessionIfc session = getSession();
            String key = getHandlesRootName(handle.getDestination(),
                handle.getConsumerName());


            session.getCurrentTransaction().begin();

            if ((vector = handleTable(key, session)) != null) {
                try {
                    session.acquireLock(vector, MAX_WAIT_TIME);

                    // linear search for the matching handle
                    Enumeration handles = vector.elements();
                    while (handles.hasMoreElements()) {
                        PersistentMessageHandle phdl =
                            (PersistentMessageHandle) handles.nextElement();

                        if (phdl.getMessageId().getId().equals(
                            handle.getMessageId().getId())) {
                            phdl.setDelivered(true);
                            break;
                        }
                    }
                    session.updateObject(vector);
                } catch (Exception err) {
                    throw new PersistenceException("Error in addMessageHandle " +
                        err.toString());
                }
            } else {
                throw new PersistenceException("Error in updateMessageHandle " +
                    "Failed to get handle table for " + key);
            }

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

    // implementation of PersistenceAdapter.removeMessageHandle
    public synchronized void removeMessageHandle(Connection connection,
                                                 PersistentMessageHandle handle)
        throws PersistenceException {

        try {
            PMDVector vector;
            SessionIfc session = getSession();
            String key = getHandlesRootName(handle.getDestination(),
                handle.getConsumerName());


            session.getCurrentTransaction().begin();

            if ((vector = handleTable(key, session)) != null) {
                try {
                    session.acquireLock(vector, MAX_WAIT_TIME);
                    vector.remove(handle);
                    session.updateObject(vector);
                } catch (Exception err) {
                    throw new PersistenceException("Error in removeMessageHandle " +
                        err.toString());
                }
            } else {
                throw new PersistenceException("Error in removeMessageHandle " +
                    "Failed to get the handle table for " + key);
            }

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

    // implementation of PersistenceAdapter.getMessageHandles
    public synchronized Vector getMessageHandles(Connection connection,
                                                 JmsDestination destination, String name)
        throws PersistenceException {
        Vector result = new Vector();

        try {
            PMDVector vector;
            SessionIfc session = getSession();
            String key = getHandlesRootName(destination, name);


            session.getCurrentTransaction().begin();

            if ((vector = handleTable(key, session)) != null) {
                try {
                    session.acquireLock(vector, MAX_WAIT_TIME);

                    Enumeration handles = vector.elements();
                    while (handles.hasMoreElements()) {
                        PersistentMessageHandle handle =
                            (PersistentMessageHandle) handles.nextElement();
                        result.addElement(handle.clone());
                    }
                } catch (Exception err) {
                    throw new PersistenceException("Error in getMessageHandles " +
                        err.toString());
                }
            }

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

    // implementation of PersistenceAdapter.addDurableConsumer
    public synchronized void addDurableConsumer(Connection connection,
                                                String topic, String consumer)
        throws PersistenceException {

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

            session.getCurrentTransaction().begin();

            if ((vector = destinationTable(session)) != null) {
                try {
                    session.acquireLock(vector, MAX_WAIT_TIME);
                    String target = "@" + consumer;

                    boolean found = false;
                    Enumeration entries = vector.elements();
                    while (entries.hasMoreElements()) {
                        PersistentString entry = (PersistentString) entries.nextElement();
                        if (entry.toString().endsWith(target)) {
                            found = true;
                            break;
                        }
                    }

                    String key = null;
                    if (!found) {
                        // add the target and also create a new table
                        // to hold handles for this consumer
                        vector.addElement(new PersistentString(TOPIC + topic + target));
                        session.updateObject(vector);

                        key = getHandlesRootName(consumer);
                        PMDVector handles =
                            (PMDVector) session.getCollectionManager().createVector();
                        session.createObject(handles);
                        session.bind(key, handles);
                    } else {
                        throw new PersistenceException("Error in addDurableConsumer " +
                            consumer + " already exists.");
                    }
                } catch (Exception err) {
                    throw new PersistenceException("Error in addDurableConsumer " +
                        err.toString());
                }
            } else {
                throw new PersistenceException("Error in addDurableConsumer " +
                    "Failed to get the destination table.");
            }

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

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

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

            session.getCurrentTransaction().begin();

            if ((vector = destinationTable(session)) != null) {
                try {
                    session.acquireLock(vector, MAX_WAIT_TIME);
                    String target = "@" + consumer;

                    boolean found = false;
                    PersistentString entry = null;
                    Enumeration entries = vector.elements();
                    while (entries.hasMoreElements()) {
                        entry = (PersistentString) entries.nextElement();