import javax.jms.ResourceAllocationException; //导入依赖的package包/类
@Test(timeout = 60000)
public void testAddressIsBlockedForOtherProdudersWhenFull() throws Exception {
   Connection connection = createConnection();
   Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   Destination d = session.createQueue(getQueueName());
   MessageProducer p = session.createProducer(d);

   fillAddress(getQueueName());

   Exception e = null;
   try {
      p.send(session.createBytesMessage());
   } catch (ResourceAllocationException rae) {
      e = rae;
   }
   assertTrue(e instanceof ResourceAllocationException);
   assertTrue(e.getMessage().contains("resource-limit-exceeded"));

   long addressSize = server.getPagingManager().getPageStore(new SimpleString(getQueueName())).getAddressSize();
   assertTrue(addressSize >= MAX_SIZE_BYTES_REJECT_THRESHOLD);
}
 

import javax.jms.ResourceAllocationException; //导入依赖的package包/类
public static JMSRuntimeException convertToRuntimeException(JMSException e) {
    if (e instanceof javax.jms.IllegalStateException) {
        return new IllegalStateRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof InvalidClientIDException) {
        return new InvalidClientIDRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof InvalidDestinationException) {
        return new InvalidDestinationRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof InvalidSelectorException) {
        return new InvalidSelectorRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof JMSSecurityException) {
        return new JMSSecurityRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof MessageFormatException) {
        return new MessageFormatRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof MessageNotWriteableException) {
        return new MessageNotWriteableRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof ResourceAllocationException) {
        return new ResourceAllocationRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof TransactionInProgressException) {
        return new TransactionInProgressRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof TransactionRolledBackException) {
        return new TransactionRolledBackRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    return new JMSRuntimeException(e.getMessage(), e.getErrorCode(), e);
}
 

import javax.jms.ResourceAllocationException; //导入依赖的package包/类
protected final void waitForSpace(ConnectionContext context, ProducerBrokerExchange producerBrokerExchange, Usage<?> usage, int highWaterMark, String warning) throws IOException, InterruptedException, ResourceAllocationException {
    if (!context.isNetworkConnection() && systemUsage.isSendFailIfNoSpace()) {
        getLog().debug("sendFailIfNoSpace, forcing exception on send, usage: {}: {}", usage, warning);
        throw new ResourceAllocationException(warning);
    }
    if (!context.isNetworkConnection() && systemUsage.getSendFailIfNoSpaceAfterTimeout() != 0) {
        if (!usage.waitForSpace(systemUsage.getSendFailIfNoSpaceAfterTimeout(), highWaterMark)) {
            getLog().debug("sendFailIfNoSpaceAfterTimeout expired, forcing exception on send, usage: {}: {}", usage, warning);
            throw new ResourceAllocationException(warning);
        }
    } else {
        long start = System.currentTimeMillis();
        long nextWarn = start;
        producerBrokerExchange.blockingOnFlowControl(true);
        destinationStatistics.getBlockedSends().increment();
        while (!usage.waitForSpace(1000, highWaterMark)) {
            if (context.getStopping().get()) {
                throw new IOException("Connection closed, send aborted.");
            }

            long now = System.currentTimeMillis();
            if (now >= nextWarn) {
                getLog().info("{}: {} (blocking for: {}s)", new Object[]{ usage, warning, new Long(((now - start) / 1000))});
                nextWarn = now + blockedProducerWarningInterval;
            }
        }
        long finish = System.currentTimeMillis();
        long totalTimeBlocked = finish - start;
        destinationStatistics.getBlockedTime().addTime(totalTimeBlocked);
        producerBrokerExchange.incrementTimeBlocked(this,totalTimeBlocked);
        producerBrokerExchange.blockingOnFlowControl(false);
    }
}
 

import javax.jms.ResourceAllocationException; //导入依赖的package包/类
@Override
public void run() {
    TimeoutMessage timeout;
    try {
        while (true) {
            timeout = flowControlTimeoutMessages.take();
            if (timeout != null) {
                synchronized (messagesWaitingForSpace) {
                    if (messagesWaitingForSpace.remove(timeout.message.getMessageId()) != null) {
                        ExceptionResponse response = new ExceptionResponse(
                                new ResourceAllocationException(
                                        "Usage Manager Memory Limit reached. Stopping producer ("
                                                + timeout.message.getProducerId()
                                                + ") to prevent flooding "
                                                + getActiveMQDestination().getQualifiedName()
                                                + "."
                                                + " See http://activemq.apache.org/producer-flow-control.html for more info"));
                        response.setCorrelationId(timeout.message.getCommandId());
                        timeout.context.getConnection().dispatchAsync(response);
                    }
                }
            }
        }
    } catch (InterruptedException e) {
        LOG.debug(getName() + "Producer Flow Control Timeout Task is stopping");
    }
}
 

import javax.jms.ResourceAllocationException; //导入依赖的package包/类
protected ConnectionFactory getConnectionFactory() throws Exception {
   factory.setExceptionListener(new ExceptionListener() {
      @Override
      public void onException(JMSException arg0) {
         if (arg0 instanceof ResourceAllocationException) {
            gotResourceException.set(true);
         }
      }
   });
   return factory;
}
 

import javax.jms.ResourceAllocationException; //导入依赖的package包/类
/**
 * Converts instances of sub-classes of {@link JMSException} into the corresponding sub-class of
 * {@link JMSRuntimeException}.
 *
 * @param e
 * @return
 */
public static JMSRuntimeException convertToRuntimeException(JMSException e) {
   if (e instanceof javax.jms.IllegalStateException) {
      return new IllegalStateRuntimeException(e.getMessage(), e.getErrorCode(), e);
   }
   if (e instanceof InvalidClientIDException) {
      return new InvalidClientIDRuntimeException(e.getMessage(), e.getErrorCode(), e);
   }
   if (e instanceof InvalidDestinationException) {
      return new InvalidDestinationRuntimeException(e.getMessage(), e.getErrorCode(), e);
   }
   if (e instanceof InvalidSelectorException) {
      return new InvalidSelectorRuntimeException(e.getMessage(), e.getErrorCode(), e);
   }
   if (e instanceof JMSSecurityException) {
      return new JMSSecurityRuntimeException(e.getMessage(), e.getErrorCode(), e);
   }
   if (e instanceof MessageFormatException) {
      return new MessageFormatRuntimeException(e.getMessage(), e.getErrorCode(), e);
   }
   if (e instanceof MessageNotWriteableException) {
      return new MessageNotWriteableRuntimeException(e.getMessage(), e.getErrorCode(), e);
   }
   if (e instanceof ResourceAllocationException) {
      return new ResourceAllocationRuntimeException(e.getMessage(), e.getErrorCode(), e);
   }
   if (e instanceof TransactionInProgressException) {
      return new TransactionInProgressRuntimeException(e.getMessage(), e.getErrorCode(), e);
   }
   if (e instanceof TransactionRolledBackException) {
      return new TransactionRolledBackRuntimeException(e.getMessage(), e.getErrorCode(), e);
   }
   return new JMSRuntimeException(e.getMessage(), e.getErrorCode(), e);
}
 

import javax.jms.ResourceAllocationException; //导入依赖的package包/类
@Repeat(repetitions = 1)
@Test(timeout = 20000)
public void testRemotelyCloseProducerWithSendWaitingForCredit() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        Connection connection = testFixture.establishConnecton(testPeer);

        testPeer.expectBegin();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // Expect producer creation, don't give it credit.
        testPeer.expectSenderAttachWithoutGrantingCredit();

        // Producer has no credit so the send should block waiting for it, then fail when the remote close occurs
        testPeer.remotelyDetachLastOpenedLinkOnLastOpenedSession(true, true, AmqpError.RESOURCE_LIMIT_EXCEEDED, "Producer closed", 50);
        testPeer.expectClose();

        Queue queue = session.createQueue("myQueue");
        final MessageProducer producer = session.createProducer(queue);

        Message message = session.createTextMessage("myMessage");

        try {
            producer.send(message);
            fail("Expected exception to be thrown due to close of producer");
        } catch (ResourceAllocationException rae) {
            // Expected if remote close beat the send to the provider
        } catch (IllegalStateException ise) {
            // Can happen if send fires before remote close if processed.
        }

        connection.close();

        testPeer.waitForAllHandlersToComplete(3000);
    }
}
 

import javax.jms.ResourceAllocationException; //导入依赖的package包/类
public static JMSRuntimeException toRuntimeException(final JMSException e) {
    if (e instanceof javax.jms.IllegalStateException) {
        return new IllegalStateRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof InvalidClientIDException) {
        return new InvalidClientIDRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof InvalidDestinationException) {
        return new InvalidDestinationRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof InvalidSelectorException) {
        return new InvalidSelectorRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof JMSSecurityException) {
        return new JMSSecurityRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof MessageFormatException) {
        return new MessageFormatRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof MessageNotWriteableException) {
        return new MessageNotWriteableRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof ResourceAllocationException) {
        return new ResourceAllocationRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof TransactionInProgressException) {
        return new TransactionInProgressRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    if (e instanceof TransactionRolledBackException) {
        return new TransactionRolledBackRuntimeException(e.getMessage(), e.getErrorCode(), e);
    }
    return new JMSRuntimeException(e.getMessage(), e.getErrorCode(), e);
}
 

import javax.jms.ResourceAllocationException; //导入依赖的package包/类
@Test(expected = ResourceAllocationRuntimeException.class)
public void testConvertsResourceAllocationExceptionToResourceAllocationRuntimeException() {
    throw JMSExceptionSupport.createRuntimeException(new ResourceAllocationException("error"));
}
 

import javax.jms.ResourceAllocationException; //导入依赖的package包/类
@Test
public void testPublisherRecoverAfterBlockWithSyncSend() throws Exception {
   ActiveMQConnectionFactory factory = (ActiveMQConnectionFactory) getConnectionFactory();
   factory.setExceptionListener(null);
   factory.setUseAsyncSend(false);
   this.flowControlConnection = (ActiveMQConnection) factory.createConnection();
   this.flowControlConnection.start();

   final Session session = this.flowControlConnection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
   final MessageProducer producer = session.createProducer(queueA);

   final AtomicBoolean keepGoing = new AtomicBoolean(true);
   final AtomicInteger exceptionCount = new AtomicInteger(0);
   Thread thread = new Thread("Filler") {
      @Override
      public void run() {
         while (keepGoing.get()) {
            try {
               producer.send(session.createTextMessage("Test message"));
            } catch (JMSException arg0) {
               if (arg0 instanceof ResourceAllocationException) {
                  gotResourceException.set(true);
                  exceptionCount.incrementAndGet();
               }
            }
         }
      }
   };
   thread.start();
   waitForBlockedOrResourceLimit(new AtomicBoolean(false));

   // resourceException on second message, resumption if we
   // can receive 10
   MessageConsumer consumer = session.createConsumer(queueA);
   TextMessage msg;
   for (int idx = 0; idx < 10; ++idx) {
      msg = (TextMessage) consumer.receive(1000);
      if (msg != null) {
         msg.acknowledge();
      }
   }
   assertTrue("we were blocked at least 5 times", 5 < exceptionCount.get());
   keepGoing.set(false);
}
 

import javax.jms.ResourceAllocationException; //导入依赖的package包/类
@Test(expected = ResourceAllocationRuntimeException.class)
public void testConvertsResourceAllocationExceptionToResourceAllocationRuntimeException() {
    throw JmsExceptionSupport.createRuntimeException(new ResourceAllocationException("error"));
}