[jboss-cvs] JBoss Messaging SVN: r4926 - in trunk: src/main/org/jboss/messaging/core/persistence/impl/journal and 17 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Wed Sep 10 06:51:55 EDT 2008
Author: timfox
Date: 2008-09-10 06:51:55 -0400 (Wed, 10 Sep 2008)
New Revision: 4926
Added:
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ReattachSessionMessage.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java
trunk/tests/src/org/jboss/messaging/tests/integration/cluster/
trunk/tests/src/org/jboss/messaging/tests/integration/cluster/ReplicationTest.java
Modified:
trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionFactoryImpl.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java
trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionPacketHandler.java
trunk/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java
trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java
trunk/src/main/org/jboss/messaging/core/remoting/Channel.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/ConnectionRegistryImpl.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/PacketImpl.java
trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionReplicateDeliveryResponseMessage.java
trunk/src/main/org/jboss/messaging/core/server/MessagingServer.java
trunk/src/main/org/jboss/messaging/core/server/ServerSession.java
trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerImpl.java
trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerPacketHandler.java
trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java
trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java
trunk/src/main/org/jboss/messaging/core/transaction/impl/TransactionImpl.java
trunk/src/main/org/jboss/messaging/jms/client/JBossConnectionFactory.java
trunk/src/main/org/jboss/messaging/jms/server/JMSServerManager.java
trunk/src/main/org/jboss/messaging/jms/server/impl/JMSServerDeployer.java
trunk/src/main/org/jboss/messaging/jms/server/impl/JMSServerManagerImpl.java
trunk/src/main/org/jboss/messaging/jms/server/management/JMSServerControlMBean.java
trunk/src/main/org/jboss/messaging/jms/server/management/impl/JMSServerControl.java
trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/CTSMiscellaneousTest.java
trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/JMSTestCase.java
trunk/tests/jms-tests/src/org/jboss/test/messaging/tools/container/LocalTestServer.java
trunk/tests/src/org/jboss/messaging/tests/unit/core/journal/impl/AlignedJournalImplTest.java
Log:
Session replication and failover
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionFactoryImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionFactoryImpl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionFactoryImpl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -21,7 +21,8 @@
*/
package org.jboss.messaging.core.client.impl;
-import java.util.HashMap;
+import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.EXCEPTION;
+
import java.util.Map;
import org.jboss.messaging.core.client.ClientSession;
@@ -32,14 +33,12 @@
import org.jboss.messaging.core.remoting.Channel;
import org.jboss.messaging.core.remoting.ChannelHandler;
import org.jboss.messaging.core.remoting.ConnectionRegistry;
-import org.jboss.messaging.core.remoting.Interceptor;
import org.jboss.messaging.core.remoting.Packet;
import org.jboss.messaging.core.remoting.RemotingConnection;
import org.jboss.messaging.core.remoting.impl.ConnectionRegistryImpl;
import org.jboss.messaging.core.remoting.impl.wireformat.CreateSessionMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.CreateSessionResponseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.MessagingExceptionMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl;
import org.jboss.messaging.core.remoting.spi.ConnectorFactory;
import org.jboss.messaging.core.version.Version;
import org.jboss.messaging.util.VersionLoader;
@@ -90,6 +89,10 @@
private volatile Map<String, Object> transportParams;
+ private volatile ConnectorFactory backupConnectorFactory;
+
+ private volatile Map<String, Object> backupTransportParams;
+
private volatile long pingPeriod;
private volatile long callTimeout;
@@ -112,24 +115,11 @@
// Constructors ---------------------------------------------------------------------------------
- private void instantiateConnectorFactory(final String connectorFactoryClassName)
- {
- ClassLoader loader = Thread.currentThread().getContextClassLoader();
- try
- {
- Class<?> clazz = loader.loadClass(connectorFactoryClassName);
- connectorFactory = (ConnectorFactory)clazz.newInstance();
- }
- catch (Exception e)
- {
- log.warn("Error instantiating connector factory \"" + connectorFactoryClassName + "\"", e);
- }
- }
-
/**
* Create a ClientSessionFactoryImpl specifying all attributes
*/
public ClientSessionFactoryImpl(final TransportConfiguration connectorConfig,
+ final TransportConfiguration backupConfig,
final long pingPeriod,
final long callTimeout,
final int consumerWindowSize, final int consumerMaxRate,
@@ -138,8 +128,13 @@
final boolean blockOnNonPersistentSend,
final boolean blockOnPersistentSend)
{
- instantiateConnectorFactory(connectorConfig.getFactoryClassName());
+ this.connectorFactory = instantiateConnectorFactory(connectorConfig.getFactoryClassName());
this.transportParams = connectorConfig.getParams();
+ if (backupConfig != null)
+ {
+ this.backupConnectorFactory = instantiateConnectorFactory(backupConfig.getFactoryClassName());
+ this.backupTransportParams = backupConfig.getParams();
+ }
this.pingPeriod = pingPeriod;
this.callTimeout = callTimeout;
this.consumerWindowSize = consumerWindowSize;
@@ -152,12 +147,34 @@
this.connectionRegistry = ConnectionRegistryImpl.instance;
}
+ public ClientSessionFactoryImpl(final TransportConfiguration connectorConfig,
+ final TransportConfiguration backupConfig)
+ {
+ this.connectorFactory = instantiateConnectorFactory(connectorConfig.getFactoryClassName());
+ this.transportParams = connectorConfig.getParams();
+ if (backupConfig != null)
+ {
+ this.backupConnectorFactory = instantiateConnectorFactory(backupConfig.getFactoryClassName());
+ this.backupTransportParams = backupConfig.getParams();
+ }
+ pingPeriod = DEFAULT_PING_PERIOD;
+ callTimeout = DEFAULT_CALL_TIMEOUT;
+ consumerWindowSize = DEFAULT_CONSUMER_WINDOW_SIZE;
+ consumerMaxRate = DEFAULT_CONSUMER_MAX_RATE;
+ producerWindowSize = DEFAULT_PRODUCER_WINDOW_SIZE;
+ producerMaxRate = DEFAULT_PRODUCER_MAX_RATE;
+ blockOnAcknowledge = DEFAULT_BLOCK_ON_ACKNOWLEDGE;
+ blockOnPersistentSend = DEFAULT_BLOCK_ON_PERSISTENT_SEND;
+ blockOnNonPersistentSend = DEFAULT_BLOCK_ON_NON_PERSISTENT_SEND;
+ this.connectionRegistry = ConnectionRegistryImpl.instance;
+ }
+
/**
* Create a ClientSessionFactoryImpl specify transport type and using defaults
*/
public ClientSessionFactoryImpl(final TransportConfiguration connectorConfig)
{
- instantiateConnectorFactory(connectorConfig.getFactoryClassName());
+ this.connectorFactory = instantiateConnectorFactory(connectorConfig.getFactoryClassName());
this.transportParams = connectorConfig.getParams();
pingPeriod = DEFAULT_PING_PERIOD;
callTimeout = DEFAULT_CALL_TIMEOUT;
@@ -280,7 +297,27 @@
{
this.transportParams = transportParams;
}
+
+ public ConnectorFactory getBackupConnectorFactory()
+ {
+ return backupConnectorFactory;
+ }
+ public void setBqackupConnectorFactory(final ConnectorFactory connectorFactory)
+ {
+ this.backupConnectorFactory = connectorFactory;
+ }
+
+ public Map<String, Object> getBackupTransportParams()
+ {
+ return backupTransportParams;
+ }
+
+ public void setBackupTransportParams(final Map<String, Object> transportParams)
+ {
+ this.backupTransportParams = transportParams;
+ }
+
public long getPingPeriod()
{
return pingPeriod;
@@ -314,9 +351,21 @@
// Private --------------------------------------------------------------------------------------
- //TODO for now needs to be synchronized since we can't currently cope with more than one
- //concurrent blocking send on a particular channel.
- private synchronized ClientSession createSessionInternal(final String username, final String password, final boolean xa,
+ private ConnectorFactory instantiateConnectorFactory(final String connectorFactoryClassName)
+ {
+ ClassLoader loader = Thread.currentThread().getContextClassLoader();
+ try
+ {
+ Class<?> clazz = loader.loadClass(connectorFactoryClassName);
+ return (ConnectorFactory)clazz.newInstance();
+ }
+ catch (Exception e)
+ {
+ throw new IllegalArgumentException("Error instantiating connector factory \"" + connectorFactoryClassName + "\"", e);
+ }
+ }
+
+ private ClientSession createSessionInternal(final String username, final String password, final boolean xa,
final boolean autoCommitSends, final boolean autoCommitAcks,
int lazyAckBatchSize, boolean cacheProducers)
throws MessagingException
@@ -324,10 +373,17 @@
Version clientVersion = VersionLoader.load();
RemotingConnection remotingConnection = null;
+ RemotingConnection backupConnection = null;
try
{
remotingConnection = connectionRegistry.getConnection(connectorFactory, transportParams,
pingPeriod, callTimeout);
+
+ if (backupConnectorFactory != null)
+ {
+ backupConnection = connectionRegistry.getConnection(backupConnectorFactory, backupTransportParams,
+ pingPeriod, callTimeout);
+ }
Packet request =
new CreateSessionMessage(clientVersion.getIncrementingVersion(),
@@ -338,7 +394,7 @@
Packet packet = channel1.sendBlocking(request);
- if (packet.getType() == PacketImpl.EXCEPTION)
+ if (packet.getType() == EXCEPTION)
{
MessagingExceptionMessage mem = (MessagingExceptionMessage)packet;
@@ -351,10 +407,11 @@
Channel sessionChannel = remotingConnection.getChannel(sessionID, false, response.getPacketConfirmationBatchSize());
- ClientSessionInternal session = new ClientSessionImpl(sessionID, xa, lazyAckBatchSize, cacheProducers,
+ ClientSessionInternal session = new ClientSessionImpl(xa, lazyAckBatchSize, cacheProducers,
autoCommitSends, autoCommitAcks, blockOnAcknowledge,
remotingConnection, this,
- response.getServerVersion(), sessionChannel);
+ response.getServerVersion(), sessionChannel,
+ backupConnection);
ChannelHandler handler = new ClientSessionPacketHandler(session);
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionImpl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -50,6 +50,8 @@
import org.jboss.messaging.core.remoting.RemotingConnection;
import org.jboss.messaging.core.remoting.impl.ConnectionRegistryImpl;
import org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl;
+import org.jboss.messaging.core.remoting.impl.wireformat.ReattachSessionMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.ReattachSessionResponseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionAcknowledgeMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionAddDestinationMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryMessage;
@@ -122,7 +124,7 @@
private final Executor executor;
- private final RemotingConnection remotingConnection;
+ private volatile RemotingConnection remotingConnection;
private final Map<Integer, ClientBrowser> browsers = new ConcurrentHashMap<Integer, ClientBrowser>();
@@ -166,18 +168,23 @@
// For testing only
private boolean forceNotSameRM;
-
+
+ private final FailureListener failoverListener = new FailoverListener();
+
+ private volatile RemotingConnection backupConnection;
+
// Constructors
// ---------------------------------------------------------------------------------
-
- public ClientSessionImpl(final long serverTargetID,
+
+ public ClientSessionImpl(
final boolean xa, final int lazyAckBatchSize,
final boolean cacheProducers, final boolean autoCommitSends,
final boolean autoCommitAcks, final boolean blockOnAcknowledge,
final RemotingConnection remotingConnection,
final ClientSessionFactory connectionFactory,
final int version,
- final Channel channel)
+ final Channel channel,
+ final RemotingConnection backupConnection)
throws MessagingException
{
if (lazyAckBatchSize < -1 || lazyAckBatchSize == 0)
@@ -217,6 +224,13 @@
this.version = version;
this.connectionRegistry = ConnectionRegistryImpl.instance;
+
+ this.backupConnection = backupConnection;
+
+ if (backupConnection != null)
+ {
+ this.remotingConnection.addFailureListener(failoverListener);
+ }
}
// ClientSession implementation
@@ -565,8 +579,7 @@
public ClientMessage createClientMessage(byte type, boolean durable,
long expiration, long timestamp, byte priority)
{
- MessagingBuffer body = remotingConnection
- .createBuffer(INITIAL_MESSAGE_BODY_SIZE);
+ MessagingBuffer body = remotingConnection.createBuffer(INITIAL_MESSAGE_BODY_SIZE);
return new ClientMessageImpl(type, durable, expiration, timestamp,
priority, body);
@@ -574,16 +587,14 @@
public ClientMessage createClientMessage(byte type, boolean durable)
{
- MessagingBuffer body = remotingConnection
- .createBuffer(INITIAL_MESSAGE_BODY_SIZE);
+ MessagingBuffer body = remotingConnection.createBuffer(INITIAL_MESSAGE_BODY_SIZE);
return new ClientMessageImpl(type, durable, body);
}
public ClientMessage createClientMessage(boolean durable)
{
- MessagingBuffer body = remotingConnection
- .createBuffer(INITIAL_MESSAGE_BODY_SIZE);
+ MessagingBuffer body = remotingConnection.createBuffer(INITIAL_MESSAGE_BODY_SIZE);
return new ClientMessageImpl(durable, body);
}
@@ -1049,6 +1060,11 @@
this.connectionRegistry = registry;
}
+ public RemotingConnection getConnection()
+ {
+ return remotingConnection;
+ }
+
// Protected
// ------------------------------------------------------------------------------------
@@ -1057,9 +1073,43 @@
// Private
// --------------------------------------------------------------------------------------
-
-
-
+
+ private void handleFailover(final MessagingException me)
+ {
+ log.info("Failure has been detected, initiating failover");
+
+ channel.lock();
+
+ try
+ {
+ Packet request = new ReattachSessionMessage(channel.getID(), channel.getLastReceivedCommandID());
+
+ Channel channel1 = backupConnection.getChannel(1, false, -1);
+
+ ReattachSessionResponseMessage response = (ReattachSessionResponseMessage)channel1.sendBlocking(request);
+
+ channel.transferConnection(backupConnection);
+
+ remotingConnection.removeFailureListener(failoverListener);
+
+ remotingConnection = backupConnection;
+
+ remotingConnection.addFailureListener(failoverListener);
+
+ backupConnection = null;
+
+ channel.replayCommands(response.getLastReceivedCommandID());
+ }
+ catch (Throwable t)
+ {
+ log.error("Failed to handle failover", t);
+ }
+ finally
+ {
+ channel.unlock();
+ }
+ }
+
private void checkXA() throws XAException
{
if (!xa)
@@ -1157,6 +1207,11 @@
channel.close();
connectionRegistry.returnConnection(remotingConnection.getID());
+
+ if (backupConnection != null)
+ {
+ remotingConnection.removeFailureListener(failoverListener);
+ }
closed = true;
}
@@ -1168,5 +1223,13 @@
// Inner Classes
// --------------------------------------------------------------------------------
+
+ private class FailoverListener implements FailureListener
+ {
+ public void connectionFailed(final MessagingException me)
+ {
+ handleFailover(me);
+ }
+ }
}
Modified: trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionPacketHandler.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionPacketHandler.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/client/impl/ClientSessionPacketHandler.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -48,6 +48,7 @@
public ClientSessionPacketHandler(final ClientSessionInternal clientSesssion)
{
+ // log.info("creating clientsessionpacketHandler " + System.identityHashCode(this));
this.clientSession = clientSesssion;
}
@@ -55,6 +56,8 @@
{
byte type = packet.getType();
+ // log.info(System.identityHashCode(this) + "handling packet");
+
try
{
switch (type)
Modified: trunk/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/persistence/impl/journal/JournalStorageManager.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -288,7 +288,7 @@
for (PreparedTransactionInfo preparedTransaction : preparedTransactions)
{
log.trace(preparedTransaction);
- EncodingXid encodingXid = new EncodingXid(preparedTransaction.xidData);
+ EncodingXid encodingXid = new EncodingXid(preparedTransaction.extraData);
Xid xid = encodingXid.xid;
Transaction tx = new TransactionImpl(preparedTransaction.id, xid, this, postOffice);
Modified: trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/postoffice/impl/PostOfficeImpl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -317,11 +317,14 @@
public void setBackup(final boolean backup)
{
- this.backup = backup;
-
- for (Binding binding: nameMap.values())
+ if (this.backup != backup)
{
- binding.getQueue().setBackup(backup);
+ this.backup = backup;
+
+ for (Binding binding: nameMap.values())
+ {
+ binding.getQueue().setBackup(backup);
+ }
}
}
Modified: trunk/src/main/org/jboss/messaging/core/remoting/Channel.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/Channel.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/remoting/Channel.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -31,6 +31,8 @@
*/
public interface Channel
{
+ long getID();
+
void send(Packet packet);
Packet sendBlocking(Packet packet) throws MessagingException;
@@ -40,4 +42,14 @@
void close();
Channel getReplicatingChannel();
+
+ void transferConnection(RemotingConnection newConnection);
+
+ int replayCommands(int lastReceivedCommandID);
+
+ int getLastReceivedCommandID();
+
+ void lock();
+
+ void unlock();
}
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/ConnectionRegistryImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/ConnectionRegistryImpl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/ConnectionRegistryImpl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -287,8 +287,12 @@
}
else
{
- if (this.params.size() == kother.params.size())
+ if (kother.params == null)
{
+ return false;
+ }
+ else if (this.params.size() == kother.params.size())
+ {
for (Map.Entry<String, Object> entry: this.params.entrySet())
{
Object thisVal = entry.getValue();
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -27,6 +27,8 @@
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.EXCEPTION;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.PING;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.PONG;
+import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.REATTACH_SESSION;
+import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.REATTACH_SESSION_RESP;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_ACKNOWLEDGE;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_ADD_DESTINATION;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.SESS_BINDINGQUERY;
@@ -88,6 +90,8 @@
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.jboss.messaging.core.exception.MessagingException;
import org.jboss.messaging.core.logging.Logger;
@@ -106,6 +110,8 @@
import org.jboss.messaging.core.remoting.impl.wireformat.Ping;
import org.jboss.messaging.core.remoting.impl.wireformat.Pong;
import org.jboss.messaging.core.remoting.impl.wireformat.ProducerFlowCreditMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.ReattachSessionMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.ReattachSessionResponseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.ReceiveMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SendMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionAcknowledgeMessage;
@@ -150,6 +156,7 @@
import org.jboss.messaging.core.remoting.spi.Connection;
import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
import org.jboss.messaging.util.ExecutorFactory;
+import org.jboss.messaging.util.Future;
import org.jboss.messaging.util.OrderedExecutorFactory;
/**
@@ -292,7 +299,7 @@
if (channel == null)
{
- channel = new ChannelImpl(channelID, ordered, packetConfirmationBatchSize);
+ channel = new ChannelImpl(this, channelID, ordered, packetConfirmationBatchSize);
channels.put(channelID, channel);
}
@@ -315,6 +322,8 @@
public synchronized void addFailureListener(final FailureListener listener)
{
+ checkDestroyed();
+
if (listener == null)
{
throw new IllegalStateException("FailureListener cannot be null");
@@ -335,6 +344,8 @@
public MessagingBuffer createBuffer(final int size)
{
+ checkDestroyed();
+
return transportConnection.createBuffer(size);
}
@@ -397,6 +408,8 @@
public void bufferReceived(final Object connectionID, final MessagingBuffer buffer)
{
+ //checkDestroyed();
+
final Packet packet = decode(buffer);
long channelID = packet.getChannelID();
@@ -434,13 +447,18 @@
// Private
// --------------------------------------------------------------------------------------
- private void doWrite(final Packet packet)
+ private void checkDestroyed()
{
if (destroyed)
{
- throw new IllegalStateException("Cannot write packet to connection, it is destroyed");
+ throw new IllegalStateException("Connection is destroyed");
}
-
+ }
+
+ private void doWrite(final Packet packet)
+ {
+ checkDestroyed();
+
MessagingBuffer buffer = transportConnection.createBuffer(PacketImpl.INITIAL_BUFFER_SIZE);
packet.encode(buffer);
@@ -482,6 +500,16 @@
packet = new CreateSessionResponseMessage();
break;
}
+ case REATTACH_SESSION:
+ {
+ packet = new ReattachSessionMessage();
+ break;
+ }
+ case REATTACH_SESSION_RESP:
+ {
+ packet = new ReattachSessionResponseMessage();
+ break;
+ }
case SESS_CREATECONSUMER:
{
packet = new SessionCreateConsumerMessage();
@@ -761,7 +789,8 @@
// Inner classes
// --------------------------------------------------------------------------------
- private class ChannelImpl implements Channel
+ //Needs to be static so we can re-assign it to another remotingconnection
+ private static class ChannelImpl implements Channel
{
private final long id;
@@ -783,13 +812,19 @@
private final Channel replicatingChannel;
- public ChannelImpl(final long id, final boolean ordered, final int packetConfirmationBatchSize)
+ private final ReadWriteLock lock = new ReentrantReadWriteLock(true);
+
+ private volatile RemotingConnectionImpl connection;
+
+ private ChannelImpl(final RemotingConnectionImpl connection, final long id, final boolean ordered, final int packetConfirmationBatchSize)
{
+ this.connection = connection;
+
this.id = id;
- if (ordered && executorFactory != null)
+ if (ordered && connection.executorFactory != null)
{
- executor = executorFactory.getExecutor();
+ executor = connection.executorFactory.getExecutor();
}
else
{
@@ -798,7 +833,7 @@
this.packetConfirmationBatchSize = packetConfirmationBatchSize;
- if (packetConfirmationBatchSize != -1 && (client && !backup || !client && replicatingConnection == null))
+ if (packetConfirmationBatchSize != -1 && (connection.client && !connection.backup || !connection.client && connection.replicatingConnection == null))
{
resendCache = new ConcurrentLinkedQueue<Packet>();
@@ -809,9 +844,9 @@
resendCache = null;
}
- if (replicatingConnection != null)
+ if (connection.replicatingConnection != null)
{
- replicatingChannel = replicatingConnection.getChannel(id, ordered, -1);
+ replicatingChannel = connection.replicatingConnection.getChannel(id, ordered, -1);
replicatingChannel.setHandler(new ReplicatedPacketsConfirmedChannelHandler());
}
@@ -820,72 +855,107 @@
replicatingChannel = null;
}
}
+
+ public long getID()
+ {
+ return id;
+ }
+
+ public int getLastReceivedCommandID()
+ {
+ return lastReceivedCommandID;
+ }
public void send(final Packet packet)
{
- packet.setChannelID(id);
-
- if (resendCache != null)
+ lock.readLock().lock();
+
+ try
{
- addToCache(packet);
+ packet.setChannelID(id);
+
+ if (resendCache != null)
+ {
+ addToCache(packet);
+ }
+
+ if (connection.writePackets || packet.getType() == PacketImpl.SESS_PACKETS_CONFIRMED
+ || packet.getType() == PacketImpl.SESS_REPLICATE_DELIVERY_RESP)
+ {
+ connection.doWrite(packet);
+ }
}
-
- if (writePackets || packet.getType() == PacketImpl.SESS_PACKETS_CONFIRMED
- || packet.getType() == PacketImpl.SESS_REPLICATE_DELIVERY_RESP)
- {
- doWrite(packet);
+ finally
+ {
+ lock.readLock().unlock();
}
}
+
+ private final Object blockingLock = new Object();
public synchronized Packet sendBlocking(final Packet packet) throws MessagingException
{
- response = null;
-
- packet.setChannelID(id);
-
- if (resendCache != null)
- {
- addToCache(packet);
- }
+ lock.readLock().lock();
- doWrite(packet);
-
- long toWait = blockingCallTimeout;
-
- long start = System.currentTimeMillis();
-
- while (response == null && toWait > 0)
+ try
{
- try
+ //For now we only allow one blocking request-response at a time per channel
+ //We can relax this but it will involve some kind of correlation id
+ synchronized (blockingLock)
{
- wait(toWait);
+ response = null;
+
+ packet.setChannelID(id);
+
+ if (resendCache != null)
+ {
+ addToCache(packet);
+ }
+
+ connection.doWrite(packet);
+
+ long toWait = connection.blockingCallTimeout;
+
+ long start = System.currentTimeMillis();
+
+ while (response == null && toWait > 0)
+ {
+ try
+ {
+ wait(toWait);
+ }
+ catch (InterruptedException e)
+ {
+ }
+
+ long now = System.currentTimeMillis();
+
+ toWait -= now - start;
+
+ start = now;
+ }
+
+ if (response == null)
+ {
+ throw new IllegalStateException("Timed out waiting for response");
+ }
+
+ if (response.getType() == PacketImpl.EXCEPTION)
+ {
+ MessagingExceptionMessage mem = (MessagingExceptionMessage)response;
+
+ throw mem.getException();
+ }
+ else
+ {
+ return response;
+ }
}
- catch (InterruptedException e)
- {
- }
-
- long now = System.currentTimeMillis();
-
- toWait -= now - start;
-
- start = now;
}
-
- if (response == null)
+ finally
{
- throw new IllegalStateException("Timed out waiting for response");
+ lock.readLock().unlock();
}
-
- if (response.getType() == PacketImpl.EXCEPTION)
- {
- MessagingExceptionMessage mem = (MessagingExceptionMessage)response;
-
- throw mem.getException();
- }
- else
- {
- return response;
- }
}
public void setHandler(final ChannelHandler handler)
@@ -895,7 +965,7 @@
public void close()
{
- if (!destroyed && channels.remove(id) == null)
+ if (!connection.destroyed && connection.channels.remove(id) == null)
{
throw new IllegalArgumentException("Cannot find channel with id " + id + " to close");
}
@@ -919,8 +989,63 @@
return replicatingChannel;
}
+ public void transferConnection(final RemotingConnection newConnection)
+ {
+ if (executor != null)
+ {
+ //First wait for anything in the executor to complete
+ Future future = new Future();
+
+ executor.execute(future);
+
+ boolean ok = future.await(10000);
+
+ if (!ok)
+ {
+ throw new IllegalStateException("Timed out waiting for executor to complete");
+ }
+ }
+
+ RemotingConnectionImpl rnewConnection = (RemotingConnectionImpl)newConnection;
+
+ connection.channels.remove(id);
+
+ rnewConnection.channels.put(id, this);
+
+ connection = rnewConnection;
+ }
+
+ public int replayCommands(final int otherLastReceivedCommandID)
+ {
+ clearUpTo(otherLastReceivedCommandID);
+
+ Packet packet = null;
+
+ int count = 0;
+
+ while ((packet = resendCache.poll()) != null)
+ {
+ connection.doWrite(packet);
+
+ count++;
+ }
+
+ return this.lastReceivedCommandID;
+ }
+
+ public void lock()
+ {
+ lock.writeLock().lock();
+ }
+
+ public void unlock()
+ {
+ lock.writeLock().unlock();
+ }
+
private void handlePacket(final Packet packet)
- {
+ {
+ // log.info("handling packet client " + connection.client + " backup " + connection.backup);
if (packet.getType() == PacketImpl.SESS_PACKETS_CONFIRMED)
{
if (resendCache != null)
@@ -942,7 +1067,7 @@
});
}
}
- else if (replicatingConnection != null)
+ else if (connection.replicatingConnection != null)
{
replicatingChannel.send(packet);
}
@@ -960,13 +1085,13 @@
replicatingChannel.send(packet);
}
- if (interceptors != null)
+ if (connection.interceptors != null)
{
- for (Interceptor interceptor : interceptors)
+ for (Interceptor interceptor : connection.interceptors)
{
try
{
- boolean callNext = interceptor.intercept(packet, RemotingConnectionImpl.this);
+ boolean callNext = interceptor.intercept(packet, connection);
if (!callNext)
{
@@ -1022,11 +1147,11 @@
}
private void checkConfirmation(final Packet packet)
- {
+ {
if (packet.isUsesConfirmations() && resendCache != null)
- {
+ {
lastReceivedCommandID++;
-
+
if (lastReceivedCommandID == nextConfirmation)
{
Packet confirmed = new PacketsConfirmedMessage(lastReceivedCommandID);
@@ -1034,8 +1159,8 @@
nextConfirmation += packetConfirmationBatchSize;
confirmed.setChannelID(id);
-
- doWrite(confirmed);
+
+ connection.doWrite(confirmed);
}
}
}
@@ -1046,7 +1171,7 @@
}
private void clearUpTo(final int lastReceivedCommandID)
- {
+ {
int numberToClear = 1 + lastReceivedCommandID - firstStoredCommandID;
if (numberToClear == -1)
@@ -1074,7 +1199,7 @@
if (packet.getType() == SESS_PACKETS_CONFIRMED)
{
//Send it straight back to the client
- doWrite(packet);
+ connection.doWrite(packet);
}
else if (packet.getType() == PacketImpl.SESS_REPLICATE_DELIVERY_RESP)
{
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/PacketImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/PacketImpl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/PacketImpl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -58,9 +58,8 @@
// Server
public static final byte CREATESESSION = 30;
public static final byte CREATESESSION_RESP = 31;
- public static final byte AUTHENTICATE = 32;
- public static final byte AUTHENTICATE_RESP = 33;
-
+ public static final byte REATTACH_SESSION = 32;
+ public static final byte REATTACH_SESSION_RESP = 33;
// Session
public static final byte SESS_CREATECONSUMER = 40;
Added: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ReattachSessionMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ReattachSessionMessage.java (rev 0)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ReattachSessionMessage.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -0,0 +1,107 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jboss.messaging.core.remoting.impl.wireformat;
+
+import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
+
+/**
+ *
+ * A ReattachSessionMessage
+ *
+ * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
+ *
+ */
+public class ReattachSessionMessage extends PacketImpl
+{
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ private long sessionID;
+
+ private int lastReceivedCommandID;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ public ReattachSessionMessage(final long sessionID, final int lastReceivedCommandID)
+ {
+ super(REATTACH_SESSION);
+
+ this.sessionID = sessionID;
+
+ this.lastReceivedCommandID = lastReceivedCommandID;
+ }
+
+ public ReattachSessionMessage()
+ {
+ super(REATTACH_SESSION);
+ }
+
+ // Public --------------------------------------------------------
+
+ public long getSessionID()
+ {
+ return sessionID;
+ }
+
+ public int getLastReceivedCommandID()
+ {
+ return lastReceivedCommandID;
+ }
+
+ public void encodeBody(final MessagingBuffer buffer)
+ {
+ buffer.putLong(sessionID);
+ buffer.putInt(lastReceivedCommandID);
+ }
+
+ public void decodeBody(final MessagingBuffer buffer)
+ {
+ sessionID = buffer.getLong();
+ lastReceivedCommandID = buffer.getInt();
+ }
+
+ public boolean equals(Object other)
+ {
+ if (other instanceof ReattachSessionMessage == false)
+ {
+ return false;
+ }
+
+ ReattachSessionMessage r = (ReattachSessionMessage)other;
+
+ return super.equals(other) && this.lastReceivedCommandID == r.lastReceivedCommandID &&
+ this.sessionID == r.sessionID;
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+}
+
Added: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java (rev 0)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/ReattachSessionResponseMessage.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -0,0 +1,100 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jboss.messaging.core.remoting.impl.wireformat;
+
+import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
+
+/**
+ *
+ * A ReattachSessionResponseMessage
+ *
+ * @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
+ *
+ */
+public class ReattachSessionResponseMessage extends PacketImpl
+{
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ private int lastReceivedCommandID;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ public ReattachSessionResponseMessage(final int lastReceivedCommandID)
+ {
+ super(REATTACH_SESSION_RESP);
+
+ this.lastReceivedCommandID = lastReceivedCommandID;
+ }
+
+ public ReattachSessionResponseMessage()
+ {
+ super(REATTACH_SESSION_RESP);
+ }
+
+ // Public --------------------------------------------------------
+
+ public int getLastReceivedCommandID()
+ {
+ return lastReceivedCommandID;
+ }
+
+ public void encodeBody(final MessagingBuffer buffer)
+ {
+ buffer.putInt(lastReceivedCommandID);
+ }
+
+ public void decodeBody(final MessagingBuffer buffer)
+ {
+ lastReceivedCommandID = buffer.getInt();
+ }
+
+ public boolean isResponse()
+ {
+ return true;
+ }
+
+ public boolean equals(Object other)
+ {
+ if (other instanceof ReattachSessionResponseMessage == false)
+ {
+ return false;
+ }
+
+ ReattachSessionResponseMessage r = (ReattachSessionResponseMessage)other;
+
+ return super.equals(other) && this.lastReceivedCommandID == r.lastReceivedCommandID;
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+}
+
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionReplicateDeliveryResponseMessage.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionReplicateDeliveryResponseMessage.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/wireformat/SessionReplicateDeliveryResponseMessage.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -23,7 +23,6 @@
package org.jboss.messaging.core.remoting.impl.wireformat;
-
/**
* @author <a href="mailto:tim.fox at jboss.com">Tim Fox</a>
*
Modified: trunk/src/main/org/jboss/messaging/core/server/MessagingServer.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/MessagingServer.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/server/MessagingServer.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -32,6 +32,7 @@
import org.jboss.messaging.core.remoting.RemotingConnection;
import org.jboss.messaging.core.remoting.RemotingService;
import org.jboss.messaging.core.remoting.impl.wireformat.CreateSessionResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.ReattachSessionResponseMessage;
import org.jboss.messaging.core.security.JBMSecurityManager;
import org.jboss.messaging.core.security.Role;
import org.jboss.messaging.core.settings.HierarchicalRepository;
@@ -73,13 +74,18 @@
void setManagementService(ManagementService managementService);
Version getVersion();
+
+ ReattachSessionResponseMessage reattachSession(RemotingConnection connection,
+ long sessionID, int lastReceivedCommandID);
- CreateSessionResponseMessage createSession(final String username, final String password,
- final int incrementingVersion,
- final RemotingConnection remotingConnection,
- final boolean autoCommitSends,
- final boolean autoCommitAcks,
- final boolean xa) throws Exception;
+ CreateSessionResponseMessage createSession(String username, String password,
+ int incrementingVersion,
+ RemotingConnection remotingConnection,
+ boolean autoCommitSends,
+ boolean autoCommitAcks,
+ boolean xa) throws Exception;
+
+ void removeSession(long sessionID);
boolean isStarted();
Modified: trunk/src/main/org/jboss/messaging/core/server/ServerSession.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/ServerSession.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/server/ServerSession.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -26,6 +26,7 @@
import javax.transaction.xa.Xid;
+import org.jboss.messaging.core.remoting.RemotingConnection;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionBindingQueryResponseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateConsumerResponseMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.SessionCreateProducerResponseMessage;
@@ -138,4 +139,8 @@
void handleReplicateDelivery(long messageID, int consumerID) throws Exception;
void handleDeferredDelivery();
+
+ void transferConnection(RemotingConnection newConnection);
+
+ int replayCommands(int lastReceivedCommandID);
}
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerImpl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerImpl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -23,7 +23,9 @@
package org.jboss.messaging.core.server.impl;
import java.util.HashSet;
+import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -51,6 +53,7 @@
import org.jboss.messaging.core.remoting.RemotingService;
import org.jboss.messaging.core.remoting.impl.ConnectionRegistryImpl;
import org.jboss.messaging.core.remoting.impl.wireformat.CreateSessionResponseMessage;
+import org.jboss.messaging.core.remoting.impl.wireformat.ReattachSessionResponseMessage;
import org.jboss.messaging.core.remoting.spi.ConnectorFactory;
import org.jboss.messaging.core.security.JBMSecurityManager;
import org.jboss.messaging.core.security.Role;
@@ -58,6 +61,7 @@
import org.jboss.messaging.core.security.impl.SecurityStoreImpl;
import org.jboss.messaging.core.server.MessagingServer;
import org.jboss.messaging.core.server.QueueFactory;
+import org.jboss.messaging.core.server.ServerSession;
import org.jboss.messaging.core.settings.HierarchicalRepository;
import org.jboss.messaging.core.settings.impl.HierarchicalObjectRepository;
import org.jboss.messaging.core.settings.impl.QueueSettings;
@@ -107,6 +111,7 @@
private MessagingServerControlMBean serverManagement;
private RemotingConnection replicatingConnection;
private final AtomicInteger sessionIDSequence = new AtomicInteger(2);
+ private final Map<Long, ServerSession> sessions = new ConcurrentHashMap<Long, ServerSession>();
// plugins
@@ -370,6 +375,29 @@
{
return started;
}
+
+ public ReattachSessionResponseMessage reattachSession(final RemotingConnection connection,
+ final long sessionID,
+ final int lastReceivedCommandID)
+ {
+ ServerSession session = sessions.get(sessionID);
+
+ if (session == null)
+ {
+ throw new IllegalArgumentException("Cannot find session with id " + sessionID + " to reattach");
+ }
+
+ postOffice.setBackup(false);
+
+ //Reconnect the channel to the new connection
+ session.transferConnection(connection);
+
+ int serverLastReceivedCommandID = session.replayCommands(lastReceivedCommandID);
+
+ connection.setBackup(false);
+
+ return new ReattachSessionResponseMessage(serverLastReceivedCommandID);
+ }
public CreateSessionResponseMessage createSession(final String username, final String password,
final int incrementingVersion,
@@ -407,7 +435,10 @@
resourceManager,
securityStore,
executorFactory.getExecutor(),
- channel);
+ channel,
+ this);
+
+ sessions.put(sessionID, session);
ChannelHandler handler = new ServerSessionPacketHandler(session, channel);
@@ -416,8 +447,17 @@
remotingConnection.addFailureListener(session);
return
- new CreateSessionResponseMessage(sessionID, version.getIncrementingVersion(), configuration.getPacketConfirmationBatchSize());
+ new CreateSessionResponseMessage(sessionID, version.getIncrementingVersion(),
+ configuration.getPacketConfirmationBatchSize());
}
+
+ public void removeSession(final long sessionID)
+ {
+ if (sessions.remove(sessionID) == null)
+ {
+ throw new IllegalArgumentException("Cannot find session with id " + sessionID + " to remove");
+ }
+ }
public MessagingServerControlMBean getServerManagement()
{
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerPacketHandler.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerPacketHandler.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/MessagingServerPacketHandler.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -23,6 +23,7 @@
package org.jboss.messaging.core.server.impl;
import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.CREATESESSION;
+import static org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl.REATTACH_SESSION;
import org.jboss.messaging.core.exception.MessagingException;
import org.jboss.messaging.core.logging.Logger;
@@ -32,7 +33,7 @@
import org.jboss.messaging.core.remoting.RemotingConnection;
import org.jboss.messaging.core.remoting.impl.wireformat.CreateSessionMessage;
import org.jboss.messaging.core.remoting.impl.wireformat.MessagingExceptionMessage;
-import org.jboss.messaging.core.remoting.impl.wireformat.PacketImpl;
+import org.jboss.messaging.core.remoting.impl.wireformat.ReattachSessionMessage;
import org.jboss.messaging.core.server.MessagingServer;
/**
@@ -82,6 +83,13 @@
request.isAutoCommitAcks(),
request.isXA());
}
+ else if (type == REATTACH_SESSION)
+ {
+ ReattachSessionMessage request = (ReattachSessionMessage)packet;
+
+ response =
+ server.reattachSession(connection, request.getSessionID(), request.getLastReceivedCommandID());
+ }
else
{
response = new MessagingExceptionMessage(new MessagingException(MessagingException.UNSUPPORTED_PACKET,
@@ -105,7 +113,7 @@
response = new MessagingExceptionMessage(me);
}
-
+
channel1.send(response);
}
}
\ No newline at end of file
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionImpl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -61,6 +61,7 @@
import org.jboss.messaging.core.security.SecurityStore;
import org.jboss.messaging.core.server.Delivery;
import org.jboss.messaging.core.server.MessageReference;
+import org.jboss.messaging.core.server.MessagingServer;
import org.jboss.messaging.core.server.Queue;
import org.jboss.messaging.core.server.ServerConsumer;
import org.jboss.messaging.core.server.ServerMessage;
@@ -106,7 +107,7 @@
private final boolean autoCommitAcks;
- private final RemotingConnection remotingConnection;
+ private volatile RemotingConnection remotingConnection;
private final Map<Integer, ServerConsumer> consumers = new ConcurrentHashMap<Integer, ServerConsumer>();
@@ -140,10 +141,14 @@
private final java.util.Queue<Delivery> deferredDeliveries = new ConcurrentLinkedQueue<Delivery>();
+ private final MessagingServer server;
+
private volatile boolean started = false;
private volatile int objectIDSequence;
-
+
+ private final List<Runnable> failureRunners = new ArrayList<Runnable>();
+
// Constructors
// ---------------------------------------------------------------------------------
@@ -161,7 +166,8 @@
final ResourceManager resourceManager,
final SecurityStore securityStore,
final Executor executor,
- final Channel channel) throws Exception
+ final Channel channel,
+ final MessagingServer server) throws Exception
{
this.id = id;
@@ -197,6 +203,8 @@
this.channel = channel;
this.replicatingChannel = channel.getReplicatingChannel();
+
+ this.server = server;
}
// ServerSession implementation
@@ -328,6 +336,8 @@
deliveries.clear();
channel.close();
+
+ server.removeSession(id);
}
public void promptDelivery(final Queue queue)
@@ -915,23 +925,25 @@
// session is closed.
// It is up to the user to delete the address when finished with it
- remotingConnection.addFailureListener(new FailureListener()
- {
- public void connectionFailed(final MessagingException me)
- {
- try
- {
- postOffice.removeDestination(address, durable);
- }
- catch (Exception e)
- {
- log.error("Failed to remove temporary address " + address);
- }
- }
- });
+ failureRunners.add(
+ new Runnable()
+ {
+ public void run()
+ {
+ try
+ {
+ postOffice.removeDestination(address, durable);
+ }
+ catch (Exception e)
+ {
+ log.error("Failed to remove temporary address " + address);
+ }
+ }
+ }
+ );
}
}
-
+
public void removeDestination(final SimpleString address,
final boolean durable) throws Exception
{
@@ -977,22 +989,24 @@
// It is up to the user to delete the queue when finished with it
final Queue queue = binding.getQueue();
-
- remotingConnection.addFailureListener(new FailureListener()
- {
- public void connectionFailed(final MessagingException me)
- {
- try
- {
- postOffice.removeBinding(queue.getName());
- }
- catch (Exception e)
- {
- log.error("Failed to remove temporary queue "
- + queue.getName());
- }
- }
- });
+
+ failureRunners.add(
+ new Runnable()
+ {
+ public void run()
+ {
+ try
+ {
+ postOffice.removeBinding(queue.getName());
+ }
+ catch (Exception e)
+ {
+ log.error("Failed to remove temporary queue "
+ + queue.getName());
+ }
+ }
+ }
+ );
}
}
@@ -1247,6 +1261,22 @@
delivery.deliver();
}
+ public void transferConnection(final RemotingConnection newConnection)
+ {
+ channel.transferConnection(newConnection);
+
+ remotingConnection.removeFailureListener(this);
+
+ remotingConnection = newConnection;
+
+ remotingConnection.addFailureListener(this);
+ }
+
+ public int replayCommands(final int lastReceivedCommandID)
+ {
+ return channel.replayCommands(lastReceivedCommandID);
+ }
+
// FailureListener implementation
// --------------------------------------------------------------------
@@ -1254,6 +1284,18 @@
{
try
{
+ for (Runnable runner: failureRunners)
+ {
+ try
+ {
+ runner.run();
+ }
+ catch (Throwable t)
+ {
+ log.error("Failed to execute failure runner", t);
+ }
+ }
+
close();
}
catch (Throwable t)
Modified: trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/server/impl/ServerSessionPacketHandler.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -105,7 +105,7 @@
{
return session.getID();
}
-
+
public void handlePacket(final Packet packet)
{
Packet response = null;
Modified: trunk/src/main/org/jboss/messaging/core/transaction/impl/TransactionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/transaction/impl/TransactionImpl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/core/transaction/impl/TransactionImpl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -48,7 +48,6 @@
{
private static final Logger log = Logger.getLogger(TransactionImpl.class);
-
private final StorageManager storageManager;
private final PostOffice postOffice;
Modified: trunk/src/main/org/jboss/messaging/jms/client/JBossConnectionFactory.java
===================================================================
--- trunk/src/main/org/jboss/messaging/jms/client/JBossConnectionFactory.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/jms/client/JBossConnectionFactory.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -75,6 +75,8 @@
private final TransportConfiguration connectorConfig;
+ private final TransportConfiguration backupConnectorConfig;
+
private final String clientID;
private final int dupsOKBatchSize;
@@ -100,6 +102,7 @@
// Constructors ---------------------------------------------------------------------------------
public JBossConnectionFactory(final TransportConfiguration connectorConfig,
+ final TransportConfiguration backupConnectorConfig,
final long pingPeriod,
final long callTimeout,
final String clientID,
@@ -113,6 +116,7 @@
final boolean blockOnPersistentSend)
{
this.connectorConfig = connectorConfig;
+ this.backupConnectorConfig = backupConnectorConfig;
this.clientID = clientID;
this.dupsOKBatchSize = dupsOKBatchSize;
this.pingPeriod = pingPeriod;
@@ -293,6 +297,7 @@
//It doesn't matter if more than one is created due to a race
sessionFactory = new ClientSessionFactoryImpl(
connectorConfig,
+ backupConnectorConfig,
pingPeriod,
callTimeout,
consumerWindowSize,
Modified: trunk/src/main/org/jboss/messaging/jms/server/JMSServerManager.java
===================================================================
--- trunk/src/main/org/jboss/messaging/jms/server/JMSServerManager.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/jms/server/JMSServerManager.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -96,6 +96,7 @@
boolean destroyTopic(String name) throws Exception;
boolean createConnectionFactory(String name, TransportConfiguration connectorConfig,
+ TransportConfiguration backupConnectorConfig,
long pingPeriod, long callTimeout, String clientID,
int dupsOKBatchSize, int consumerWindowSize, int consumerMaxRate,
int producerWindowSize, int producerMaxRate,
@@ -106,6 +107,7 @@
boolean createConnectionFactory(String name, TransportConfiguration connectorConfig,
+ TransportConfiguration backupConnectorConfig,
long pingPeriod, long callTimeout, String clientID,
int dupsOKBatchSize, int consumerWindowSize, int consumerMaxRate,
int producerWindowSize, int producerMaxRate,
Modified: trunk/src/main/org/jboss/messaging/jms/server/impl/JMSServerDeployer.java
===================================================================
--- trunk/src/main/org/jboss/messaging/jms/server/impl/JMSServerDeployer.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/jms/server/impl/JMSServerDeployer.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -61,6 +61,7 @@
private static final String SEND_NP_MESSAGES_SYNCHRONOUSLY_ELEMENT = "send-np-messages-synchronously";
private static final String SEND_P_MESSAGES_SYNCHRONOUSLY_ELEMENT = "send-p-messages-synchronously";
private static final String CONNECTOR_ELEMENT = "connector";
+ private static final String BACKUP_CONNECTOR_ELEMENT = "backup-connector";
private static final String FACTORY_CLASS_ELEMENT = "factory-class";
private static final String PARAMS_ELEMENT = "params";
private static final String PARAM_ELEMENT = "param";
@@ -128,6 +129,8 @@
List<String> jndiBindings = new ArrayList<String>();
String connectorFactoryClassName = null;
Map<String, Object> params = new HashMap<String, Object>();
+ String backupConnectorFactoryClassName = null;
+ Map<String, Object> backupParams = new HashMap<String, Object>();
for (int j = 0; j < children.getLength(); j++)
{
@@ -261,6 +264,87 @@
}
}
}
+ else if (BACKUP_CONNECTOR_ELEMENT.equalsIgnoreCase(children.item(j).getNodeName()))
+ {
+ NodeList children2 = children.item(j).getChildNodes();
+
+ for (int l = 0; l < children2.getLength(); l++)
+ {
+ String nodeName = children2.item(l).getNodeName();
+
+ if (FACTORY_CLASS_ELEMENT.equalsIgnoreCase(nodeName))
+ {
+ backupConnectorFactoryClassName = children2.item(l).getTextContent();
+ }
+ else if (PARAMS_ELEMENT.equalsIgnoreCase(nodeName))
+ {
+ NodeList nlParams = children2.item(l).getChildNodes();
+
+ for (int m = 0; m < nlParams.getLength(); m++)
+ {
+ if (PARAM_ELEMENT.equalsIgnoreCase(nlParams.item(m).getNodeName()))
+ {
+ Node paramNode = nlParams.item(m);
+
+ NamedNodeMap attributes = paramNode.getAttributes();
+
+ Node nkey = attributes.getNamedItem("key");
+
+ String key = nkey.getTextContent();
+
+ Node nValue = attributes.getNamedItem("value");
+
+ String value = nValue.getTextContent();
+
+ Node nType = attributes.getNamedItem("type");
+
+ String type = nType.getTextContent();
+
+ if (type.equalsIgnoreCase("Integer"))
+ {
+ try
+ {
+ Integer iVal = Integer.parseInt(value);
+
+ backupParams.put(key, iVal);
+ }
+ catch (NumberFormatException e2)
+ {
+ throw new IllegalArgumentException("Remoting acceptor parameter " + value + " is not a valid Integer");
+ }
+ }
+ else if (type.equalsIgnoreCase("Long"))
+ {
+ try
+ {
+ Long lVal = Long.parseLong(value);
+
+ backupParams.put(key, lVal);
+ }
+ catch (NumberFormatException e2)
+ {
+ throw new IllegalArgumentException("Remoting acceptor parameter " + value + " is not a valid Long");
+ }
+ }
+ else if (type.equalsIgnoreCase("String"))
+ {
+ backupParams.put(key, value);
+ }
+ else if (type.equalsIgnoreCase("Boolean"))
+ {
+ Boolean lVal = Boolean.parseBoolean(value);
+
+ backupParams.put(key, lVal);
+ }
+ else
+ {
+ throw new IllegalArgumentException("Invalid parameter type " + type);
+ }
+ }
+ }
+ }
+ }
+ }
}
if (connectorFactoryClassName == null)
@@ -270,10 +354,17 @@
TransportConfiguration connectorConfig =
new TransportConfiguration(connectorFactoryClassName, params);
+
+ TransportConfiguration backupConnectorConfig = null;
+
+ if (backupConnectorFactoryClassName != null)
+ {
+ backupConnectorConfig = new TransportConfiguration(backupConnectorFactoryClassName, backupParams);
+ }
String name = node.getAttributes().getNamedItem(getKeyAttribute()).getNodeValue();
- jmsServerManager.createConnectionFactory(name, connectorConfig,
+ jmsServerManager.createConnectionFactory(name, connectorConfig, backupConnectorConfig,
pingPeriod, callTimeout, clientID, dupsOKBatchSize,
consumerWindowSize, consumerMaxRate, producerWindowSize, producerMaxRate,
blockOnAcknowledge, blockOnNonPersistentSend,
Modified: trunk/src/main/org/jboss/messaging/jms/server/impl/JMSServerManagerImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/jms/server/impl/JMSServerManagerImpl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/jms/server/impl/JMSServerManagerImpl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -190,6 +190,7 @@
public boolean createConnectionFactory(String name,
TransportConfiguration connectorConfig,
+ TransportConfiguration backupConnectorConfig,
long pingPeriod, long callTimeout, String clientID,
int dupsOKBatchSize, int consumerWindowSize, int consumerMaxRate,
int producerWindowSize, int producerMaxRate,
@@ -202,6 +203,7 @@
if (cf == null)
{
cf = new JBossConnectionFactory(connectorConfig,
+ backupConnectorConfig,
pingPeriod, callTimeout,
clientID, dupsOKBatchSize,
consumerWindowSize, consumerMaxRate, producerWindowSize,
@@ -229,6 +231,7 @@
public boolean createConnectionFactory(String name,
TransportConfiguration connectorConfig,
+ TransportConfiguration backupConnectorConfig,
long pingPeriod, long callTimeout, String clientID,
int dupsOKBatchSize, int consumerWindowSize, int consumerMaxRate,
int producerWindowSize, int producerMaxRate,
@@ -241,6 +244,7 @@
if (cf == null)
{
cf = new JBossConnectionFactory(connectorConfig,
+ backupConnectorConfig,
pingPeriod, callTimeout,
clientID, dupsOKBatchSize,
consumerWindowSize, consumerMaxRate, producerWindowSize,
Modified: trunk/src/main/org/jboss/messaging/jms/server/management/JMSServerControlMBean.java
===================================================================
--- trunk/src/main/org/jboss/messaging/jms/server/management/JMSServerControlMBean.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/jms/server/management/JMSServerControlMBean.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -69,7 +69,8 @@
@Operation(desc = "Create a JMS ConnectionFactory", impact = ACTION)
void createConnectionFactory(
@Parameter(name = "name", desc = "Name of the ConnectionFactory to create") String name,
- @Parameter(name = "transportConfiguration", desc = "The remoting connector configuration") TransportConfiguration connectorConfig,
+ @Parameter(name = "transportConfiguration", desc = "The remoting connector configuration") TransportConfiguration connectorConfig,
+ @Parameter(name = "backupTransportConfiguration", desc = "The backup remoting connector configuration") TransportConfiguration backupConnectorConfig,
@Parameter(name = "pingPeriod", desc = "The ping period in m") long pingPeriod,
@Parameter(name = "callTimeout", desc = "The call timeout in m") long callTimeout,
@Parameter(name = "clientID", desc = "ClientID for created connections") String clientID,
Modified: trunk/src/main/org/jboss/messaging/jms/server/management/impl/JMSServerControl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/jms/server/management/impl/JMSServerControl.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/src/main/org/jboss/messaging/jms/server/management/impl/JMSServerControl.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -77,6 +77,7 @@
// JMSServerControlMBean implementation --------------------------
public void createConnectionFactory(String name, TransportConfiguration connectorConfig,
+ TransportConfiguration backupConnectorConfig,
long pingPeriod, long callTimeout, String clientID,
int dupsOKBatchSize, int consumerWindowSize, int consumerMaxRate,
int producerWindowSize, int producerMaxRate,
@@ -88,6 +89,7 @@
bindings.add(jndiBinding);
boolean created = server.createConnectionFactory(name, connectorConfig,
+ backupConnectorConfig,
pingPeriod, callTimeout, clientID, dupsOKBatchSize,
consumerWindowSize, consumerMaxRate, producerWindowSize, producerMaxRate,
blockOnAcknowledge, blockOnNonPersistentSend,
Modified: trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/CTSMiscellaneousTest.java
===================================================================
--- trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/CTSMiscellaneousTest.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/CTSMiscellaneousTest.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -86,7 +86,7 @@
bindings.add("StrictTCKConnectionFactory");
getJmsServerManager().createConnectionFactory("StrictTCKConnectionFactory",
- new TransportConfiguration("org.jboss.messaging.core.remoting.impl.netty.NettyConnectorFactory"), 5000, 5000,
+ new TransportConfiguration("org.jboss.messaging.core.remoting.impl.netty.NettyConnectorFactory"), null, 5000, 5000,
null,
1000, 1024 * 1024, -1, 1000, -1, true, true, true, "/StrictTCKConnectionFactory");
Modified: trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/JMSTestCase.java
===================================================================
--- trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/JMSTestCase.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/tests/jms-tests/src/org/jboss/test/messaging/jms/JMSTestCase.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -39,7 +39,7 @@
//both np and p messages are sent synchronously
getJmsServerManager().createConnectionFactory("testsuitecf",
- new TransportConfiguration("org.jboss.messaging.core.remoting.impl.netty.NettyConnectorFactory"), 5000, 5000,
+ new TransportConfiguration("org.jboss.messaging.core.remoting.impl.netty.NettyConnectorFactory"), null, 5000, 5000,
null, 1000, 1024 * 1024, -1, 1000, -1, true, true, true, "/testsuitecf");
cf = (JBossConnectionFactory) getInitialContext().lookup("/testsuitecf");
Modified: trunk/tests/jms-tests/src/org/jboss/test/messaging/tools/container/LocalTestServer.java
===================================================================
--- trunk/tests/jms-tests/src/org/jboss/test/messaging/tools/container/LocalTestServer.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/tests/jms-tests/src/org/jboss/test/messaging/tools/container/LocalTestServer.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -529,7 +529,7 @@
{
log.info("deploying connection factory with name: " + objectName + " and dupsok: " + dupsOkBatchSize);
getJMSServerManager().createConnectionFactory(objectName,
- new TransportConfiguration("org.jboss.messaging.core.remoting.impl.netty.NettyConnectorFactory"), 5000, 5000,
+ new TransportConfiguration("org.jboss.messaging.core.remoting.impl.netty.NettyConnectorFactory"), null, 5000, 5000,
clientId, dupsOkBatchSize,
prefetchSize, -1, 1000, -1, blockOnAcknowledge, true, true, jndiBindings);
}
Added: trunk/tests/src/org/jboss/messaging/tests/integration/cluster/ReplicationTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/cluster/ReplicationTest.java (rev 0)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/cluster/ReplicationTest.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -0,0 +1,229 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jboss.messaging.tests.integration.cluster;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import junit.framework.TestCase;
+
+import org.jboss.messaging.core.client.ClientConsumer;
+import org.jboss.messaging.core.client.ClientMessage;
+import org.jboss.messaging.core.client.ClientProducer;
+import org.jboss.messaging.core.client.ClientSession;
+import org.jboss.messaging.core.client.ClientSessionFactory;
+import org.jboss.messaging.core.client.impl.ClientSessionFactoryImpl;
+import org.jboss.messaging.core.client.impl.ClientSessionImpl;
+import org.jboss.messaging.core.config.Configuration;
+import org.jboss.messaging.core.config.TransportConfiguration;
+import org.jboss.messaging.core.config.impl.ConfigurationImpl;
+import org.jboss.messaging.core.exception.MessagingException;
+import org.jboss.messaging.core.logging.Logger;
+import org.jboss.messaging.core.remoting.RemotingConnection;
+import org.jboss.messaging.core.remoting.impl.invm.TransportConstants;
+import org.jboss.messaging.core.server.MessagingService;
+import org.jboss.messaging.core.server.impl.MessagingServiceImpl;
+import org.jboss.messaging.jms.client.JBossTextMessage;
+import org.jboss.messaging.util.SimpleString;
+
+public class ReplicationTest extends TestCase
+{
+ private static final Logger log = Logger.getLogger(ReplicationTest.class);
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ // Public --------------------------------------------------------
+
+ public void testReplication() throws Exception
+ {
+ final SimpleString QUEUE = new SimpleString("CoreClientTestQueue");
+
+ Configuration backupConf = new ConfigurationImpl();
+ backupConf.setSecurityEnabled(false);
+ backupConf.setPacketConfirmationBatchSize(1);
+ Map<String, Object> backupParams = new HashMap<String, Object>();
+ backupParams.put(TransportConstants.SERVER_ID_PROP_NAME, 1);
+ backupConf.getAcceptorConfigurations().add(new TransportConfiguration("org.jboss.messaging.core.remoting.impl.invm.InVMAcceptorFactory", backupParams));
+ backupConf.setBackup(true);
+ MessagingService backupService = MessagingServiceImpl.newNullStorageMessagingServer(backupConf);
+ backupService.start();
+
+ Configuration liveConf = new ConfigurationImpl();
+ liveConf.setSecurityEnabled(false);
+ liveConf.setPacketConfirmationBatchSize(1);
+ liveConf.getAcceptorConfigurations().add(new TransportConfiguration("org.jboss.messaging.core.remoting.impl.invm.InVMAcceptorFactory"));
+ liveConf.setBackupConnectorConfiguration(new TransportConfiguration("org.jboss.messaging.core.remoting.impl.invm.InVMConnectorFactory", backupParams));
+ MessagingService liveService = MessagingServiceImpl.newNullStorageMessagingServer(liveConf);
+ liveService.start();
+
+ ClientSessionFactory sf = new ClientSessionFactoryImpl(new TransportConfiguration("org.jboss.messaging.core.remoting.impl.invm.InVMConnectorFactory"));
+
+ ClientSession session = sf.createSession(false, true, true, -1, false);
+
+ session.createQueue(QUEUE, QUEUE, null, false, false);
+
+
+ ClientProducer producer = session.createProducer(QUEUE);
+
+ final int numMessages = 1000;
+
+ for (int i = 0; i < numMessages; i++)
+ {
+ ClientMessage message = session.createClientMessage(JBossTextMessage.TYPE, false, 0,
+ System.currentTimeMillis(), (byte) 1);
+ message.getBody().putString("testINVMCoreClient");
+ message.getBody().flip();
+ producer.send(message);
+ }
+
+ ClientConsumer consumer = session.createConsumer(QUEUE);
+
+ session.start();
+
+ for (int i = 0; i < numMessages; i++)
+ {
+ ClientMessage message2 = consumer.receive();
+
+ assertEquals("testINVMCoreClient", message2.getBody().getString());
+
+ session.acknowledge();
+ }
+
+ session.close();
+
+ liveService.stop();
+ backupService.stop();
+ }
+
+ public void testFailover() throws Exception
+ {
+ final SimpleString QUEUE = new SimpleString("CoreClientTestQueue");
+
+ Configuration backupConf = new ConfigurationImpl();
+ backupConf.setSecurityEnabled(false);
+ backupConf.setPacketConfirmationBatchSize(1);
+ Map<String, Object> backupParams = new HashMap<String, Object>();
+ backupParams.put(TransportConstants.SERVER_ID_PROP_NAME, 1);
+ backupConf.getAcceptorConfigurations().add(new TransportConfiguration("org.jboss.messaging.core.remoting.impl.invm.InVMAcceptorFactory", backupParams));
+ backupConf.setBackup(true);
+ MessagingService backupService = MessagingServiceImpl.newNullStorageMessagingServer(backupConf);
+ backupService.start();
+
+ Configuration liveConf = new ConfigurationImpl();
+ liveConf.setSecurityEnabled(false);
+ liveConf.setPacketConfirmationBatchSize(1);
+ liveConf.getAcceptorConfigurations().add(new TransportConfiguration("org.jboss.messaging.core.remoting.impl.invm.InVMAcceptorFactory"));
+ liveConf.setBackupConnectorConfiguration(new TransportConfiguration("org.jboss.messaging.core.remoting.impl.invm.InVMConnectorFactory", backupParams));
+ MessagingService liveService = MessagingServiceImpl.newNullStorageMessagingServer(liveConf);
+ liveService.start();
+
+ ClientSessionFactory sf =
+ new ClientSessionFactoryImpl(new TransportConfiguration("org.jboss.messaging.core.remoting.impl.invm.InVMConnectorFactory"),
+ new TransportConfiguration("org.jboss.messaging.core.remoting.impl.invm.InVMConnectorFactory", backupParams));
+
+ ClientSession session = sf.createSession(false, true, true, -1, false);
+
+ session.createQueue(QUEUE, QUEUE, null, false, false);
+
+ ClientProducer producer = session.createProducer(QUEUE);
+
+ final int numMessages = 10;
+
+ for (int i = 0; i < numMessages; i++)
+ {
+ ClientMessage message = session.createClientMessage(JBossTextMessage.TYPE, false, 0,
+ System.currentTimeMillis(), (byte) 1);
+ message.putIntProperty(new SimpleString("blah"), i);
+ message.getBody().putString("testINVMCoreClient");
+ message.getBody().flip();
+ producer.send(message);
+ }
+
+ RemotingConnection conn = ((ClientSessionImpl)session).getConnection();
+
+ //Simulate failure on connection
+ conn.fail(new MessagingException(MessagingException.NOT_CONNECTED));
+
+ ClientConsumer consumer = session.createConsumer(QUEUE);
+
+ session.start();
+
+ for (int i = 0; i < numMessages / 2; i++)
+ {
+ ClientMessage message2 = consumer.receive();
+
+ assertEquals("testINVMCoreClient", message2.getBody().getString());
+
+ session.acknowledge();
+
+ log.info("got message " + message2.getProperty(new SimpleString("blah")));
+ }
+
+ session.close();
+
+ sf =
+ new ClientSessionFactoryImpl(new TransportConfiguration("org.jboss.messaging.core.remoting.impl.invm.InVMConnectorFactory", backupParams));
+
+ session = sf.createSession(false, true, true, -1, false);
+
+ consumer = session.createConsumer(QUEUE);
+
+ session.start();
+
+ for (int i = 0; i < numMessages / 2; i++)
+ {
+ ClientMessage message2 = consumer.receive();
+
+ assertEquals("testINVMCoreClient", message2.getBody().getString());
+
+ session.acknowledge();
+
+ log.info("got message " + message2.getProperty(new SimpleString("blah")));
+ }
+
+ ClientMessage message3 = consumer.receive(1000);
+
+ assertNull(message3);
+
+ liveService.stop();
+ backupService.stop();
+
+ // todo - do we need to failover connection factories too?????
+
+
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+}
Modified: trunk/tests/src/org/jboss/messaging/tests/unit/core/journal/impl/AlignedJournalImplTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/unit/core/journal/impl/AlignedJournalImplTest.java 2008-09-10 09:29:43 UTC (rev 4925)
+++ trunk/tests/src/org/jboss/messaging/tests/unit/core/journal/impl/AlignedJournalImplTest.java 2008-09-10 10:51:55 UTC (rev 4926)
@@ -899,11 +899,11 @@
assertEquals(1, transactions.size());
assertEquals(1, records.size());
- assertEquals(10, transactions.get(0).xidData.length);
+ assertEquals(10, transactions.get(0).extraData.length);
for (int i = 0; i < 10; i++)
{
- assertEquals((byte)1, transactions.get(0).xidData[i]);
+ assertEquals((byte)1, transactions.get(0).extraData[i]);
}
@@ -938,10 +938,10 @@
assertEquals(0, records.size());
assertEquals(1, transactions.size());
- assertEquals(10, transactions.get(0).xidData.length);
+ assertEquals(10, transactions.get(0).extraData.length);
for (int i = 0; i < 10; i++)
{
- assertEquals((byte)1, transactions.get(0).xidData[i]);
+ assertEquals((byte)1, transactions.get(0).extraData[i]);
}
journalImpl.checkAndReclaimFiles();
@@ -971,11 +971,11 @@
assertEquals(1, transactions.size());
- assertEquals(15, transactions.get(0).xidData.length);
+ assertEquals(15, transactions.get(0).extraData.length);
- for (int i = 0; i < transactions.get(0).xidData.length; i++)
+ for (int i = 0; i < transactions.get(0).extraData.length; i++)
{
- assertEquals(2, transactions.get(0).xidData[i]);
+ assertEquals(2, transactions.get(0).extraData[i]);
}
assertEquals(10, journalImpl.getDataFilesCount());
More information about the jboss-cvs-commits
mailing list