rdbmsadapter.java

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

            _destinationLock.readLock().acquire();
            Messages.instance().add(connection, message);
        } catch (InterruptedException exception) {
            throw new PersistenceException("Failed to acquire lock",
                exception);
        } finally {
            _destinationLock.readLock().release();

            if (_log.isDebugEnabled()) {
                _log.debug("addMessage," +
                    (System.currentTimeMillis() - start));
            }
        }
    }

    // implementation of PersistenceMessage.addMessage
    public void updateMessage(Connection connection, MessageImpl message)
        throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            _destinationLock.readLock().acquire();
            Messages.instance().update(connection, message);
        } catch (InterruptedException exception) {
            throw new PersistenceException("Failed to acquire lock",
                exception);
        } finally {
            _destinationLock.readLock().release();
            if (_log.isDebugEnabled()) {
                _log.debug("updateMessage," +
                    (System.currentTimeMillis() - start));
            }
        }
    }

    // implementation of PersistenceAdapter.getUnprocessedMessages
    public Vector getUnprocessedMessages(Connection connection)
        throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            return Messages.instance().getUnprocessedMessages(connection);
        } finally {
            if (_log.isDebugEnabled()) {
                _log.debug("getUnprocessedMessages," + (System.currentTimeMillis() - start));
            }
        }
    }


    // implementation of PersistenceAdapter.removeMessage
    public void removeMessage(Connection connection, String id)
        throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            _destinationLock.readLock().acquire();
            Messages.instance().remove(connection, id);
        } catch (InterruptedException exception) {
            throw new PersistenceException("Failed to acquire lock",
                exception);
        } finally {
            _destinationLock.readLock().release();
            if (_log.isDebugEnabled()) {
                _log.debug("removeMessage," +
                    (System.currentTimeMillis() - start));
            }
        }
    }

    // implementation of PersistenceAdapter.getMessage
    public MessageImpl getMessage(Connection connection, String id)
        throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            return Messages.instance().get(connection, id);
        } finally {
            if (_log.isDebugEnabled()) {
                _log.debug("getMessage," + (System.currentTimeMillis() - start));
            }
        }
    }

    // implementation of PersistenceAdapter.getMessages
    public Vector getMessages(Connection connection, PersistentMessageHandle handle)
        throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            return Messages.instance().getMessages(connection,
                handle.getDestination().getName(), handle.getPriority(),
                handle.getAcceptedTime());
        } finally {
            if (_log.isDebugEnabled()) {
                _log.debug("getMessages," + (System.currentTimeMillis() - start));
            }
        }
    }

    // implementation of PersistenceAdapter.addMessageHandle
    public void addMessageHandle(Connection connection, PersistentMessageHandle handle)
        throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            _destinationLock.readLock().acquire();
            MessageHandles.instance().addMessageHandle(connection, handle);
        } catch (InterruptedException exception) {
            throw new PersistenceException("Failed to acquire lock",
                exception);
        } finally {
            _destinationLock.readLock().release();
            if (_log.isDebugEnabled()) {
                _log.debug("addMessageHandle," + (System.currentTimeMillis() - start));
            }
        }
    }

    // implementation of PersistenceAdapter.updateMessageHandle
    public void updateMessageHandle(Connection connection, PersistentMessageHandle handle)
        throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            _destinationLock.readLock().acquire();
            MessageHandles.instance().updateMessageHandle(connection, handle);
        } catch (InterruptedException exception) {
            throw new PersistenceException("Failed to acquire lock",
                exception);
        } finally {
            _destinationLock.readLock().release();
            if (_log.isDebugEnabled()) {
                _log.debug("updateMessageHandle," + (System.currentTimeMillis() - start));
            }
        }
    }

    // implementation of PersistenceAdapter.removeMessageHandle
    public void removeMessageHandle(Connection connection, PersistentMessageHandle handle)
        throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            _destinationLock.readLock().acquire();
            MessageHandles.instance().removeMessageHandle(connection, handle);
        } catch (InterruptedException exception) {
            throw new PersistenceException("Failed to acquire lock",
                exception);
        } finally {
            _destinationLock.readLock().release();
            if (_log.isDebugEnabled()) {
                _log.debug("removeMessageHandle," + (System.currentTimeMillis() - start));
            }
        }
    }

    // implementation of PersistenceAdapter.getMessageHandles
    public Vector getMessageHandles(Connection connection,
                                    JmsDestination destination, String name)
        throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            return MessageHandles.instance().getMessageHandles(connection,
                destination.getName(), name);
        } finally {
            if (_log.isDebugEnabled()) {
                _log.debug("getMessageHandles," 
                          + (System.currentTimeMillis() - start));
            }
        }
    }

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

        try {
            _destinationLock.readLock().acquire();
            Consumers.instance().add(connection, topic, consumer);
        } catch (InterruptedException exception) {
            throw new PersistenceException("Failed to acquire lock",
                exception);
        } finally {
            _destinationLock.readLock().release();
        }
    }

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

        try {
            _destinationLock.readLock().acquire();
            Consumers.instance().remove(connection, consumer);
        } catch (InterruptedException exception) {
            throw new PersistenceException("Failed to acquire lock",
                exception);
        } finally {
            _destinationLock.readLock().release();
        }
    }

    // implementation of PersistenceAdapter.getDurableConsumers
    public Enumeration getDurableConsumers(Connection connection, String topic)
        throws PersistenceException {
        return Consumers.instance().getDurableConsumers(topic).elements();
    }

    // implementation of PersistenceAdapter.getAllDurableConsumers
    public HashMap getAllDurableConsumers(Connection connection)
        throws PersistenceException {

        return Consumers.instance().getAllDurableConsumers();
    }

    // implementation of PersistenceAdapter.durableConsumerExists
    public boolean durableConsumerExists(Connection connection, String name)
        throws PersistenceException {

        return Consumers.instance().exists(name);
    }

    // implementation of PersistenceAdapter.addDestination
    public void addDestination(Connection connection, String name,
                               boolean queue)
        throws PersistenceException {

        JmsDestination destination = (queue)
            ? (JmsDestination) new JmsQueue(name)
            : (JmsDestination) new JmsTopic(name);

        // create the destination. If the destination is also
        // a queue create a special consumer for it.
        try {
            _destinationLock.readLock().acquire();
            Destinations.instance().add(connection, destination);
            if (queue) {
                Consumers.instance().add(connection, name, name);
            }
        } catch (InterruptedException exception) {
            throw new PersistenceException("Failed to acquire lock",
                exception);
        } finally {
            _destinationLock.readLock().release();
        }
    }

    // implementation of PersistenceAdapter.removeDestination
    public void removeDestination(Connection connection, String name)
        throws PersistenceException {

        JmsDestination destination = Destinations.instance().get(name);
        if (destination != null) {
            try {
                _destinationLock.writeLock().acquire();
                Destinations.instance().remove(connection, destination);
            } catch (InterruptedException exception) {
                throw new PersistenceException("Failed to acquire lock",
                    exception);
            } finally {
                _destinationLock.writeLock().release();
            }
        }
    }

    // implementation of PersistenceAdapter.getAllDestinations
    public Enumeration getAllDestinations(Connection connection)
        throws PersistenceException {

        return Destinations.instance().getDestinations().elements();
    }

    // implementation of PersistenceAdapter.checkDestination
    public boolean checkDestination(Connection connection, String name)
        throws PersistenceException {

        return (Destinations.instance().get(name) != null);
    }

    // implementation of getQueueMessageCount
    public int getQueueMessageCount(Connection connection, String name)
        throws PersistenceException {

        return MessageHandles.instance().getMessageCount(
            connection, name, name);
    }

    // implementation of PersistenceAdapter.getQueueMessageCount
    public int getDurableConsumerMessageCount(Connection connection,
                                              String destination, String name)
        throws PersistenceException {

        return MessageHandles.instance().getMessageCount(connection,
            destination, name);
    }