import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test
public void testDefaultsToLocalURI() {
    load(EmptyConfiguration.class);

    JmsTemplate jmsTemplate = this.context.getBean(JmsTemplate.class);
    ConnectionFactory connectionFactory =
        this.context.getBean(ConnectionFactory.class);

    assertTrue(connectionFactory instanceof JmsConnectionFactory);

    JmsConnectionFactory qpidJmsFactory = (JmsConnectionFactory) connectionFactory;

    assertEquals(jmsTemplate.getConnectionFactory(), connectionFactory);
    assertEquals("amqp://localhost:5672", qpidJmsFactory.getRemoteURI());
    assertNull(qpidJmsFactory.getUsername());
    assertNull(qpidJmsFactory.getPassword());
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test
public void testCustomConnectionFactorySettings() {
    load(EmptyConfiguration.class,
         "spring.qpidjms.remoteURL=amqp://127.0.0.1:5672",
         "spring.qpidjms.username=foo",
         "spring.qpidjms.password=bar");

    JmsTemplate jmsTemplate = this.context.getBean(JmsTemplate.class);
    JmsConnectionFactory connectionFactory =
        this.context.getBean(JmsConnectionFactory.class);

    assertEquals(jmsTemplate.getConnectionFactory(), connectionFactory);
    assertEquals("amqp://127.0.0.1:5672", connectionFactory.getRemoteURI());
    assertEquals("foo", connectionFactory.getUsername());
    assertEquals("bar", connectionFactory.getPassword());
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test
public void testReceiveLocalOnlyOptionsAppliedFromEnvOverridesURI() {
    load(EmptyConfiguration.class,
         "spring.qpidjms.remoteURL=amqp://127.0.0.1:5672" +
             "?jms.receiveLocalOnly=false&jms.receiveNoWaitLocalOnly=false",
         "spring.qpidjms.receiveLocalOnly=true",
         "spring.qpidjms.receiveNoWaitLocalOnly=true");

    JmsTemplate jmsTemplate = this.context.getBean(JmsTemplate.class);
    JmsConnectionFactory connectionFactory =
        this.context.getBean(JmsConnectionFactory.class);

    assertEquals(jmsTemplate.getConnectionFactory(), connectionFactory);

    assertTrue(connectionFactory.isReceiveLocalOnly());
    assertTrue(connectionFactory.isReceiveNoWaitLocalOnly());
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
   Connection connection = null;
   String csvData = System.getProperty(CSVDATA);
   if(CSVDATA == null || CSVDATA.equals(""))
       throw new RuntimeException("LoyaltyCardManager.main() must pass the "+CSVDATA +" system property With format  OPERATION;USERID;FIRSTNAME;LASTNAME;TRXID;TRXFEESAMOUNT;CURRENCY");
   System.out.println("LoyaltyCardManager() will connect to router: "+ROUTER_URL+" : at the following address: "+QUEUE_NAME);
   ConnectionFactory connectionFactory = new JmsConnectionFactory(ROUTER_URL);
   try {
      // Step 1. Create an AMQP qpid connection
      connection = connectionFactory.createConnection();
      // Step 2. Create a JMS session
      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      // Step 3. Create a Producer
      Queue fidelityRequestQueue = session.createQueue(QUEUE_NAME);
      MessageProducer beosbankFidelityRequestProducer = session.createProducer(fidelityRequestQueue);
      // Step 4. send a CSV Text Data on user transactions 
      beosbankFidelityRequestProducer.send(session.createTextMessage(csvData));
      System.out.println("\nmessage sent:"+ csvData+" \n");
   } finally {
      if (connection != null) {
         // Step 9. close the connection
         connection.close();
      }
   }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test
public void testCustomConnectionFactorySettings() {
    load(EmptyConfiguration.class,
         "amqphub.amqp10jms.remote-url=amqp://127.0.0.1:5672",
         "amqphub.amqp10jms.username=foo",
         "amqphub.amqp10jms.password=bar");

    JmsTemplate jmsTemplate = this.context.getBean(JmsTemplate.class);
    JmsConnectionFactory connectionFactory =
        this.context.getBean(JmsConnectionFactory.class);

    assertEquals(jmsTemplate.getConnectionFactory(), connectionFactory);
    assertEquals("amqp://127.0.0.1:5672", connectionFactory.getRemoteURI());
    assertEquals("foo", connectionFactory.getUsername());
    assertEquals("bar", connectionFactory.getPassword());
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test
public void testReceiveLocalOnlyOptionsAppliedFromEnvOverridesURI() {
    load(EmptyConfiguration.class,
         "amqphub.amqp10jms.remote-url=amqp://127.0.0.1:5672" +
             "?jms.receiveLocalOnly=false&jms.receiveNoWaitLocalOnly=false",
         "amqphub.amqp10jms.receiveLocalOnly=true",
         "amqphub.amqp10jms.receiveNoWaitLocalOnly=true");

    JmsTemplate jmsTemplate = this.context.getBean(JmsTemplate.class);
    JmsConnectionFactory connectionFactory =
        this.context.getBean(JmsConnectionFactory.class);

    assertEquals(jmsTemplate.getConnectionFactory(), connectionFactory);

    assertTrue(connectionFactory.isReceiveLocalOnly());
    assertTrue(connectionFactory.isReceiveNoWaitLocalOnly());
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
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;
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test(timeout = 30000)
public void testSendReceiveOverWS() throws Exception {
   JmsConnectionFactory factory = new JmsConnectionFactory(getBrokerQpidJMSConnectionURI());
   JmsConnection connection = (JmsConnection) factory.createConnection();

   try {
      Session session = connection.createSession();
      Queue queue = session.createQueue(getQueueName());

      MessageProducer producer = session.createProducer(queue);
      producer.send(session.createMessage());
      producer.close();

      connection.start();

      MessageConsumer consumer = session.createConsumer(queue);
      Message message = consumer.receive(1000);

      assertNotNull(message);
   } finally {
      connection.close();
   }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test(timeout = 30000)
public void testSendLargeMessageToClientFromOpenWire() throws Exception {
   JmsConnectionFactory factory = new JmsConnectionFactory(getBrokerQpidJMSConnectionURI());
   JmsConnection connection = (JmsConnection) factory.createConnection();

   sendLargeMessageViaOpenWire();

   try {
      Session session = connection.createSession();
      Queue queue = session.createQueue(getQueueName());
      connection.start();

      MessageConsumer consumer = session.createConsumer(queue);
      Message message = consumer.receive(1000);

      assertNotNull(message);
      assertTrue(message instanceof BytesMessage);
   } finally {
      connection.close();
   }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Ignore("Broker can't accept messages over 65535 right now")
@Test(timeout = 30000)
public void testSendLargeMessageToClientFromAMQP() throws Exception {
   JmsConnectionFactory factory = new JmsConnectionFactory(getBrokerQpidJMSConnectionURI());
   JmsConnection connection = (JmsConnection) factory.createConnection();

   sendLargeMessageViaAMQP();

   try {
      Session session = connection.createSession();
      Queue queue = session.createQueue(getQueueName());
      connection.start();

      MessageConsumer consumer = session.createConsumer(queue);
      Message message = consumer.receive(1000);

      assertNotNull(message);
      assertTrue(message instanceof BytesMessage);
   } finally {
      connection.close();
   }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test(timeout = 120000)
public void testFailoverListWithAMQP() throws Exception {
   JmsConnectionFactory factory = getJmsConnectionFactory();
   try (Connection connection = factory.createConnection()) {
      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      javax.jms.Queue queue = session.createQueue(ADDRESS.toString());
      MessageProducer producer = session.createProducer(queue);
      producer.send(session.createTextMessage("hello before failover"));
      liveServer.crash(true, true);
      producer.send(session.createTextMessage("hello after failover"));
      MessageConsumer consumer = session.createConsumer(queue);
      connection.start();
      TextMessage receive = (TextMessage) consumer.receive(5000);
      Assert.assertNotNull(receive);
      Assert.assertEquals("hello before failover", receive.getText());
      receive = (TextMessage) consumer.receive(5000);
      Assert.assertEquals("hello after failover", receive.getText());
      Assert.assertNotNull(receive);
   }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
private void createConnectionFactories(Hashtable<Object, Object> environment, Map<String, Object> bindings) throws NamingException {
    Map<String, String> factories = getConnectionFactoryNamesAndURIs(environment);
    Map<String, String> defaults = getConnectionFactoryDefaults(environment);
    for (Entry<String, String> entry : factories.entrySet()) {
        String name = entry.getKey();
        String uri = entry.getValue();

        JmsConnectionFactory factory = null;
        try {
            factory = createConnectionFactory(name, uri, defaults, environment);
        } catch (Exception e) {
            NamingException ne = new NamingException("Exception while creating ConnectionFactory '" + name + "'.");
            ne.initCause(e);
            throw ne;
        }

        bindings.put(name, factory);
    }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
protected JmsConnectionFactory createConnectionFactory(String name, String uri, Map<String, String> defaults, Hashtable<Object, Object> environment) throws URISyntaxException {
    Map<String, String> props = new LinkedHashMap<String, String>();

    // Add the defaults which apply to all connection factories
    props.putAll(defaults);

    // Add any URI entry for this specific factory name
    if (uri != null && !uri.trim().isEmpty()) {
        props.put(JmsConnectionFactory.REMOTE_URI_PROP, uri);
    }

    // Add any factory-specific additional properties
    props.putAll(getConnectionFactoryProperties(name, environment));

    return createConnectionFactory(props);
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test(timeout = 20000)
public void testConnectionPropertiesContainExpectedMetaData() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {

        Matcher<?> connPropsMatcher = allOf(hasEntry(AmqpSupport.PRODUCT, MetaDataSupport.PROVIDER_NAME),
                hasEntry(AmqpSupport.VERSION, MetaDataSupport.PROVIDER_VERSION),
                hasEntry(AmqpSupport.PLATFORM, MetaDataSupport.PLATFORM_DETAILS));

        testPeer.expectSaslAnonymous();
        testPeer.expectOpen(connPropsMatcher, null, false);
        testPeer.expectBegin();

        ConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:" + testPeer.getServerPort() + "?jms.clientID=foo");
        Connection connection = factory.createConnection();

        testPeer.waitForAllHandlersToComplete(1000);
        assertNull(testPeer.getThrowable());

        testPeer.expectClose();
        connection.close();
    }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
private void doMaxFrameSizeOptionTestImpl(int uriOption, UnsignedInteger transmittedValue) throws JMSException, InterruptedException, Exception, IOException {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        testPeer.expectSaslLayerDisabledConnect(equalTo(transmittedValue));
        // Each connection creates a session for managing temporary destinations etc
        testPeer.expectBegin();

        String uri = "amqp://localhost:" + testPeer.getServerPort() + "?amqp.saslLayer=false&amqp.maxFrameSize=" + uriOption;
        ConnectionFactory factory = new JmsConnectionFactory(uri);
        Connection connection = factory.createConnection();
        connection.start();

        testPeer.waitForAllHandlersToComplete(3000);
        assertNull(testPeer.getThrowable());

        testPeer.expectClose();
        connection.close();
    }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
private void doAmqpHostnameTestImpl(String amqpHostname, boolean setHostnameOption, Matcher<?> hostnameMatcher) throws JMSException, InterruptedException, Exception, IOException {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        testPeer.expectSaslAnonymous();
        testPeer.expectOpen(null, hostnameMatcher, false);
        // Each connection creates a session for managing temporary destinations etc
        testPeer.expectBegin();

        String uri = "amqp://localhost:" + testPeer.getServerPort();
        if(setHostnameOption) {
            uri += "?amqp.vhost=" + amqpHostname;
        }

        ConnectionFactory factory = new JmsConnectionFactory(uri);
        Connection connection = factory.createConnection();
        // Set a clientID to provoke the actual AMQP connection process to occur.
        connection.setClientID("clientName");

        testPeer.waitForAllHandlersToComplete(1000);
        assertNull(testPeer.getThrowable());

        testPeer.expectClose();
        connection.close();
    }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test(timeout = 20000)
public void testCreateConnectionWithServerSendingPreemptiveData() throws Exception {
    boolean sendServerSaslHeaderPreEmptively = true;
    try (TestAmqpPeer testPeer = new TestAmqpPeer(null, false, sendServerSaslHeaderPreEmptively);) {
        // Don't use test fixture, handle the connection directly to control sasl behaviour
        testPeer.expectSaslAnonymousWithPreEmptiveServerHeader();
        testPeer.expectOpen();
        testPeer.expectBegin();

        JmsConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:" + testPeer.getServerPort());
        Connection connection = factory.createConnection();
        connection.start();

        testPeer.expectClose();
        connection.close();
    }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test
public void testCreateQueueSender() throws Exception {
    JmsConnectionFactory factory = new JmsConnectionFactory(getBrokerAmqpConnectionURI());
    QueueConnection connection = factory.createQueueConnection();
    assertNotNull(connection);

    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);
    Queue queue = session.createQueue(name.getMethodName());
    QueueSender sender = session.createSender(queue);
    assertNotNull(sender);

    QueueViewMBean proxy = getProxyToQueue(name.getMethodName());
    assertEquals(0, proxy.getQueueSize());
    connection.close();
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test(timeout=20000)
public void testCreateConnectionGoodProviderURI() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        // Ignore errors from peer close due to not sending any Open / Close frames
        testPeer.setSuppressReadExceptionOnClose(true);

        // DONT create a test fixture, we will drive everything directly.
        testPeer.expectSaslAnonymous();

        JmsConnectionFactory factory = new JmsConnectionFactory(new URI("amqp://127.0.0.1:" + testPeer.getServerPort()));
        Connection connection = factory.createConnection();
        assertNotNull(connection);

        testPeer.waitForAllHandlersToComplete(1000);

        testPeer.expectOpen();
        testPeer.expectClose();

        connection.close();

        testPeer.waitForAllHandlersToCompleteNoAssert(1000);
    }
}
 

import org.apache.qpid.jms.JmsConnectionFactory; //导入依赖的package包/类
@Test(timeout=20000)
public void testCreateConnectionGoodProviderString() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        // Ignore errors from peer close due to not sending any Open / Close frames
        testPeer.setSuppressReadExceptionOnClose(true);

        // DONT create a test fixture, we will drive everything directly.
        testPeer.expectSaslAnonymous();

        JmsConnectionFactory factory = new JmsConnectionFactory("amqp://127.0.0.1:" + testPeer.getServerPort());
        Connection connection = factory.createConnection();
        assertNotNull(connection);

        testPeer.waitForAllHandlersToComplete(1000);

        testPeer.expectOpen();
        testPeer.expectClose();

        connection.close();

        testPeer.waitForAllHandlersToCompleteNoAssert(1000);
    }
}