JBoss hornetq SVN: r9381 - trunk.
by do-not-reply＠jboss.org
Author: jmesnil
Date: 2010-07-07 04:16:50 -0400 (Wed, 07 Jul 2010)
New Revision: 9381
Modified:
trunk/build-hornetq.xml
Log:
fix twitter test
* pass twitter username / password as sysprops before running the test in a forked VM
Modified: trunk/build-hornetq.xml
===================================================================
--- trunk/build-hornetq.xml 2010-07-06 16:45:43 UTC (rev 9380)
+++ trunk/build-hornetq.xml 2010-07-07 08:16:50 UTC (rev 9381)
@@ -1382,7 +1382,9 @@
timeout="${junit.timeout}">
<sysproperty key="user.home" value="${user.home}"/>
<sysproperty key="java.io.tmpdir" value="${java.io.tmpdir}"/>
- <jvmarg value="-Djava.library.path=native/bin"/>
+ <sysproperty key="twitter.username" value="${twitter.username}"/>
+ <sysproperty key="twitter.password" value="${twitter.password}"/>
+ <jvmarg value="-Djava.library.path=native/bin"/>
<jvmarg value="-Dmodule.output=./"/>
<jvmarg value="-Djava.util.logging.config.file=src/config/trunk/non-clustered/logging.properties"/>
<jvmarg value="-XX:+PrintClassHistogram"/>
13 years, 10 months
- 1
- 0
JBoss hornetq SVN: r9380 - in trunk: tests/src/org/hornetq/tests/integration/ra and 1 other directory.
by do-not-reply＠jboss.org
Author: ataylor
Date: 2010-07-06 12:45:43 -0400 (Tue, 06 Jul 2010)
New Revision: 9380
Added:
trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java
trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java
trunk/tests/src/org/hornetq/tests/integration/ra/HornetQRATestBase.java
Modified:
trunk/src/main/org/hornetq/ra/inflow/HornetQActivation.java
trunk/src/main/org/hornetq/ra/inflow/HornetQActivationSpec.java
trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java
Log:
HORNETQ-140 - JCA Integration tests and a couple of tweaks
Modified: trunk/src/main/org/hornetq/ra/inflow/HornetQActivation.java
===================================================================
--- trunk/src/main/org/hornetq/ra/inflow/HornetQActivation.java 2010-07-02 12:01:33 UTC (rev 9379)
+++ trunk/src/main/org/hornetq/ra/inflow/HornetQActivation.java 2010-07-06 16:45:43 UTC (rev 9380)
@@ -105,10 +105,6 @@
// Whether we are in the failure recovery loop
private AtomicBoolean inFailure = new AtomicBoolean(false);
-
- private final int setupAttempts = 10;
-
- private final long setupInterval = 2 * 1000;
static
{
@@ -520,13 +516,13 @@
return;
try
{
- while (deliveryActive.get() && reconnectCount < setupAttempts)
+ while (deliveryActive.get() && reconnectCount < spec.getSetupAttempts())
{
teardown();
try
{
- Thread.sleep(setupInterval);
+ Thread.sleep(spec.getSetupInterval());
}
catch (InterruptedException e)
{
Modified: trunk/src/main/org/hornetq/ra/inflow/HornetQActivationSpec.java
===================================================================
--- trunk/src/main/org/hornetq/ra/inflow/HornetQActivationSpec.java 2010-07-02 12:01:33 UTC (rev 9379)
+++ trunk/src/main/org/hornetq/ra/inflow/HornetQActivationSpec.java 2010-07-06 16:45:43 UTC (rev 9380)
@@ -40,6 +40,10 @@
{
private static final int DEFAULT_MAX_SESSION = 15;
+ private static final int DEFAULT_SETUP_ATTEMPTS = 10;
+
+ private static final long DEFAULT_SETUP_INTERVAL = 2 * 1000;
+
/** The logger */
private static final Logger log = Logger.getLogger(HornetQActivationSpec.class);
@@ -89,6 +93,10 @@
/* use local tx instead of XA*/
private Boolean localTx;
+ private int setupAttempts;
+
+ private long setupInterval;
+
/**
* Constructor
*/
@@ -110,6 +118,8 @@
password = null;
maxSession = DEFAULT_MAX_SESSION;
transactionTimeout = 0;
+ setupAttempts = DEFAULT_SETUP_ATTEMPTS;
+ setupInterval = DEFAULT_SETUP_INTERVAL;
}
/**
@@ -530,6 +540,26 @@
this.localTx = localTx;
}
+ public int getSetupAttempts()
+ {
+ return setupAttempts;
+ }
+
+ public void setSetupAttempts(int setupAttempts)
+ {
+ this.setupAttempts = setupAttempts;
+ }
+
+ public long getSetupInterval()
+ {
+ return setupInterval;
+ }
+
+ public void setSetupInterval(long setupInterval)
+ {
+ this.setupInterval = setupInterval;
+ }
+
/**
* Validate
* @exception InvalidPropertyException Thrown if a validation exception occurs
Added: trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java (rev 0)
+++ trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerSecurityTest.java 2010-07-06 16:45:43 UTC (rev 9380)
@@ -0,0 +1,95 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package org.hornetq.tests.integration.ra;
+
+import org.hornetq.api.core.HornetQException;
+import org.hornetq.api.core.client.ClientMessage;
+import org.hornetq.api.core.client.ClientProducer;
+import org.hornetq.api.core.client.ClientSession;
+import org.hornetq.core.postoffice.Binding;
+import org.hornetq.core.postoffice.impl.LocalQueueBinding;
+import org.hornetq.core.security.Role;
+import org.hornetq.core.server.impl.QueueImpl;
+import org.hornetq.ra.HornetQResourceAdapter;
+import org.hornetq.ra.inflow.HornetQActivationSpec;
+
+import javax.resource.ResourceException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
+ * Created Jul 6, 2010
+ */
+public class HornetQMessageHandlerSecurityTest extends HornetQRATestBase
+{
+ @Override
+ public boolean isSecure()
+ {
+ return true;
+ }
+
+ public void testSimpleMessageReceivedOnQueueWithSecurityFails() throws Exception
+ {
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Queue");
+ spec.setDestination(MDBQUEUE);
+ spec.setUser("dodgyuser");
+ spec.setPassword("dodgypassword");
+ spec.setSetupAttempts(0);
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ Binding binding = server.getPostOffice().getBinding(MDBQUEUEPREFIXEDSIMPLE);
+ assertEquals(((LocalQueueBinding)binding).getQueue().getConsumerCount(), 0);
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
+ }
+
+ public void testSimpleMessageReceivedOnQueueWithSecuritySucceeds() throws Exception
+ {
+ server.getSecurityManager().addUser("testuser", "testpassword");
+ server.getSecurityManager().addRole("testuser", "arole");
+ Role role = new Role("arole", false, true, false, false, false, false, false);
+ Set<Role> roles = new HashSet<Role>();
+ roles.add(role);
+ server.getSecurityRepository().addMatch(MDBQUEUEPREFIXED, roles);
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Queue");
+ spec.setDestination(MDBQUEUE);
+ spec.setUser("testuser");
+ spec.setPassword("testpassword");
+ spec.setSetupAttempts(0);
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ Binding binding = server.getPostOffice().getBinding(MDBQUEUEPREFIXEDSIMPLE);
+ assertEquals(((LocalQueueBinding)binding).getQueue().getConsumerCount(), 15);
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
+ }
+}
Modified: trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java 2010-07-02 12:01:33 UTC (rev 9379)
+++ trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerTest.java 2010-07-06 16:45:43 UTC (rev 9380)
@@ -15,25 +15,10 @@
import org.hornetq.api.core.client.ClientMessage;
import org.hornetq.api.core.client.ClientProducer;
import org.hornetq.api.core.client.ClientSession;
-import org.hornetq.core.config.Configuration;
-import org.hornetq.core.server.HornetQServer;
-import org.hornetq.jms.client.HornetQMessage;
+import org.hornetq.core.postoffice.Binding;
+import org.hornetq.core.postoffice.impl.LocalQueueBinding;
import org.hornetq.ra.HornetQResourceAdapter;
import org.hornetq.ra.inflow.HornetQActivationSpec;
-import org.hornetq.tests.util.ServiceTestBase;
-
-import javax.jms.Message;
-import javax.jms.MessageListener;
-import javax.resource.ResourceException;
-import javax.resource.spi.BootstrapContext;
-import javax.resource.spi.UnavailableException;
-import javax.resource.spi.XATerminator;
-import javax.resource.spi.endpoint.MessageEndpoint;
-import javax.resource.spi.endpoint.MessageEndpointFactory;
-import javax.resource.spi.work.*;
-import javax.transaction.xa.XAResource;
-import java.lang.reflect.Method;
-import java.util.Timer;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -41,65 +26,172 @@
* @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
* Created May 20, 2010
*/
-public class HornetQMessageHandlerTest extends ServiceTestBase
+public class HornetQMessageHandlerTest extends HornetQRATestBase
{
- private Configuration configuration;
+ @Override
+ public boolean isSecure()
+ {
+ return false;
+ }
- private HornetQServer server;
+ public void testSimpleMessageReceivedOnQueue() throws Exception
+ {
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Queue");
+ spec.setDestination(MDBQUEUE);
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ ClientSession session = createFactory(false).createSession();
+ ClientProducer clientProducer = session.createProducer(MDBQUEUEPREFIXED);
+ ClientMessage message = session.createMessage(true);
+ message.getBodyBuffer().writeString("teststring");
+ clientProducer.send(message);
+ session.close();
+ latch.await(5, TimeUnit.SECONDS);
- @Override
- protected void setUp() throws Exception
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "teststring");
+
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
+ }
+
+ public void testSimpleMessageReceivedOnQueueWithSelector() throws Exception
{
- super.setUp();
- clearData();
- configuration = createDefaultConfig();
- configuration.setSecurityEnabled(false);
- server = createServer(true, configuration);
- server.start();
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Queue");
+ spec.setDestination(MDBQUEUE);
+ spec.setMessageSelector("color='red'");
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ ClientSession session = createFactory(false).createSession();
+ ClientProducer clientProducer = session.createProducer(MDBQUEUEPREFIXED);
+ ClientMessage message = session.createMessage(true);
+ message.getBodyBuffer().writeString("blue");
+ message.putStringProperty("color", "blue");
+ clientProducer.send(message);
+ message = session.createMessage(true);
+ message.getBodyBuffer().writeString("red");
+ message.putStringProperty("color", "red");
+ clientProducer.send(message);
+ session.close();
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "red");
+
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
}
- @Override
- protected void tearDown() throws Exception
+ public void testEndpointDeactivated() throws Exception
{
- if (server != null)
- {
- try
- {
- server.stop();
- server = null;
- }
- catch (Exception e)
- {
- // ignore
- }
- }
- super.tearDown();
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Queue");
+ spec.setDestination(MDBQUEUE);
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ Binding binding = server.getPostOffice().getBinding(MDBQUEUEPREFIXEDSIMPLE);
+ assertEquals(((LocalQueueBinding) binding).getQueue().getConsumerCount(), 15);
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
+ assertEquals(((LocalQueueBinding) binding).getQueue().getConsumerCount(), 0);
+ assertTrue(endpoint.released);
}
- public void testSelectorChangedWithTopic() throws Exception
+ public void testMaxSessions() throws Exception
{
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setMaxSession(1);
spec.setResourceAdapter(qResourceAdapter);
spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Queue");
+ spec.setDestination(MDBQUEUE);
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ Binding binding = server.getPostOffice().getBinding(MDBQUEUEPREFIXEDSIMPLE);
+ assertEquals(((LocalQueueBinding) binding).getQueue().getConsumerCount(), 1);
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
+ }
+
+ public void testSimpleTopic() throws Exception
+ {
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
spec.setDestinationType("javax.jms.Topic");
spec.setDestination("mdbTopic");
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ ClientSession session = createFactory(false).createSession();
+ ClientProducer clientProducer = session.createProducer("jms.topic.mdbTopic");
+ ClientMessage message = session.createMessage(true);
+ message.getBodyBuffer().writeString("test");
+ clientProducer.send(message);
+
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "test");
+
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
+ }
+
+ public void testDurableSubscription() throws Exception
+ {
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Topic");
+ spec.setDestination("mdbTopic");
spec.setSubscriptionDurability("Durable");
spec.setSubscriptionName("durable-mdb");
spec.setClientID("id-1");
- spec.setMessageSelector("foo='bar'");
qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
CountDownLatch latch = new CountDownLatch(1);
DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
- DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
qResourceAdapter.endpointActivation(endpointFactory, spec);
ClientSession session = createFactory(false).createSession();
ClientProducer clientProducer = session.createProducer("jms.topic.mdbTopic");
ClientMessage message = session.createMessage(true);
message.getBodyBuffer().writeString("1");
- message.putStringProperty("foo", "bar");
clientProducer.send(message);
latch.await(5, TimeUnit.SECONDS);
@@ -111,18 +203,20 @@
message = session.createMessage(true);
message.getBodyBuffer().writeString("2");
- message.putStringProperty("foo", "bar");
clientProducer.send(message);
latch = new CountDownLatch(1);
endpoint = new DummyMessageEndpoint(latch);
- //change the selector forcing the queue to be recreated
- spec.setMessageSelector("foo='abar'");
- endpointFactory = new DummyMessageEndpointFactory(endpoint);
+ endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
qResourceAdapter.endpointActivation(endpointFactory, spec);
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "2");
+ latch = new CountDownLatch(1);
+ endpoint.reset(latch);
message = session.createMessage(true);
message.getBodyBuffer().writeString("3");
- message.putStringProperty("foo", "abar");
clientProducer.send(message);
latch.await(5, TimeUnit.SECONDS);
@@ -131,8 +225,8 @@
qResourceAdapter.endpointDeactivation(endpointFactory, spec);
}
- public void testSelectorNotChangedWithTopic() throws Exception
- {
+ public void testNonDurableSubscription() throws Exception
+ {
HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
MyBootstrapContext ctx = new MyBootstrapContext();
qResourceAdapter.start(ctx);
@@ -141,20 +235,15 @@
spec.setUseJNDI(false);
spec.setDestinationType("javax.jms.Topic");
spec.setDestination("mdbTopic");
- spec.setSubscriptionDurability("Durable");
- spec.setSubscriptionName("durable-mdb");
- spec.setClientID("id-1");
- spec.setMessageSelector("foo='bar'");
qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
CountDownLatch latch = new CountDownLatch(1);
DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
- DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
qResourceAdapter.endpointActivation(endpointFactory, spec);
ClientSession session = createFactory(false).createSession();
ClientProducer clientProducer = session.createProducer("jms.topic.mdbTopic");
ClientMessage message = session.createMessage(true);
message.getBodyBuffer().writeString("1");
- message.putStringProperty("foo", "bar");
clientProducer.send(message);
latch.await(5, TimeUnit.SECONDS);
@@ -166,123 +255,126 @@
message = session.createMessage(true);
message.getBodyBuffer().writeString("2");
- message.putStringProperty("foo", "bar");
clientProducer.send(message);
latch = new CountDownLatch(1);
endpoint = new DummyMessageEndpoint(latch);
- endpointFactory = new DummyMessageEndpointFactory(endpoint);
+ endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
qResourceAdapter.endpointActivation(endpointFactory, spec);
+ message = session.createMessage(true);
+ message.getBodyBuffer().writeString("3");
+ clientProducer.send(message);
latch.await(5, TimeUnit.SECONDS);
assertNotNull(endpoint.lastMessage);
- assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "2");
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "3");
qResourceAdapter.endpointDeactivation(endpointFactory, spec);
-
}
- class DummyMessageEndpointFactory implements MessageEndpointFactory
+ public void testSelectorChangedWithTopic() throws Exception
{
- private DummyMessageEndpoint endpoint;
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Topic");
+ spec.setDestination("mdbTopic");
+ spec.setSubscriptionDurability("Durable");
+ spec.setSubscriptionName("durable-mdb");
+ spec.setClientID("id-1");
+ spec.setMessageSelector("foo='bar'");
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ ClientSession session = createFactory(false).createSession();
+ ClientProducer clientProducer = session.createProducer("jms.topic.mdbTopic");
+ ClientMessage message = session.createMessage(true);
+ message.getBodyBuffer().writeString("1");
+ message.putStringProperty("foo", "bar");
+ clientProducer.send(message);
- public DummyMessageEndpointFactory(DummyMessageEndpoint endpoint)
- {
- this.endpoint = endpoint;
- }
+ latch.await(5, TimeUnit.SECONDS);
- public MessageEndpoint createEndpoint(XAResource xaResource) throws UnavailableException
- {
- return endpoint;
- }
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "1");
- public boolean isDeliveryTransacted(Method method) throws NoSuchMethodException
- {
- return false;
- }
- }
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
- class DummyMessageEndpoint implements MessageEndpoint, MessageListener
- {
- public CountDownLatch latch;
+ message = session.createMessage(true);
+ message.getBodyBuffer().writeString("2");
+ message.putStringProperty("foo", "bar");
+ clientProducer.send(message);
- private HornetQMessage lastMessage;
+ latch = new CountDownLatch(1);
+ endpoint = new DummyMessageEndpoint(latch);
+ //change the selector forcing the queue to be recreated
+ spec.setMessageSelector("foo='abar'");
+ endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ message = session.createMessage(true);
+ message.getBodyBuffer().writeString("3");
+ message.putStringProperty("foo", "abar");
+ clientProducer.send(message);
+ latch.await(5, TimeUnit.SECONDS);
- public DummyMessageEndpoint(CountDownLatch latch)
- {
- this.latch = latch;
- }
-
- public void beforeDelivery(Method method) throws NoSuchMethodException, ResourceException
- {
-
- }
-
- public void afterDelivery() throws ResourceException
- {
- if(latch != null)
- {
- latch.countDown();
- }
- }
-
- public void release()
- {
-
- }
-
- public void onMessage(Message message)
- {
- lastMessage = (HornetQMessage) message;
- }
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "3");
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
}
- class MyBootstrapContext implements BootstrapContext
+ public void testSelectorNotChangedWithTopic() throws Exception
{
- WorkManager workManager = new DummyWorkManager();
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Topic");
+ spec.setDestination("mdbTopic");
+ spec.setSubscriptionDurability("Durable");
+ spec.setSubscriptionName("durable-mdb");
+ spec.setClientID("id-1");
+ spec.setMessageSelector("foo='bar'");
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ DummyMessageEndpoint endpoint = new DummyMessageEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ ClientSession session = createFactory(false).createSession();
+ ClientProducer clientProducer = session.createProducer("jms.topic.mdbTopic");
+ ClientMessage message = session.createMessage(true);
+ message.getBodyBuffer().writeString("1");
+ message.putStringProperty("foo", "bar");
+ clientProducer.send(message);
- public Timer createTimer() throws UnavailableException
- {
- return null;
- }
+ latch.await(5, TimeUnit.SECONDS);
- public WorkManager getWorkManager()
- {
- return workManager;
- }
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "1");
- public XATerminator getXATerminator()
- {
- return null;
- }
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
- class DummyWorkManager implements WorkManager
- {
- public void doWork(Work work) throws WorkException
- {
- }
+ message = session.createMessage(true);
+ message.getBodyBuffer().writeString("2");
+ message.putStringProperty("foo", "bar");
+ clientProducer.send(message);
- public void doWork(Work work, long l, ExecutionContext executionContext, WorkListener workListener) throws WorkException
- {
- }
+ latch = new CountDownLatch(1);
+ endpoint = new DummyMessageEndpoint(latch);
+ endpointFactory = new DummyMessageEndpointFactory(endpoint, false);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ latch.await(5, TimeUnit.SECONDS);
- public long startWork(Work work) throws WorkException
- {
- return 0;
- }
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "2");
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
- public long startWork(Work work, long l, ExecutionContext executionContext, WorkListener workListener) throws WorkException
- {
- return 0;
- }
+ }
- public void scheduleWork(Work work) throws WorkException
- {
- work.run();
- }
- public void scheduleWork(Work work, long l, ExecutionContext executionContext, WorkListener workListener) throws WorkException
- {
- }
- }
- }
}
Added: trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java (rev 0)
+++ trunk/tests/src/org/hornetq/tests/integration/ra/HornetQMessageHandlerXATest.java 2010-07-06 16:45:43 UTC (rev 9380)
@@ -0,0 +1,163 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package org.hornetq.tests.integration.ra;
+
+import org.hornetq.api.core.client.ClientMessage;
+import org.hornetq.api.core.client.ClientProducer;
+import org.hornetq.api.core.client.ClientSession;
+import org.hornetq.core.transaction.impl.XidImpl;
+import org.hornetq.ra.HornetQResourceAdapter;
+import org.hornetq.ra.inflow.HornetQActivationSpec;
+import org.hornetq.utils.UUIDGenerator;
+
+import javax.resource.ResourceException;
+import javax.transaction.xa.XAException;
+import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+import java.lang.reflect.Method;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
+ * Created Jul 6, 2010
+ */
+public class HornetQMessageHandlerXATest extends HornetQRATestBase
+{
+ @Override
+ public boolean isSecure()
+ {
+ return false;
+ }
+
+ public void testXACommit() throws Exception
+ {
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Queue");
+ spec.setDestination(MDBQUEUE);
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ XADummyEndpoint endpoint = new XADummyEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, true);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ ClientSession session = createFactory(false).createSession();
+ ClientProducer clientProducer = session.createProducer(MDBQUEUEPREFIXED);
+ ClientMessage message = session.createMessage(true);
+ message.getBodyBuffer().writeString("teststring");
+ clientProducer.send(message);
+ session.close();
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "teststring");
+ endpoint.prepare();
+ endpoint.commit();
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
+ }
+
+ public void testXARollback() throws Exception
+ {
+ HornetQResourceAdapter qResourceAdapter = new HornetQResourceAdapter();
+ MyBootstrapContext ctx = new MyBootstrapContext();
+ qResourceAdapter.start(ctx);
+ HornetQActivationSpec spec = new HornetQActivationSpec();
+ spec.setResourceAdapter(qResourceAdapter);
+ spec.setMaxSession(1);
+ spec.setUseJNDI(false);
+ spec.setDestinationType("javax.jms.Queue");
+ spec.setDestination(MDBQUEUE);
+ qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
+ CountDownLatch latch = new CountDownLatch(1);
+ XADummyEndpoint endpoint = new XADummyEndpoint(latch);
+ DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, true);
+ qResourceAdapter.endpointActivation(endpointFactory, spec);
+ ClientSession session = createFactory(false).createSession();
+ ClientProducer clientProducer = session.createProducer(MDBQUEUEPREFIXED);
+ ClientMessage message = session.createMessage(true);
+ message.getBodyBuffer().writeString("teststring");
+ clientProducer.send(message);
+ session.close();
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "teststring");
+ latch = new CountDownLatch(1);
+ endpoint.reset(latch);
+ endpoint.rollback();
+ latch.await(5, TimeUnit.SECONDS);
+
+ assertNotNull(endpoint.lastMessage);
+ assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "teststring");
+ qResourceAdapter.endpointDeactivation(endpointFactory, spec);
+ }
+
+ class XADummyEndpoint extends DummyMessageEndpoint
+ {
+ private Xid xid;
+
+ public XADummyEndpoint(CountDownLatch latch)
+ {
+ super(latch);
+ xid = new XidImpl("xa1".getBytes(), 1, UUIDGenerator.getInstance().generateStringUUID().getBytes());
+ }
+
+ @Override
+ public void beforeDelivery(Method method) throws NoSuchMethodException, ResourceException
+ {
+ super.beforeDelivery(method);
+ try
+ {
+ xaResource.start(xid, XAResource.TMNOFLAGS);
+ }
+ catch (XAException e)
+ {
+ throw new ResourceException(e.getMessage(), e);
+ }
+ }
+
+ @Override
+ public void afterDelivery() throws ResourceException
+ {
+ try
+ {
+ xaResource.end(xid, XAResource.TMSUCCESS);
+ }
+ catch (XAException e)
+ {
+ throw new ResourceException(e.getMessage(), e);
+ }
+
+ super.afterDelivery();
+ }
+
+ public void rollback() throws XAException
+ {
+ xaResource.rollback(xid);
+ }
+
+ public void prepare() throws XAException
+ {
+ xaResource.prepare(xid);
+ }
+
+ public void commit() throws XAException
+ {
+ xaResource.commit(xid, false);
+ }
+ }
+}
Added: trunk/tests/src/org/hornetq/tests/integration/ra/HornetQRATestBase.java
===================================================================
--- trunk/tests/src/org/hornetq/tests/integration/ra/HornetQRATestBase.java (rev 0)
+++ trunk/tests/src/org/hornetq/tests/integration/ra/HornetQRATestBase.java 2010-07-06 16:45:43 UTC (rev 9380)
@@ -0,0 +1,207 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+package org.hornetq.tests.integration.ra;
+
+import org.hornetq.api.core.SimpleString;
+import org.hornetq.core.config.Configuration;
+import org.hornetq.core.server.HornetQServer;
+import org.hornetq.jms.client.HornetQMessage;
+import org.hornetq.tests.util.ServiceTestBase;
+
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.resource.ResourceException;
+import javax.resource.spi.BootstrapContext;
+import javax.resource.spi.UnavailableException;
+import javax.resource.spi.XATerminator;
+import javax.resource.spi.endpoint.MessageEndpoint;
+import javax.resource.spi.endpoint.MessageEndpointFactory;
+import javax.resource.spi.work.*;
+import javax.transaction.xa.XAResource;
+import java.lang.reflect.Method;
+import java.util.Timer;
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * @author <a href="mailto:andy.taylor@jboss.org">Andy Taylor</a>
+ * Created Jul 6, 2010
+ */
+public abstract class HornetQRATestBase extends ServiceTestBase
+{
+ protected Configuration configuration;
+
+ protected HornetQServer server;
+
+ protected static final String MDBQUEUE = "mdbQueue";
+ protected static final String MDBQUEUEPREFIXED = "jms.queue.mdbQueue";
+ protected static final SimpleString MDBQUEUEPREFIXEDSIMPLE = new SimpleString("jms.queue.mdbQueue");
+
+ @Override
+ protected void setUp() throws Exception
+ {
+ super.setUp();
+ clearData();
+ configuration = createDefaultConfig();
+ configuration.setSecurityEnabled(isSecure());
+ server = createServer(true, configuration);
+ server.start();
+ server.createQueue(MDBQUEUEPREFIXEDSIMPLE, MDBQUEUEPREFIXEDSIMPLE, null, true, false);
+ }
+
+ @Override
+ protected void tearDown() throws Exception
+ {
+ if (server != null)
+ {
+ try
+ {
+ server.stop();
+ server = null;
+ }
+ catch (Exception e)
+ {
+ // ignore
+ }
+ }
+ super.tearDown();
+ }
+ public abstract boolean isSecure();
+
+ class DummyMessageEndpointFactory implements MessageEndpointFactory
+ {
+ private DummyMessageEndpoint endpoint;
+
+ private final boolean isDeliveryTransacted;
+
+ public DummyMessageEndpointFactory(DummyMessageEndpoint endpoint, boolean deliveryTransacted)
+ {
+ this.endpoint = endpoint;
+ isDeliveryTransacted = deliveryTransacted;
+ }
+
+ public MessageEndpoint createEndpoint(XAResource xaResource) throws UnavailableException
+ {
+ if(xaResource != null)
+ {
+ endpoint.setXAResource(xaResource);
+ }
+ return endpoint;
+ }
+
+ public boolean isDeliveryTransacted(Method method) throws NoSuchMethodException
+ {
+ return isDeliveryTransacted;
+ }
+ }
+
+ class DummyMessageEndpoint implements MessageEndpoint, MessageListener
+ {
+ public CountDownLatch latch;
+
+ public HornetQMessage lastMessage;
+
+ public boolean released = false;
+
+ public XAResource xaResource;
+
+ public DummyMessageEndpoint(CountDownLatch latch)
+ {
+ this.latch = latch;
+ }
+
+ public void beforeDelivery(Method method) throws NoSuchMethodException, ResourceException
+ {
+
+ }
+
+ public void afterDelivery() throws ResourceException
+ {
+ if(latch != null)
+ {
+ latch.countDown();
+ }
+ System.out.println("lastMessage = " + lastMessage);
+ }
+
+ public void release()
+ {
+ released = true;
+ }
+
+ public void onMessage(Message message)
+ {
+ lastMessage = (HornetQMessage) message;
+ }
+
+ public void reset(CountDownLatch latch)
+ {
+ this.latch = latch;
+ lastMessage = null;
+ }
+
+ public void setXAResource(XAResource xaResource)
+ {
+ this.xaResource = xaResource;
+ }
+ }
+
+ class MyBootstrapContext implements BootstrapContext
+ {
+ WorkManager workManager = new DummyWorkManager();
+
+ public Timer createTimer() throws UnavailableException
+ {
+ return null;
+ }
+
+ public WorkManager getWorkManager()
+ {
+ return workManager;
+ }
+
+ public XATerminator getXATerminator()
+ {
+ return null;
+ }
+
+ class DummyWorkManager implements WorkManager
+ {
+ public void doWork(Work work) throws WorkException
+ {
+ }
+
+ public void doWork(Work work, long l, ExecutionContext executionContext, WorkListener workListener) throws WorkException
+ {
+ }
+
+ public long startWork(Work work) throws WorkException
+ {
+ return 0;
+ }
+
+ public long startWork(Work work, long l, ExecutionContext executionContext, WorkListener workListener) throws WorkException
+ {
+ return 0;
+ }
+
+ public void scheduleWork(Work work) throws WorkException
+ {
+ work.run();
+ }
+
+ public void scheduleWork(Work work, long l, ExecutionContext executionContext, WorkListener workListener) throws WorkException
+ {
+ }
+ }
+ }
+}
13 years, 10 months
- 1
- 0
JBoss hornetq SVN: r9379 - trunk.
by do-not-reply＠jboss.org
Author: jmesnil
Date: 2010-07-02 08:01:33 -0400 (Fri, 02 Jul 2010)
New Revision: 9379
Modified:
trunk/
Log:
add target to svn:ignore
Property changes on: trunk
___________________________________________________________________
Name: svn:ignore
- build
eclipse-output
thirdparty
logs
ObjectStore
tmp
data
junit*.properties
+ build
eclipse-output
thirdparty
logs
ObjectStore
tmp
data
junit*.properties
target
13 years, 10 months
- 1
- 0
JBoss hornetq SVN: r9378 - in branches/2_2_0_HA_Improvements/src/main/org/hornetq/core: server/cluster/impl and 1 other directory.
by do-not-reply＠jboss.org
Author: timfox
Date: 2010-07-01 14:00:35 -0400 (Thu, 01 Jul 2010)
New Revision: 9378
Added:
branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/SubscribeClusterTopologyUpdatesMessage.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java
Log:
HA improvements
Added: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/SubscribeClusterTopologyUpdatesMessage.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/SubscribeClusterTopologyUpdatesMessage.java (rev 0)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/SubscribeClusterTopologyUpdatesMessage.java 2010-07-01 18:00:35 UTC (rev 9378)
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package org.hornetq.core.protocol.core.impl.wireformat;
+
+import org.hornetq.api.core.HornetQBuffer;
+import org.hornetq.core.logging.Logger;
+import org.hornetq.core.protocol.core.impl.PacketImpl;
+
+/**
+ * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
+ *
+ */
+public class SubscribeClusterTopologyUpdatesMessage extends PacketImpl
+{
+ // Constants -----------------------------------------------------
+
+ private static final Logger log = Logger.getLogger(SubscribeClusterTopologyUpdatesMessage.class);
+
+ // Attributes ----------------------------------------------------
+
+ private boolean clusterConnection;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ public SubscribeClusterTopologyUpdatesMessage(final boolean clusterConnection)
+ {
+ super(PacketImpl.SUBSCRIBE_TOPOLOGY);
+
+ this.clusterConnection = clusterConnection;
+ }
+
+ public SubscribeClusterTopologyUpdatesMessage()
+ {
+ super(PacketImpl.SUBSCRIBE_TOPOLOGY);
+ }
+
+ // Public --------------------------------------------------------
+
+ public boolean isClusterConnection()
+ {
+ return clusterConnection;
+ }
+
+ @Override
+ public void encodeRest(final HornetQBuffer buffer)
+ {
+ buffer.writeBoolean(clusterConnection);
+ }
+
+ @Override
+ public void decodeRest(final HornetQBuffer buffer)
+ {
+ clusterConnection = buffer.readBoolean();
+ }
+
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+}
Added: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java (rev 0)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java 2010-07-01 18:00:35 UTC (rev 9378)
@@ -0,0 +1,227 @@
+/*
+ * Copyright 2010 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package org.hornetq.core.server.cluster.impl;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
+
+import org.hornetq.api.core.HornetQException;
+import org.hornetq.api.core.SimpleString;
+import org.hornetq.api.core.TransportConfiguration;
+import org.hornetq.api.core.client.ClientConsumer;
+import org.hornetq.api.core.client.ClientMessage;
+import org.hornetq.api.core.client.ClientProducer;
+import org.hornetq.api.core.client.ClientSessionFactory;
+import org.hornetq.api.core.client.ServerLocator;
+import org.hornetq.api.core.management.ManagementHelper;
+import org.hornetq.api.core.management.NotificationType;
+import org.hornetq.api.core.management.ResourceNames;
+import org.hornetq.core.logging.Logger;
+import org.hornetq.core.message.impl.MessageImpl;
+import org.hornetq.core.persistence.StorageManager;
+import org.hornetq.core.postoffice.BindingType;
+import org.hornetq.core.server.Queue;
+import org.hornetq.core.server.ServerMessage;
+import org.hornetq.core.server.cluster.MessageFlowRecord;
+import org.hornetq.core.server.cluster.Transformer;
+import org.hornetq.utils.UUID;
+import org.hornetq.utils.UUIDGenerator;
+
+/**
+ * A ClusterConnectionBridge
+ *
+ * @author tim
+ *
+ *
+ */
+public class ClusterConnectionBridge extends BridgeImpl
+{
+ private static final Logger log = Logger.getLogger(ClusterConnectionBridge.class);
+
+ private final MessageFlowRecord flowRecord;
+
+ private final SimpleString managementAddress;
+
+ private final SimpleString managementNotificationAddress;
+
+ private ClientConsumer notifConsumer;
+
+ private final SimpleString idsHeaderName;
+
+ private final TransportConfiguration connector;
+
+ public ClusterConnectionBridge(final ServerLocator serverLocator,
+ final UUID nodeUUID,
+ final SimpleString name,
+ final Queue queue,
+ final Executor executor,
+ final SimpleString filterString,
+ final SimpleString forwardingAddress,
+ final ScheduledExecutorService scheduledExecutor,
+ final Transformer transformer,
+ final boolean useDuplicateDetection,
+ final String user,
+ final String password,
+ final boolean activated,
+ final StorageManager storageManager,
+ final SimpleString managementAddress,
+ final SimpleString managementNotificationAddress,
+ final MessageFlowRecord flowRecord,
+ final TransportConfiguration connector) throws Exception
+ {
+ super(serverLocator,
+ nodeUUID,
+ name,
+ queue,
+ executor,
+ filterString,
+ forwardingAddress,
+ scheduledExecutor,
+ transformer,
+ useDuplicateDetection,
+ user,
+ password,
+ activated,
+ storageManager);
+
+ idsHeaderName = MessageImpl.HDR_ROUTE_TO_IDS.concat(name);
+
+ this.managementAddress = managementAddress;
+ this.managementNotificationAddress = managementNotificationAddress;
+ this.flowRecord = flowRecord;
+ this.connector = connector;
+ }
+
+ @Override
+ protected ServerMessage beforeForward(ServerMessage message)
+ {
+ // We make a copy of the message, then we strip out the unwanted routing id headers and leave
+ // only
+ // the one pertinent for the address node - this is important since different queues on different
+ // nodes could have same queue ids
+ // Note we must copy since same message may get routed to other nodes which require different headers
+ message = message.copy();
+
+ // TODO - we can optimise this
+
+ Set<SimpleString> propNames = new HashSet<SimpleString>(message.getPropertyNames());
+
+ byte[] queueIds = message.getBytesProperty(idsHeaderName);
+
+ for (SimpleString propName : propNames)
+ {
+ if (propName.startsWith(MessageImpl.HDR_ROUTE_TO_IDS))
+ {
+ message.removeProperty(propName);
+ }
+ }
+
+ message.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, queueIds);
+
+ message = super.beforeForward(message);
+
+ return message;
+ }
+
+ private void setupNotificationConsumer() throws Exception
+ {
+ if (flowRecord != null)
+ {
+ flowRecord.reset();
+
+ if (notifConsumer != null)
+ {
+ try
+ {
+ notifConsumer.close();
+
+ notifConsumer = null;
+ }
+ catch (HornetQException e)
+ {
+ log.error("Failed to close consumer", e);
+ }
+ }
+
+ // Get the queue data
+
+ String qName = "notif." + UUIDGenerator.getInstance().generateStringUUID();
+
+ SimpleString notifQueueName = new SimpleString(qName);
+
+ SimpleString filter = new SimpleString(ManagementHelper.HDR_BINDING_TYPE + "<>" +
+ BindingType.DIVERT.toInt() +
+ " AND " +
+ ManagementHelper.HDR_NOTIFICATION_TYPE +
+ " IN ('" +
+ NotificationType.BINDING_ADDED +
+ "','" +
+ NotificationType.BINDING_REMOVED +
+ "','" +
+ NotificationType.CONSUMER_CREATED +
+ "','" +
+ NotificationType.CONSUMER_CLOSED +
+ "','" +
+ NotificationType.PROPOSAL +
+ "','" +
+ NotificationType.PROPOSAL_RESPONSE +
+ "') AND " +
+ ManagementHelper.HDR_DISTANCE +
+ "<" +
+ flowRecord.getMaxHops() +
+ " AND (" +
+ ManagementHelper.HDR_ADDRESS +
+ " LIKE '" +
+ flowRecord.getAddress() +
+ "%')");
+
+ session.createQueue(managementNotificationAddress, notifQueueName, filter, false);
+
+ notifConsumer = session.createConsumer(notifQueueName);
+
+ notifConsumer.setMessageHandler(flowRecord);
+
+ session.start();
+
+ ClientMessage message = session.createMessage(false);
+
+ ManagementHelper.putOperationInvocation(message,
+ ResourceNames.CORE_SERVER,
+ "sendQueueInfoToQueue",
+ notifQueueName.toString(),
+ flowRecord.getAddress());
+
+ ClientProducer prod = session.createProducer(managementAddress);
+
+ prod.send(message);
+ }
+ }
+
+ @Override
+ protected void afterConnect() throws Exception
+ {
+ setupNotificationConsumer();
+ }
+
+ @Override
+ protected ClientSessionFactory createSessionFactory() throws Exception
+ {
+ //We create the session factory using the specified connector
+
+ return serverLocator.createSessionFactory(connector);
+ }
+
+}
13 years, 10 months
- 1
- 0
JBoss hornetq SVN: r9377 - in branches/2_2_0_HA_Improvements/src/main/org/hornetq: api/jms and 12 other directories.
by do-not-reply＠jboss.org
Author: timfox
Date: 2010-07-01 12:14:07 -0400 (Thu, 01 Jul 2010)
New Revision: 9377
Added:
branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/NodeAnnounceMessage.java
Modified:
branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/HornetQClient.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/jms/HornetQJMSClient.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/jms/management/JMSServerControl.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/management/impl/JMSServerControlImpl.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/JMSServerManager.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerConfigParserImpl.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/ConnectionFactoryProperties.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQRAMCFProperties.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQResourceAdapter.java
branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/inflow/HornetQActivation.java
Log:
HA improvements
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/HornetQClient.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/HornetQClient.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/HornetQClient.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -93,6 +93,8 @@
public static final int DEFAULT_INITIAL_MESSAGE_PACKET_SIZE = 1500;
+ public static final boolean DEFAULT_HA = false;
+
/**
* Create a ServerLocator which creates session factories using a static list of transportConfigurations, the ServerLocator is not updated automatically
* as the cluster topology changes, and no HA backup information is propagated to the client
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/jms/HornetQJMSClient.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/jms/HornetQJMSClient.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/jms/HornetQJMSClient.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -79,9 +79,9 @@
* @param transportConfigurations
* @return the HornetQConnectionFactory
*/
- public static HornetQConnectionFactory createConnectionFactoryWithoutHA(final TransportConfiguration... initialServers)
+ public static HornetQConnectionFactory createConnectionFactoryWithoutHA(final TransportConfiguration... transportConfigurations)
{
- return new HornetQConnectionFactory(false, initialServers);
+ return new HornetQConnectionFactory(false, transportConfigurations);
}
/**
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/jms/management/JMSServerControl.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/jms/management/JMSServerControl.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/jms/management/JMSServerControl.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -64,8 +64,8 @@
*
* @return {@code true} if the queue was created, {@code false} else
*/
- @Operation(desc = "Create a JMS Queue", impact = MBeanOperationInfo.ACTION)
- boolean createQueue(@Parameter(name = "name", desc = "Name of the queue to create") String name) throws Exception;
+ @Operation(desc = "Create a JMS Queue", impact = MBeanOperationInfo.ACTION)
+ boolean createQueue(@Parameter(name = "name", desc = "Name of the queue to create") String name) throws Exception;
/**
* Creates a JMS Queue with the specified name and JNDI binding.
@@ -81,12 +81,11 @@
*
* @return {@code true} if the queue was created, {@code false} else
*/
- @Operation(desc = "Create a JMS Queue", impact = MBeanOperationInfo.ACTION)
- boolean createQueue(@Parameter(name = "name", desc = "Name of the queue to create") String name,
- @Parameter(name = "jndiBindings", desc = "comma-separated list of JNDI bindings (use ',' if u need to use commas in your jndi name)") String jndiBindings,
- @Parameter(name = "selector", desc = "the jms selector") String selector) throws Exception;
+ @Operation(desc = "Create a JMS Queue", impact = MBeanOperationInfo.ACTION)
+ boolean createQueue(@Parameter(name = "name", desc = "Name of the queue to create") String name,
+ @Parameter(name = "jndiBindings", desc = "comma-separated list of JNDI bindings (use ',' if u need to use commas in your jndi name)") String jndiBindings,
+ @Parameter(name = "selector", desc = "the jms selector") String selector) throws Exception;
-
/**
* Destroys a JMS Queue with the specified name.
*
@@ -125,41 +124,40 @@
* <br>
* The ConnectionFactory is bound to JNDI for all the specified bindings.
*/
- void createConnectionFactory(final String name,
- final String liveTransportClassName,
- final Map<String, Object> liveTransportParams,
- final Object[] jndiBindings) throws Exception;
+ void createConnectionFactory(String name,
+ boolean ha,
+ String liveTransportClassName,
+ Map<String, Object> liveTransportParams,
+ Object[] jndiBindings) throws Exception;
+
/**
* Create a JMS ConnectionFactory with the specified name connected to a static list of live-backup servers.
* <br>
* The ConnectionFactory is bound to JNDI for all the specified bindings Strings.
* <br>
- * {@code liveConnectorsTransportClassNames} (resp. {@code backupConnectorsTransportClassNames}) are the class names
- * of the {@link ConnectorFactory} to connect to the live (resp. backup) servers
- * and {@code liveConnectorTransportParams} (resp. backupConnectorTransportParams) are Map<String, Object> for the corresponding {@link TransportConfiguration}'s parameters.
+ * {@code liveConnectorsTransportClassNames} are the class names
+ * of the {@link ConnectorFactory} to connect to the live servers
+ * and {@code liveConnectorTransportParams} are Map<String, Object> for the corresponding {@link TransportConfiguration}'s parameters.
*
* @see ClientSessionFactory#setStaticConnectors(java.util.List)
*/
void createConnectionFactory(String name,
+ boolean ha,
Object[] liveConnectorsTransportClassNames,
- Object[] liveConnectorTransportParams,
- Object[] backupConnectorsTransportClassNames,
- Object[] backupConnectorTransportParams,
+ Object[] liveConnectorTransportParams,
Object[] bindings) throws Exception;
/**
* Create a JMS ConnectionFactory with the specified name connected to a single live-backup pair of servers.
* <br>
* The ConnectionFactory is bound to JNDI for all the specified bindings Strings.
- * <br>
- * {@code backupTransportClassNames} and {@code backupTransportParams} can be {@code null} if there is no backup server.
+ *
*/
@Operation(desc = "Create a JMS ConnectionFactory", impact = MBeanOperationInfo.ACTION)
void createConnectionFactory(@Parameter(name = "name") String name,
+ @Parameter(name = "ha") boolean ha,
@Parameter(name = "liveTransportClassNames", desc = "comma-separated list of class names for transport to live servers") String liveTransportClassNames,
@Parameter(name = "liveTransportParams", desc = "comma-separated list of key=value parameters for the live transports (enclosed between { } for each transport)") String liveTransportParams,
- @Parameter(name = "backupTransportClassNames", desc = "comma-separated list of class names for transport to backup servers") String backupTransportClassNames,
- @Parameter(name = "backupTransportParams", desc = "comma-separated list of key=value parameters for the backup transports (enclosed between { } for each transport)") String backupTransportParams,
@Parameter(name = "jndiBindings", desc = "comma-separated list of JNDI bindings (use ',' if u need to use commas in your jndi name)") String jndiBindings) throws Exception;
/**
@@ -171,10 +169,7 @@
*
* @see #createConnectionFactory(String, Object[], Object[], Object[], Object[])
*/
- void createConnectionFactory(String name,
- String discoveryAddress,
- int discoveryPort,
- Object[] bindings) throws Exception;
+ void createConnectionFactory(String name, boolean ha, String discoveryAddress, int discoveryPort, Object[] bindings) throws Exception;
/**
* Create a JMS ConnectionFactory with the specified name using a discovery group to discover HornetQ servers.
@@ -187,6 +182,7 @@
*/
@Operation(desc = "Create a JMS ConnectionFactory", impact = MBeanOperationInfo.ACTION)
void createConnectionFactory(@Parameter(name = "name") String name,
+ @Parameter(name = "ha") boolean ha,
@Parameter(name = "discoveryAddress") String discoveryAddress,
@Parameter(name = "discoveryPort") int discoveryPort,
@Parameter(name = "jndiBindings") String jndiBindings) throws Exception;
@@ -227,5 +223,4 @@
@Operation(desc = "List the sessions for the given connectionID", impact = MBeanOperationInfo.INFO)
String[] listSessions(@Parameter(desc = "a connection ID", name = "connectionID") String connectionID) throws Exception;
-
}
Added: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/NodeAnnounceMessage.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/NodeAnnounceMessage.java (rev 0)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/NodeAnnounceMessage.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -0,0 +1,103 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ * Red Hat licenses this file to you under the Apache License, version
+ * 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+ * implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package org.hornetq.core.protocol.core.impl.wireformat;
+
+import org.hornetq.api.core.HornetQBuffer;
+import org.hornetq.api.core.TransportConfiguration;
+import org.hornetq.core.logging.Logger;
+import org.hornetq.core.protocol.core.impl.PacketImpl;
+
+/**
+ * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
+ *
+ */
+public class NodeAnnounceMessage extends PacketImpl
+{
+ // Constants -----------------------------------------------------
+
+ private static final Logger log = Logger.getLogger(NodeAnnounceMessage.class);
+
+ // Attributes ----------------------------------------------------
+
+ private String nodeID;
+
+ private boolean backup;
+
+ private TransportConfiguration connector;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ public NodeAnnounceMessage(final String nodeID, final boolean backup, final TransportConfiguration tc)
+ {
+ super(PacketImpl.NODE_ANNOUNCE);
+
+ this.nodeID = nodeID;
+
+ this.backup = backup;
+
+ this.connector = tc;
+ }
+
+ public NodeAnnounceMessage()
+ {
+ super(PacketImpl.NODE_ANNOUNCE);
+ }
+
+ // Public --------------------------------------------------------
+
+
+ public String getNodeID()
+ {
+ return nodeID;
+ }
+
+ public boolean isBackup()
+ {
+ return backup;
+ }
+
+ public TransportConfiguration getConnector()
+ {
+ return connector;
+ }
+
+
+ @Override
+ public void encodeRest(final HornetQBuffer buffer)
+ {
+ buffer.writeString(nodeID);
+ buffer.writeBoolean(backup);
+ connector.encode(buffer);
+ }
+
+ @Override
+ public void decodeRest(final HornetQBuffer buffer)
+ {
+ this.nodeID = buffer.readString();
+ this.backup = buffer.readBoolean();
+ connector = new TransportConfiguration();
+ connector.decode(buffer);
+ }
+
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+}
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -208,7 +208,7 @@
private boolean initialised;
- private FailoverManager replicationFailoverManager;
+ // private FailoverManager replicationFailoverManager;
private ReplicationManager replicationManager;
@@ -1153,26 +1153,26 @@
* Protected so tests can change this behaviour
* @param backupConnector
*/
- protected FailoverManagerImpl createBackupConnectionFailoverManager(final TransportConfiguration backupConnector,
- final ExecutorService threadPool,
- final ScheduledExecutorService scheduledPool)
- {
- return new FailoverManagerImpl((ClientSessionFactory)null,
- backupConnector,
- null,
- false,
- HornetQClient.DEFAULT_CALL_TIMEOUT,
- HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
- HornetQClient.DEFAULT_CONNECTION_TTL,
- 0,
- 1.0d,
- 0,
- 1,
- false,
- threadPool,
- scheduledPool,
- null);
- }
+// protected FailoverManagerImpl createBackupConnectionFailoverManager(final TransportConfiguration backupConnector,
+// final ExecutorService threadPool,
+// final ScheduledExecutorService scheduledPool)
+// {
+// return new FailoverManagerImpl((ClientSessionFactory)null,
+// backupConnector,
+// null,
+// false,
+// HornetQClient.DEFAULT_CALL_TIMEOUT,
+// HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD,
+// HornetQClient.DEFAULT_CONNECTION_TTL,
+// 0,
+// 1.0d,
+// 0,
+// 1,
+// false,
+// threadPool,
+// scheduledPool,
+// null);
+// }
protected PagingManager createPagingManager()
{
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/client/HornetQConnectionFactory.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -569,7 +569,12 @@
{
serverLocator.close();
}
-
+
+ public ServerLocator getServerLocator()
+ {
+ return serverLocator;
+ }
+
// Package protected ----------------------------------------------------------------------------
// Protected ------------------------------------------------------------------------------------
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/management/impl/JMSServerControlImpl.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/management/impl/JMSServerControlImpl.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/management/impl/JMSServerControlImpl.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -28,7 +28,6 @@
import javax.management.NotificationListener;
import javax.management.StandardMBean;
-import org.hornetq.api.core.Pair;
import org.hornetq.api.core.TransportConfiguration;
import org.hornetq.api.core.management.ManagementHelper;
import org.hornetq.api.core.management.Parameter;
@@ -87,12 +86,10 @@
return trimmed;
}
- private static List<Pair<TransportConfiguration, TransportConfiguration>> convertToConnectorPairs(final Object[] liveConnectorsTransportClassNames,
- final Object[] liveConnectorTransportParams,
- final Object[] backupConnectorsTransportClassNames,
- final Object[] backupConnectorTransportParams)
- {
- List<Pair<TransportConfiguration, TransportConfiguration>> pairs = new ArrayList<Pair<TransportConfiguration, TransportConfiguration>>();
+ private static List<TransportConfiguration> convertToTransportConfigurationArray(final Object[] liveConnectorsTransportClassNames,
+ final Object[] liveConnectorTransportParams)
+ {
+ List<TransportConfiguration> tcs = new ArrayList<TransportConfiguration>();
for (int i = 0; i < liveConnectorsTransportClassNames.length; i++)
{
@@ -105,25 +102,11 @@
TransportConfiguration tcLive = new TransportConfiguration(liveConnectorsTransportClassNames[i].toString(),
liveParams);
- Map<String, Object> backupParams = null;
- if (backupConnectorTransportParams.length > i)
- {
- backupParams = (Map<String, Object>)backupConnectorTransportParams[i];
- }
-
- TransportConfiguration tcBackup = null;
- if (backupConnectorsTransportClassNames.length > i)
- {
- new TransportConfiguration(backupConnectorsTransportClassNames[i].toString(), backupParams);
- }
- Pair<TransportConfiguration, TransportConfiguration> pair = new Pair<TransportConfiguration, TransportConfiguration>(tcLive,
- tcBackup);
-
- pairs.add(pair);
+ tcs.add(tcLive);
}
- return pairs;
- }
+ return tcs;
+ }
public static MBeanNotificationInfo[] getNotificationInfos()
{
@@ -135,7 +118,7 @@
}
return new MBeanNotificationInfo[] { new MBeanNotificationInfo(names,
JMSServerControl.class.getName(),
- "Notifications emitted by a JMS Server") };
+ "Notifications emitted by a JMS Server") };
}
// Constructors --------------------------------------------------
@@ -151,10 +134,11 @@
// JMSServerControlMBean implementation --------------------------
public void createConnectionFactory(final String name,
+ final boolean ha,
final String liveTransportClassName,
final Map<String, Object> liveTransportParams,
final Object[] jndiBindings) throws Exception
- {
+ {
checkStarted();
clearIO();
@@ -162,8 +146,12 @@
try
{
TransportConfiguration liveTC = new TransportConfiguration(liveTransportClassName, liveTransportParams);
+
+ List<TransportConfiguration> list = new ArrayList<TransportConfiguration>(1);
+
+ list.add(liveTC);
- server.createConnectionFactory(name, liveTC, JMSServerControlImpl.convert(jndiBindings));
+ server.createConnectionFactory(name, ha, list, JMSServerControlImpl.convert(jndiBindings));
sendNotification(NotificationType.CONNECTION_FACTORY_CREATED, name);
}
@@ -171,26 +159,23 @@
{
blockOnIO();
}
- }
+ }
public void createConnectionFactory(final String name,
+ final boolean ha,
final Object[] liveConnectorsTransportClassNames,
final Object[] liveConnectorTransportParams,
- final Object[] backupConnectorsTransportClassNames,
- final Object[] backupConnectorTransportParams,
final Object[] jndiBindings) throws Exception
- {
+ {
checkStarted();
clearIO();
try
{
- List<Pair<TransportConfiguration, TransportConfiguration>> pairs = JMSServerControlImpl.convertToConnectorPairs(liveConnectorsTransportClassNames,
- liveConnectorTransportParams,
- backupConnectorsTransportClassNames,
- backupConnectorTransportParams);
- server.createConnectionFactory(name, pairs, JMSServerControlImpl.convert(jndiBindings));
+ List<TransportConfiguration> pairs = JMSServerControlImpl.convertToTransportConfigurationArray(liveConnectorsTransportClassNames,
+ liveConnectorTransportParams);
+ server.createConnectionFactory(name, ha, pairs, JMSServerControlImpl.convert(jndiBindings));
sendNotification(NotificationType.CONNECTION_FACTORY_CREATED, name);
}
@@ -198,15 +183,14 @@
{
blockOnIO();
}
- }
+ }
public void createConnectionFactory(final String name,
+ final boolean ha,
final String liveTransportClassNames,
- final String liveTransportParams,
- final String backupTransportClassNames,
- final String backupTransportParams,
+ final String liveTransportParams,
final String jndiBindings) throws Exception
- {
+ {
checkStarted();
clearIO();
@@ -214,32 +198,33 @@
try
{
Object[] liveClassNames = JMSServerControlImpl.toArray(liveTransportClassNames);
- Object[] liveParams = ManagementHelper.fromCommaSeparatedArrayOfCommaSeparatedKeyValues(liveTransportParams);
- Object[] backupClassNames = JMSServerControlImpl.toArray(backupTransportClassNames);
- Object[] backupParams = ManagementHelper.fromCommaSeparatedArrayOfCommaSeparatedKeyValues(backupTransportParams);;
+ Object[] liveParams = ManagementHelper.fromCommaSeparatedArrayOfCommaSeparatedKeyValues(liveTransportParams);
Object[] bindings = JMSServerControlImpl.toArray(jndiBindings);
- createConnectionFactory(name, liveClassNames, liveParams, backupClassNames, backupParams, bindings);
+ createConnectionFactory(name, ha, liveClassNames, liveParams, bindings);
}
finally
{
blockOnIO();
}
- }
+ }
-
-
public void createConnectionFactory(final String name,
+ final boolean ha,
final String discoveryAddress,
final int discoveryPort,
final Object[] jndiBindings) throws Exception
- {
+ {
checkStarted();
clearIO();
try
{
- server.createConnectionFactory(name, discoveryAddress, discoveryPort, JMSServerControlImpl.convert(jndiBindings));
+ server.createConnectionFactory(name,
+ ha,
+ discoveryAddress,
+ discoveryPort,
+ JMSServerControlImpl.convert(jndiBindings));
sendNotification(NotificationType.CONNECTION_FACTORY_CREATED, name);
}
@@ -247,13 +232,14 @@
{
blockOnIO();
}
- }
+ }
public void createConnectionFactory(final String name,
+ final boolean ha,
final String discoveryAddress,
final int discoveryPort,
final String jndiBindings) throws Exception
- {
+ {
checkStarted();
clearIO();
@@ -262,29 +248,28 @@
{
Object[] bindings = JMSServerControlImpl.toArray(jndiBindings);
- createConnectionFactory(name, discoveryAddress, discoveryPort, bindings);
+ createConnectionFactory(name, ha, discoveryAddress, discoveryPort, bindings);
}
finally
{
blockOnIO();
}
- }
+ }
-
public boolean createQueue(@Parameter(name = "name", desc = "Name of the queue to create") String name) throws Exception
{
return createQueue(name, null, null);
}
-
-
public boolean createQueue(final String name, final String jndiBindings) throws Exception
{
return createQueue(name, jndiBindings, null);
}
- public boolean createQueue(@Parameter(name = "name", desc = "Name of the queue to create") String name, @Parameter(name = "jndiBindings", desc = "comma-separated list of JNDI bindings (use ',' if u need to use commas in your jndi name)") String jndiBindings, @Parameter(name = "selector", desc = "the jms selector") String selector) throws Exception
+ public boolean createQueue(@Parameter(name = "name", desc = "Name of the queue to create") String name,
+ @Parameter(name = "jndiBindings", desc = "comma-separated list of JNDI bindings (use ',' if u need to use commas in your jndi name)") String jndiBindings,
+ @Parameter(name = "selector", desc = "the jms selector") String selector) throws Exception
{
checkStarted();
@@ -460,8 +445,8 @@
try
{
Object[] cfControls = server.getHornetQServer()
- .getManagementService()
- .getResources(ConnectionFactoryControl.class);
+ .getManagementService()
+ .getResources(ConnectionFactoryControl.class);
String[] names = new String[cfControls.length];
for (int i = 0; i < cfControls.length; i++)
{
@@ -481,9 +466,9 @@
public void removeNotificationListener(final NotificationListener listener,
final NotificationFilter filter,
final Object handback) throws ListenerNotFoundException
- {
+ {
broadcaster.removeNotificationListener(listener, filter, handback);
- }
+ }
public void removeNotificationListener(final NotificationListener listener) throws ListenerNotFoundException
{
@@ -493,9 +478,9 @@
public void addNotificationListener(final NotificationListener listener,
final NotificationFilter filter,
final Object handback) throws IllegalArgumentException
- {
+ {
broadcaster.addNotificationListener(listener, filter, handback);
- }
+ }
public MBeanNotificationInfo[] getNotificationInfo()
{
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/JMSServerManager.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/JMSServerManager.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/JMSServerManager.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -18,7 +18,6 @@
import javax.naming.Context;
-import org.hornetq.api.core.Pair;
import org.hornetq.api.core.TransportConfiguration;
import org.hornetq.core.security.Role;
import org.hornetq.core.server.HornetQComponent;
@@ -158,41 +157,17 @@
*/
boolean destroyTopic(String name) throws Exception;
- void createConnectionFactory(String name, String discoveryAddress, int discoveryPort, String ... jndiBindings) throws Exception;
+ void createConnectionFactory(String name, boolean ha, String discoveryAddress, int discoveryPort, String ... jndiBindings) throws Exception;
void createConnectionFactory(String name,
- List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs,
+ boolean ha,
+ List<TransportConfiguration> connectorConfigs,
String ... jndiBindings) throws Exception;
void createConnectionFactory(String name,
- TransportConfiguration liveTC,
- TransportConfiguration backupTC,
- String ... jndiBindings) throws Exception;
-
- void createConnectionFactory(String name, TransportConfiguration liveTC, String ... jndiBindings) throws Exception;
-
- void createConnectionFactory(String name,
+ boolean ha,
+ List<TransportConfiguration> connectorConfigs,
String clientID,
- String discoveryAddress,
- int discoveryPort,
- String ... jndiBindings) throws Exception;
-
- void createConnectionFactory(String name,
- String clientID,
- List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs,
- String ... jndiBindings) throws Exception;
-
- void createConnectionFactory(String name,
- String clientID,
- TransportConfiguration liveTC,
- TransportConfiguration backupTC,
- String ... jndiBindings) throws Exception;
-
- void createConnectionFactory(String name, String clientID, TransportConfiguration liveTC, String ... jndiBindings) throws Exception;
-
- void createConnectionFactory(String name,
- List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs,
- String clientID,
long clientFailureCheckPeriod,
long connectionTTL,
long callTimeout,
@@ -224,6 +199,7 @@
String ... jndiBindings) throws Exception;
void createConnectionFactory(String name,
+ boolean ha,
String localBindAdress,
String discoveryAddress,
int discoveryPort,
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/config/ConnectionFactoryConfiguration.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -15,10 +15,8 @@
import java.util.List;
-import org.hornetq.api.core.Pair;
import org.hornetq.api.core.TransportConfiguration;
import org.hornetq.core.journal.EncodingSupport;
-import org.hornetq.jms.server.JMSServerManager;
/**
* A ConnectionFactoryConfiguration
@@ -47,38 +45,14 @@
void setDiscoveryPort(int discoveryPort);
+ List<TransportConfiguration> getConnectorConfigs();
- /**
- * A Reference to the group configuration.
- */
- String getDiscoveryGroupName();
+ void setConnectorConfigs(List<TransportConfiguration> connectorConfigs);
- /**
- * A Reference to the group configuration.
- */
- void setDiscoveryGroupName(String groupName);
+ boolean isHA();
-
- /**
- * A List of connector names that will be converted into ConnnectorConfigs.
- * This is useful when using the method {@link JMSServerManager#createConnectionFactory(ConnectionFactoryConfiguration)}
- *
- * @return
- */
- List<Pair<String, String>> getConnectorNames();
+ void setHA(boolean ha);
- /**
- * A List of connector names that will be converted into ConnnectorConfigs.
- * This is useful when using the method {@link JMSServerManager#createConnectionFactory(ConnectionFactoryConfiguration)}
- *
- * @return
- */
- void setConnectorNames(List<Pair<String, String>> connectors);
-
- List<Pair<TransportConfiguration, TransportConfiguration>> getConnectorConfigs();
-
- void setConnectorConfigs(List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs);
-
String getClientID();
void setClientID(String clientID);
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/config/impl/ConnectionFactoryConfigurationImpl.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -17,7 +17,6 @@
import java.util.List;
import org.hornetq.api.core.HornetQBuffer;
-import org.hornetq.api.core.Pair;
import org.hornetq.api.core.TransportConfiguration;
import org.hornetq.api.core.client.HornetQClient;
import org.hornetq.jms.server.config.ConnectionFactoryConfiguration;
@@ -41,20 +40,18 @@
private String[] bindings;
- private String discoveryGroupName;
-
private String localBindAddress;
private String discoveryAddress;
private int discoveryPort;
- private List<Pair<String, String>> connectorNames;
+ private List<TransportConfiguration> connectorConfigs;
- private List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs;
-
private String clientID = null;
+ private boolean ha = HornetQClient.DEFAULT_HA;
+
private long discoveryRefreshTimeout = HornetQClient.DEFAULT_DISCOVERY_REFRESH_TIMEOUT;
private long clientFailureCheckPeriod = HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD;
@@ -112,9 +109,9 @@
private boolean failoverOnInitialConnection = HornetQClient.DEFAULT_FAILOVER_ON_INITIAL_CONNECTION;
private boolean failoverOnServerShutdown = HornetQClient.DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN;
-
+
private String groupID = null;
-
+
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
@@ -123,58 +120,31 @@
public ConnectionFactoryConfigurationImpl()
{
}
-
+
public ConnectionFactoryConfigurationImpl(final String name,
+ final boolean ha,
final String discoveryAddress,
final int discoveryPort,
final String... bindings)
{
- this(name, bindings);
+ this(name, ha, bindings);
this.discoveryAddress = discoveryAddress;
this.discoveryPort = discoveryPort;
}
-
- public ConnectionFactoryConfigurationImpl(final String name,
- final String localBindAddress,
- final String discoveryAddress,
- final int discoveryPort,
- final String... bindings)
- {
- this(name, bindings);
- this.localBindAddress = localBindAddress;
- this.discoveryAddress = discoveryAddress;
- this.discoveryPort = discoveryPort;
- }
public ConnectionFactoryConfigurationImpl(final String name,
- final TransportConfiguration liveConfig,
+ final boolean ha,
+ final List<TransportConfiguration> transportConfigs,
final String... bindings)
{
- this(name, liveConfig, null, bindings);
+ this(name, ha, bindings);
+ connectorConfigs = transportConfigs;
}
- public ConnectionFactoryConfigurationImpl(final String name,
- final TransportConfiguration liveConfig,
- final TransportConfiguration backupConfig,
- final String... bindings)
+ private ConnectionFactoryConfigurationImpl(final String name, final boolean ha, final String... bindings)
{
- this(name, bindings);
- connectorConfigs = new ArrayList<Pair<TransportConfiguration, TransportConfiguration>>();
- connectorConfigs.add(new Pair<TransportConfiguration, TransportConfiguration>(liveConfig, backupConfig));
- }
-
- public ConnectionFactoryConfigurationImpl(final String name,
- final List<Pair<TransportConfiguration, TransportConfiguration>> transportConfigs,
- final String... bindings)
- {
- this(name, bindings);
- connectorConfigs = new ArrayList<Pair<TransportConfiguration, TransportConfiguration>>();
- connectorConfigs.addAll(transportConfigs);
- }
-
- public ConnectionFactoryConfigurationImpl(final String name, final String... bindings)
- {
this.name = name;
+ this.ha = ha;
this.bindings = new String[bindings.length];
System.arraycopy(bindings, 0, this.bindings, 0, bindings.length);
}
@@ -195,7 +165,7 @@
{
return name;
}
-
+
public String getLocalBindAddress()
{
return localBindAddress;
@@ -225,17 +195,27 @@
{
this.discoveryPort = discoveryPort;
}
-
- public List<Pair<TransportConfiguration, TransportConfiguration>> getConnectorConfigs()
+
+ public List<TransportConfiguration> getConnectorConfigs()
{
- return connectorConfigs;
+ return this.connectorConfigs;
}
- public void setConnectorConfigs(final List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs)
+ public void setConnectorConfigs(final List<TransportConfiguration> connectorConfigs)
{
this.connectorConfigs = connectorConfigs;
}
+
+ public boolean isHA()
+ {
+ return ha;
+ }
+ public void setHA(boolean ha)
+ {
+ this.ha = ha;
+ }
+
public String getClientID()
{
return clientID;
@@ -515,7 +495,7 @@
{
this.reconnectAttempts = reconnectAttempts;
}
-
+
public boolean isFailoverOnInitialConnection()
{
return failoverOnInitialConnection;
@@ -546,70 +526,37 @@
this.groupID = groupID;
}
- /* (non-Javadoc)
- * @see org.hornetq.jms.server.config.ConnectionFactoryConfiguration#getConnectorNames()
- */
- public List<Pair<String, String>> getConnectorNames()
- {
- return connectorNames;
- }
+ // Encoding Support Implementation --------------------------------------------------------------
/* (non-Javadoc)
- * @see org.hornetq.jms.server.config.ConnectionFactoryConfiguration#setConnectorNames(java.util.List)
- */
- public void setConnectorNames(List<Pair<String, String>> connectors)
- {
- this.connectorNames = connectors;
- }
-
- /* (non-Javadoc)
- * @see org.hornetq.jms.server.config.ConnectionFactoryConfiguration#getDiscoveryGroupName()
- */
- public String getDiscoveryGroupName()
- {
- return discoveryGroupName;
- }
-
- /* (non-Javadoc)
- * @see org.hornetq.jms.server.config.ConnectionFactoryConfiguration#setDiscoveryGroupName(java.lang.String)
- */
- public void setDiscoveryGroupName(String groupName)
- {
- this.discoveryGroupName = groupName;
- }
-
- // Encoding Support Implementation --------------------------------------------------------------
-
- /* (non-Javadoc)
* @see org.hornetq.core.journal.EncodingSupport#decode(org.hornetq.api.core.HornetQBuffer)
*/
public void decode(HornetQBuffer buffer)
{
name = buffer.readSimpleString().toString();
- discoveryGroupName = BufferHelper.readNullableSimpleStringAsString(buffer);
-
localBindAddress = BufferHelper.readNullableSimpleStringAsString(buffer);
-
+
discoveryAddress = BufferHelper.readNullableSimpleStringAsString(buffer);
discoveryPort = buffer.readInt();
int nConnectors = buffer.readInt();
- connectorNames = new ArrayList<Pair<String, String>>(nConnectors);
-
- for (int i = 0; i < nConnectors; i++)
+ if (nConnectors > 0)
{
- String a = BufferHelper.readNullableSimpleStringAsString(buffer);
-
- String b = BufferHelper.readNullableSimpleStringAsString(buffer);
-
- connectorNames.add(new Pair<String, String>(a, b));
- }
+ connectorConfigs = new ArrayList<TransportConfiguration>(nConnectors);
- connectorConfigs = TransportConfigurationEncodingSupport.decodeConfigs(buffer);
+ for (int i = 0; i < nConnectors; i++)
+ {
+ TransportConfiguration tc = new TransportConfiguration();
+ tc.decode(buffer);
+
+ connectorConfigs.add(tc);
+ }
+ }
+
clientID = BufferHelper.readNullableSimpleStringAsString(buffer);
discoveryRefreshTimeout = buffer.readLong();
@@ -665,7 +612,7 @@
maxRetryInterval = buffer.readLong();
reconnectAttempts = buffer.readInt();
-
+
failoverOnInitialConnection = buffer.readBoolean();
failoverOnServerShutdown = buffer.readBoolean();
@@ -680,27 +627,26 @@
{
BufferHelper.writeAsSimpleString(buffer, name);
- BufferHelper.writeAsNullableSimpleString(buffer, discoveryGroupName);
-
BufferHelper.writeAsNullableSimpleString(buffer, localBindAddress);
-
+
BufferHelper.writeAsNullableSimpleString(buffer, discoveryAddress);
buffer.writeInt(discoveryPort);
- buffer.writeInt(connectorNames == null ? 0 : connectorNames.size());
-
- if (connectorNames != null)
+ if (connectorConfigs == null)
{
- for (Pair<String, String> namePair : connectorNames)
+ buffer.writeInt(0);
+ }
+ else
+ {
+ buffer.writeInt(connectorConfigs.size());
+
+ for (TransportConfiguration tc : connectorConfigs)
{
- BufferHelper.writeAsNullableSimpleString(buffer, namePair.a);
- BufferHelper.writeAsNullableSimpleString(buffer, namePair.b);
+ tc.encode(buffer);
}
}
- TransportConfigurationEncodingSupport.encodeConfigs(buffer, connectorConfigs);
-
BufferHelper.writeAsNullableSimpleString(buffer, clientID);
buffer.writeLong(discoveryRefreshTimeout);
@@ -756,7 +702,7 @@
buffer.writeLong(maxRetryInterval);
buffer.writeInt(reconnectAttempts);
-
+
buffer.writeBoolean(failoverOnInitialConnection);
buffer.writeBoolean(failoverOnServerShutdown);
@@ -764,105 +710,119 @@
BufferHelper.writeAsNullableSimpleString(buffer, groupID);
}
-
- private int sizeOfConnectors()
+ /* (non-Javadoc)
+ * @see org.hornetq.core.journal.EncodingSupport#getEncodeSize()
+ */
+ public int getEncodeSize()
{
- int size = DataConstants.SIZE_INT; // for the number of connectors persisted
+ int size = BufferHelper.sizeOfSimpleString(name) +
- if (connectorNames != null)
+ BufferHelper.sizeOfNullableSimpleString(localBindAddress) +
+
+ BufferHelper.sizeOfNullableSimpleString(discoveryAddress) +
+
+ DataConstants.SIZE_INT; // discoveryPort
+
+ if (this.connectorConfigs != null)
{
- for (Pair<String, String> pair : connectorNames)
+ for (TransportConfiguration tc : this.connectorConfigs)
{
- size += BufferHelper.sizeOfNullableSimpleString(pair.a);
- size += BufferHelper.sizeOfNullableSimpleString(pair.b);
+ size += TransportConfigurationEncodingSupport.getEncodeSize(tc);
}
}
+ size += BufferHelper.sizeOfNullableSimpleString(clientID) +
- return size;
- }
+ DataConstants.SIZE_LONG +
+ // discoveryRefreshTimeout
- /* (non-Javadoc)
- * @see org.hornetq.core.journal.EncodingSupport#getEncodeSize()
- */
- public int getEncodeSize()
- {
- return BufferHelper.sizeOfSimpleString(name) +
-
- BufferHelper.sizeOfNullableSimpleString(discoveryGroupName) +
-
- BufferHelper.sizeOfNullableSimpleString(localBindAddress)+
-
- BufferHelper.sizeOfNullableSimpleString(discoveryAddress)+
-
- DataConstants.SIZE_INT + // discoveryPort
-
- sizeOfConnectors() +
-
- TransportConfigurationEncodingSupport.getEncodeSize(connectorConfigs) +
+ DataConstants.SIZE_LONG +
+ // clientFailureCheckPeriod
- BufferHelper.sizeOfNullableSimpleString(clientID) +
-
- DataConstants.SIZE_LONG + // discoveryRefreshTimeout
-
- DataConstants.SIZE_LONG + // clientFailureCheckPeriod
+ DataConstants.SIZE_LONG +
+ // connectionTTL
- DataConstants.SIZE_LONG + // connectionTTL
+ DataConstants.SIZE_LONG +
+ // callTimeout
- DataConstants.SIZE_LONG + // callTimeout
+ DataConstants.SIZE_BOOLEAN +
+ // cacheLargeMessagesClient
- DataConstants.SIZE_BOOLEAN + // cacheLargeMessagesClient
-
- DataConstants.SIZE_INT + // minLargeMessageSize
+ DataConstants.SIZE_INT +
+ // minLargeMessageSize
- DataConstants.SIZE_INT + // consumerWindowSize
+ DataConstants.SIZE_INT +
+ // consumerWindowSize
- DataConstants.SIZE_INT + // consumerMaxRate
+ DataConstants.SIZE_INT +
+ // consumerMaxRate
- DataConstants.SIZE_INT + // confirmationWindowSize
+ DataConstants.SIZE_INT +
+ // confirmationWindowSize
- DataConstants.SIZE_INT + // producerWindowSize
+ DataConstants.SIZE_INT +
+ // producerWindowSize
- DataConstants.SIZE_INT + // producerMaxRate
+ DataConstants.SIZE_INT +
+ // producerMaxRate
- DataConstants.SIZE_BOOLEAN + // blockOnAcknowledge
+ DataConstants.SIZE_BOOLEAN +
+ // blockOnAcknowledge
- DataConstants.SIZE_BOOLEAN + // blockOnDurableSend
+ DataConstants.SIZE_BOOLEAN +
+ // blockOnDurableSend
- DataConstants.SIZE_BOOLEAN + // blockOnNonDurableSend
+ DataConstants.SIZE_BOOLEAN +
+ // blockOnNonDurableSend
- DataConstants.SIZE_BOOLEAN + // autoGroup
+ DataConstants.SIZE_BOOLEAN +
+ // autoGroup
- DataConstants.SIZE_BOOLEAN + // preAcknowledge
+ DataConstants.SIZE_BOOLEAN +
+ // preAcknowledge
- BufferHelper.sizeOfSimpleString(loadBalancingPolicyClassName) +
+ BufferHelper.sizeOfSimpleString(loadBalancingPolicyClassName) +
- DataConstants.SIZE_INT + // transactionBatchSize
+ DataConstants.SIZE_INT +
+ // transactionBatchSize
- DataConstants.SIZE_INT + // dupsOKBatchSize
+ DataConstants.SIZE_INT +
+ // dupsOKBatchSize
- DataConstants.SIZE_LONG + // initialWaitTimeout
+ DataConstants.SIZE_LONG +
+ // initialWaitTimeout
- DataConstants.SIZE_BOOLEAN + // useGlobalPools
+ DataConstants.SIZE_BOOLEAN +
+ // useGlobalPools
- DataConstants.SIZE_INT + // scheduledThreadPoolMaxSize
+ DataConstants.SIZE_INT +
+ // scheduledThreadPoolMaxSize
- DataConstants.SIZE_INT + // threadPoolMaxSize
+ DataConstants.SIZE_INT +
+ // threadPoolMaxSize
- DataConstants.SIZE_LONG + // retryInterval
+ DataConstants.SIZE_LONG +
+ // retryInterval
- DataConstants.SIZE_DOUBLE + // retryIntervalMultiplier
+ DataConstants.SIZE_DOUBLE +
+ // retryIntervalMultiplier
- DataConstants.SIZE_LONG + // maxRetryInterval
+ DataConstants.SIZE_LONG +
+ // maxRetryInterval
- DataConstants.SIZE_INT + // reconnectAttempts
-
- DataConstants.SIZE_BOOLEAN + // failoverOnInitialConnection
+ DataConstants.SIZE_INT +
+ // reconnectAttempts
- DataConstants.SIZE_BOOLEAN + // failoverOnServerShutdown
-
- BufferHelper.sizeOfNullableSimpleString(groupID);
+ DataConstants.SIZE_BOOLEAN +
+ // failoverOnInitialConnection
+
+ DataConstants.SIZE_BOOLEAN +
+ // failoverOnServerShutdown
+
+ BufferHelper.sizeOfNullableSimpleString(groupID);
+
+ return size;
}
-
+
// Public --------------------------------------------------------
// Package protected ---------------------------------------------
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerConfigParserImpl.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerConfigParserImpl.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerConfigParserImpl.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -20,8 +20,10 @@
import java.util.List;
import org.hornetq.api.core.HornetQException;
-import org.hornetq.api.core.Pair;
+import org.hornetq.api.core.TransportConfiguration;
import org.hornetq.api.core.client.HornetQClient;
+import org.hornetq.core.config.Configuration;
+import org.hornetq.core.config.DiscoveryGroupConfiguration;
import org.hornetq.core.config.impl.Validators;
import org.hornetq.core.logging.Logger;
import org.hornetq.jms.server.JMSServerConfigParser;
@@ -57,12 +59,19 @@
// Attributes ----------------------------------------------------
+ private final Configuration config;
+
// Static --------------------------------------------------------
// Constructors --------------------------------------------------
// Public --------------------------------------------------------
+ public JMSServerConfigParserImpl(final Configuration config)
+ {
+ this.config = config;
+ }
+
/**
* Parse the JMS Configuration XML as a JMSConfiguration object
*/
@@ -188,6 +197,32 @@
return newQueue(queueName, selectorString, durable, jndiArray);
}
+ // private void lookupDiscovery(final String discoveryGroupName, final ConnectionFactoryConfiguration cfConfig)
+ // throws HornetQException
+ // {
+ // Configuration configuration = server.getConfiguration();
+ //
+ // DiscoveryGroupConfiguration discoveryGroupConfiguration = configuration.getDiscoveryGroupConfigurations()
+ // .get(cfConfig.getDiscoveryGroupName());
+ //
+ // if (discoveryGroupConfiguration == null)
+ // {
+ // JMSServerManagerImpl.log.warn("There is no discovery group with name '" + cfConfig.getDiscoveryGroupName() +
+ // "' deployed.");
+ //
+ // throw new HornetQException(HornetQException.ILLEGAL_STATE,
+ // "There is no discovery group with name '" + cfConfig.getDiscoveryGroupName() +
+ // "' deployed.");
+ // }
+ //
+ // cfConfig.setLocalBindAddress(discoveryGroupConfiguration.getLocalBindAddress());
+ // cfConfig.setDiscoveryAddress(discoveryGroupConfiguration.getGroupAddress());
+ // cfConfig.setDiscoveryPort(discoveryGroupConfiguration.getGroupPort());
+ // cfConfig.setDiscoveryRefreshTimeout(discoveryGroupConfiguration.getRefreshTimeout());
+ //
+ //
+ // }
+
/**
* Parse the Connection Configuration node as a ConnectionFactoryConfiguration object
* @param node
@@ -310,9 +345,11 @@
HornetQClient.DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT,
Validators.GT_ZERO);
+ boolean ha = XMLConfigurationUtil.getBoolean(e, "ha", HornetQClient.DEFAULT_HA);
+
String groupid = XMLConfigurationUtil.getString(e, "group-id", null, Validators.NO_CHECK);
List<String> jndiBindings = new ArrayList<String>();
- List<Pair<String, String>> connectorNames = new ArrayList<Pair<String, String>>();
+ List<String> connectorNames = new ArrayList<String>();
String discoveryGroupName = null;
NodeList children = node.getChildNodes();
@@ -345,16 +382,7 @@
{
String connectorName = entry.getAttributes().getNamedItem("connector-name").getNodeValue();
- String backupConnectorName = null;
-
- Node backupNode = entry.getAttributes().getNamedItem("backup-connector-name");
- if (backupNode != null)
- {
- backupConnectorName = backupNode.getNodeValue();
- }
-
- connectorNames.add(new Pair<String, String>(connectorName, backupConnectorName));
-
+ connectorNames.add(connectorName);
}
}
}
@@ -371,14 +399,43 @@
if (discoveryGroupName != null)
{
- cfConfig = new ConnectionFactoryConfigurationImpl(name, strbindings);
+ DiscoveryGroupConfiguration discoveryGroupConfiguration = config.getDiscoveryGroupConfigurations()
+ .get(discoveryGroupName);
+
+ if (discoveryGroupConfiguration == null)
+ {
+ log.warn("There is no discovery group with name '" + discoveryGroupName + "' deployed.");
+
+ throw new IllegalArgumentException("There is no discovery group with name '" + discoveryGroupName +
+ "' deployed.");
+ }
+
+ cfConfig = new ConnectionFactoryConfigurationImpl(name,
+ ha,
+ discoveryGroupConfiguration.getGroupAddress(),
+ discoveryGroupConfiguration.getGroupPort(),
+ strbindings);
+ cfConfig.setLocalBindAddress(discoveryGroupConfiguration.getLocalBindAddress());
cfConfig.setInitialWaitTimeout(discoveryInitialWaitTimeout);
- cfConfig.setDiscoveryGroupName(discoveryGroupName);
}
else
{
- cfConfig = new ConnectionFactoryConfigurationImpl(name, strbindings);
- cfConfig.setConnectorNames(connectorNames);
+ List<TransportConfiguration> tcList = new ArrayList<TransportConfiguration>();
+
+ for (String connectorName : connectorNames)
+ {
+ TransportConfiguration tc = config.getConnectorConfigurations().get(connectorName);
+
+ if (tc == null)
+ {
+ log.warn("There is no connector with name '" + connectorName + "' deployed.");
+
+ throw new IllegalArgumentException("There is no connector with name '" + connectorName + "' deployed.");
+ }
+
+ tcList.add(tc);
+ }
+ cfConfig = new ConnectionFactoryConfigurationImpl(name, ha, tcList, strbindings);
}
cfConfig.setClientID(clientID);
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerDeployer.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -44,7 +44,7 @@
protected static final String ENTRY_NODE_NAME = "entry";
- protected static final String CONNECTORS_NODE_NAME = "connectors";
+ protected static final String CONNECTORS_NODE_NAME = "static-connectors";
protected static final String CONNECTION_FACTORY_NODE_NAME = "connection-factory";
@@ -63,7 +63,7 @@
this.jmsServerManager = jmsServerManager;
- parser = new JMSServerConfigParserImpl();
+ parser = new JMSServerConfigParserImpl(jmsServerManager.getHornetQServer().getConfiguration());
}
/**
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/impl/JMSServerManagerImpl.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -27,14 +27,12 @@
import javax.naming.NamingException;
import org.hornetq.api.core.HornetQException;
-import org.hornetq.api.core.Pair;
import org.hornetq.api.core.SimpleString;
import org.hornetq.api.core.TransportConfiguration;
import org.hornetq.api.core.management.AddressControl;
import org.hornetq.api.core.management.ResourceNames;
import org.hornetq.api.jms.HornetQJMSClient;
import org.hornetq.core.config.Configuration;
-import org.hornetq.core.config.DiscoveryGroupConfiguration;
import org.hornetq.core.deployers.DeploymentManager;
import org.hornetq.core.deployers.impl.FileDeploymentManager;
import org.hornetq.core.deployers.impl.XmlDeployer;
@@ -170,9 +168,9 @@
public void preActivate()
{
-
+
}
-
+
public synchronized void activated()
{
active = true;
@@ -676,36 +674,25 @@
}
public synchronized void createConnectionFactory(final String name,
- final List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs,
+ final boolean ha,
+ final List<TransportConfiguration> connectorConfigs,
String... jndiBindings) throws Exception
{
checkInitialised();
HornetQConnectionFactory cf = connectionFactories.get(name);
if (cf == null)
{
- ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, connectorConfigs);
+ ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name,
+ ha,
+ connectorConfigs);
createConnectionFactory(true, configuration, jndiBindings);
}
}
public synchronized void createConnectionFactory(final String name,
+ final boolean ha,
+ final List<TransportConfiguration> connectorConfigs,
final String clientID,
- final List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs,
- String... jndiBindings) throws Exception
- {
- checkInitialised();
- HornetQConnectionFactory cf = connectionFactories.get(name);
- if (cf == null)
- {
- ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, connectorConfigs);
- configuration.setClientID(clientID);
- createConnectionFactory(true, configuration, jndiBindings);
- }
- }
-
- public synchronized void createConnectionFactory(final String name,
- final List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs,
- final String clientID,
final long clientFailureCheckPeriod,
final long connectionTTL,
final long callTimeout,
@@ -740,7 +727,9 @@
HornetQConnectionFactory cf = connectionFactories.get(name);
if (cf == null)
{
- ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, connectorConfigs);
+ ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name,
+ ha,
+ connectorConfigs);
configuration.setClientID(clientID);
configuration.setClientFailureCheckPeriod(clientFailureCheckPeriod);
configuration.setConnectionTTL(connectionTTL);
@@ -775,6 +764,7 @@
}
public synchronized void createConnectionFactory(final String name,
+ final boolean ha,
final String localBindAddress,
final String discoveryAddress,
final int discoveryPort,
@@ -798,7 +788,7 @@
final boolean preAcknowledge,
final String loadBalancingPolicyClassName,
final int transactionBatchSize,
- final int dupsOKBatchSize,
+ final int dupsOKBatchSize,
final boolean useGlobalPools,
final int scheduledThreadPoolMaxSize,
final int threadPoolMaxSize,
@@ -810,15 +800,16 @@
final boolean failoverOnServerShutdown,
final String groupId,
final String... jndiBindings) throws Exception
- {
+ {
checkInitialised();
HornetQConnectionFactory cf = connectionFactories.get(name);
if (cf == null)
{
ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name,
- localBindAddress,
+ ha,
discoveryAddress,
discoveryPort);
+ configuration.setLocalBindAddress(localBindAddress);
configuration.setClientID(clientID);
configuration.setDiscoveryRefreshTimeout(discoveryRefreshTimeout);
configuration.setInitialWaitTimeout(discoveryInitialWaitTimeout);
@@ -854,6 +845,7 @@
}
public synchronized void createConnectionFactory(final String name,
+ final boolean ha,
final String discoveryAddress,
final int discoveryPort,
final String... jndiBindings) throws Exception
@@ -863,31 +855,13 @@
if (cf == null)
{
ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name,
+ ha,
discoveryAddress,
discoveryPort);
createConnectionFactory(true, configuration, jndiBindings);
}
}
- public synchronized void createConnectionFactory(final String name,
- final String clientID,
- final String discoveryAddress,
- final int discoveryPort,
- final String... jndiBindings) throws Exception
- {
- checkInitialised();
-
- HornetQConnectionFactory cf = connectionFactories.get(name);
- if (cf == null)
- {
- ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name,
- discoveryAddress,
- discoveryPort);
- configuration.setClientID(clientID);
- createConnectionFactory(true, configuration, jndiBindings);
- }
- }
-
public synchronized void createConnectionFactory(final boolean storeConfig,
final ConnectionFactoryConfiguration cfConfig,
String... jndi) throws Exception
@@ -914,153 +888,6 @@
}
}
- private HornetQConnectionFactory internalCreateConnectionFactory(final String name,
- final String localBindAddress,
- final String discoveryAddress,
- final int discoveryPort,
- final String clientID,
- final long discoveryRefreshTimeout,
- final long discoveryInitialWaitTimeout,
- final long clientFailureCheckPeriod,
- final long connectionTTL,
- final long callTimeout,
- final boolean cacheLargeMessagesClient,
- final int minLargeMessageSize,
- final int consumerWindowSize,
- final int consumerMaxRate,
- final int confirmationWindowSize,
- final int producerWindowSize,
- final int producerMaxRate,
- final boolean blockOnAcknowledge,
- final boolean blockOnDurableSend,
- final boolean blockOnNonDurableSend,
- final boolean autoGroup,
- final boolean preAcknowledge,
- final String loadBalancingPolicyClassName,
- final int transactionBatchSize,
- final int dupsOKBatchSize,
- final boolean useGlobalPools,
- final int scheduledThreadPoolMaxSize,
- final int threadPoolMaxSize,
- final long retryInterval,
- final double retryIntervalMultiplier,
- final long maxRetryInterval,
- final int reconnectAttempts,
- final boolean failoverOnInitialConnection,
- final boolean failoverOnServerShutdown,
- final String groupId) throws Exception
- {
- checkInitialised();
- HornetQConnectionFactory cf = connectionFactories.get(name);
- if (cf == null)
- {
- cf = (HornetQConnectionFactory)HornetQJMSClient.createConnectionFactory(discoveryAddress, discoveryPort);
- cf.setClientID(clientID);
- cf.setLocalBindAddress(localBindAddress);
- cf.setDiscoveryRefreshTimeout(discoveryRefreshTimeout);
- cf.setDiscoveryInitialWaitTimeout(discoveryInitialWaitTimeout);
- cf.setClientFailureCheckPeriod(clientFailureCheckPeriod);
- cf.setConnectionTTL(connectionTTL);
- cf.setCallTimeout(callTimeout);
- cf.setCacheLargeMessagesClient(cacheLargeMessagesClient);
- cf.setMinLargeMessageSize(minLargeMessageSize);
- cf.setConsumerWindowSize(consumerWindowSize);
- cf.setConsumerMaxRate(consumerMaxRate);
- cf.setConfirmationWindowSize(confirmationWindowSize);
- cf.setProducerWindowSize(producerWindowSize);
- cf.setProducerMaxRate(producerMaxRate);
- cf.setBlockOnAcknowledge(blockOnAcknowledge);
- cf.setBlockOnDurableSend(blockOnDurableSend);
- cf.setBlockOnNonDurableSend(blockOnNonDurableSend);
- cf.setAutoGroup(autoGroup);
- cf.setPreAcknowledge(preAcknowledge);
- cf.setConnectionLoadBalancingPolicyClassName(loadBalancingPolicyClassName);
- cf.setTransactionBatchSize(transactionBatchSize);
- cf.setDupsOKBatchSize(dupsOKBatchSize);
- cf.setUseGlobalPools(useGlobalPools);
- cf.setScheduledThreadPoolMaxSize(scheduledThreadPoolMaxSize);
- cf.setThreadPoolMaxSize(threadPoolMaxSize);
- cf.setRetryInterval(retryInterval);
- cf.setRetryIntervalMultiplier(retryIntervalMultiplier);
- cf.setMaxRetryInterval(maxRetryInterval);
- cf.setReconnectAttempts(reconnectAttempts);
- cf.setFailoverOnInitialConnection(failoverOnInitialConnection);
- cf.setFailoverOnServerShutdown(failoverOnServerShutdown);
- }
-
- return cf;
- }
-
- private HornetQConnectionFactory internalCreateConnectionFactory(final String name,
- final List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs,
- final String clientID,
- final long clientFailureCheckPeriod,
- final long connectionTTL,
- final long callTimeout,
- final boolean cacheLargeMessagesClient,
- final int minLargeMessageSize,
- final int consumerWindowSize,
- final int consumerMaxRate,
- final int confirmationWindowSize,
- final int producerWindowSize,
- final int producerMaxRate,
- final boolean blockOnAcknowledge,
- final boolean blockOnDurableSend,
- final boolean blockOnNonDurableSend,
- final boolean autoGroup,
- final boolean preAcknowledge,
- final String loadBalancingPolicyClassName,
- final int transactionBatchSize,
- final int dupsOKBatchSize,
- final boolean useGlobalPools,
- final int scheduledThreadPoolMaxSize,
- final int threadPoolMaxSize,
- final long retryInterval,
- final double retryIntervalMultiplier,
- final long maxRetryInterval,
- final int reconnectAttempts,
- final boolean failoverOnInitialConnection,
- final boolean failoverOnServerShutdown,
- final String groupId) throws Exception
- {
- checkInitialised();
- HornetQConnectionFactory cf = connectionFactories.get(name);
- if (cf == null)
- {
- cf = (HornetQConnectionFactory)HornetQJMSClient.createConnectionFactory(connectorConfigs);
- cf.setClientID(clientID);
- cf.setClientFailureCheckPeriod(clientFailureCheckPeriod);
- cf.setConnectionTTL(connectionTTL);
- cf.setCallTimeout(callTimeout);
- cf.setCacheLargeMessagesClient(cacheLargeMessagesClient);
- cf.setMinLargeMessageSize(minLargeMessageSize);
- cf.setConsumerWindowSize(consumerWindowSize);
- cf.setConsumerMaxRate(consumerMaxRate);
- cf.setConfirmationWindowSize(confirmationWindowSize);
- cf.setProducerWindowSize(producerWindowSize);
- cf.setProducerMaxRate(producerMaxRate);
- cf.setBlockOnAcknowledge(blockOnAcknowledge);
- cf.setBlockOnDurableSend(blockOnDurableSend);
- cf.setBlockOnNonDurableSend(blockOnNonDurableSend);
- cf.setAutoGroup(autoGroup);
- cf.setPreAcknowledge(preAcknowledge);
- cf.setConnectionLoadBalancingPolicyClassName(loadBalancingPolicyClassName);
- cf.setTransactionBatchSize(transactionBatchSize);
- cf.setDupsOKBatchSize(dupsOKBatchSize);
- cf.setUseGlobalPools(useGlobalPools);
- cf.setScheduledThreadPoolMaxSize(scheduledThreadPoolMaxSize);
- cf.setThreadPoolMaxSize(threadPoolMaxSize);
- cf.setRetryInterval(retryInterval);
- cf.setRetryIntervalMultiplier(retryIntervalMultiplier);
- cf.setMaxRetryInterval(maxRetryInterval);
- cf.setReconnectAttempts(reconnectAttempts);
- cf.setFailoverOnInitialConnection(failoverOnInitialConnection);
- cf.setFailoverOnServerShutdown(failoverOnServerShutdown);
- cf.setGroupID(groupId);
- }
- return cf;
- }
-
private String[] getJNDIList(final Map<String, List<String>> map, final String name)
{
List<String> result = map.get(name);
@@ -1151,82 +978,73 @@
private HornetQConnectionFactory internalCreateCF(final ConnectionFactoryConfiguration cfConfig) throws HornetQException,
Exception
{
- List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs = lookupConnectors(cfConfig);
+ checkInitialised();
- lookupDiscovery(cfConfig);
- HornetQConnectionFactory cf;
- if (cfConfig.getDiscoveryAddress() != null)
- {
- cf = internalCreateConnectionFactory(cfConfig.getName(),
- cfConfig.getLocalBindAddress(),
- cfConfig.getDiscoveryAddress(),
- cfConfig.getDiscoveryPort(),
- cfConfig.getClientID(),
- cfConfig.getDiscoveryRefreshTimeout(),
- cfConfig.getInitialWaitTimeout(),
- cfConfig.getClientFailureCheckPeriod(),
- cfConfig.getConnectionTTL(),
- cfConfig.getCallTimeout(),
- cfConfig.isCacheLargeMessagesClient(),
- cfConfig.getMinLargeMessageSize(),
- cfConfig.getConsumerWindowSize(),
- cfConfig.getConsumerMaxRate(),
- cfConfig.getConfirmationWindowSize(),
- cfConfig.getProducerWindowSize(),
- cfConfig.getProducerMaxRate(),
- cfConfig.isBlockOnAcknowledge(),
- cfConfig.isBlockOnDurableSend(),
- cfConfig.isBlockOnNonDurableSend(),
- cfConfig.isAutoGroup(),
- cfConfig.isPreAcknowledge(),
- cfConfig.getLoadBalancingPolicyClassName(),
- cfConfig.getTransactionBatchSize(),
- cfConfig.getDupsOKBatchSize(),
- cfConfig.isUseGlobalPools(),
- cfConfig.getScheduledThreadPoolMaxSize(),
- cfConfig.getThreadPoolMaxSize(),
- cfConfig.getRetryInterval(),
- cfConfig.getRetryIntervalMultiplier(),
- cfConfig.getMaxRetryInterval(),
- cfConfig.getReconnectAttempts(),
- cfConfig.isFailoverOnInitialConnection(),
- cfConfig.isFailoverOnServerShutdown(),
- cfConfig.getGroupID());
- }
- else
+ HornetQConnectionFactory cf = connectionFactories.get(cfConfig.getName());
+
+ if (cf == null)
{
- cf = internalCreateConnectionFactory(cfConfig.getName(),
- connectorConfigs,
- cfConfig.getClientID(),
- cfConfig.getClientFailureCheckPeriod(),
- cfConfig.getConnectionTTL(),
- cfConfig.getCallTimeout(),
- cfConfig.isCacheLargeMessagesClient(),
- cfConfig.getMinLargeMessageSize(),
- cfConfig.getConsumerWindowSize(),
- cfConfig.getConsumerMaxRate(),
- cfConfig.getConfirmationWindowSize(),
- cfConfig.getProducerWindowSize(),
- cfConfig.getProducerMaxRate(),
- cfConfig.isBlockOnAcknowledge(),
- cfConfig.isBlockOnDurableSend(),
- cfConfig.isBlockOnNonDurableSend(),
- cfConfig.isAutoGroup(),
- cfConfig.isPreAcknowledge(),
- cfConfig.getLoadBalancingPolicyClassName(),
- cfConfig.getTransactionBatchSize(),
- cfConfig.getDupsOKBatchSize(),
- cfConfig.isUseGlobalPools(),
- cfConfig.getScheduledThreadPoolMaxSize(),
- cfConfig.getThreadPoolMaxSize(),
- cfConfig.getRetryInterval(),
- cfConfig.getRetryIntervalMultiplier(),
- cfConfig.getMaxRetryInterval(),
- cfConfig.getReconnectAttempts(),
- cfConfig.isFailoverOnInitialConnection(),
- cfConfig.isFailoverOnServerShutdown(),
- cfConfig.getGroupID());
+ if (cfConfig.getDiscoveryAddress() != null)
+ {
+
+ if (cfConfig.isHA())
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithHA(cfConfig.getDiscoveryAddress(),
+ cfConfig.getDiscoveryPort());
+ }
+ else
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithoutHA(cfConfig.getDiscoveryAddress(),
+ cfConfig.getDiscoveryPort());
+ }
+ cf.setLocalBindAddress(cfConfig.getLocalBindAddress());
+ cf.setDiscoveryRefreshTimeout(cfConfig.getDiscoveryRefreshTimeout());
+ cf.setDiscoveryInitialWaitTimeout(cfConfig.getInitialWaitTimeout());
+ }
+ else
+ {
+ if (cfConfig.isHA())
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithHA(cfConfig.getConnectorConfigs()
+ .toArray(new TransportConfiguration[0]));
+ }
+ else
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithoutHA(cfConfig.getConnectorConfigs()
+ .toArray(new TransportConfiguration[0]));
+ }
+ }
+
+ cf.setClientID(cfConfig.getClientID());
+ cf.setClientFailureCheckPeriod(cfConfig.getClientFailureCheckPeriod());
+ cf.setConnectionTTL(cfConfig.getConnectionTTL());
+ cf.setCallTimeout(cfConfig.getCallTimeout());
+ cf.setCacheLargeMessagesClient(cfConfig.isCacheLargeMessagesClient());
+ cf.setMinLargeMessageSize(cfConfig.getMinLargeMessageSize());
+ cf.setConsumerWindowSize(cfConfig.getConsumerWindowSize());
+ cf.setConsumerMaxRate(cfConfig.getConsumerMaxRate());
+ cf.setConfirmationWindowSize(cfConfig.getConfirmationWindowSize());
+ cf.setProducerWindowSize(cfConfig.getProducerWindowSize());
+ cf.setProducerMaxRate(cfConfig.getProducerMaxRate());
+ cf.setBlockOnAcknowledge(cfConfig.isBlockOnAcknowledge());
+ cf.setBlockOnDurableSend(cfConfig.isBlockOnDurableSend());
+ cf.setBlockOnNonDurableSend(cfConfig.isBlockOnNonDurableSend());
+ cf.setAutoGroup(cfConfig.isAutoGroup());
+ cf.setPreAcknowledge(cfConfig.isPreAcknowledge());
+ cf.setConnectionLoadBalancingPolicyClassName(cfConfig.getLoadBalancingPolicyClassName());
+ cf.setTransactionBatchSize(cfConfig.getTransactionBatchSize());
+ cf.setDupsOKBatchSize(cfConfig.getDupsOKBatchSize());
+ cf.setUseGlobalPools(cfConfig.isUseGlobalPools());
+ cf.setScheduledThreadPoolMaxSize(cfConfig.getScheduledThreadPoolMaxSize());
+ cf.setThreadPoolMaxSize(cfConfig.getThreadPoolMaxSize());
+ cf.setRetryInterval(cfConfig.getRetryInterval());
+ cf.setRetryIntervalMultiplier(cfConfig.getRetryIntervalMultiplier());
+ cf.setMaxRetryInterval(cfConfig.getMaxRetryInterval());
+ cf.setReconnectAttempts(cfConfig.getReconnectAttempts());
+ cf.setFailoverOnInitialConnection(cfConfig.isFailoverOnInitialConnection());
+ cf.setFailoverOnServerShutdown(cfConfig.isFailoverOnServerShutdown());
}
+
connectionFactories.put(cfConfig.getName(), cf);
jmsManagementService.registerConnectionFactory(cfConfig.getName(), cf);
@@ -1234,64 +1052,6 @@
return cf;
}
- public synchronized void createConnectionFactory(final String name,
- final TransportConfiguration liveTC,
- final String... jndiBindings) throws Exception
- {
- checkInitialised();
- HornetQConnectionFactory cf = connectionFactories.get(name);
- if (cf == null)
- {
- ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, liveTC);
- createConnectionFactory(true, configuration, jndiBindings);
- }
- }
-
- public synchronized void createConnectionFactory(final String name,
- final String clientID,
- final TransportConfiguration liveTC,
- final String... jndiBindings) throws Exception
- {
- checkInitialised();
- HornetQConnectionFactory cf = connectionFactories.get(name);
- if (cf == null)
- {
- ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, liveTC);
- configuration.setClientID(clientID);
- createConnectionFactory(true, configuration, jndiBindings);
- }
- }
-
- public synchronized void createConnectionFactory(final String name,
- final TransportConfiguration liveTC,
- final TransportConfiguration backupTC,
- final String... jndiBindings) throws Exception
- {
- checkInitialised();
- HornetQConnectionFactory cf = connectionFactories.get(name);
- if (cf == null)
- {
- ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, liveTC, backupTC);
- createConnectionFactory(true, configuration, jndiBindings);
- }
- }
-
- public synchronized void createConnectionFactory(final String name,
- final String clientID,
- final TransportConfiguration liveTC,
- final TransportConfiguration backupTC,
- final String... jndiBindings) throws Exception
- {
- checkInitialised();
- HornetQConnectionFactory cf = connectionFactories.get(name);
- if (cf == null)
- {
- ConnectionFactoryConfiguration configuration = new ConnectionFactoryConfigurationImpl(name, liveTC, backupTC);
- configuration.setClientID(clientID);
- createConnectionFactory(true, configuration, jndiBindings);
- }
- }
-
public synchronized boolean destroyConnectionFactory(final String name) throws Exception
{
checkInitialised();
@@ -1412,38 +1172,6 @@
return true;
}
- /**
- * @param cfConfig
- * @throws HornetQException
- */
- private void lookupDiscovery(final ConnectionFactoryConfiguration cfConfig) throws HornetQException
- {
- if (cfConfig.getDiscoveryGroupName() != null)
- {
- Configuration configuration = server.getConfiguration();
-
- DiscoveryGroupConfiguration discoveryGroupConfiguration = null;
- discoveryGroupConfiguration = configuration.getDiscoveryGroupConfigurations()
- .get(cfConfig.getDiscoveryGroupName());
-
- if (discoveryGroupConfiguration == null)
- {
- JMSServerManagerImpl.log.warn("There is no discovery group with name '" + cfConfig.getDiscoveryGroupName() +
- "' deployed.");
-
- throw new HornetQException(HornetQException.ILLEGAL_STATE,
- "There is no discovery group with name '" + cfConfig.getDiscoveryGroupName() +
- "' deployed.");
- }
-
- cfConfig.setLocalBindAddress(discoveryGroupConfiguration.getLocalBindAddress());
- cfConfig.setDiscoveryAddress(discoveryGroupConfiguration.getGroupAddress());
- cfConfig.setDiscoveryPort(discoveryGroupConfiguration.getGroupPort());
- cfConfig.setDiscoveryRefreshTimeout(discoveryGroupConfiguration.getRefreshTimeout());
-
- }
- }
-
private void deploy() throws Exception
{
if (config == null)
@@ -1656,60 +1384,4 @@
}
}
- /**
- * @param cfConfig
- * @return
- * @throws HornetQException
- */
- private List<Pair<TransportConfiguration, TransportConfiguration>> lookupConnectors(final ConnectionFactoryConfiguration cfConfig) throws HornetQException
- {
- if (cfConfig.getConnectorConfigs() != null && cfConfig.getConnectorConfigs().size() > 0)
- {
- return cfConfig.getConnectorConfigs();
- }
- else if (cfConfig.getConnectorNames() != null)
- {
- Configuration configuration = server.getConfiguration();
- List<Pair<TransportConfiguration, TransportConfiguration>> connectorConfigs = new ArrayList<Pair<TransportConfiguration, TransportConfiguration>>();
-
- for (Pair<String, String> configConnector : cfConfig.getConnectorNames())
- {
- String connectorName = configConnector.a;
- String backupConnectorName = configConnector.b;
-
- TransportConfiguration connector = configuration.getConnectorConfigurations().get(connectorName);
-
- if (connector == null)
- {
- JMSServerManagerImpl.log.warn("There is no connector with name '" + connectorName + "' deployed.");
- throw new HornetQException(HornetQException.ILLEGAL_STATE,
- "There is no connector with name '" + connectorName + "' deployed.");
- }
-
- TransportConfiguration backupConnector = null;
-
- if (backupConnectorName != null)
- {
- backupConnector = configuration.getConnectorConfigurations().get(backupConnectorName);
-
- if (backupConnector == null)
- {
- JMSServerManagerImpl.log.warn("There is no backup connector with name '" + backupConnectorName +
- "' deployed.");
- throw new HornetQException(HornetQException.ILLEGAL_STATE,
- "There is no backup connector with name '" + backupConnectorName +
- "' deployed.");
- }
- }
-
- connectorConfigs.add(new Pair<TransportConfiguration, TransportConfiguration>(connector, backupConnector));
- }
- return connectorConfigs;
-
- }
- else
- {
- return null;
- }
- }
}
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/jms/server/recovery/HornetQXAResourceWrapper.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -21,7 +21,11 @@
import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.TransportConfiguration;
-import org.hornetq.api.core.client.*;
+import org.hornetq.api.core.client.ClientSession;
+import org.hornetq.api.core.client.ClientSessionFactory;
+import org.hornetq.api.core.client.HornetQClient;
+import org.hornetq.api.core.client.ServerLocator;
+import org.hornetq.api.core.client.SessionFailureListener;
import org.hornetq.core.logging.Logger;
/**
@@ -56,6 +60,8 @@
private final String password;
+ private ServerLocator serverLocator;
+
private ClientSessionFactory csf;
private XAResource delegate;
@@ -274,7 +280,8 @@
}
TransportConfiguration config = new TransportConfiguration(connectorFactoryClassName, connectorConfig);
- csf = HornetQClient.createClientSessionFactory(config);
+ serverLocator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration[]{config});
+ csf = serverLocator.createSessionFactory();
ClientSession cs = null;
if (username == null)
@@ -302,16 +309,20 @@
{
try
{
+ ServerLocator oldServerLocator = null;
ClientSessionFactory oldCSF = null;
synchronized (HornetQXAResourceWrapper.lock)
{
oldCSF = csf;
csf = null;
delegate = null;
+ oldServerLocator = serverLocator;
+ serverLocator = null;
}
if (oldCSF != null)
{
oldCSF.close();
+ oldServerLocator.close();
}
}
catch (Exception ignored)
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/ConnectionFactoryProperties.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/ConnectionFactoryProperties.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/ConnectionFactoryProperties.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -42,14 +42,9 @@
* The transport config, changing the default configured from the RA
*/
private Map<String, Object> connectionParameters;
+
+ private Boolean ha;
- /**
- * The transport config, changing the default configured from the RA
- */
- private Map<String, Object> backupConnectionParameters;
-
- private String backupConnectorClassName;
-
private String connectionLoadBalancingPolicyClassName;
private String discoveryAddress;
@@ -131,35 +126,23 @@
hasBeenUpdated = true;
}
- public String getBackupConnectorClassName()
+ public Boolean isHA()
{
- return backupConnectorClassName;
+ return ha;
}
-
- public Map<String, Object> getParsedBackupConnectionParameters()
+
+ public void setHA(final Boolean ha)
{
- return backupConnectionParameters;
- }
-
- public void setParsedBackupConnectionParameters(final Map<String, Object> backupConnectionParameters)
- {
- this.backupConnectionParameters = backupConnectionParameters;
hasBeenUpdated = true;
+ this.ha = ha;
}
-
- public void setBackupConnectorClassName(final String backupConnectorClassName)
- {
- this.backupConnectorClassName = backupConnectorClassName;
- hasBeenUpdated = true;
- }
-
+
public String getConnectionLoadBalancingPolicyClassName()
{
if (ConnectionFactoryProperties.trace)
{
ConnectionFactoryProperties.log.trace("getConnectionLoadBalancingPolicyClassName()");
}
- hasBeenUpdated = true;
return connectionLoadBalancingPolicyClassName;
}
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQRAMCFProperties.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQRAMCFProperties.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQRAMCFProperties.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -113,20 +113,6 @@
}
/**
- * @return the connectionParameters
- */
- public String getBackupConnectionParameters()
- {
- return strBackupConnectionParameters;
- }
-
- public void setBackupConnectionParameters(final String configuration)
- {
- strBackupConnectionParameters = configuration;
- setParsedBackupConnectionParameters(Util.parseConfig(configuration));
- }
-
- /**
* Set the default session type.
*
* @param defaultType either javax.jms.Topic or javax.jms.Queue
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQRAManagedConnectionFactory.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -396,26 +396,6 @@
mcfProperties.setConnectionParameters(configuration);
}
- public String getBackupConnectorClassName()
- {
- return mcfProperties.getBackupConnectorClassName();
- }
-
- public void setBackupConnectorClassName(final String backupConnectorClassName)
- {
- mcfProperties.setBackupConnectorClassName(backupConnectorClassName);
- }
-
- public String getBackupConnectionParameters()
- {
- return mcfProperties.getBackupConnectionParameters();
- }
-
- public void setBackupConnectionParameters(final String configuration)
- {
- mcfProperties.setBackupConnectionParameters(configuration);
- }
-
/**
* @return the transportType
*/
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQResourceAdapter.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/HornetQResourceAdapter.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -33,9 +33,9 @@
import org.hornetq.api.core.client.ClientSession;
import org.hornetq.api.core.client.ClientSessionFactory;
import org.hornetq.api.core.client.HornetQClient;
-import org.hornetq.jms.client.HornetQConnectionFactory;
import org.hornetq.api.jms.HornetQJMSClient;
import org.hornetq.core.logging.Logger;
+import org.hornetq.jms.client.HornetQConnectionFactory;
import org.hornetq.ra.inflow.HornetQActivation;
import org.hornetq.ra.inflow.HornetQActivationSpec;
@@ -80,11 +80,6 @@
private String unparsedProperties;
/**
- * The JBoss connection factory
- */
- private ClientSessionFactory sessionFactory;
-
- /**
* Have the factory been configured
*/
private final AtomicBoolean configured;
@@ -107,7 +102,6 @@
}
raProperties = new HornetQRAProperties();
- sessionFactory = null;
configured = new AtomicBoolean(false);
activations = new ConcurrentHashMap<ActivationSpec, HornetQActivation>();
}
@@ -254,33 +248,6 @@
}
}
- public String getBackupConnectorClassName()
- {
- return raProperties.getBackupConnectorClassName();
- }
-
- public void setBackupConnectorClassName(final String backupConnector)
- {
- if (HornetQResourceAdapter.trace)
- {
- HornetQResourceAdapter.log.trace("setBackUpTransportType(" + backupConnector + ")");
- }
- raProperties.setBackupConnectorClassName(backupConnector);
- }
-
- public Map<String, Object> getBackupConnectionParameters()
- {
- return raProperties.getParsedBackupConnectionParameters();
- }
-
- public void setBackupTransportConfiguration(final String config)
- {
- if (config != null)
- {
- raProperties.setParsedBackupConnectionParameters(Util.parseConfig(config));
- }
- }
-
/**
* Get the discovery group name
*
@@ -340,6 +307,16 @@
raProperties.setDiscoveryPort(dgp);
}
+
+ public Boolean isHA()
+ {
+ return raProperties.isHA();
+ }
+
+ public void setHA(final Boolean ha)
+ {
+ this.raProperties.setHA(ha);
+ }
/**
* Get discovery refresh timeout
@@ -1331,7 +1308,6 @@
protected void setup() throws HornetQException
{
defaultHornetQConnectionFactory = createHornetQConnectionFactory(raProperties);
- sessionFactory = defaultHornetQConnectionFactory.getCoreFactory();
}
public HornetQConnectionFactory getDefaultHornetQConnectionFactory() throws ResourceException
@@ -1350,6 +1326,7 @@
return defaultHornetQConnectionFactory;
}
+ //TODO - currently RA only allows a single target server to be specified we should allow a list of servers to be passed in
public HornetQConnectionFactory createHornetQConnectionFactory(final ConnectionFactoryProperties overrideProperties)
{
HornetQConnectionFactory cf;
@@ -1357,28 +1334,38 @@
: getConnectorClassName();
String discoveryAddress = overrideProperties.getDiscoveryAddress() != null ? overrideProperties.getDiscoveryAddress()
: getDiscoveryAddress();
+
+ Boolean ha = overrideProperties.isHA() != null ? overrideProperties.isHA() : isHA();
+
if (connectorClassName != null)
{
Map<String, Object> connectionParams =
overrideConnectionParameters(overrideProperties.getParsedConnectionParameters(),raProperties.getParsedConnectionParameters());
+
TransportConfiguration transportConf = new TransportConfiguration(connectorClassName, connectionParams);
-
- String backUpCOnnectorClassname = overrideProperties.getBackupConnectorClassName() != null ? overrideProperties.getBackupConnectorClassName()
- : getBackupConnectorClassName();
- Map<String, Object> backupConnectionParams =
- overrideConnectionParameters(overrideProperties.getParsedBackupConnectionParameters(),
- getBackupConnectionParameters());
- TransportConfiguration backup = backUpCOnnectorClassname == null ? null
- : new TransportConfiguration(backUpCOnnectorClassname,
- backupConnectionParams);
-
- cf = HornetQJMSClient.createConnectionFactory(transportConf, backup);
+
+ if (ha)
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithHA(new TransportConfiguration[] {transportConf});
+ }
+ else
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithoutHA(new TransportConfiguration[] {transportConf});
+ }
}
else if (discoveryAddress != null)
{
Integer discoveryPort = overrideProperties.getDiscoveryPort() != null ? overrideProperties.getDiscoveryPort()
: getDiscoveryPort();
- cf = HornetQJMSClient.createConnectionFactory(discoveryAddress, discoveryPort);
+
+ if (ha)
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithHA(discoveryAddress, discoveryPort);
+ }
+ else
+ {
+ cf = HornetQJMSClient.createConnectionFactoryWithoutHA(discoveryAddress, discoveryPort);
+ }
}
else
{
Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/inflow/HornetQActivation.java
===================================================================
--- branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/inflow/HornetQActivation.java 2010-06-30 17:22:16 UTC (rev 9376)
+++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/ra/inflow/HornetQActivation.java 2010-07-01 16:14:07 UTC (rev 9377)
@@ -32,10 +32,10 @@
import org.hornetq.api.core.HornetQException;
import org.hornetq.api.core.SimpleString;
import org.hornetq.api.core.client.ClientSession;
+import org.hornetq.api.jms.HornetQJMSClient;
+import org.hornetq.core.logging.Logger;
import org.hornetq.jms.client.HornetQConnectionFactory;
import org.hornetq.jms.client.HornetQDestination;
-import org.hornetq.api.jms.HornetQJMSClient;
-import org.hornetq.core.logging.Logger;
import org.hornetq.ra.HornetQResourceAdapter;
import org.hornetq.ra.Util;
@@ -352,7 +352,7 @@
try
{
- result = ra.createSession(factory.getCoreFactory(),
+ result = ra.createSession(factory.getServerLocator().createSessionFactory(),
spec.getAcknowledgeModeInt(),
spec.getUser(),
spec.getPassword(),
13 years, 10 months
- 1
- 0