javax.jms.DeliveryMode#PERSISTENT源码实例Demo

@Override
public void setDeliveryMode(final int deliveryMode) throws JMSException {
   checkClosed();
   if (deliveryMode != DeliveryMode.NON_PERSISTENT && deliveryMode != DeliveryMode.PERSISTENT) {
      throw ActiveMQJMSClientBundle.BUNDLE.illegalDeliveryMode(deliveryMode);
   }

   defaultDeliveryMode = deliveryMode;
}
 

/**
 * Sets up a test with a topic destination, durable suscriber and persistent
 * delivery mode.
 *
 * @see junit.framework.TestCase#setUp()
 */
@Override
protected void setUp() throws Exception {
   topic = true;
   durable = true;
   deliveryMode = DeliveryMode.PERSISTENT;
   super.setUp();
}
 

@Override
public void setDeliveryMode(int deliveryMode) throws JMSException {
    checkClosed();
    switch (deliveryMode) {
        case DeliveryMode.PERSISTENT:
        case DeliveryMode.NON_PERSISTENT:
            this.deliveryMode = deliveryMode;
            break;
        default:
            throw new JMSException(String.format("Invalid DeliveryMode specified: %d", deliveryMode));
    }
}
 

@Override
public JMSProducer setDeliveryMode(int deliveryMode) {
    switch (deliveryMode) {
        case DeliveryMode.PERSISTENT:
        case DeliveryMode.NON_PERSISTENT:
            this.deliveryMode = deliveryMode;
            return this;
        default:
            throw new JMSRuntimeException(String.format("Invalid DeliveryMode specified: %d", deliveryMode));
    }
}
 

private void runTest(Map<String,String> options) throws Exception
{
    String resultsFile = options.get(RESULTS_FILE_ARG);
    String jndiProperties = options.get(JNDI_PROPERTIES_ARG);
    String connectionFactoryString = options.get(JNDI_CONNECTION_FACTORY_ARG);
    int numConnections = Integer.parseInt(options.get(CONNECTIONS_ARG));
    int numSessions = Integer.parseInt(options.get(SESSIONS_ARG));
    int numProducers = Integer.parseInt(options.get(PRODUCERS_ARG));
    int numMessage = Integer.parseInt(options.get(MESSAGE_COUNT_ARG));
    int messageSize = Integer.parseInt(options.get(MESSAGE_SIZE_ARG));
    String queueString = options.get(JNDI_DESTINATION_ARG);
    int deliveryMode = Boolean.valueOf(options.get(PERSISTENT_ARG)) ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT;
    long receiveTimeout = Long.parseLong(options.get(TIMEOUT_ARG));
    boolean transacted = Boolean.valueOf(options.get(TRANSACTED_ARG));

    LOGGER.info("Using options: " + options);


    // Load JNDI properties
    Context ctx = getInitialContext(jndiProperties);
    final ConnectionFactory conFac = (ConnectionFactory) ctx.lookup(connectionFactoryString);

    Destination destination = ensureQueueCreated(queueString, conFac);
    Map<Connection, List<Session>> connectionsAndSessions = openConnectionsAndSessions(numConnections, numSessions, transacted, conFac);
    publish(numMessage, messageSize, numProducers, deliveryMode, destination, connectionsAndSessions);
    MemoryStatistic memoryStatistics = collectMemoryStatistics(options);
    generateCSV(memoryStatistics, numConnections, numSessions, transacted, numMessage, messageSize, numProducers, deliveryMode, resultsFile);
    purgeQueue(conFac, queueString, receiveTimeout);
    closeConnections(connectionsAndSessions.keySet());
    System.exit(0);
}
 

public void testDurableQueueMessage() throws Exception {
   super.topic = false;
   deliveryMode = DeliveryMode.PERSISTENT;
   durableSubscriber = false;
   doTest();
   validateConsumerPrefetch(this.getDestinationString(), 0);
}
 

/**
 * Set up the test with a queue and persistent delivery mode.
 *
 * @see junit.framework.TestCase#setUp()
 */
@Override
protected void setUp() throws Exception {
   topic = false;
   deliveryMode = DeliveryMode.PERSISTENT;
   super.setUp();
}
 

protected int getDeliveryMode() {
   return DeliveryMode.PERSISTENT;
}
 

public static void main(String[] args) throws Exception {
    CommandLine cmd = parseAndValidateCommandLineArguments(args);
    final WrapInt count = new WrapInt();
    final long ds = System.currentTimeMillis();

    final int interval = Integer.parseInt(cmd.getOptionValue("interval", "1000"));
    String name = cmd.getOptionValue("name", UUID.randomUUID().toString());
    int deliveryMode = cmd.hasOption("notPersistent") ? DeliveryMode.NON_PERSISTENT : DeliveryMode.PERSISTENT;
    registerShutdownHook(count, ds, interval);

    try {
        String user = null;
        String password = null;
        String secrets = null;            
        if (cmd.hasOption("user") && cmd.hasOption("password")) {
            user = cmd.getOptionValue("user");
            password = cmd.getOptionValue("password");                
        } else {
            secrets = getUserPassword("MQBrokerUserPassword");
            if (secrets!=null && !secrets.isEmpty()) {
                user = secrets.split(",")[0];
                password = secrets.split(",")[1];
            }
        }
        JmsConnectionFactory connFact = new JmsConnectionFactory(user, password, cmd.getOptionValue("url"));
        JmsConnection conn = (JmsConnection) connFact.createConnection();
        conn.setClientID("AmazonMQWorkshop-" + System.currentTimeMillis());
        conn.start();

        Session session = conn.createSession(false, Session.CLIENT_ACKNOWLEDGE);

        if (cmd.getOptionValue("mode").contentEquals("sender")) {
            if (cmd.getOptionValue("type").contentEquals("queue")) {
                MessageProducer queueMessageProducer = session.createProducer(session.createQueue(cmd.getOptionValue("type") + "://" + cmd.getOptionValue("destination")));
                sendMessages(session, queueMessageProducer, name, interval, deliveryMode, count);
            } else {
                MessageProducer topicMessageProducer = session.createProducer(session.createTopic(cmd.getOptionValue("type") + "://" + cmd.getOptionValue("destination")));
                sendMessages(session, topicMessageProducer, name, interval, deliveryMode, count);
            }
        } else {
            if (cmd.getOptionValue("type").contentEquals("queue")) {
                MessageConsumer queueConsumer = session.createConsumer(session.createQueue(cmd.getOptionValue("destination")));
                receiveMessages(session, queueConsumer);
            } else {
                MessageConsumer topicConsumer = session.createConsumer(session.createTopic(cmd.getOptionValue("destination")));
                receiveMessages(session, topicConsumer);
            }
        }
    } catch (javax.jms.JMSSecurityException ex) {
        System.out.println(String.format("Error: %s", ex.getMessage()));
        System.exit(1);
    }
}
 

@Override
public int getJMSDeliveryMode() throws JMSException {
   return message.isDurable() ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT;
}
 

@Override
public int getJMSDeliveryMode() throws JMSException {
    return persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT;
}
 

@Override
protected void setUp() throws Exception {
   this.topic = false;
   this.deliveryMode = DeliveryMode.PERSISTENT;
   super.setUp();
}
 

public void createProducer(final CreateProducerCommand command)
{
    try
    {
        final Session session = _testSessions.get(command.getSessionName());
        if (session == null)
        {
            throw new DistributedTestException("No test session found called: " + command.getSessionName(), command);
        }

        synchronized(session)
        {
            final Destination destination;
            if(command.isTopic())
            {
                destination = session.createTopic(command.getDestinationName());
            }
            else
            {
                destination = session.createQueue(command.getDestinationName());
            }

            final MessageProducer jmsProducer = session.createProducer(destination);

            if (command.getPriority() != -1)
            {
                jmsProducer.setPriority(command.getPriority());
            }
            if (command.getTimeToLive() > 0)
            {
                jmsProducer.setTimeToLive(command.getTimeToLive());
            }

            if (command.getDeliveryMode() == DeliveryMode.NON_PERSISTENT
                    || command.getDeliveryMode() == DeliveryMode.PERSISTENT)
            {
                jmsProducer.setDeliveryMode(command.getDeliveryMode());
            }

            addProducer(command.getParticipantName(), jmsProducer);
        }
    }
    catch (final JMSException jmse)
    {
        throw new DistributedTestException("Unable to create new producer: " + command, jmse);
    }
}
 

@Test
public void testCreateForProducer()
{
    CreateProducerCommand command = new CreateProducerCommand();
    setCommonCommandFields(command);

    int deliveryMode = DeliveryMode.PERSISTENT;
    command.setDeliveryMode(deliveryMode);

    int priority = 5;
    command.setPriority(priority);

    long producerInterval = 50;
    command.setInterval(producerInterval);

    long timeToLive = 60;
    command.setTimeToLive(timeToLive);

    int totalNumberOfConsumers = 0;
    int totalNumberOfProducers = 1;

    int acknowledgeMode = 1;

    ProducerParticipantResult result = _participantResultFactory.createForProducer(PARTICIPANT_NAME,
                                                                                   REGISTERED_CLIENT_NAME,
                                                                                   command,
                                                                                   acknowledgeMode,
                                                                                   NUMBER_OF_MESSAGES_PROCESSED,
                                                                                   PAYLOAD_SIZE,
                                                                                   TOTAL_PAYLOAD_PROCESSED,
                                                                                   START, END,
                                                                                   PROVIDER_VERSION,
                                                                                   PROTOCOL_VERSION);

    assertCommonResultProperties(result);

    assertEquals((long) deliveryMode, (long) result.getDeliveryMode());
    assertEquals((long) acknowledgeMode, (long) result.getAcknowledgeMode());
    assertEquals((long) priority, (long) result.getPriority());
    assertEquals(producerInterval, result.getInterval());
    assertEquals(timeToLive, result.getTimeToLive());
    assertEquals((long) totalNumberOfConsumers, (long) result.getTotalNumberOfConsumers());
}
 

protected int getDeliveryMode() {
   return DeliveryMode.PERSISTENT;
}
 

public void testDurableTopicMessage() throws Exception {
   super.topic = true;
   deliveryMode = DeliveryMode.PERSISTENT;
   durableSubscriber = true;
   doTest();
}
 

/**
 * Render a {@link Message}.
 */
public String doRender(final Object o) {

   if (o instanceof Message) {
      final StringBuilder sbuf = new StringBuilder(100);
      final Message m = (Message) o;
      try {
         sbuf.append("DeliveryMode=");
         switch (m.getJMSDeliveryMode()) {
            case DeliveryMode.NON_PERSISTENT:
               sbuf.append("NON_PERSISTENT");
               break;
            case DeliveryMode.PERSISTENT:
               sbuf.append("PERSISTENT");
               break;
            default:
               sbuf.append("UNKNOWN");
         }
         sbuf.append(", CorrelationID=");
         sbuf.append(m.getJMSCorrelationID());

         sbuf.append(", Destination=");
         sbuf.append(m.getJMSDestination());

         sbuf.append(", Expiration=");
         sbuf.append(m.getJMSExpiration());

         sbuf.append(", MessageID=");
         sbuf.append(m.getJMSMessageID());

         sbuf.append(", Priority=");
         sbuf.append(m.getJMSPriority());

         sbuf.append(", Redelivered=");
         sbuf.append(m.getJMSRedelivered());

         sbuf.append(", ReplyTo=");
         sbuf.append(m.getJMSReplyTo());

         sbuf.append(", Timestamp=");
         sbuf.append(m.getJMSTimestamp());

         sbuf.append(", Type=");
         sbuf.append(m.getJMSType());

         //Enumeration enum = m.getPropertyNames();
         //while(enum.hasMoreElements()) {
         //  String key = (String) enum.nextElement();
         //  sbuf.append("; "+key+"=");
         //  sbuf.append(m.getStringProperty(key));
         //}

      } catch (final JMSException e) {
         LogLog.error("Could not parse Message.", e);
      }
      return sbuf.toString();
   } else {
      return o.toString();
   }
}
 

@Override
public final int getJMSDeliveryMode() throws JMSException {
   return message.isDurable() ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT;
}
 

/**
 * Returns delivery mode.
 *
 * @return int - persistent delivery mode.
 */
@Override
protected int getDeliveryMode() {
   return DeliveryMode.PERSISTENT;
}
 

/**
 * Returns delivery mode.
 *
 * @return int - persistent delivery mode.
 */
protected int getDeliveryMode() {
   return DeliveryMode.PERSISTENT;
}