javax.jms.XASession#createProducer ( )源码实例Demo

@Test
public void testXAPrepare() throws Exception {
   try {

      XAConnection connection = xaFactory.createXAConnection();

      XASession xasession = connection.createXASession();

      Xid xid = newXID();
      xasession.getXAResource().start(xid, XAResource.TMNOFLAGS);
      Queue queue = xasession.createQueue(queueName);
      MessageProducer producer = xasession.createProducer(queue);
      producer.send(xasession.createTextMessage("hello"));
      producer.send(xasession.createTextMessage("hello"));
      xasession.getXAResource().end(xid, XAResource.TMSUCCESS);

      xasession.getXAResource().prepare(xid);

      connection.close();

      System.err.println("Done!!!");
   } catch (Exception e) {
      e.printStackTrace();
   }
}
 

@Test
public void testXAResourceCommittedRemoved() throws Exception {
   Queue queue = null;

   Xid xid = newXID();
   try (XAConnection xaconnection = xaFactory.createXAConnection()) {
      XASession session = xaconnection.createXASession();
      queue = session.createQueue(queueName);
      session.getXAResource().start(xid, XAResource.TMNOFLAGS);
      MessageProducer producer = session.createProducer(queue);
      producer.send(session.createTextMessage("xa message"));
      session.getXAResource().end(xid, XAResource.TMSUCCESS);
      session.getXAResource().commit(xid, true);
   }
   XidImpl xid1 = new XidImpl(xid);
   Transaction transaction = server.getResourceManager().getTransaction(xid1);
   assertNull(transaction);
}
 

@Test
public void testXAResourceRolledBackRemoved() throws Exception {
   Queue queue = null;

   Xid xid = newXID();
   try (XAConnection xaconnection = xaFactory.createXAConnection()) {
      XASession session = xaconnection.createXASession();
      queue = session.createQueue(queueName);
      session.getXAResource().start(xid, XAResource.TMNOFLAGS);
      MessageProducer producer = session.createProducer(queue);
      producer.send(session.createTextMessage("xa message"));
      session.getXAResource().end(xid, XAResource.TMSUCCESS);
      session.getXAResource().rollback(xid);
   }
   XidImpl xid1 = new XidImpl(xid);
   Transaction transaction = server.getResourceManager().getTransaction(xid1);
   assertNull(transaction);
}
 

public void testCloseSendConnection() throws Exception {
   String brokerName = "closeSend";
   BrokerService broker = BrokerFactory.createBroker(new URI("broker:(tcp://localhost:0)/" + brokerName));
   broker.start();
   broker.waitUntilStarted();
   ActiveMQXAConnectionFactory cf = new ActiveMQXAConnectionFactory(broker.getTransportConnectors().get(0).getConnectUri());
   XAConnection connection = (XAConnection) cf.createConnection();
   connection.start();
   XASession session = connection.createXASession();
   XAResource resource = session.getXAResource();
   Destination dest = new ActiveMQQueue(getName());

   // publish a message
   Xid tid = createXid();
   resource.start(tid, XAResource.TMNOFLAGS);
   MessageProducer producer = session.createProducer(dest);
   ActiveMQTextMessage message = new ActiveMQTextMessage();
   message.setText(getName());
   producer.send(message);

   connection.close();

   //comment out this check as it doesn't apply to artemis
   //assertTransactionGoneFromBroker(tid);

   broker.stop();
}
 

/** */
public static void main(String[] args) throws Exception {
    XAConnectionFactory factory = getXAConnectionFactory();

    if (factory == null) {
        throw new IllegalArgumentException("Provide a factory for the broker in the getXAConnectionFactory() method");
    }

    // create a connection, session and XA transaction
    XAConnection conn = factory.createXAConnection();
    XASession sess = conn.createXASession();
    XAResource xaRes = sess.getXAResource();
    Xid xid = new MyXid(1);

    // start the transaction and produce one message
    xaRes.start(xid, XAResource.TMNOFLAGS);
    MessageProducer producer = sess.createProducer(sess.createQueue("queue"));
    producer.send(sess.createTextMessage("foo"));
    xaRes.end(xid, XAResource.TMSUCCESS);

    // prepare the transaction
    xaRes.prepare(xid);

    // now disconnect. Some brokers roll back the transaction, but this is not
    // compatible with Jet's fault tolerance.
    conn.close();

    // connect again
    conn = factory.createXAConnection();
    conn.start();
    sess = conn.createXASession();
    xaRes = sess.getXAResource();

    // commit the prepared transaction
    xaRes.commit(xid, false);

    // check that the message is there
    MessageConsumer cons = sess.createConsumer(sess.createQueue("queue"));
    TextMessage msg = (TextMessage) cons.receive(TIMEOUT);
    if (msg == null || !msg.getText().equals("foo")) {
        System.err.println("Message is missing or has wrong text, transaction probably lost");
    } else {
        System.out.println("Success!");
    }
    conn.close();
}
 

@Test
public void test2PCSendFailOnPrepare() throws Exception {
   XAConnection conn = null;
   Connection conn2 = null;
   try {
      conn = xacf.createXAConnection();

      tm.begin();

      XASession sess = conn.createXASession();
      XAResource res = sess.getXAResource();

      // prevent 1Pc optimisation
      // res.setForceNotSameRM(true);

      XAResource res2 = new DummyXAResource(true);
      XAResource res3 = new DummyXAResource();
      XAResource res4 = new DummyXAResource();

      Transaction tx = tm.getTransaction();
      tx.enlistResource(res);
      tx.enlistResource(res2);
      tx.enlistResource(res3);
      tx.enlistResource(res4);

      MessageProducer prod = sess.createProducer(null);
      prod.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
      Message m = sess.createTextMessage("XATest1");
      prod.send(queue1, m);
      m = sess.createTextMessage("XATest2");
      prod.send(queue1, m);

      tx.delistResource(res, XAResource.TMSUCCESS);
      tx.delistResource(res2, XAResource.TMSUCCESS);
      tx.delistResource(res3, XAResource.TMSUCCESS);
      tx.delistResource(res4, XAResource.TMSUCCESS);

      try {
         tm.commit();

         Assert.fail("should not get here");
      } catch (Exception e) {
         // We should expect this
      }

      conn2 = cf.createConnection();
      conn2.start();
      Session sessReceiver = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer cons = sessReceiver.createConsumer(queue1);
      Message m2 = cons.receive(100);
      Assert.assertNull(m2);
   } finally {
      if (conn != null) {
         conn.close();
      }
      if (conn2 != null) {
         conn2.close();
      }
   }
}
 

@Test
public void test2PCSendRollback() throws Exception {
   XAConnection conn = null;
   Connection conn2 = null;
   try {
      conn = xacf.createXAConnection();

      tm.begin();

      XASession sess = conn.createXASession();
      XAResource res = sess.getXAResource();

      // prevent 1Pc optimisation
      // res.setForceNotSameRM(true);

      XAResource res2 = new DummyXAResource();

      Transaction tx = tm.getTransaction();
      tx.enlistResource(res);
      tx.enlistResource(res2);

      MessageProducer prod = sess.createProducer(null);
      prod.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
      Message m = sess.createTextMessage("XATest1");
      prod.send(queue1, m);
      m = sess.createTextMessage("XATest2");
      prod.send(queue1, m);

      tx.delistResource(res, XAResource.TMSUCCESS);
      tx.delistResource(res2, XAResource.TMSUCCESS);

      tm.rollback();

      conn2 = cf.createConnection();
      conn2.start();
      Session sessReceiver = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer cons = sessReceiver.createConsumer(queue1);
      Message m2 = cons.receive(100);
      Assert.assertNull(m2);

   } finally {
      if (conn != null) {
         conn.close();
      }
      if (conn2 != null) {
         conn2.close();
      }
   }
}
 

@Test
public void testMultipleSessionsOneTxCommitSend1PCOptimization() throws Exception {
   // Since both resources have some RM, TM will probably use 1PC optimization

   XAConnection conn = null;

   Connection conn2 = null;

   try {
      conn = xacf.createXAConnection();
      conn.start();

      tm.begin();

      // Create 2 sessions and enlist them
      XASession sess1 = conn.createXASession();
      XAResource res1 = sess1.getXAResource();
      XASession sess2 = conn.createXASession();
      XAResource res2 = sess2.getXAResource();

      Transaction tx = tm.getTransaction();
      tx.enlistResource(res1);
      tx.enlistResource(res2);

      // Send 2 messages - one from each session

      MessageProducer prod1 = sess1.createProducer(queue1);
      MessageProducer prod2 = sess2.createProducer(queue1);

      prod1.send(sess1.createTextMessage("echidna1"));
      prod2.send(sess2.createTextMessage("echidna2"));

      tx.delistResource(res1, XAResource.TMSUCCESS);
      tx.delistResource(res2, XAResource.TMSUCCESS);

      // commit
      tm.commit();

      // Messages should be in queue

      conn2 = cf.createConnection();
      Session sess = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer cons = sess.createConsumer(queue1);
      conn2.start();

      TextMessage r1 = (TextMessage) cons.receive(5000);
      Assert.assertNotNull(r1);
      Assert.assertEquals("echidna1", r1.getText());

      TextMessage r2 = (TextMessage) cons.receive(5000);
      Assert.assertNotNull(r2);
      Assert.assertEquals("echidna2", r2.getText());

   } finally {
      if (conn != null) {
         conn.close();
      }
      if (conn2 != null) {
         conn2.close();
      }
   }
}
 

@Test
public void testMultipleSessionsOneTxCommitSend() throws Exception {
   // Since both resources have some RM, TM will probably use 1PC optimization

   XAConnection conn = null;

   Connection conn2 = null;

   try {

      conn = xacf.createXAConnection();
      conn.start();

      tm.begin();

      // Create 2 sessions and enlist them
      XASession sess1 = conn.createXASession();
      ClientSessionInternal res1 = (ClientSessionInternal) sess1.getXAResource();
      XASession sess2 = conn.createXASession();
      ClientSessionInternal res2 = (ClientSessionInternal) sess2.getXAResource();
      res1.setForceNotSameRM(true);
      res2.setForceNotSameRM(true);

      Transaction tx = tm.getTransaction();
      tx.enlistResource(res1);
      tx.enlistResource(res2);

      // Send 2 messages - one from each session

      MessageProducer prod1 = sess1.createProducer(queue1);
      MessageProducer prod2 = sess2.createProducer(queue1);

      prod1.send(sess1.createTextMessage("echidna1"));
      prod2.send(sess2.createTextMessage("echidna2"));

      tx.delistResource(res1, XAResource.TMSUCCESS);
      tx.delistResource(res2, XAResource.TMSUCCESS);

      // commit
      tm.commit();

      // Messages should be in queue

      conn2 = cf.createConnection();
      Session sess = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer cons = sess.createConsumer(queue1);
      conn2.start();

      TextMessage r1 = (TextMessage) cons.receive(5000);
      Assert.assertNotNull(r1);
      Assert.assertEquals("echidna1", r1.getText());

      TextMessage r2 = (TextMessage) cons.receive(5000);
      Assert.assertNotNull(r2);
      Assert.assertEquals("echidna2", r2.getText());

   } finally {
      if (conn != null) {
         conn.close();
      }
      if (conn2 != null) {
         conn2.close();
      }

   }

}
 

@Test
public void testMultipleSessionsOneTxRollbackSend1PCOptimization() throws Exception {
   // Since both resources have some RM, TM will probably use 1PC optimization

   XAConnection conn = null;

   Connection conn2 = null;

   try {
      conn = xacf.createXAConnection();
      conn.start();

      tm.begin();

      // Create 2 sessions and enlist them
      XASession sess1 = conn.createXASession();
      XAResource res1 = sess1.getXAResource();
      XASession sess2 = conn.createXASession();
      XAResource res2 = sess2.getXAResource();

      Transaction tx = tm.getTransaction();
      tx.enlistResource(res1);
      tx.enlistResource(res2);

      // Send 2 messages - one from each session

      MessageProducer prod1 = sess1.createProducer(queue1);
      MessageProducer prod2 = sess2.createProducer(queue1);

      prod1.send(sess1.createTextMessage("echidna1"));
      prod2.send(sess2.createTextMessage("echidna2"));

      tx.delistResource(res1, XAResource.TMSUCCESS);
      tx.delistResource(res2, XAResource.TMSUCCESS);

      // rollback
      tm.rollback();

      // Messages should not be in queue

      conn2 = cf.createConnection();
      Session sess = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer cons = sess.createConsumer(queue1);
      conn2.start();

      TextMessage r1 = (TextMessage) cons.receive(100);
      Assert.assertNull(r1);

   } finally {
      if (conn != null) {
         conn.close();
      }
      if (conn2 != null) {
         conn2.close();
      }
   }
}
 

@Test
public void testMultipleSessionsOneTxRollbackSend() throws Exception {
   XAConnection conn = null;

   Connection conn2 = null;

   try {

      conn = xacf.createXAConnection();
      conn.start();

      tm.begin();

      // Create 2 sessions and enlist them
      XASession sess1 = conn.createXASession();
      ClientSessionInternal res1 = (ClientSessionInternal) sess1.getXAResource();
      XASession sess2 = conn.createXASession();
      ClientSessionInternal res2 = (ClientSessionInternal) sess2.getXAResource();
      res1.setForceNotSameRM(true);
      res2.setForceNotSameRM(true);

      Transaction tx = tm.getTransaction();
      tx.enlistResource(res1);
      tx.enlistResource(res2);

      // Send 2 messages - one from each session

      MessageProducer prod1 = sess1.createProducer(queue1);
      MessageProducer prod2 = sess2.createProducer(queue1);

      prod1.send(sess1.createTextMessage("echidna1"));
      prod2.send(sess2.createTextMessage("echidna2"));

      tx.delistResource(res1, XAResource.TMSUCCESS);
      tx.delistResource(res2, XAResource.TMSUCCESS);

      // rollback
      tm.rollback();

      // Messages should not be in queue

      conn2 = cf.createConnection();
      Session sess = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer cons = sess.createConsumer(queue1);
      conn2.start();

      TextMessage r1 = (TextMessage) cons.receive(100);
      Assert.assertNull(r1);
   } finally {
      if (conn != null) {
         conn.close();
      }
      if (conn2 != null) {
         conn2.close();
      }
   }
}
 

@Test
public void testOneSessionTwoTransactionsCommitSend() throws Exception {
   XAConnection conn = null;

   Connection conn2 = null;

   try {
      conn = xacf.createXAConnection();

      // Create a session
      XASession sess1 = conn.createXASession();
      XAResource res1 = sess1.getXAResource();

      MessageProducer prod1 = sess1.createProducer(queue1);

      tm.begin();

      Transaction tx1 = tm.getTransaction();
      tx1.enlistResource(res1);

      // Send a message
      prod1.send(sess1.createTextMessage("kangaroo1"));

      tx1.delistResource(res1, XAResource.TMSUCCESS);

      // suspend the tx
      Transaction suspended = tm.suspend();

      tm.begin();

      // Send another message in another tx using the same session
      Transaction tx2 = tm.getTransaction();
      tx2.enlistResource(res1);

      // Send a message
      prod1.send(sess1.createTextMessage("kangaroo2"));

      tx2.delistResource(res1, XAResource.TMSUCCESS);

      // commit this transaction
      tm.commit();

      // verify only kangaroo2 message is sent
      conn2 = cf.createConnection();
      Session sess = conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);
      conn2.start();
      MessageConsumer cons = sess.createConsumer(queue1);
      TextMessage r1 = (TextMessage) cons.receive(5000);
      Assert.assertNotNull(r1);
      Assert.assertEquals("kangaroo2", r1.getText());
      TextMessage r2 = (TextMessage) cons.receive(100);
      Assert.assertNull(r2);

      // now resume the first tx and then commit it
      tm.resume(suspended);

      tm.commit();

      // verify that the first text message is received
      TextMessage r3 = (TextMessage) cons.receive(5000);
      Assert.assertNotNull(r3);
      Assert.assertEquals("kangaroo1", r3.getText());

   } finally {
      if (conn != null) {
         conn.close();
      }
      if (conn2 != null) {
         conn2.close();
      }

   }

}
 

public void testConsumerCloseTransactionalSendReceive() throws Exception {

      ActiveMQXAConnectionFactory cf1 = new ActiveMQXAConnectionFactory("vm://localhost?broker.persistent=false");
      XAConnection connection1 = (XAConnection) cf1.createConnection();
      connection1.start();
      XASession session = connection1.createXASession();
      XAResource resource = session.getXAResource();
      Destination dest = new ActiveMQQueue(getName());

      // publish a message
      Xid tid = createXid();
      resource.start(tid, XAResource.TMNOFLAGS);
      MessageProducer producer = session.createProducer(dest);
      ActiveMQTextMessage message = new ActiveMQTextMessage();
      message.setText(getName());
      producer.send(message);
      producer.close();
      resource.end(tid, XAResource.TMSUCCESS);
      resource.commit(tid, true);
      session.close();

      session = connection1.createXASession();
      MessageConsumer consumer = session.createConsumer(dest);
      tid = createXid();
      resource = session.getXAResource();
      resource.start(tid, XAResource.TMNOFLAGS);
      TextMessage receivedMessage = (TextMessage) consumer.receive(1000);
      consumer.close();
      assertNotNull(receivedMessage);
      assertEquals(getName(), receivedMessage.getText());
      resource.end(tid, XAResource.TMSUCCESS);
      resource.commit(tid, true);

      session = connection1.createXASession();
      consumer = session.createConsumer(dest);
      tid = createXid();
      resource = session.getXAResource();
      resource.start(tid, XAResource.TMNOFLAGS);
      assertNull(consumer.receive(1000));
      resource.end(tid, XAResource.TMSUCCESS);
      resource.commit(tid, true);

   }
 

public void testSessionCloseTransactionalSendReceive() throws Exception {

      ActiveMQXAConnectionFactory cf1 = new ActiveMQXAConnectionFactory("vm://localhost?broker.persistent=false");
      XAConnection connection1 = (XAConnection) cf1.createConnection();
      connection1.start();
      XASession session = connection1.createXASession();
      XAResource resource = session.getXAResource();
      Destination dest = new ActiveMQQueue(getName());

      // publish a message
      Xid tid = createXid();
      resource.start(tid, XAResource.TMNOFLAGS);
      MessageProducer producer = session.createProducer(dest);
      ActiveMQTextMessage message = new ActiveMQTextMessage();
      message.setText(getName());
      producer.send(message);
      session.close();
      resource.end(tid, XAResource.TMSUCCESS);
      resource.commit(tid, true);

      session = connection1.createXASession();
      MessageConsumer consumer = session.createConsumer(dest);
      tid = createXid();
      resource = session.getXAResource();
      resource.start(tid, XAResource.TMNOFLAGS);
      TextMessage receivedMessage = (TextMessage) consumer.receive(1000);
      session.close();
      assertNotNull(receivedMessage);
      assertEquals(getName(), receivedMessage.getText());
      resource.end(tid, XAResource.TMSUCCESS);
      resource.commit(tid, true);

      session = connection1.createXASession();
      consumer = session.createConsumer(dest);
      tid = createXid();
      resource = session.getXAResource();
      resource.start(tid, XAResource.TMNOFLAGS);
      assertNull(consumer.receive(1000));
      resource.end(tid, XAResource.TMSUCCESS);
      resource.commit(tid, true);
   }