rdbmsadapter.java

来自「OpenJMS是一个开源的Java Message Service API 1.」· Java 代码 · 共 959 行 · 第 1/3 页

            throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            _destinationLock.readLock().acquire();
            _messages.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.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.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.get(connection, id);
        } finally {
            if (_log.isDebugEnabled()) {
                _log.debug(
                        "getMessage," + (System.currentTimeMillis() - start));
            }
        }
    }

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

        try {
            return _messages.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, MessageHandle handle)
            throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            _destinationLock.readLock().acquire();
            _handles.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,
                                    MessageHandle handle)
            throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            _destinationLock.readLock().acquire();
            _handles.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,
                                    MessageHandle handle)
            throws PersistenceException {
        long start = 0;
        if (_log.isDebugEnabled()) {
            start = System.currentTimeMillis();
        }

        try {
            _destinationLock.readLock().acquire();
            _handles.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 _handles.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.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.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.getDurableConsumers(topic).elements();
    }

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

        return _consumers.getAllDurableConsumers();
    }

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

        return _consumers.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.add(connection, destination);
            if (queue) {
                _consumers.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.get(name);
        if (destination != null) {
            try {
                _destinationLock.writeLock().acquire();
                _destinations.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.getDestinations().elements();
    }

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

        return (_destinations.get(name) != null);
    }

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

        return _handles.getMessageCount(connection, name,
                                                         name);
    }

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

        return _handles.getMessageCount(connection,