Java 类javax.jms.ExceptionListener 实例源码

@Test(timeout = 60000)
public void testExceptionListenerGetsNotified() throws Exception {
    final CountDownLatch signal = new CountDownLatch(1);
    Connection connection = cf.createConnection();
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            LOG.info("ExceptionListener called with error: {}", exception.getMessage());
            signal.countDown();
        }
    });

    assertNotNull(connection.getExceptionListener());

    MockJMSConnection mockJMSConnection = (MockJMSConnection) ((JmsPoolConnection) connection).getConnection();
    mockJMSConnection.injectConnectionError(new JMSException("Some non-fatal error"));

    assertTrue(signal.await(10, TimeUnit.SECONDS));
}

@Test
public void testFailedConnectThenSucceedsWithListener() throws JMSException {
    Connection connection = pooledConnFact.createConnection("invalid", "credentials");
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            LOG.warn("Connection get error: {}", exception.getMessage());
        }
    });

    try {
        connection.start();
        fail("Should fail to connect");
    } catch (JMSSecurityException ex) {
        LOG.info("Caught expected security error");
    }

    connection = pooledConnFact.createConnection("system", "manager");
    connection.start();

    LOG.info("Successfully create new connection.");

    connection.close();
}

/**
 * Prepare the given Connection before it is exposed.
 * <p>The default implementation applies ExceptionListener and client id.
 * Can be overridden in subclasses.
 * @param con the Connection to prepare
 * @throws JMSException if thrown by JMS API methods
 * @see #setExceptionListener
 * @see #setReconnectOnException
 */
protected void prepareConnection(Connection con) throws JMSException {
    if (getClientId() != null) {
        con.setClientID(getClientId());
    }
    if (this.aggregatedExceptionListener != null) {
        con.setExceptionListener(this.aggregatedExceptionListener);
    }
    else if (getExceptionListener() != null || isReconnectOnException()) {
        ExceptionListener listenerToUse = getExceptionListener();
        if (isReconnectOnException()) {
            this.aggregatedExceptionListener = new AggregatedExceptionListener();
            this.aggregatedExceptionListener.delegates.add(this);
            if (listenerToUse != null) {
                this.aggregatedExceptionListener.delegates.add(listenerToUse);
            }
            listenerToUse = this.aggregatedExceptionListener;
        }
        con.setExceptionListener(listenerToUse);
    }
}

private Connection createConnection(URI remoteURI, String username, String password, String clientId, boolean start) throws JMSException {
   JmsConnectionFactory factory = new JmsConnectionFactory(remoteURI);

   Connection connection = trackJMSConnection(factory.createConnection(username, password));

   connection.setExceptionListener(new ExceptionListener() {
      @Override
      public void onException(JMSException exception) {
         exception.printStackTrace();
      }
   });

   if (clientId != null && !clientId.isEmpty()) {
      connection.setClientID(clientId);
   }

   if (start) {
      connection.start();
   }

   return connection;
}

private Connection createCoreConnection(String connectionString, String username, String password, String clientId, boolean start) throws JMSException {
   ActiveMQJMSConnectionFactory factory = new ActiveMQJMSConnectionFactory(connectionString);

   Connection connection = trackJMSConnection(factory.createConnection(username, password));

   connection.setExceptionListener(new ExceptionListener() {
      @Override
      public void onException(JMSException exception) {
         exception.printStackTrace();
      }
   });

   if (clientId != null && !clientId.isEmpty()) {
      connection.setClientID(clientId);
   }

   if (start) {
      connection.start();
   }

   return connection;
}

private Connection createOpenWireConnection(String connectionString, String username, String password, String clientId, boolean start) throws JMSException {
   ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(connectionString);

   Connection connection = trackJMSConnection(factory.createConnection(username, password));

   connection.setExceptionListener(new ExceptionListener() {
      @Override
      public void onException(JMSException exception) {
         exception.printStackTrace();
      }
   });

   if (clientId != null && !clientId.isEmpty()) {
      connection.setClientID(clientId);
   }

   if (start) {
      connection.start();
   }

   return connection;
}

public void testSetExceptionListener() throws Exception {
   ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
   connection = (ActiveMQConnection) cf.createConnection();
   assertNull(connection.getExceptionListener());

   ExceptionListener exListener = new ExceptionListener() {
      @Override
      public void onException(JMSException arg0) {
      }
   };
   cf.setExceptionListener(exListener);
   connection.close();

   connection = (ActiveMQConnection) cf.createConnection();
   assertNotNull(connection.getExceptionListener());
   assertEquals(exListener, connection.getExceptionListener());
   connection.close();

   connection = (ActiveMQConnection) cf.createConnection();
   assertEquals(exListener, connection.getExceptionListener());

   assertEquals(exListener, cf.getExceptionListener());
   connection.close();

}

private void updateInternal(Map<String, ?> configuration) throws JMSException {
  // get JMS up and running
  jmsConnection = connectionFactory.createQueueConnection();
  jmsConnection.setExceptionListener(new ExceptionListener() {
    @Override
    public void onException(JMSException e) {
      log.error("There was an error while working with JMS.", e);
    }
  });
  jmsSession = jmsConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  Destination destination = jmsSession.createQueue("test");

  producer = new JMSProducer(jmsSession, destination);
  consumer = new JMSConsumer(jmsSession, destination);
  jmsConnection.start();
}

/**
 * Creates a connection to the broker, and sets a connection listener to prevent failover and an exception listener 
 * with a {@link CountDownLatch} to synchronise in the {@link #check403Exception(Throwable)} method and allow the
 * {@link #tearDown()} method to complete properly.
 */
public Connection getConnection(String vhost, String username, String password) throws NamingException, JMSException, URLSyntaxException
{
    AMQConnection connection = (AMQConnection) getConnection(createConnectionURL(vhost, username, password));

    //Prevent Failover
    connection.setConnectionListener(this);

    //QPID-2081: use a latch to sync on exception causing connection close, to work 
    //around the connection close race during tearDown() causing sporadic failures
    _exceptionReceived = new CountDownLatch(1);

    connection.setExceptionListener(new ExceptionListener()
    {
        public void onException(JMSException e)
        {
            _exceptionReceived.countDown();
        }
    });

    return (Connection) connection;
}

public void testBrokerDeath() throws Exception
{
    Connection conn = getConnection("guest", "guest");

    conn.start();

    final CountDownLatch fired = new CountDownLatch(1);
    conn.setExceptionListener(new ExceptionListener()
    {
        public void onException(JMSException e)
        {
            fired.countDown();
        }
    });

    stopBroker();

    if (!fired.await(3, TimeUnit.SECONDS))
    {
        fail("exception listener was not fired");
    }
}

@Override
protected MessageConsumer createConsumer() throws Exception {
    connection = createConnectionToMockProvider();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue destination = session.createQueue(_testName.getMethodName());
    MessageConsumer consumer = session.createConsumer(destination);
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
        }
    });
    connection.start();
    providerListener.onConnectionFailure(new IOException());

    final JmsConnection jmsConnection = connection;
    assertTrue(Wait.waitFor(new Wait.Condition() {

        @Override
        public boolean isSatisified() throws Exception {
            return !jmsConnection.isConnected();
        }
    }, TimeUnit.SECONDS.toMillis(30), TimeUnit.MILLISECONDS.toMillis(2)));

    return consumer;
}

@Override
protected void createTestResources() throws Exception {
    connection = createConnectionToMockProvider();
    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
        }
    });
    Queue destination = session.createQueue(_testName.getMethodName());

    sender = session.createProducer(destination);
    receiver = session.createConsumer(destination);
    connection.start();
    providerListener.onConnectionFailure(new IOException());

    final JmsConnection jmsConnection = connection;
    assertTrue(Wait.waitFor(new Wait.Condition() {

        @Override
        public boolean isSatisified() throws Exception {
            return !jmsConnection.isConnected();
        }
    }, TimeUnit.SECONDS.toMillis(30), TimeUnit.MILLISECONDS.toMillis(2)));
}

@Test
public void testSetExceptionListenerPassthrough() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);

    ExceptionListener listener = new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
        }
    };

    try {
        context.setExceptionListener(listener);
    } finally {
        context.close();
    }

    Mockito.verify(connection, Mockito.times(1)).setExceptionListener(listener);
}

@Test
public void testGlobalExceptionListenerIsAppliedToCreatedConnection() throws Exception {
    JmsConnectionFactory factory = new JmsConnectionFactory(new URI("mock://127.0.0.1:5763"));

    ExceptionListener listener = new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
        }
    };

    factory.setExceptionListener(listener);
    Connection connection = factory.createConnection();
    assertNotNull(connection);
    assertNotNull(connection.getExceptionListener());
    assertSame(listener, connection.getExceptionListener());

    connection.close();
}

/**
 * Verify that the 'global' exception listener set on the connection factory
 * is ignored when the factory gets serialized.
 *
 * @throws Exception if an error occurs during the test.
 */
@Test
public void testSerializeThenDeserializeIgnoresGlobalExceptionListener() throws Exception {
    String uri = "amqp://localhost:1234";

    JmsConnectionFactory cf = new JmsConnectionFactory(uri);
    cf.setExceptionListener(new ExceptionListener() {
        @Override
        public void onException(JMSException exception) {
            // Nothing
        }
    });

    Map<String, String> props = cf.getProperties();

    Object roundTripped = roundTripSerialize(cf);

    assertNotNull("Null object returned", roundTripped);
    assertEquals("Unexpected type", JmsConnectionFactory.class, roundTripped.getClass());
    assertEquals("Unexpected uri", uri, ((JmsConnectionFactory)roundTripped).getRemoteURI());

    Map<String, String> props2 = ((JmsConnectionFactory)roundTripped).getProperties();

    assertFalse("Properties map should not contain ExceptionListener", props.containsKey("exceptionListener"));
    assertEquals("Properties were not equal", props, props2);
}

@Override
protected JmsConnection createConnection() throws Exception {
    connection = createConnectionToMockProvider();
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
        }
    });
    connection.start();

    providerListener.onConnectionFailure(new IOException());

    final JmsConnection jmsConnection = connection;
    assertTrue(Wait.waitFor(new Wait.Condition() {

        @Override
        public boolean isSatisified() throws Exception {
            return !jmsConnection.isConnected();
        }
    }, TimeUnit.SECONDS.toMillis(30), TimeUnit.MILLISECONDS.toMillis(10)));

    return connection;
}

@Test(timeout=60000)
public void testConnectionExceptionBrokerStop() throws Exception {
    final CountDownLatch latch = new CountDownLatch(1);
    connection = createAmqpConnection();
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            latch.countDown();
        }
    });
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);

    stopPrimaryBroker();

    assertTrue(latch.await(10, TimeUnit.SECONDS));

    connection.close();
}

@Test(timeout=60000)
public void testStartFailureWithAsyncExceptionListener() throws Exception {
    URI brokerURI = new URI(getAmqpFailoverURI() +
        "?failover.maxReconnectAttempts=5" +
        "&failover.useReconnectBackOff=false");

    final CountDownLatch failed = new CountDownLatch(1);
    JmsConnectionFactory factory = new JmsConnectionFactory(brokerURI);
    factory.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            LOG.info("Connection got exception: {}", exception.getMessage());
            failed.countDown();
        }
    });
    connection = factory.createConnection();
    connection.start();

    stopPrimaryBroker();

    assertTrue("No async exception", failed.await(15, TimeUnit.SECONDS));
}

@Test(timeout=60000)
public void testConnectionExceptionBrokerStop() throws Exception {
    final CountDownLatch latch = new CountDownLatch(1);
    Connection connection = createStompConnection();
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            latch.countDown();
        }
    });
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);

    stopPrimaryBroker();

    assertTrue(latch.await(10, TimeUnit.SECONDS));

    connection.close();
}

@Test(timeout=60000)
public void testStartFailureWithAsyncExceptionListener() throws Exception {
    URI brokerURI = new URI(getAmqpFailoverURI() + "?maxReconnectDelay=1000&maxReconnectAttempts=5");

    final CountDownLatch failed = new CountDownLatch(1);
    JmsConnectionFactory factory = new JmsConnectionFactory(brokerURI);
    factory.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            LOG.info("Connection got exception: {}", exception.getMessage());
            failed.countDown();
        }
    });
    Connection connection = factory.createConnection();
    connection.start();

    stopPrimaryBroker();

    assertTrue("No async exception", failed.await(15, TimeUnit.SECONDS));
}

@Override
protected MessageProducer createProducer() throws Exception {
    final CountDownLatch latch = new CountDownLatch(1);
    connection = createAmqpConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    message = session.createMessage();
    destination = session.createQueue("test");
    MessageProducer producer = session.createProducer(destination);
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            latch.countDown();
        }
    });
    connection.start();
    stopPrimaryBroker();
    assertTrue(latch.await(10, TimeUnit.SECONDS));
    return producer;
}

@Override
protected MessageConsumer createConsumer() throws Exception {
    final CountDownLatch latch = new CountDownLatch(1);
    connection = createAmqpConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue destination = session.createQueue(name.getMethodName());
    MessageConsumer consumer = session.createConsumer(destination);
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            latch.countDown();
        }
    });
    connection.start();
    stopPrimaryBroker();
    assertTrue(latch.await(10, TimeUnit.SECONDS));
    return consumer;
}

@Test(timeout=60000)
public void testConnectionExceptionBrokerStop() throws Exception {
    final CountDownLatch latch = new CountDownLatch(1);
    Connection connection = createAmqpConnection();
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            latch.countDown();
        }
    });
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);

    stopPrimaryBroker();

    assertTrue(latch.await(10, TimeUnit.SECONDS));

    connection.close();
}

@Override
protected Session createSession() throws Exception {
    final CountDownLatch latch = new CountDownLatch(1);
    connection = createAmqpConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            latch.countDown();
        }
    });
    connection.start();
    stopPrimaryBroker();
    assertTrue(latch.await(20, TimeUnit.SECONDS));
    TimeUnit.MILLISECONDS.sleep(500);
    return session;
}

@Override
protected Connection createConnection() throws Exception {
    final CountDownLatch latch = new CountDownLatch(1);
    connection = createAmqpConnection();
    connection.setExceptionListener(new ExceptionListener() {

        @Override
        public void onException(JMSException exception) {
            latch.countDown();
        }
    });
    connection.start();
    stopPrimaryBroker();
    assertTrue(latch.await(20, TimeUnit.SECONDS));
    TimeUnit.MILLISECONDS.sleep(500);
    return connection;
}

@Override
public ExceptionListener getExceptionListener() {
    try {
        return connection.getExceptionListener();
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public void setExceptionListener(ExceptionListener listener) {
    try {
        connection.setExceptionListener(listener);
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public ExceptionListener getExceptionListener() {
    try {
        return connection.getExceptionListener();
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public void setExceptionListener(ExceptionListener listener) {
    try {
        connection.setExceptionListener(listener);
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Test(timeout = 60000)
public void testSetGetExceptionListener() throws Exception {
    pcf = new JmsPoolConnectionFactory();
    pcf.setConnectionFactory(new ActiveMQConnectionFactory(
        "vm://test?broker.persistent=false&broker.useJmx=false"));

    connection = (TopicConnection) pcf.createConnection();
    ExceptionListener listener = new ExceptionListener() {
        @Override
        public void onException(JMSException exception) {
        }
    };
    connection.setExceptionListener(listener);
    assertEquals(listener, connection.getExceptionListener());
}

@Override
public ExceptionListener getExceptionListener() {
    try {
        return connection.getExceptionListener();
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

@Override
public void setExceptionListener(ExceptionListener listener) {
    try {
        connection.setExceptionListener(listener);
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

/**
 * Invoke the registered JMS ExceptionListener, if any.
 * @param ex the exception that arose during JMS processing
 * @see #setExceptionListener
 */
protected void invokeExceptionListener(JMSException ex) {
    ExceptionListener exceptionListener = getExceptionListener();
    if (exceptionListener != null) {
        exceptionListener.onException(ex);
    }
}

@Override
public void onException(JMSException ex) {
    synchronized (connectionMonitor) {
        // Iterate over temporary copy in order to avoid ConcurrentModificationException,
        // since listener invocations may in turn trigger registration of listeners...
        for (ExceptionListener listener : new LinkedHashSet<ExceptionListener>(this.delegates)) {
            listener.onException(ex);
        }
    }
}

@Test
public void testWithConnectionFactoryAndExceptionListener() throws JMSException {
    ConnectionFactory cf = mock(ConnectionFactory.class);
    Connection con = mock(Connection.class);

    ExceptionListener listener = new ChainedExceptionListener();
    given(cf.createConnection()).willReturn(con);
    given(con.getExceptionListener()).willReturn(listener);

    SingleConnectionFactory scf = new SingleConnectionFactory(cf);
    scf.setExceptionListener(listener);
    Connection con1 = scf.createConnection();
    assertEquals(listener, con1.getExceptionListener());
    con1.start();
    con1.stop();
    con1.close();
    Connection con2 = scf.createConnection();
    con2.start();
    con2.stop();
    con2.close();
    scf.destroy();  // should trigger actual close

    verify(con).setExceptionListener(listener);
    verify(con, times(2)).start();
    verify(con, times(2)).stop();
    verify(con).close();
}

/**
 * Test sendRequest with null request object - should throw exception.
 * Also calls the lifecycle start() method and checks connection and session
 * calls.
 * @throws JMSException
 * @throws InterruptedException 
 */
@Test(expected = NullPointerException.class)
public void testStartAndSendRequestNullRequest() throws JMSException, InterruptedException { 
  //need to simulate start
  EasyMock.expect(connectionFactory.createConnection()).andReturn(connection).times(2);        
  EasyMock.expect(connection.createSession(false, Session.AUTO_ACKNOWLEDGE)).andReturn(session).times(3);
  connection.setExceptionListener(EasyMock.isA(ExceptionListener.class));  
  connection.start();       

  MessageConsumer messageConsumer = EasyMock.createMock(MessageConsumer.class);
  EasyMock.expect(session.createConsumer(EasyMock.isA(Destination.class))).andReturn(messageConsumer).times(2);    
  messageConsumer.setMessageListener(EasyMock.isA(MessageListener.class));
  messageConsumer.setMessageListener(EasyMock.isA(MessageListener.class));
  session.close();

  EasyMock.replay(connectionFactory);
  EasyMock.replay(connection);
  EasyMock.replay(session);
  EasyMock.replay(messageConsumer);
  ((JmsProxyImpl) jmsProxy).init();
  Thread.sleep(2000); //leave time for connection thread to run (and set connected flag to true)
  jmsProxy.sendRequest(null, "test.queue", 1000);
  EasyMock.verify(connectionFactory);
  EasyMock.verify(connection);
  EasyMock.verify(session);
  EasyMock.verify(messageConsumer);
}

/**
 * Test sendRequest with null queue name - should throw exception.
 * @throws JMSException
 * @throws InterruptedException 
 */
@Test(expected = NullPointerException.class)
public void testSendRequestNullQueue() throws JMSException, InterruptedException { 
  JsonRequest<ClientRequestResult> jsonRequest = EasyMock.createMock(JsonRequest.class);
  //need to simulate start
  EasyMock.expect(connectionFactory.createConnection()).andReturn(connection).times(2);        
  EasyMock.expect(connection.createSession(false, Session.AUTO_ACKNOWLEDGE)).andReturn(session).times(3);
  connection.setExceptionListener(EasyMock.isA(ExceptionListener.class));  
  connection.start();       

  MessageConsumer messageConsumer = EasyMock.createMock(MessageConsumer.class);
  EasyMock.expect(session.createConsumer(EasyMock.isA(Destination.class))).andReturn(messageConsumer).times(2);    
  messageConsumer.setMessageListener(EasyMock.isA(MessageListener.class));
  messageConsumer.setMessageListener(EasyMock.isA(MessageListener.class));
  session.close();

  EasyMock.replay(connectionFactory);
  EasyMock.replay(connection);
  EasyMock.replay(session);
  EasyMock.replay(messageConsumer);
  ((JmsProxyImpl) jmsProxy).init();
  Thread.sleep(2000); //leave time for connection thread to run (and set connected flag to true)
  jmsProxy.sendRequest(jsonRequest, null, 1000);
  EasyMock.verify(connectionFactory);
  EasyMock.verify(connection);
  EasyMock.verify(session);
  EasyMock.verify(messageConsumer);
}

protected ConnectionFactory getConnectionFactory() throws Exception {
   factory.setExceptionListener(new ExceptionListener() {
      @Override
      public void onException(JMSException arg0) {
         if (arg0 instanceof ResourceAllocationException) {
            gotResourceException.set(true);
         }
      }
   });
   return factory;
}

@Test
@BMRules(
   rules = {@BMRule(
      name = "Corrupt Decoding",
      targetClass = "org.apache.activemq.artemis.core.protocol.core.impl.PacketDecoder",
      targetMethod = "decode(byte)",
      targetLocation = "ENTRY",
      action = "org.apache.activemq.artemis.tests.extras.byteman.DisconnectOnCriticalFailureTest.doThrow();")})
public void testSendDisconnect() throws Exception {
   createQueue("queue1");
   final Connection producerConnection = nettyCf.createConnection();
   final CountDownLatch latch = new CountDownLatch(1);

   try {
      producerConnection.setExceptionListener(new ExceptionListener() {
         @Override
         public void onException(JMSException e) {
            latch.countDown();
         }
      });

      corruptPacket.set(true);
      producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);

      assertTrue(latch.await(5, TimeUnit.SECONDS));
   } finally {
      corruptPacket.set(false);

      if (producerConnection != null) {
         producerConnection.close();
      }
   }
}

@Ignore("See AMQ-4286")
@Test(timeout = 60 * 1000)
public void testLotsOfConcurrentConnections() throws Exception {
   ExecutorService executor = Executors.newCachedThreadPool();
   final ConnectionFactory factory = createConnectionFactory();
   int connectionCount = 400;
   final AtomicInteger threadId = new AtomicInteger(0);
   for (int i = 0; i < connectionCount; i++) {
      executor.execute(new Runnable() {
         @Override
         public void run() {
            final int innerId = threadId.incrementAndGet();
            try {
               ExceptionListener listener = new NioQueueSubscriptionTestListener(innerId, exceptions, LOG);
               ActiveMQConnection connection = (ActiveMQConnection) factory.createConnection();
               connection.setExceptionListener(listener);
               connection.start();
               assertNotNull(connection.getBrokerName());
               connections.add(connection);
            } catch (Exception e) {
               LOG.error(">>>> Exception in run() on thread " + innerId, e);
               exceptions.put(Thread.currentThread(), e);
            }
         }
      });
   }

   executor.shutdown();
   executor.awaitTermination(30, TimeUnit.SECONDS);

   if (!exceptions.isEmpty()) {
      LOG.error(">>>> " + exceptions.size() + " exceptions like", exceptions.values().iterator().next());
      fail("unexpected exceptions in worker threads: " + exceptions.values().iterator().next());
   }
   LOG.info("created " + connectionCount + " connections");
}