Java 类javax.jms.XAConnectionFactory 实例源码

@Override
protected XAJMSContext createProviderContext(String username, String password, int sessionMode) {
    if (connectionFactory instanceof ConnectionFactory) {
        if (username == null && password == null) {
            return ((XAConnectionFactory) connectionFactory).createXAContext();
        } else {
            return ((XAConnectionFactory) connectionFactory).createXAContext(username, password);
        }
    } else {
        throw new javax.jms.IllegalStateRuntimeException("connectionFactory should implement javax.jms.ConnectionFactory");
    }
}

@Override
public void setConnectionFactory(Object toUse) {
    if (toUse instanceof XAConnectionFactory) {
        try {
            toUse.getClass().getMethod("createContext", String.class, String.class);
            LOG.info("Porovided ConnectionFactory is JMS 2.0+ capable.");
            jmsContextSupported = true;
        } catch (NoSuchMethodException | SecurityException e) {
            LOG.info("Porovided ConnectionFactory is not JMS 2.0+ capable.");
        }

        connectionFactory = toUse;
    } else {
        throw new IllegalArgumentException("connectionFactory should implement javax.xml.XAConnectionFactory");
    }
}

protected JMSSpecificContext createXAContext(final XAConnectionFactory cf, final Options<CreateContextOption> options) {

        if (TransactionUtil.tm == null) {
            throw new NullPointerException("TransactionManager not found; is transactions module on the classpath?");
        }

        XAConnection connection = (XAConnection) DestinationUtil.mightThrow(new Callable() {
            @Override
            public Object call() throws Exception {
                if (options.has(CreateContextOption.USERNAME)) {
                    return cf.createXAConnection(options.getString(CreateContextOption.USERNAME),
                                                 options.getString(CreateContextOption.PASSWORD));
                } else {
                    return cf.createXAConnection();
                }
            }
        });

        return new JMSXAContext(connection, this,
                                (Context.Mode)options.get(CreateContextOption.MODE),
                                options.has(CreateContextOption.HOST));
    }

private void buildXAPool() throws Exception {
    if (pool != null)
        return;

    if (log.isDebugEnabled()) { log.debug("building JMS XA pool for " + getUniqueName() + " with " + getMinPoolSize() + " connection(s)"); }
    pool = new XAPool<DualSessionWrapper, JmsPooledConnection>(this, this, xaConnectionFactory);
    boolean builtXaFactory = false;
    if (this.xaConnectionFactory == null) {
        this.xaConnectionFactory = (XAConnectionFactory) pool.getXAFactory();
        builtXaFactory = true;
    }
    try {
        ResourceRegistrar.register(this);
    } catch (RecoveryException ex) {
        if (builtXaFactory) xaConnectionFactory = null;
        pool = null;
        throw ex;
    }
}

@Override
public JmsPooledConnection createPooledConnection(Object xaFactory, ResourceBean bean) throws Exception {
    if (!(xaFactory instanceof XAConnectionFactory))
        throw new IllegalArgumentException("class '" + xaFactory.getClass().getName() + "' does not implement " + XAConnectionFactory.class.getName());
    XAConnectionFactory xaConnectionFactory = (XAConnectionFactory) xaFactory;

    XAConnection xaConnection;
    if (user == null || password == null) {
        if (log.isDebugEnabled()) { log.debug("creating new JMS XAConnection with no credentials"); }
        xaConnection = xaConnectionFactory.createXAConnection();
    }
    else {
        if (log.isDebugEnabled()) { log.debug("creating new JMS XAConnection with user <" + user + "> and password <" + password + ">"); }
        xaConnection = xaConnectionFactory.createXAConnection(user, password);
    }

    JmsPooledConnection jmsPooledConnection = new JmsPooledConnection(this, xaConnection);
    xaStatefulHolders.add(jmsPooledConnection);
    return jmsPooledConnection;
}

/**
 * Create an unitialized {@link XAResourceProducer} implementation which depends on the XA resource class name.
 * @param xaResourceClassName an XA resource class name.
 * @return a {@link XAResourceProducer} implementation.
 * @throws ClassNotFoundException if the {@link XAResourceProducer} cannot be instantiated.
 * @throws IllegalAccessException if the {@link XAResourceProducer} cannot be instantiated.
 * @throws InstantiationException if the {@link XAResourceProducer} cannot be instantiated.
 */
private static XAResourceProducer instantiate(String xaResourceClassName) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
    Class<?> clazz = ClassLoaderUtils.loadClass(xaResourceClassName);

    // resource classes are instantiated via reflection so that there is no hard class binding between this internal
    // transaction manager service and 3rd party libraries like the JMS ones.
    // This allows using the TM with a 100% JDBC application without requiring JMS libraries.

    if (XADataSource.class.isAssignableFrom(clazz)) {
        return (XAResourceProducer) ClassLoaderUtils.loadClass(JDBC_RESOURCE_CLASSNAME).newInstance();
    }
    else if (XAConnectionFactory.class.isAssignableFrom(clazz)) {
        return (XAResourceProducer) ClassLoaderUtils.loadClass(JMS_RESOURCE_CLASSNAME).newInstance();
    }
    else
        return null;
}

@Override
protected XAConnection createProviderConnection(PooledConnectionKey key) throws JMSException {
    if (connectionFactory instanceof XAConnectionFactory) {
        if (key.getUserName() == null && key.getPassword() == null) {
            return ((XAConnectionFactory) connectionFactory).createXAConnection();
        } else {
            return ((XAConnectionFactory) connectionFactory).createXAConnection(key.getUserName(), key.getPassword());
        }
    } else {
        throw new IllegalStateException("connectionFactory should implement javax.jms.XAConnectionFactory");
    }
}

public ManagedConnectionFactoryImpl(ConnectionFactory connectionFactory, XAConnectionFactory xaConnectionFactory, ExceptionSorter exceptionSorter) {
    assert connectionFactory != null;
    assert xaConnectionFactory != null;
    assert exceptionSorter != null;
    this.connectionFactory = connectionFactory;
    this.xaConnectionFactory = xaConnectionFactory;
    this.exceptionSorter = exceptionSorter;
}

@Override
public XAConnectionFactory createXAConnectionFactory() throws JMSException {
    if (_userName != null && _password != null) {
        return new ActiveMQXAConnectionFactory(_userName, _password, _brokerUrl);
    } else {
        return new ActiveMQXAConnectionFactory(_brokerUrl);
    }
}

@Override
public XAConnectionFactory createXAConnectionFactory() throws JMSException {
    if (_userName != null && _password != null) {
        return new ActiveMQXAConnectionFactory(_userName, _password, _brokerUrl);
    } else {
        return new ActiveMQXAConnectionFactory(_brokerUrl);
    }
}

/**
 * Create a JMS connection factory for XA connections.
 *
 * @param url The Oracle JDBC URL.
 * @param user The database user.
 * @param password The database password.
 * @return XA connection factory.
 * @throws JMSException on JMS errors.
 * @throws IllegalStateException if the Oracle AQAPI is missing.
 */
public static XAConnectionFactory createXAConnectionFactory(String url, String user, String password)
        throws JMSException {
    Method m = resolveAqJmsFactoryMethod("getXAConnectionFactory");
    try {
        return (XAConnectionFactory) m.invoke(null, url, createPropsWithUserAndPassword(user, password));
    }
    catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) {
        throw new JMSException("Failed to open AQ JMS connection factory");
    }
}

@Override
public ConnectionFactory wrapConnectionFactory(
        XAConnectionFactory connectionFactory) {
    PoolingConnectionFactoryBean pool = new PoolingConnectionFactoryBean();
    pool.setConnectionFactory(connectionFactory);
    return pool;
}

@Override
public ConnectionFactory wrapConnectionFactory(
        XAConnectionFactory connectionFactory) {
    AtomikosConnectionFactoryBean bean = new AtomikosConnectionFactoryBean();
    bean.setXaConnectionFactory(connectionFactory);
    return bean;
}

@Override
public ConnectionFactory wrapConnectionFactory(
        XAConnectionFactory xaConnectionFactory) {
    XAResourceRecoveryHelper recoveryHelper = getRecoveryHelper(xaConnectionFactory);
    this.recoveryManager.registerXAResourceRecoveryHelper(recoveryHelper);
    return new ConnectionFactoryProxy(xaConnectionFactory,
            new TransactionHelperImpl(this.transactionManager));
}

private XAResourceRecoveryHelper getRecoveryHelper(
        XAConnectionFactory xaConnectionFactory) {
    if (this.properties.getRecoveryJmsUser() == null
            && this.properties.getRecoveryJmsPass() == null) {
        return new JmsXAResourceRecoveryHelper(xaConnectionFactory);
    }
    return new JmsXAResourceRecoveryHelper(xaConnectionFactory,
            this.properties.getRecoveryJmsUser(),
            this.properties.getRecoveryJmsPass());
}

@Test
public void setConnectionFactory() throws Exception {
    XAConnectionFactory factory = mock(XAConnectionFactory.class);
    this.bean.setConnectionFactory(factory);
    this.bean.setBeanName("beanName");
    this.bean.afterPropertiesSet();
    this.bean.init();
    this.bean.createPooledConnection(factory, this.bean);
    verify(factory).createXAConnection();
}

@Test
public void wrap() {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    BitronixXAConnectionFactoryWrapper wrapper = new BitronixXAConnectionFactoryWrapper();
    ConnectionFactory wrapped = wrapper.wrapConnectionFactory(connectionFactory);
    assertThat(wrapped).isInstanceOf(PoolingConnectionFactoryBean.class);
    assertThat(((PoolingConnectionFactoryBean) wrapped).getConnectionFactory())
            .isSameAs(connectionFactory);
}

@Test
public void wrap() {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    AtomikosXAConnectionFactoryWrapper wrapper = new AtomikosXAConnectionFactoryWrapper();
    ConnectionFactory wrapped = wrapper.wrapConnectionFactory(connectionFactory);
    assertThat(wrapped).isInstanceOf(AtomikosConnectionFactoryBean.class);
    assertThat(((AtomikosConnectionFactoryBean) wrapped).getXaConnectionFactory())
            .isSameAs(connectionFactory);
}

@Bean
public ConnectionFactory pooledConnectionFactory(
        XAConnectionFactoryWrapper wrapper) throws Exception {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    XAConnection connection = mock(XAConnection.class);
    XASession session = mock(XASession.class);
    TemporaryQueue queue = mock(TemporaryQueue.class);
    XAResource resource = mock(XAResource.class);
    given(connectionFactory.createXAConnection()).willReturn(connection);
    given(connection.createXASession()).willReturn(session);
    given(session.createTemporaryQueue()).willReturn(queue);
    given(session.getXAResource()).willReturn(resource);
    return wrapper.wrapConnectionFactory(connectionFactory);
}

@Override
public ConnectionFactory wrapConnectionFactory(
        XAConnectionFactory connectionFactory) {
    PoolingConnectionFactoryBean pool = new PoolingConnectionFactoryBean();
    pool.setConnectionFactory(connectionFactory);
    return pool;
}

@Override
public ConnectionFactory wrapConnectionFactory(
        XAConnectionFactory connectionFactory) {
    AtomikosConnectionFactoryBean bean = new AtomikosConnectionFactoryBean();
    bean.setXaConnectionFactory(connectionFactory);
    return bean;
}

@Override
public ConnectionFactory wrapConnectionFactory(
        XAConnectionFactory xaConnectionFactory) {
    XAResourceRecoveryHelper recoveryHelper = getRecoveryHelper(xaConnectionFactory);
    this.recoveryManager.registerXAResourceRecoveryHelper(recoveryHelper);
    return new ConnectionFactoryProxy(xaConnectionFactory,
            new TransactionHelperImpl(this.transactionManager));
}

private XAResourceRecoveryHelper getRecoveryHelper(
        XAConnectionFactory xaConnectionFactory) {
    if (this.properties.getRecoveryJmsUser() == null
            && this.properties.getRecoveryJmsPass() == null) {
        return new JmsXAResourceRecoveryHelper(xaConnectionFactory);
    }
    return new JmsXAResourceRecoveryHelper(xaConnectionFactory,
            this.properties.getRecoveryJmsUser(),
            this.properties.getRecoveryJmsPass());
}

@Test
public void setConnectionFactory() throws Exception {
    XAConnectionFactory factory = mock(XAConnectionFactory.class);
    this.bean.setConnectionFactory(factory);
    this.bean.setBeanName("beanName");
    this.bean.afterPropertiesSet();
    this.bean.init();
    this.bean.createPooledConnection(factory, this.bean);
    verify(factory).createXAConnection();
}

@Test
public void wrap() {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    BitronixXAConnectionFactoryWrapper wrapper = new BitronixXAConnectionFactoryWrapper();
    ConnectionFactory wrapped = wrapper.wrapConnectionFactory(connectionFactory);
    assertThat(wrapped).isInstanceOf(PoolingConnectionFactoryBean.class);
    assertThat(((PoolingConnectionFactoryBean) wrapped).getConnectionFactory())
            .isSameAs(connectionFactory);
}

@Test
public void wrap() {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    AtomikosXAConnectionFactoryWrapper wrapper = new AtomikosXAConnectionFactoryWrapper();
    ConnectionFactory wrapped = wrapper.wrapConnectionFactory(connectionFactory);
    assertThat(wrapped).isInstanceOf(AtomikosConnectionFactoryBean.class);
    assertThat(((AtomikosConnectionFactoryBean) wrapped).getXaConnectionFactory())
            .isSameAs(connectionFactory);
}

@Bean
public ConnectionFactory pooledConnectionFactory(
        XAConnectionFactoryWrapper wrapper) throws Exception {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    XAConnection connection = mock(XAConnection.class);
    XASession session = mock(XASession.class);
    TemporaryQueue queue = mock(TemporaryQueue.class);
    XAResource resource = mock(XAResource.class);
    given(connectionFactory.createXAConnection()).willReturn(connection);
    given(connection.createXASession()).willReturn(session);
    given(session.createTemporaryQueue()).willReturn(queue);
    given(session.getXAResource()).willReturn(resource);
    return wrapper.wrapConnectionFactory(connectionFactory);
}

/**
 * Creates a new EncryptedCredentialsConnectionFactory object.
 */
public EncryptedCredentialsXAConnectionFactory(PasswordDecryptor passwordDecryptor,
    XAConnectionFactory targetConnectionFactory) {
    super(passwordDecryptor);
    Assert.notNull(targetConnectionFactory, "targetConnectionFactory is null");
    this.targetConnectionFactory = targetConnectionFactory;
}

@Override
public ConnectionFactory wrapConnectionFactory(
        XAConnectionFactory connectionFactory) {
    PoolingConnectionFactoryBean pool = new PoolingConnectionFactoryBean();
    pool.setConnectionFactory(connectionFactory);
    return pool;
}

@Override
public ConnectionFactory wrapConnectionFactory(
        XAConnectionFactory connectionFactory) {
    AtomikosConnectionFactoryBean bean = new AtomikosConnectionFactoryBean();
    bean.setXaConnectionFactory(connectionFactory);
    return bean;
}

@Test
public void setConnectionFactory() throws Exception {
    XAConnectionFactory factory = mock(XAConnectionFactory.class);
    this.bean.setConnectionFactory(factory);
    this.bean.setBeanName("beanName");
    this.bean.afterPropertiesSet();
    this.bean.init();
    this.bean.createPooledConnection(factory, this.bean);
    verify(factory).createXAConnection();
}

@Test
public void wrap() {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    BitronixXAConnectionFactoryWrapper wrapper = new BitronixXAConnectionFactoryWrapper();
    ConnectionFactory wrapped = wrapper.wrapConnectionFactory(connectionFactory);
    assertThat(wrapped, instanceOf(PoolingConnectionFactoryBean.class));
    assertThat(((PoolingConnectionFactoryBean) wrapped).getConnectionFactory(),
            sameInstance(connectionFactory));
}

@Test
public void wrap() {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    AtomikosXAConnectionFactoryWrapper wrapper = new AtomikosXAConnectionFactoryWrapper();
    ConnectionFactory wrapped = wrapper.wrapConnectionFactory(connectionFactory);
    assertThat(wrapped, instanceOf(AtomikosConnectionFactoryBean.class));
    assertThat(((AtomikosConnectionFactoryBean) wrapped).getXaConnectionFactory(),
            sameInstance(connectionFactory));
}

@Bean
public ConnectionFactory pooledConnectionFactory(
        XAConnectionFactoryWrapper wrapper) throws Exception {
    XAConnectionFactory connectionFactory = mock(XAConnectionFactory.class);
    XAConnection connection = mock(XAConnection.class);
    XASession session = mock(XASession.class);
    TemporaryQueue queue = mock(TemporaryQueue.class);
    XAResource resource = mock(XAResource.class);
    given(connectionFactory.createXAConnection()).willReturn(connection);
    given(connection.createXASession()).willReturn(session);
    given(session.createTemporaryQueue()).willReturn(queue);
    given(session.getXAResource()).willReturn(resource);
    return wrapper.wrapConnectionFactory(connectionFactory);
}

private void assertNTypes(ActiveMQConnectionFactory factory, final int total) {
   StringBuilder text = new StringBuilder();
   text.append(factory + "\n is instance of ");
   int num = 0;
   if (factory instanceof ConnectionFactory) {
      num++;
      text.append("ConnectionFactory ");
   }
   if (factory instanceof XAConnectionFactory) {
      num++;
      text.append("XAConnectionFactory ");
   }
   if (factory instanceof QueueConnectionFactory) {
      num++;
      text.append("QueueConnectionFactory ");
   }
   if (factory instanceof TopicConnectionFactory) {
      num++;
      text.append("TopicConnectionFactory ");
   }
   if (factory instanceof XAQueueConnectionFactory) {
      num++;
      text.append("XAQueueConnectionFactory ");
   }
   if (factory instanceof XATopicConnectionFactory) {
      num++;
      text.append("XATopicConnectionFactory ");
   }
   Assert.assertEquals(text.toString(), total, num);
}

@Override
public Context createContext(Map<CreateContextOption, Object> options) throws Exception {
    final Options<CreateContextOption> opts = new Options<>(options);
    JMSSpecificContext context;

    boolean xa = opts.getBoolean(CreateContextOption.XA);
    ConnectionFactory cf;
    if (opts.has(CreateContextOption.HOST)) {
        cf = createRemoteConnectionFactory(opts);

    }  else {
        start();

        cf = (ConnectionFactory)lookupJNDI(xa ? JNDI_XA_CF_NAME : JNDI_CF_NAME);
    }

    if (xa) {
        context = createXAContext((XAConnectionFactory)cf, opts);
    } else {
        context = createContext(cf, opts);
    }

    if (opts.has(CreateContextOption.CLIENT_ID)) {
        context.jmsConnection().setClientID(opts.getString(CreateContextOption.CLIENT_ID));
    }

    return context;
}

@Override
public XAConnectionFactory createXAConnectionFactory(Map<String, Object> props) throws JMSRuntimeException {
    Map<String, Object> properties = new HashMap<>(props);
    try {
        XAConnectionFactory cf = XAConnectionFactory.class.cast(ibmMqXaConnectionFactoryClass.newInstance());
        setProperties(cf, ibmMqXaConnectionFactoryClass, properties);
        ConnectionFactoryAdapter cfa = new ConnectionFactoryAdapter();
        cfa.setXaTargetConnectionFactory(cf);
        return cfa;
    } catch (Exception ex) {
        throw new JMSRuntimeException("", "", ex);
    }
}

@Override
public XAConnectionFactory createXAConnectionFactory(Map<String, Object> props) throws JMSRuntimeException {
    props = new HashMap<>(props);
    String url  = (String) props.remove(ConnectionFactoryFactory.JMS_URL);
    if (url == null) {
        throw new JMSRuntimeException("The url property must be set");
    }
    ActiveMQXAConnectionFactory xaCf = new ActiveMQXAConnectionFactory(url);
    try {
        BeanSupport.setData(xaCf, props);
    } catch (Exception e) {
        throw (JMSRuntimeException) new JMSRuntimeException("Unable to build Artemis ConnectionFactory").initCause(e);
    }
    return xaCf;
}

private ConnectionFactory createCF(ConnectionFactoryFactory cff, Dictionary<String, Object> decryptedConfig) throws Exception {
    Objects.requireNonNull(cff, "Must provide a ConnectionFactoryFactory");
    Map<String, Object> props = toMap(decryptedConfig);
    String user = (String) props.remove(ConnectionFactoryFactory.JMS_USER);
    String pswd = (String) props.remove(ConnectionFactoryFactory.JMS_PASSWORD);
    ConnectionFactory cf = cff.createConnectionFactory(props);
    XAConnectionFactory xaCf = cff.createXAConnectionFactory(props);
    ManagedConnectionFactoryBuilder builder = ManagedConnectionFactoryBuilder.builder()
            .transactionManager(transactionManager)
            .connectionFactory(cf, xaCf)
            .userName(user)
            .password(pswd);
    return builder.build();
}

/**
 * Setup a Generic JMS Connection
 *
 * @param ctx               the naming context
 * @param user              the user
 * @param pass              the password
 * @param clientID          the client id
 * @param connectionFactory the connection factory from JNDI
 * @return the connection
 * @throws Exception for any error
 */
protected Connection setupConnection(Context ctx, String user, String pass, String clientID, String connectionFactory) throws Exception {
    log.debug("Attempting to lookup connection factory " + connectionFactory);
    ConnectionFactory gcf = (ConnectionFactory) lookup(ctx, connectionFactory, ConnectionFactory.class);
    log.debug("Got connection factory " + gcf + " from " + connectionFactory);
    log.debug("Attempting to create connection with user " + user);
    Connection result;
    if (gcf instanceof XAConnectionFactory && isDeliveryTransacted) {
        XAConnectionFactory xagcf = (XAConnectionFactory) gcf;
        if (user != null) {
            result = xagcf.createXAConnection(user, pass);
        } else {
            result = xagcf.createXAConnection();
        }
    } else {
        if (user != null) {
            result = gcf.createConnection(user, pass);
        } else {
            result = gcf.createConnection();
        }
    }
    try {
        if (clientID != null) {
            result.setClientID(clientID);
        }
        result.setExceptionListener(this);
        log.debug("Using generic connection " + result);
        return result;
    } catch (Throwable t) {
        try {
            result.close();
        } catch (Exception e) {
            log.trace("Ignored error closing connection", e);
        }
        if (t instanceof Exception) {
            throw (Exception) t;
        }
        throw new RuntimeException("Error configuring connection", t);
    }
}