bookkeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [35/49] bookkeeper git commit: BOOKKEEPER-769: Remove the Hedwig Code
Date Wed, 16 Mar 2016 03:44:45 GMT
http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client-jms/src/test/java/org/apache/activemq/test/TestSupport.java
----------------------------------------------------------------------
diff --git a/hedwig-client-jms/src/test/java/org/apache/activemq/test/TestSupport.java b/hedwig-client-jms/src/test/java/org/apache/activemq/test/TestSupport.java
deleted file mode 100644
index 6de9021..0000000
--- a/hedwig-client-jms/src/test/java/org/apache/activemq/test/TestSupport.java
+++ /dev/null
@@ -1,261 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.test;
-
-import javax.jms.Topic;
-import java.io.File;
-
-import org.apache.hedwig.JmsTestBase;
-import org.apache.hedwig.jms.SessionImpl;
-import java.lang.reflect.Array;
-
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.TextMessage;
-
-import junit.framework.TestCase;
-
-import org.apache.hedwig.jms.spi.HedwigConnectionFactoryImpl;
-import org.apache.hedwig.jms.message.MessageImpl;
-
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Useful base class for unit test cases
- */
-public abstract class TestSupport extends JmsTestBase {
-    private static final Logger LOG = LoggerFactory.getLogger(TestSupport.class);
-
-    protected HedwigConnectionFactoryImpl connectionFactory;
-    protected boolean topic = true;
-
-    public TestSupport() {
-        super();
-    }
-
-    public TestSupport(String name) {
-        super(name);
-    }
-
-    /**
-     * Creates an MessageImpl.
-     *
-     * @return MessageImpl
-     */
-    protected MessageImpl createMessage() {
-        return new MessageImpl(null);
-    }
-
-    /**
-     * Creates a destination.
-     *
-     * @param subject - topic or queue name.
-     * @return Destination - either an Topic or ActiveMQQUeue.
-     */
-    protected Destination createDestination(String subject) {
-        if (topic) {
-            return SessionImpl.asTopic(subject);
-        } else {
-            return SessionImpl.asTopic(subject);
-        }
-    }
-
-    /**
-     * Tests if firstSet and secondSet are equal.
-     *
-     * @param messsage - string to be displayed when the assertion fails.
-     * @param firstSet[] - set of messages to be compared with its counterpart
-     *                in the secondset.
-     * @param secondSet[] - set of messages to be compared with its counterpart
-     *                in the firstset.
-     * @throws JMSException
-     */
-    protected void assertTextMessagesEqual(Message[] firstSet, Message[] secondSet) throws JMSException {
-        assertTextMessagesEqual("", firstSet, secondSet);
-    }
-
-    /**
-     * Tests if firstSet and secondSet are equal.
-     *
-     * @param messsage - string to be displayed when the assertion fails.
-     * @param firstSet[] - set of messages to be compared with its counterpart
-     *                in the secondset.
-     * @param secondSet[] - set of messages to be compared with its counterpart
-     *                in the firstset.
-     */
-    protected void assertTextMessagesEqual(String messsage, Message[] firstSet,
-                                           Message[] secondSet) throws JMSException {
-        assertEquals("Message count does not match: " + messsage, firstSet.length, secondSet.length);
-
-        for (int i = 0; i < secondSet.length; i++) {
-            TextMessage m1 = (TextMessage)firstSet[i];
-            TextMessage m2 = (TextMessage)secondSet[i];
-            assertTextMessageEqual("Message " + (i + 1) + " did not match : ", m1, m2);
-        }
-    }
-
-    /**
-     * Tests if m1 and m2 are equal.
-     *
-     * @param m1 - message to be compared with m2.
-     * @param m2 - message to be compared with m1.
-     * @throws JMSException
-     */
-    protected void assertEquals(TextMessage m1, TextMessage m2) throws JMSException {
-        assertEquals("", m1, m2);
-    }
-
-    /**
-     * Tests if m1 and m2 are equal.
-     *
-     * @param message - string to be displayed when the assertion fails.
-     * @param m1 - message to be compared with m2.
-     * @param m2 - message to be compared with m1.
-     */
-    protected void assertTextMessageEqual(String message, TextMessage m1, TextMessage m2) throws JMSException {
-        assertFalse(message + ": expected {" + m1 + "}, but was {" + m2 + "}", m1 == null ^ m2 == null);
-
-        if (m1 == null) {
-            return;
-        }
-
-        assertEquals(message, m1.getText(), m2.getText());
-    }
-
-    /**
-     * Tests if m1 and m2 are equal.
-     *
-     * @param m1 - message to be compared with m2.
-     * @param m2 - message to be compared with m1.
-     * @throws JMSException
-     */
-    protected void assertEquals(Message m1, Message m2) throws JMSException {
-        assertEquals("", m1, m2);
-    }
-
-    /**
-     * Tests if m1 and m2 are equal.
-     *
-     * @param message - error message.
-     * @param m1 - message to be compared with m2.
-     * @param m2 -- message to be compared with m1.
-     */
-    protected void assertEquals(String message, Message m1, Message m2) throws JMSException {
-        assertFalse(message + ": expected {" + m1 + "}, but was {" + m2 + "}", m1 == null ^ m2 == null);
-
-        if (m1 == null) {
-            return;
-        }
-
-        assertTrue(message + ": expected {" + m1 + "}, but was {" + m2 + "}", m1.getClass() == m2.getClass());
-
-        if (m1 instanceof TextMessage) {
-            assertTextMessageEqual(message, (TextMessage)m1, (TextMessage)m2);
-        } else {
-            assertEquals(message, m1, m2);
-        }
-    }
-
-    /**
-     * Test if base directory contains spaces
-     */
-    protected void assertBaseDirectoryContainsSpaces() {
-        assertFalse("Base directory cannot contain spaces.",
-                    new File(System.getProperty("basedir", ".")).getAbsoluteFile().toString().contains(" "));
-    }
-
-    /**
-     * Creates an HedwigConnectionFactoryImpl.
-     *
-     * @return HedwigConnectionFactoryImpl
-     * @throws Exception
-     */
-    protected HedwigConnectionFactoryImpl createConnectionFactory() throws Exception {
-        return new HedwigConnectionFactoryImpl();
-    }
-
-    /**
-     * Factory method to create a new connection.
-     *
-     * @return connection
-     * @throws Exception
-     */
-    protected Connection createConnection() throws Exception {
-        return getConnectionFactory().createConnection();
-    }
-
-    /**
-     * Creates an ActiveMQ connection factory.
-     *
-     * @return connectionFactory
-     * @throws Exception
-     */
-    public HedwigConnectionFactoryImpl getConnectionFactory() throws Exception {
-        if (connectionFactory == null) {
-            connectionFactory = createConnectionFactory();
-            assertTrue("Should have created a connection factory!", connectionFactory != null);
-        }
-
-        return connectionFactory;
-    }
-
-    /**
-     * Returns the consumer subject.
-     *
-     * @return String
-     */
-    protected String getConsumerSubject() {
-        return getSubject();
-    }
-
-    /**
-     * Returns the producer subject.
-     *
-     * @return String
-     */
-    protected String getProducerSubject() {
-        return getSubject();
-    }
-
-    /**
-     * Returns the subject.
-     *
-     * @return String
-     */
-    protected String getSubject() {
-        return getClass().getName() + "." + getName();
-    }
-
-    protected void assertArrayEqual(String message, Object[] expected, Object[] actual) {
-        assertEquals(message + ". Array length", expected.length, actual.length);
-        for (int i = 0; i < expected.length; i++) {
-            assertEquals(message + ". element: " + i, expected[i], actual[i]);
-        }
-    }
-
-    protected void assertPrimitiveArrayEqual(String message, Object expected, Object actual) {
-        int length = Array.getLength(expected);
-        assertEquals(message + ". Array length", length, Array.getLength(actual));
-        for (int i = 0; i < length; i++) {
-            assertEquals(message + ". element: " + i, Array.get(expected, i), Array.get(actual, i));
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client-jms/src/test/java/org/apache/activemq/test/rollback/DelegatingTransactionalMessageListener.java
----------------------------------------------------------------------
diff --git a/hedwig-client-jms/src/test/java/org/apache/activemq/test/rollback/DelegatingTransactionalMessageListener.java b/hedwig-client-jms/src/test/java/org/apache/activemq/test/rollback/DelegatingTransactionalMessageListener.java
deleted file mode 100644
index 0bc8a8e..0000000
--- a/hedwig-client-jms/src/test/java/org/apache/activemq/test/rollback/DelegatingTransactionalMessageListener.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.test.rollback;
-
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageListener;
-import javax.jms.Session;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class DelegatingTransactionalMessageListener implements MessageListener {
-    private static final transient Logger LOG = LoggerFactory.getLogger(DelegatingTransactionalMessageListener.class);
-
-    private final MessageListener underlyingListener;
-    private boolean transacted = true;
-    private int ackMode = Session.AUTO_ACKNOWLEDGE;
-    private Session session;
-
-    public DelegatingTransactionalMessageListener(MessageListener underlyingListener,
-                                                  Connection connection, Destination destination) {
-        this.underlyingListener = underlyingListener;
-
-        try {
-            session = connection.createSession(transacted, ackMode);
-            MessageConsumer consumer = session.createConsumer(destination);
-            consumer.setMessageListener(this);
-        } catch (JMSException e) {
-            throw new IllegalStateException("Could not listen to " + destination, e);
-        }
-    }
-
-    public void onMessage(Message message) {
-        try {
-            underlyingListener.onMessage(message);
-            session.commit();
-        } catch (Throwable e) {
-            rollback();
-        }
-    }
-
-    private void rollback() {
-        try {
-            session.rollback();
-        } catch (JMSException e) {
-            LOG.error("Failed to rollback: " + e, e);
-        }
-    }
-
-    public Session getSession() {
-        return session;
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ChangeSentMessageTest.java
----------------------------------------------------------------------
diff --git a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ChangeSentMessageTest.java b/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ChangeSentMessageTest.java
deleted file mode 100644
index 5bc1a07..0000000
--- a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ChangeSentMessageTest.java
+++ /dev/null
@@ -1,62 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.activemq.usecases;
-
-import java.util.HashMap;
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.ObjectMessage;
-import javax.jms.Session;
-
-import org.apache.activemq.test.TestSupport;
-
-public class ChangeSentMessageTest extends TestSupport {
-    private static final int COUNT = 200;
-    private static final String VALUE_NAME = "value";
-
-    /**
-     * test Object messages can be changed after sending with no side-affects
-     *
-     * @throws Exception
-     */
-    public void testDoChangeSentMessage() throws Exception {
-        Destination destination = createDestination("test-" + ChangeSentMessageTest.class.getName());
-        Connection connection = createConnection();
-        connection.start();
-        Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        MessageConsumer consumer = consumerSession.createConsumer(destination);
-        Session publisherSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        MessageProducer producer = publisherSession.createProducer(destination);
-        HashMap<String, Integer> map = new HashMap<String, Integer>();
-        ObjectMessage message = publisherSession.createObjectMessage();
-        for (int i = 0; i < COUNT; i++) {
-            map.put(VALUE_NAME, Integer.valueOf(i));
-            message.setObject(map);
-            producer.send(message);
-            assertTrue(message.getObject() == map);
-        }
-        for (int i = 0; i < COUNT; i++) {
-            ObjectMessage msg = (ObjectMessage)consumer.receive();
-            HashMap receivedMap = (HashMap)msg.getObject();
-            Integer intValue = (Integer)receivedMap.get(VALUE_NAME);
-            assertTrue(intValue.intValue() == i);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ChangeSessionDeliveryModeTest.java
----------------------------------------------------------------------
diff --git a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ChangeSessionDeliveryModeTest.java b/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ChangeSessionDeliveryModeTest.java
deleted file mode 100644
index b4da7a7..0000000
--- a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ChangeSessionDeliveryModeTest.java
+++ /dev/null
@@ -1,72 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.activemq.usecases;
-
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.IllegalStateException;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageListener;
-import javax.jms.Session;
-
-import org.apache.activemq.test.TestSupport;
-
-public class ChangeSessionDeliveryModeTest extends TestSupport implements MessageListener {
-
-    /**
-     * test following condition- which are defined by JMS Spec 1.1:
-     * MessageConsumers cannot use a MessageListener and receive() from the same
-     * session
-     *
-     * @throws Exception
-     */
-    public void testDoChangeSessionDeliveryMode() throws Exception {
-        Destination destination = createDestination("foo.bar");
-        Connection connection = createConnection();
-        connection.start();
-        Session consumerSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-
-        /*
-        MessageConsumer consumer1 = consumerSession.createConsumer(destination);
-        consumer1.setMessageListener(this);
-        JMSException jmsEx = null;
-        MessageConsumer consumer2 = consumerSession.createConsumer(destination);
-
-        try {
-            consumer2.receive(10);
-            fail("Did not receive expected exception.");
-        } catch (JMSException e) {
-            assertTrue(e instanceof IllegalStateException);
-        }
-        */
-        MessageConsumer consumer1 = consumerSession.createConsumer(destination);
-        consumer1.setMessageListener(this);
-
-        try {
-            consumer1.receive(10);
-            fail("Did not receive expected exception.");
-        } catch (JMSException e) {
-            assertTrue(e instanceof IllegalStateException);
-        }
-    }
-
-    public void onMessage(Message msg) {
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/CompositeConsumeTest.java
----------------------------------------------------------------------
diff --git a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/CompositeConsumeTest.java b/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/CompositeConsumeTest.java
deleted file mode 100644
index 7ea8cc0..0000000
--- a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/CompositeConsumeTest.java
+++ /dev/null
@@ -1,73 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.usecases;
-
-import javax.jms.Destination;
-import javax.jms.Message;
-
-
-import org.apache.activemq.test.JmsTopicSendReceiveWithTwoConnectionsTest;
-
-import org.apache.hedwig.jms.SessionImpl;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class CompositeConsumeTest extends JmsTopicSendReceiveWithTwoConnectionsTest {
-    private static final Logger LOG = LoggerFactory.getLogger(CompositeConsumeTest.class);
-
-    public void testSendReceive() throws Exception {
-        messages.clear();
-
-        Destination[] destinations = getDestinations();
-        int destIdx = 0;
-
-        for (int i = 0; i < data.length; i++) {
-            Message message = session.createTextMessage(data[i]);
-
-            if (verbose) {
-                LOG.info("About to send a message: " + message + " with text: " + data[i]);
-            }
-
-            producer.send(destinations[destIdx], message);
-
-            if (++destIdx >= destinations.length) {
-                destIdx = 0;
-            }
-        }
-
-        assertMessagesAreReceived();
-    }
-
-    /**
-     * Returns the subscription subject
-     */
-    protected String getSubject() {
-        // return getPrefix() + "FOO.BAR," + getPrefix() + "FOO.X.Y," + getPrefix() + "BAR.>";
-        return getPrefix() + "FOO.BAR";
-    }
-
-    /**
-     * Returns the destinations on which we publish
-     */
-    protected Destination[] getDestinations() {
-        return new Destination[]{SessionImpl.asTopic(getSubject())};
-    }
-
-    protected String getPrefix() {
-        return super.getSubject() + ".";
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/CompositePublishTest.java
----------------------------------------------------------------------
diff --git a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/CompositePublishTest.java b/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/CompositePublishTest.java
deleted file mode 100644
index 7833e01..0000000
--- a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/CompositePublishTest.java
+++ /dev/null
@@ -1,145 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.usecases;
-
-import java.util.List;
-
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageListener;
-import javax.jms.Session;
-
-import org.apache.hedwig.jms.SessionImpl;
-import org.apache.hedwig.jms.spi.HedwigConnectionFactoryImpl;
-
-import org.apache.activemq.test.JmsSendReceiveTestSupport;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class CompositePublishTest extends JmsSendReceiveTestSupport {
-    private static final Logger LOG = LoggerFactory.getLogger(CompositePublishTest.class);
-
-    protected Connection sendConnection;
-    protected Connection receiveConnection;
-    protected Session receiveSession;
-    protected MessageConsumer[] consumers;
-    protected List[] messageLists;
-
-    @SuppressWarnings("unchecked")
-    protected void setUp() throws Exception {
-        super.setUp();
-
-        connectionFactory = createConnectionFactory();
-
-        sendConnection = createConnection(false);
-        sendConnection.start();
-
-        receiveConnection = createConnection(false);
-        receiveConnection.start();
-
-        LOG.info("Created sendConnection: " + sendConnection);
-        LOG.info("Created receiveConnection: " + receiveConnection);
-
-        session = sendConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        receiveSession = receiveConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-
-        LOG.info("Created sendSession: " + session);
-        LOG.info("Created receiveSession: " + receiveSession);
-
-        producer = session.createProducer(null);
-
-        LOG.info("Created producer: " + producer);
-
-        consumerDestination = session.createTopic(getConsumerSubject());
-        producerDestination = session.createTopic(getProducerSubject());
-
-        LOG.info("Created  consumer destination: " + consumerDestination
-                 + " of type: " + consumerDestination.getClass());
-        LOG.info("Created  producer destination: " + producerDestination
-                 + " of type: " + producerDestination.getClass());
-
-        Destination[] destinations = getDestinations();
-        consumers = new MessageConsumer[destinations.length];
-        messageLists = new List[destinations.length];
-        for (int i = 0; i < destinations.length; i++) {
-            Destination dest = destinations[i];
-            messageLists[i] = createConcurrentList();
-            consumers[i] = receiveSession.createConsumer(dest);
-            consumers[i].setMessageListener(createMessageListener(i, messageLists[i]));
-        }
-
-        LOG.info("Started connections");
-    }
-
-    protected MessageListener createMessageListener(int i, final List<Message> messageList) {
-        return new MessageListener() {
-            public void onMessage(Message message) {
-                consumeMessage(message, messageList);
-            }
-        };
-    }
-
-    /**
-     * Returns the subject on which we publish
-     */
-    protected String getSubject() {
-        // return getPrefix() + "FOO.BAR," + getPrefix() + "FOO.X.Y";
-        return getPrefix() + "FOO.BAR";
-    }
-
-    /**
-     * Returns the destinations to which we consume
-     */
-    protected Destination[] getDestinations() {
-        // return new Destination[] {SessionImpl.asTopic(getPrefix() + "FOO.BAR"),
-        // SessionImpl.asTopic(getPrefix() + "FOO.*"), SessionImpl.asTopic(getPrefix() + "FOO.X.Y")};
-        return new Destination[] {SessionImpl.asTopic(getPrefix() + "FOO.BAR")};
-    }
-
-    protected String getPrefix() {
-        return super.getSubject() + ".";
-    }
-
-    @SuppressWarnings("unchecked")
-    protected void assertMessagesAreReceived() throws JMSException {
-        waitForMessagesToBeDelivered();
-        int size = messageLists.length;
-        for (int i = 0; i < size; i++) {
-            LOG.info("Message list: " + i + " contains: " + messageLists[i].size() + " message(s)");
-        }
-        size = messageLists.length;
-        for (int i = 0; i < size; i++) {
-            assertMessagesReceivedAreValid(messageLists[i]);
-        }
-    }
-
-    protected HedwigConnectionFactoryImpl createConnectionFactory() {
-        return new HedwigConnectionFactoryImpl();
-    }
-
-    protected void tearDown() throws Exception {
-        session.close();
-        receiveSession.close();
-
-        sendConnection.close();
-        receiveConnection.close();
-        super.tearDown();
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ConcurrentProducerDurableConsumerTest.java
----------------------------------------------------------------------
diff --git a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ConcurrentProducerDurableConsumerTest.java b/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ConcurrentProducerDurableConsumerTest.java
deleted file mode 100644
index 087c5ef..0000000
--- a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/ConcurrentProducerDurableConsumerTest.java
+++ /dev/null
@@ -1,413 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.usecases;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-import javax.jms.DeliveryMode;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageListener;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-import javax.jms.TopicSubscriber;
-import junit.framework.Test;
-import org.apache.hedwig.jms.spi.HedwigConnectionFactoryImpl;
-import org.apache.activemq.TestSupport;
-
-import org.apache.activemq.util.MessageIdList;
-import org.apache.activemq.util.Wait;
-//import org.apache.commons.dbcp.BasicDataSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class ConcurrentProducerDurableConsumerTest extends TestSupport {
-    private static final Logger LOG = LoggerFactory.getLogger(ConcurrentProducerDurableConsumerTest.class);
-    private int consumerCount = 5;
-    protected List<Connection> connections = Collections.synchronizedList(new ArrayList<Connection>());
-    protected Map<MessageConsumer, TimedMessageListener> consumers
-        = new HashMap<MessageConsumer, TimedMessageListener>();
-    protected MessageIdList allMessagesList = new MessageIdList();
-    private int messageSize = 1024;
-
-    public void testSendRateWithActivatingConsumers() throws Exception {
-        final Destination destination = createDestination();
-        final ConnectionFactory factory = createConnectionFactory();
-        startInactiveConsumers(factory, destination);
-
-        Connection connection = factory.createConnection();
-        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        MessageProducer producer = createMessageProducer(session, destination);
-
-        // preload the durable consumers
-        double[] inactiveConsumerStats = produceMessages(destination, 500, 10, session, producer, null);
-        LOG.info("With inactive consumers: ave: " + inactiveConsumerStats[1]
-                + ", max: " + inactiveConsumerStats[0] + ", multiplier: "
-                 + (inactiveConsumerStats[0]/inactiveConsumerStats[1]));
-
-        // periodically start a durable sub that has a backlog
-        final int consumersToActivate = 5;
-        final Object addConsumerSignal = new Object();
-        Executors.newCachedThreadPool(new ThreadFactory() {
-            @Override
-            public Thread newThread(Runnable r) {
-                return new Thread(r, "ActivateConsumer" + this);
-            }
-        }).execute(new Runnable() {
-            @Override
-            public void run() {
-                try {
-                    MessageConsumer consumer = null;
-                    for (int i = 0; i < consumersToActivate; i++) {
-                        LOG.info("Waiting for add signal from producer...");
-                        synchronized (addConsumerSignal) {
-                            addConsumerSignal.wait(30 * 60 * 1000);
-                        }
-                        TimedMessageListener listener = new TimedMessageListener();
-                        consumer = createDurableSubscriber(factory.createConnection(),
-                                                           destination, "consumer" + (i + 1));
-                        LOG.info("Created consumer " + consumer);
-                        consumer.setMessageListener(listener);
-                        consumers.put(consumer, listener);
-                    }
-                } catch (Exception e) {
-                    LOG.error("failed to start consumer", e);
-                }
-            }
-        });
-
-
-        double[] statsWithActive = produceMessages(destination, 500, 10, session, producer, addConsumerSignal);
-
-        LOG.info(" with concurrent activate, ave: " + statsWithActive[1]
-                 + ", max: " + statsWithActive[0] + ", multiplier: " + (statsWithActive[0]/ statsWithActive[1]));
-
-        while(consumers.size() < consumersToActivate) {
-            TimeUnit.SECONDS.sleep(2);
-        }
-
-        long timeToFirstAccumulator = 0;
-        for (TimedMessageListener listener : consumers.values()) {
-            long time = listener.getFirstReceipt();
-            timeToFirstAccumulator += time;
-            LOG.info("Time to first " + time);
-        }
-        LOG.info("Ave time to first message =" + timeToFirstAccumulator/consumers.size());
-
-        for (TimedMessageListener listener : consumers.values()) {
-            LOG.info("Ave batch receipt time: " + listener.waitForReceivedLimit(10000)
-                     + " max receipt: " + listener.maxReceiptTime);
-        }
-
-        //assertTrue("max (" + statsWithActive[0] + ") within reasonable
-        // multiplier of ave (" + statsWithActive[1] + ")",
-        //        statsWithActive[0] < 5 * statsWithActive[1]);
-
-        // compare no active to active
-        LOG.info("Ave send time with active: " + statsWithActive[1]
-                + " as multiplier of ave with none active: " + inactiveConsumerStats[1]
-                + ", multiplier=" + (statsWithActive[1]/inactiveConsumerStats[1]));
-
-        assertTrue("Ave send time with active: " + statsWithActive[1]
-                + " within reasonable multpler of ave with none active: " + inactiveConsumerStats[1]
-                + ", multiplier " + (statsWithActive[1]/inactiveConsumerStats[1]),
-                statsWithActive[1] < 15 * inactiveConsumerStats[1]);
-    }
-
-
-    public void x_testSendWithInactiveAndActiveConsumers() throws Exception {
-        Destination destination = createDestination();
-        ConnectionFactory factory = createConnectionFactory();
-        startInactiveConsumers(factory, destination);
-
-        Connection connection = factory.createConnection();
-        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        MessageProducer producer = session.createProducer(destination);
-        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
-
-        final int toSend = 100;
-        final int numIterations = 5;
-
-        double[] noConsumerStats = produceMessages(destination, toSend, numIterations, session, producer, null);
-
-        startConsumers(factory, destination);
-        LOG.info("Activated consumer");
-
-        double[] withConsumerStats = produceMessages(destination, toSend, numIterations, session, producer, null);
-
-        LOG.info("With consumer: " + withConsumerStats[1] + " , with noConsumer: " + noConsumerStats[1]
-                + ", multiplier: " + (withConsumerStats[1]/noConsumerStats[1]));
-        final int reasonableMultiplier = 15; // not so reasonable but improving
-        assertTrue("max X times as slow with consumer: " + withConsumerStats[1] + ", with no Consumer: "
-                + noConsumerStats[1] + ", multiplier: " + (withConsumerStats[1]/noConsumerStats[1]),
-                withConsumerStats[1] < noConsumerStats[1] * reasonableMultiplier);
-
-        final int toReceive = toSend * numIterations * consumerCount * 2;
-        Wait.waitFor(new Wait.Condition() {
-            public boolean isSatisified() throws Exception {
-                LOG.info("count: " + allMessagesList.getMessageCount());
-                return toReceive == allMessagesList.getMessageCount();
-            }
-        }, 60 * 1000);
-
-        assertEquals("got all messages", toReceive, allMessagesList.getMessageCount());
-    }
-
-
-    private MessageProducer createMessageProducer(Session session, Destination destination) throws JMSException {
-        MessageProducer producer = session.createProducer(destination);
-        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
-        return producer;
-    }
-
-
-    private void startInactiveConsumers(ConnectionFactory factory, Destination destination) throws Exception {
-        // create off line consumers
-        startConsumers(factory, destination);
-        for (Connection connection: connections) {
-            connection.close();
-        }
-        connections.clear();
-        consumers.clear();
-    }
-
-
-    protected void startConsumers(ConnectionFactory factory, Destination dest) throws Exception {
-        MessageConsumer consumer;
-        for (int i = 0; i < consumerCount; i++) {
-            TimedMessageListener list = new TimedMessageListener();
-            consumer = createDurableSubscriber(factory.createConnection(), dest, "consumer" + (i + 1));
-            consumer.setMessageListener(list);
-            consumers.put(consumer, list);
-        }
-    }
-
-    protected TopicSubscriber createDurableSubscriber(Connection conn,
-                                                      Destination dest, String name) throws Exception {
-        conn.setClientID(name);
-        connections.add(conn);
-        conn.start();
-
-        Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        final TopicSubscriber consumer = sess.createDurableSubscriber((javax.jms.Topic)dest, name);
-
-        return consumer;
-    }
-
-    /**
-     * @return max and ave send time
-     * @throws Exception
-     */
-    private double[] produceMessages(Destination destination,
-                                     final int toSend,
-                                     final int numIterations,
-                                     Session session,
-                                     MessageProducer producer,
-                                     Object addConsumerSignal) throws Exception {
-        long start;
-        long count = 0;
-        double batchMax = 0, max = 0, sum = 0;
-        for (int i=0; i<numIterations; i++) {
-            start = System.currentTimeMillis();
-            for (int j=0; j < toSend; j++) {
-                long singleSendstart = System.currentTimeMillis();
-                TextMessage msg = createTextMessage(session, "" + j);
-                // rotate
-                int priority = ((int)count%10);
-                producer.send(msg, DeliveryMode.PERSISTENT, priority, 0);
-                max = Math.max(max, (System.currentTimeMillis() - singleSendstart));
-                if (++count % 500 == 0) {
-                    if (addConsumerSignal != null) {
-                        synchronized (addConsumerSignal) {
-                            addConsumerSignal.notifyAll();
-                            LOG.info("Signalled add consumer");
-                        }
-                    }
-                }
-                ;
-                if (count % 5000 == 0) {
-                    LOG.info("Sent " + count + ", singleSendMax:" + max);
-                }
-
-            }
-            long duration = System.currentTimeMillis() - start;
-            batchMax = Math.max(batchMax, duration);
-            sum += duration;
-            LOG.info("Iteration " + i + ", sent " + toSend + ", time: "
-                    + duration + ", batchMax:" + batchMax + ", singleSendMax:" + max);
-        }
-
-        LOG.info("Sent: " + toSend * numIterations + ", batchMax: " + batchMax + " singleSendMax: " + max);
-        return new double[]{batchMax, sum/numIterations};
-    }
-
-    protected TextMessage createTextMessage(Session session, String initText) throws Exception {
-        TextMessage msg = session.createTextMessage();
-
-        // Pad message text
-        if (initText.length() < messageSize) {
-            char[] data = new char[messageSize - initText.length()];
-            Arrays.fill(data, '*');
-            String str = new String(data);
-            msg.setText(initText + str);
-
-            // Do not pad message text
-        } else {
-            msg.setText(initText);
-        }
-
-        return msg;
-    }
-
-    @Override
-    protected void setUp() throws Exception {
-        topic = true;
-        super.setUp();
-    }
-
-    @Override
-    protected void tearDown() throws Exception {
-        for (Iterator<Connection> iter = connections.iterator(); iter.hasNext();) {
-            Connection conn = iter.next();
-            try {
-                conn.close();
-            } catch (Throwable e) {
-            }
-        }
-        allMessagesList.flushMessages();
-        consumers.clear();
-        super.tearDown();
-    }
-
-
-    protected HedwigConnectionFactoryImpl createConnectionFactory() throws Exception {
-        HedwigConnectionFactoryImpl factory = new HedwigConnectionFactoryImpl();
-        return factory;
-    }
-
-    public static Test suite() {
-        return suite(ConcurrentProducerDurableConsumerTest.class);
-    }
-
-    class TimedMessageListener implements MessageListener {
-        final int batchSize = 1000;
-        CountDownLatch firstReceiptLatch = new CountDownLatch(1);
-        long mark = System.currentTimeMillis();
-        long firstReceipt = 0l;
-        long receiptAccumulator = 0;
-        long batchReceiptAccumulator = 0;
-        long maxReceiptTime = 0;
-        AtomicLong count = new AtomicLong(0);
-        Map<Integer, MessageIdList> messageLists
-            = new ConcurrentHashMap<Integer, MessageIdList>(new HashMap<Integer, MessageIdList>());
-
-        @Override
-        public void onMessage(Message message) {
-            final long current = System.currentTimeMillis();
-            final long duration = current - mark;
-            receiptAccumulator += duration;
-            int priority = 0;
-            try {
-                priority = message.getJMSPriority();
-            } catch (JMSException ignored) {}
-            if (!messageLists.containsKey(priority)) {
-                messageLists.put(priority, new MessageIdList());
-            }
-            messageLists.get(priority).onMessage(message);
-            if (count.incrementAndGet() == 1) {
-                firstReceipt = duration;
-                firstReceiptLatch.countDown();
-                LOG.info("First receipt in " + firstReceipt + "ms");
-            } else if (count.get() % batchSize == 0) {
-                LOG.info("Consumed " + count.get() + " in "
-                         + batchReceiptAccumulator + "ms" + ", priority:" + priority);
-                batchReceiptAccumulator=0;
-            }
-            maxReceiptTime = Math.max(maxReceiptTime, duration);
-            receiptAccumulator += duration;
-            batchReceiptAccumulator += duration;
-            mark = current;
-        }
-
-        long getMessageCount() {
-            return count.get();
-        }
-
-        long getFirstReceipt() throws Exception {
-            firstReceiptLatch.await(30, TimeUnit.SECONDS);
-            return firstReceipt;
-        }
-
-        public long waitForReceivedLimit(long limit) throws Exception {
-            final long expiry = System.currentTimeMillis() + 30*60*1000;
-            while (count.get() < limit) {
-                if (System.currentTimeMillis() > expiry) {
-                    throw new RuntimeException("Expired waiting for X messages, " + limit);
-                }
-                TimeUnit.SECONDS.sleep(2);
-                String missing = findFirstMissingMessage();
-                if (missing != null) {
-                    LOG.info("first missing = " + missing);
-                    throw new RuntimeException("We have a missing message. " + missing);
-                }
-
-            }
-            return receiptAccumulator/(limit/batchSize);
-        }
-
-        private String findFirstMissingMessage() {
-            /*
-            MessageId current = new MessageId();
-            for (MessageIdList priorityList : messageLists.values()) {
-                MessageId previous = null;
-                for (String id : priorityList.getMessageIds()) {
-                    current.setValue(id);
-                    if (previous == null) {
-                        previous = current.copy();
-                    } else {
-                        if (current.getProducerSequenceId() - 1 != previous.getProducerSequenceId() &&
-                            current.getProducerSequenceId() - 10 !=  previous.getProducerSequenceId()) {
-                                return "Missing next after: " + previous + ", got: " + current;
-                        } else {
-                            previous = current.copy();
-                        }
-                    }
-                }
-            }
-            return null;
-            */
-            return null;
-        }
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DiscriminatingConsumerLoadTest.java
----------------------------------------------------------------------
diff --git a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DiscriminatingConsumerLoadTest.java b/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DiscriminatingConsumerLoadTest.java
deleted file mode 100644
index b14ef71..0000000
--- a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DiscriminatingConsumerLoadTest.java
+++ /dev/null
@@ -1,322 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.usecases;
-
-
-import javax.jms.Topic;
-import javax.jms.Connection;
-import javax.jms.DeliveryMode;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-
-import org.apache.hedwig.jms.spi.HedwigConnectionFactoryImpl;
-import org.apache.activemq.JmsConnectionStartStopTest;
-
-
-
-
-/**
- * Test case intended to demonstrate delivery interruption to queue consumers when
- * a JMS selector leaves some messages on the queue (due to use of a JMS Selector)
- *
- * testNonDiscriminatingConsumer() demonstrates proper functionality for consumers that don't use
- * a selector to qualify their input.
- *
- * testDiscriminatingConsumer() demonstrates the failure condition in which delivery to the consumer
- * eventually halts.
- *
- * The expected behavior is for the delivery to the client to be maintained regardless of the depth
- * of the queue, particularly when the messages in the queue do not meet the selector criteria of the
- * client.
- *
- * https://issues.apache.org/activemq/browse/AMQ-2217
- *
- */
-public class DiscriminatingConsumerLoadTest extends TestSupport {
-
-    private static final org.apache.commons.logging.Log LOG = org.apache.commons.logging.LogFactory
-        .getLog(DiscriminatingConsumerLoadTest.class);
-
-    private Connection producerConnection;
-    private Connection consumerConnection;
-    private int counterSent = 0;
-    private int counterReceived = 0;
-
-    public static final String JMSTYPE_EATME  = "DiscriminatingLoadClient.EatMe";
-    public static final String JMSTYPE_IGNOREME = "DiscriminatingLoadClient.IgnoreMe";
-
-    private int testSize = 5000; // setting this to a small number will pass all tests
-
-
-    protected void setUp() throws Exception {
-        super.setUp();
-        this.producerConnection = this.createConnection();
-        this.consumerConnection = this.createConnection();
-    }
-
-    /**
-     * @see junit.framework.TestCase#tearDown()
-     */
-    protected void tearDown() throws Exception {
-        if (producerConnection != null) {
-            producerConnection.close();
-            producerConnection = null;
-        }
-        if (consumerConnection != null) {
-            consumerConnection.close();
-            consumerConnection = null;
-        }
-        super.tearDown();
-    }
-
-    /**
-     * Test to check if a single consumer with no JMS selector will receive all intended messages
-     *
-     * @throws java.lang.Exception
-     */
-    public void testNonDiscriminatingConsumer() throws Exception {
-        consumerConnection = createConnection();
-        consumerConnection.start();
-        LOG.info("consumerConnection = " +consumerConnection);
-
-        try {Thread.sleep(1000); } catch (Exception e) {}
-
-        // here we pass in null for the JMS selector
-        Consumer consumer = new Consumer(consumerConnection, null);
-        Thread consumerThread = new Thread(consumer);
-
-        consumerThread.start();
-
-        producerConnection = createConnection();
-        producerConnection.start();
-        LOG.info("producerConnection = " +producerConnection);
-
-        try {Thread.sleep(3000); } catch (Exception e) {}
-
-        Producer producer = new Producer(producerConnection);
-        Thread producerThread = new Thread(producer);
-        producerThread.start();
-
-        // now that everything is running, let's wait for the consumer thread to finish ...
-        consumerThread.join();
-        producer.stop = true;
-
-        if (consumer.getCount() == testSize )
-            LOG.info("test complete .... all messsages consumed!!");
-        else
-            LOG.info("test failed .... Sent " + (testSize / 1) +
-                     " messages intended to be consumed ( " + testSize
-                     + " total), but only consumed " + consumer.getCount());
-
-
-        assertTrue("Sent " + testSize + " messages intended to be consumed, but only consumed " + consumer.getCount(),
-                   (consumer.getCount() == testSize ));
-        assertFalse("Delivery of messages to consumer was halted during this test", consumer.deliveryHalted());
-
-
-    }
-
-    /**
-     * Test to check if a single consumer with a JMS selector will receive all intended messages
-     *
-     * @throws java.lang.Exception
-     */
-    public void testDiscriminatingConsumer() throws Exception {
-
-        consumerConnection = createConnection();
-        consumerConnection.start();
-        LOG.info("consumerConnection = " +consumerConnection);
-
-        try {Thread.sleep(1000); } catch (Exception e) {}
-
-        // here we pass the JMS selector we intend to consume
-        Consumer consumer = new Consumer(consumerConnection, JMSTYPE_EATME);
-        Thread consumerThread = new Thread(consumer);
-
-        consumerThread.start();
-
-        producerConnection = createConnection();
-        producerConnection.start();
-        LOG.info("producerConnection = " +producerConnection);
-
-        try {Thread.sleep(3000); } catch (Exception e) {}
-
-        Producer producer = new Producer(producerConnection);
-        Thread producerThread = new Thread(producer);
-        producerThread.start();
-
-        // now that everything is running, let's wait for the consumer thread to finish ...
-        consumerThread.join();
-        producer.stop = true;
-
-        if (consumer.getCount() == (testSize / 2))
-        {
-            LOG.info("test complete .... all messsages consumed!!");
-        }
-        else
-            {
-                LOG.info("test failed .... Sent " + testSize
-                         + " original messages, only half of which (" + (testSize / 2) +
-                         ") were intended to be consumed: consumer paused at: " + consumer.getCount());
-                //System.out.println("test failed .... Sent " + testSize
-                //+ " original messages, only half of which (" + (testSize / 2) +
-                // ") were intended to be consumed: consumer paused at: " + consumer.getCount());
-            }
-        assertTrue("Sent " + testSize  + " original messages, only half of which (" + (testSize / 2) +
-                   ") were intended to be consumed: consumer paused at: " + consumer.getCount(),
-                   (consumer.getCount() == (testSize / 2)));
-        assertTrue("Delivery of messages to consumer was halted during this test as it only wants half",
-                   consumer.deliveryHalted());
-    }
-
-    /**
-     * Helper class that will publish 2 * testSize messages.  The messages will be distributed evenly
-     * between the following two JMS types:
-     *
-     * @see JMSTYPE_INTENDED_FOR_CONSUMPTION
-     * @see JMSTYPE_NOT_INTENDED_FOR_CONSUMPTION
-     *
-     */
-    private class Producer extends Thread
-    {
-        private int counterSent = 0;
-        private Connection connection = null;
-        public boolean stop = false;
-
-        public Producer(Connection connection)
-        {
-            this.connection = connection;
-        }
-
-        public void run() {
-            try {
-                final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-                final Topic queue = session.createTopic("test");
-
-                // wait for 10 seconds to allow consumer.receive to be run
-                // first
-                Thread.sleep(10000);
-                MessageProducer producer = session.createProducer(queue);
-
-                while (!stop && (counterSent < testSize))
-                    {
-                        // first send a message intended to be consumed ....
-                        TextMessage message = session.createTextMessage("*** Ill ....... Ini ***");  // alma mater ...
-                        message.setJMSType(JMSTYPE_EATME);
-                        //LOG.info("sending .... JMSType = " + message.getJMSType());
-                        producer.send(message,DeliveryMode.NON_PERSISTENT,0,1800000);
-
-                        counterSent++;
-
-                        // now send a message intended to be consumed by some other consumer in the the future
-                        // ... we expect these messages to accrue in the queue
-                        message = session.createTextMessage("*** Ill ....... Ini ***");  // alma mater ...
-                        message.setJMSType(JMSTYPE_IGNOREME);
-                        //LOG.info("sending .... JMSType = " + message.getJMSType());
-                        producer.send(message,DeliveryMode.NON_PERSISTENT,0,1800000);
-
-                        counterSent++;
-                    }
-
-                session.close();
-
-            } catch (Exception e) {
-                e.printStackTrace();
-            }
-            LOG.info("producer thread complete ... " + counterSent + " messages sent to the queue");
-        }
-
-        public int getCount()
-        {
-            return this.counterSent;
-        }
-    }
-
-    /**
-     * Helper class that will consume messages from the queue based on the supplied JMS selector.
-     * Thread will stop after the first receive(..) timeout, or once all expected messages have
-     * been received (see testSize).  If the thread stops due to a timeout, it is experiencing the
-     * delivery pause that is symptomatic of a bug in the broker.
-     */
-    private class Consumer extends Thread
-    {
-        protected int counterReceived = 0;
-        private Connection connection = null;
-        private String jmsSelector = null;
-        private boolean deliveryHalted = false;
-
-        public Consumer(Connection connection, String jmsSelector)
-        {
-            this.connection = connection;
-            this.jmsSelector = jmsSelector;
-        }
-
-        public void run() {
-            boolean testComplete = false;
-            try {
-                Session session = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-                final Topic queue = session.createTopic("test");
-                MessageConsumer consumer = null;
-                if (null != this.jmsSelector)
-                    {
-                        consumer = session.createConsumer(queue, "JMSType='" + this.jmsSelector + "'");
-                    }
-                else
-                    {
-                        consumer = session.createConsumer(queue);
-                    }
-
-                while (!deliveryHalted && (counterReceived < testSize))
-                    {
-                        TextMessage result = (TextMessage) consumer.receive(30000);
-                        if (result != null) {
-                            counterReceived++;
-                            //System.out.println("consuming .... JMSType = " + result.getJMSType()
-                            // + " received = " + counterReceived);
-                            LOG.info("consuming .... JMSType = " + result.getJMSType()
-                                     + " received = " + counterReceived);
-                        } else
-                            {
-                                LOG.info("consuming .... timeout while waiting for a message ."
-                                         + ".. broker must have stopped delivery ...  received = " + counterReceived);
-                                deliveryHalted = true;
-                            }
-                    }
-                session.close();
-            } catch (Exception e) {
-                e.printStackTrace();
-            }
-
-        }
-
-        public int getCount()
-        {
-            return this.counterReceived;
-        }
-
-        public boolean deliveryHalted()
-        {
-            return this.deliveryHalted;
-        }
-    }
-
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DispatchMultipleConsumersTest.java
----------------------------------------------------------------------
diff --git a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DispatchMultipleConsumersTest.java b/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DispatchMultipleConsumersTest.java
deleted file mode 100644
index fa79d69..0000000
--- a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DispatchMultipleConsumersTest.java
+++ /dev/null
@@ -1,214 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.usecases;
-
-import javax.jms.Topic;
-import java.util.concurrent.CountDownLatch;
-
-import org.apache.hedwig.JmsTestBase;
-import org.apache.hedwig.jms.SessionImpl;
-import java.util.concurrent.atomic.AtomicInteger;
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import junit.framework.TestCase;
-import org.apache.hedwig.jms.spi.HedwigConnectionFactoryImpl;
-
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class DispatchMultipleConsumersTest extends JmsTestBase {
-    private final static Logger logger = LoggerFactory.getLogger(DispatchMultipleConsumersTest.class);
-    Destination dest;
-    String destinationName = "TEST.Q";
-    String msgStr = "Test text message";
-    int messagesPerThread = 20;
-    int producerThreads = 50;
-    int consumerCount = 2;
-    AtomicInteger sentCount;
-    AtomicInteger consumedCount;
-    CountDownLatch producerLatch;
-    CountDownLatch consumerLatch;
-    String userName = "";
-    String password = "";
-
-    @Override
-    protected void setUp() throws Exception {
-        super.setUp();
-        dest = SessionImpl.asTopic(destinationName);
-    }
-
-    private void resetCounters() {
-        sentCount = new AtomicInteger(0);
-        consumedCount = new AtomicInteger(0);
-        producerLatch = new CountDownLatch(producerThreads);
-        consumerLatch = new CountDownLatch(consumerCount);
-    }
-
-    public void testDispatch1() {
-        for (int i = 1; i <= 5; i++) {
-            resetCounters();
-            dispatch();
-            assertEquals("Incorrect messages in Iteration " + i, sentCount.get() * consumerCount, consumedCount.get());
-        }
-    }
-
-    private void dispatch() {
-        startConsumers();
-        startProducers();
-        try {
-            producerLatch.await();
-            consumerLatch.await();
-        } catch (InterruptedException e) {
-            fail("test interrupted!");
-        }
-    }
-
-    private void startConsumers() {
-        HedwigConnectionFactoryImpl connFactory = new HedwigConnectionFactoryImpl();
-        Connection conn;
-        try {
-            conn = connFactory.createConnection(userName, password);
-            conn.start();
-            for (int i = 0; i < consumerCount; i++) {
-                ConsumerThread th = new ConsumerThread(conn, "ConsumerThread"+i);
-                th.start();
-            }
-        } catch (JMSException e) {
-            logger.error("Failed to start consumers", e);
-        }
-    }
-
-    private void startProducers() {
-        HedwigConnectionFactoryImpl connFactory = new HedwigConnectionFactoryImpl();
-        for (int i = 0; i < producerThreads; i++) {
-            Thread th = new ProducerThread(connFactory, messagesPerThread, "ProducerThread"+i);
-            th.start();
-        }
-    }
-
-    private class ConsumerThread extends Thread {
-        private Session session;
-        private MessageConsumer consumer;
-
-        public ConsumerThread(Connection conn, String name) {
-            super();
-            this.setName(name);
-            logger.trace("Created new consumer thread:" + name);
-            try {
-                session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-                consumer = session.createConsumer(dest);
-            } catch (JMSException e) {
-                logger.error("Failed to start consumer thread:" + name, e);
-            }
-        }
-
-        @Override
-        public void run() {
-            int msgCount = 0;
-            int nullCount = 0;
-            while (true) {
-                try {
-                    Message msg = consumer.receive(200);
-                    if (msg == null) {
-                        if (producerLatch.getCount() > 0) {
-                            continue;
-                        }
-                        nullCount++;
-                        if (nullCount > 10) {
-                            //assume that we are not getting any more messages
-                            break;
-                        } else {
-                            continue;
-                        }
-                    } else {
-                        nullCount = 0;
-                    }
-                    // Thread.sleep(100);
-                    if (logger.isTraceEnabled()) {
-                        logger.trace("Message received:" + msg.getJMSMessageID());
-                    }
-                    msgCount++;
-                } catch (JMSException e) {
-                    logger.error("Failed to consume:", e);
-                    /*
-                } catch (InterruptedException e) {
-                    logger.error("Interrupted!", e);
-                    */
-                }
-            }
-            try {
-                consumer.close();
-            } catch (JMSException e) {
-                logger.error("Failed to close consumer " + getName(), e);
-            }
-            consumedCount.addAndGet(msgCount);
-            consumerLatch.countDown();
-            logger.trace("Consumed " + msgCount + " messages using thread " + getName());
-        }
-    }
-
-    private class ProducerThread extends Thread {
-        private int count;
-        private Connection conn;
-        private Session session;
-        private MessageProducer producer;
-
-        public ProducerThread(HedwigConnectionFactoryImpl connFactory, int count, String name) {
-            super();
-            this.count = count;
-            this.setName(name);
-            logger.trace("Created new producer thread:" + name);
-            try {
-                conn = connFactory.createConnection();
-                conn.start();
-                session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
-                producer = session.createProducer(dest);
-            } catch (JMSException e) {
-                logger.error("Failed to start producer thread:" + name, e);
-            }
-        }
-
-        @Override
-        public void run() {
-            int i = 0;
-            try {
-                for (; i < count; i++) {
-                    producer.send(session.createTextMessage(msgStr));
-                    // Thread.sleep(500);
-                }
-                conn.close();
-            } catch (JMSException e) {
-                logger.error(e.getMessage(), e);
-                /*
-            } catch (InterruptedException e) {
-                logger.error("Interrupted!", e);
-                */
-            }
-            sentCount.addAndGet(i);
-            producerLatch.countDown();
-            if (logger.isTraceEnabled()) {
-                logger.trace("Sent " + i + " messages from thread " + getName());
-            }
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/bookkeeper/blob/9a8d62b1/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DurableConsumerCloseAndReconnectTest.java
----------------------------------------------------------------------
diff --git a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DurableConsumerCloseAndReconnectTest.java b/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DurableConsumerCloseAndReconnectTest.java
deleted file mode 100644
index c4fa74d..0000000
--- a/hedwig-client-jms/src/test/java/org/apache/activemq/usecases/DurableConsumerCloseAndReconnectTest.java
+++ /dev/null
@@ -1,192 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.usecases;
-
-import javax.jms.Connection;
-import javax.jms.DeliveryMode;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-import javax.jms.Topic;
-
-import org.apache.hedwig.jms.spi.HedwigConnectionFactoryImpl;
-import org.apache.activemq.test.TestSupport;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class DurableConsumerCloseAndReconnectTest extends TestSupport {
-    protected static final long RECEIVE_TIMEOUT = 5000L;
-    private static final Logger LOG = LoggerFactory.getLogger(DurableConsumerCloseAndReconnectTest.class);
-
-    protected Connection connection;
-    private Session session;
-    private MessageConsumer consumer;
-    private MessageProducer producer;
-    private Destination destination;
-    private int messageCount;
-
-    @Override
-    protected void setUp() throws Exception {
-        super.setUp();
-        deleteAllMessages();
-    }
-
-    @Override
-    protected void tearDown() throws Exception {
-        super.tearDown();
-        deleteAllMessages();
-    }
-
-    private void deleteAllMessages() throws Exception {
-        HedwigConnectionFactoryImpl fac = new HedwigConnectionFactoryImpl();
-        Connection dummyConnection = fac.createConnection();
-        dummyConnection.start();
-        dummyConnection.close();
-    }
-
-    protected HedwigConnectionFactoryImpl createConnectionFactory() throws Exception {
-        return new HedwigConnectionFactoryImpl();
-    }
-
-    public void testCreateDurableConsumerCloseThenReconnect() throws Exception {
-        // force the server to stay up across both connection tests
-        Connection dummyConnection = createConnection();
-        dummyConnection.start();
-
-        consumeMessagesDeliveredWhileConsumerClosed();
-
-        dummyConnection.close();
-
-        // now lets try again without one connection open
-        consumeMessagesDeliveredWhileConsumerClosed();
-    }
-
-    protected void consumeMessagesDeliveredWhileConsumerClosed() throws Exception {
-        makeConsumer();
-        closeConsumer();
-
-        publish();
-
-        // wait a few moments for the close to really occur
-        Thread.sleep(1000);
-
-        makeConsumer();
-
-        Message message = consumer.receive(RECEIVE_TIMEOUT);
-        assertTrue("Should have received a message!", message != null);
-
-        closeConsumer();
-
-        LOG.info("Now lets create the consumer again and because we didn't ack, we should get it again");
-        makeConsumer();
-
-        message = consumer.receive(RECEIVE_TIMEOUT);
-        assertTrue("Should have received a message!", message != null);
-        message.acknowledge();
-
-        closeConsumer();
-
-        LOG.info("Now lets create the consumer again and because we did ack, we should not get it again");
-        makeConsumer();
-
-        message = consumer.receive(2000);
-        assertTrue("Should have no more messages left!", message == null);
-
-        closeConsumer();
-
-        LOG.info("Lets publish one more message now");
-        publish();
-
-        makeConsumer();
-        message = consumer.receive(RECEIVE_TIMEOUT);
-        assertTrue("Should have received a message!", message != null);
-        message.acknowledge();
-
-        closeConsumer();
-    }
-
-    protected void publish() throws Exception {
-        connection = createConnection();
-        connection.start();
-
-        session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
-        destination = createDestination();
-
-        producer = session.createProducer(destination);
-        producer.setDeliveryMode(DeliveryMode.PERSISTENT);
-        TextMessage msg = session.createTextMessage("This is a test: " + messageCount++);
-        producer.send(msg);
-
-        producer.close();
-        producer = null;
-        closeSession();
-    }
-
-    protected Destination createDestination() throws JMSException {
-        if (isTopic()) {
-            return session.createTopic(getSubject());
-        } else {
-            return session.createTopic(getSubject());
-        }
-    }
-
-    protected boolean isTopic() {
-        return true;
-    }
-
-    protected void closeConsumer() throws JMSException {
-        consumer.close();
-        consumer = null;
-        closeSession();
-    }
-
-    protected void closeSession() throws JMSException {
-        session.close();
-        session = null;
-        connection.close();
-        connection = null;
-    }
-
-    protected void makeConsumer() throws Exception {
-        String durableName = getName();
-        String clientID = getSubject();
-        LOG.info("Creating a durable subscribe for clientID: " + clientID + " and durable name: " + durableName);
-        createSession(clientID);
-        consumer = createConsumer(durableName);
-    }
-
-    private MessageConsumer createConsumer(String durableName) throws JMSException {
-        if (destination instanceof Topic) {
-            return session.createDurableSubscriber((Topic)destination, durableName);
-        } else {
-            return session.createConsumer(destination);
-        }
-    }
-
-    protected void createSession(String clientID) throws Exception {
-        connection = createConnection();
-        connection.setClientID(clientID);
-        connection.start();
-
-        session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
-        destination = createDestination();
-    }
-}


Mime
View raw message