path: root/qpid/java/systests/src/main/java/org/apache/qpid/test/unit/client/connection/ExceptionListenerTest.java

/*
 *  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.qpid.test.unit.client.connection;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.jms.Connection;
import javax.jms.ExceptionListener;
import javax.jms.IllegalStateException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Queue;
import javax.jms.Session;

import org.apache.qpid.AMQConnectionClosedException;
import org.apache.qpid.client.AMQNoRouteException;
import org.apache.qpid.jms.ConnectionURL;
import org.apache.qpid.test.utils.QpidBrokerTestCase;
import org.apache.qpid.transport.ConnectionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ExceptionListenerTest extends QpidBrokerTestCase
{
    private static final Logger LOGGER = LoggerFactory.getLogger(ExceptionListenerTest.class);

    private volatile Throwable _lastExceptionListenerException = null;

    public void testExceptionListenerHearsBrokerShutdown() throws  Exception
    {
        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(1);
        final AtomicInteger exceptionCounter = new AtomicInteger(0);
        final ExceptionListener listener = new ExceptionListener()
        {
            public void onException(JMSException exception)
            {
                exceptionCounter.incrementAndGet();
                _lastExceptionListenerException = exception;
                exceptionReceivedLatch.countDown();
            }
        };

        Connection connection = getConnection();
        connection.setExceptionListener(listener);

        stopBroker();

        exceptionReceivedLatch.await(10, TimeUnit.SECONDS);

        assertEquals("Unexpected number of exceptions received", 1, exceptionCounter.intValue());
        LOGGER.debug("exception was", _lastExceptionListenerException);
        assertNotNull("Exception should have cause", _lastExceptionListenerException.getCause());
        Class<? extends Exception> expectedExceptionClass = isBroker010() ? ConnectionException.class : AMQConnectionClosedException.class;
        assertEquals(expectedExceptionClass, _lastExceptionListenerException.getCause().getClass());
    }

    /**
     * It is reasonable for an application to perform Connection#close within the exception
     * listener.  This test verifies that close is allowed, and proceeds without generating
     * further exceptions.
     */
    public void testExceptionListenerClosesConnection_IsAllowed() throws  Exception
    {
        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(1);
        final Connection connection = getConnection();
        final ExceptionListener listener = new ExceptionListener()
        {
            public void onException(JMSException exception)
            {
                try
                {
                    connection.close();
                    // PASS
                }
                catch (Throwable t)
                {
                    _lastExceptionListenerException = t;
                }
                finally
                {
                    exceptionReceivedLatch.countDown();
                }
            }
        };
        connection.setExceptionListener(listener);


        stopBroker();

        boolean exceptionReceived = exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
        assertTrue("Exception listener did not hear exception within timeout", exceptionReceived);
        assertNull("Connection#close() should not have thrown exception", _lastExceptionListenerException);
    }

    /**
     * Spring's SingleConnectionFactory installs an ExceptionListener that calls stop()
     * and ignores any IllegalStateException that result.  This test serves to test this
     * scenario.
     */
    public void testExceptionListenerStopsConnection_ThrowsIllegalStateException() throws  Exception
    {
        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(1);
        final Connection connection = getConnection();
        final ExceptionListener listener = new ExceptionListener()
        {
            public void onException(JMSException exception)
            {
                try
                {
                    connection.stop();
                    fail("Exception not thrown");
                }
                catch (IllegalStateException ise)
                {
                    // PASS
                }
                catch (Throwable t)
                {
                    _lastExceptionListenerException = t;
                }
                finally
                {
                    exceptionReceivedLatch.countDown();
                }
            }
        };
        connection.setExceptionListener(listener);

        stopBroker();

        boolean exceptionReceived = exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
        assertTrue("Exception listener did not hear exception within timeout", exceptionReceived);
        assertNull("Connection#stop() should not have thrown unexpected exception", _lastExceptionListenerException);
    }

    /**
     * This test reproduces a deadlock that was the subject of a support call. A Spring based
     * application was using SingleConnectionFactory.  It installed an ExceptionListener that
     * stops and closes the connection in response to any exception.  On receipt of a message
     * the application would create a new session then send a response message (within onMessage).
     * It appears that a misconfiguration in the application meant that some of these messages
     * were bounced (no-route). Bounces are treated like connection exceptions and are passed
     * back to the application via the ExceptionListener.  The deadlock occurred between the
     * ExceptionListener's call to stop() and the MessageListener's attempt to create a new
     * session.
     */
    public void testExceptionListenerConnectionStopDeadlock() throws  Exception
    {
        Queue messageQueue = getTestQueue();

        Map<String, String> options = new HashMap<String, String>();
        options.put(ConnectionURL.OPTIONS_CLOSE_WHEN_NO_ROUTE, Boolean.toString(false));

        final Connection connection = getConnectionWithOptions(options);

        Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
        session.createConsumer(messageQueue).close(); // Create queue by side-effect

        // Put 10 messages onto messageQueue
        sendMessage(session, messageQueue, 10);

        // Install an exception listener that stops/closes the connection on receipt of 2nd AMQNoRouteException.
        // (Triggering on the 2nd (rather than 1st) seems to increase the probability that the test ends in deadlock,
        // at least on my machine).
        final CountDownLatch exceptionReceivedLatch  = new CountDownLatch(2);
        final ExceptionListener listener = new ExceptionListener()
        {
            public void onException(JMSException exception)
            {
                try
                {
                    assertNotNull("JMS Exception must have cause", exception.getCause() );
                    assertEquals("JMS Exception is of wrong type", AMQNoRouteException.class, exception.getCause().getClass());
                    exceptionReceivedLatch.countDown();
                    if (exceptionReceivedLatch.getCount() == 0)
                    {
                        connection.stop(); // ** Deadlock
                        connection.close();
                    }
                }
                catch (Throwable t)
                {
                    _lastExceptionListenerException = t;
                }
            }
        };
        connection.setExceptionListener(listener);

        // Create a message listener that receives from testQueue and tries to forward them to unknown queue (thus
        // provoking AMQNoRouteException exceptions to be delivered to the ExceptionListener).
        final Queue unknownQueue = session.createQueue(getTestQueueName() + "_unknown");;
        MessageListener redirectingMessageListener = new MessageListener()
        {
            @Override
            public void onMessage(Message msg)
            {
                try
                {
                    Session mlSession = connection.createSession(true, Session.SESSION_TRANSACTED);  // ** Deadlock
                    mlSession.createProducer(unknownQueue).send(msg);
                    mlSession.commit();
                }
                catch (JMSException je)
                {
                    // Connection is closed by the listener, so exceptions here are expected.
                    LOGGER.debug("Expected exception - message listener got exception", je);
                }
            }
        };

        MessageConsumer consumer = session.createConsumer(messageQueue);
        consumer.setMessageListener(redirectingMessageListener);
        connection.start();

        // Await the 2nd exception
        boolean exceptionReceived = exceptionReceivedLatch.await(10, TimeUnit.SECONDS);
        assertTrue("Exception listener did not hear exception within timeout", exceptionReceived);
        assertNull("Exception listener should not have had experienced exception", _lastExceptionListenerException);
    }
}