6 p.m.
Author: borges
Date: 2011-07-01 12:00:03 -0400 (Fri, 01 Jul 2011)
New Revision: 10911
Modified:
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/HornetQPacketHandler.java
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/server/HornetQServer.java
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/spi/core/protocol/RemotingConnection.java
Log:
HORNETQ-720 pass CoreRemotingConnection instead of channels, as it
allows us to use the failureListener.
Modified:
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java
===================================================================
---
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java 2011-07-01
13:31:52 UTC (rev 10910)
+++
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/CoreProtocolManager.java 2011-07-01
16:00:03 UTC (rev 10911)
@@ -158,14 +158,10 @@
{
HaBackupRegistrationMessage msg = (HaBackupRegistrationMessage)packet;
System.out.println("HA_BACKUP_REGISTRATION: " + msg + "
connector=" + msg.getConnector());
- long channelID = msg.getChannelID();
- Channel channelX = rc.getChannel(CHANNEL_ID.SESSION.id, -1);
- Channel replicationChannel = rc.getChannel(CHANNEL_ID.REPLICATION.id,
-1);
- System.out.println("msg channelID: " + channelID);
System.out.println("HA_BR: " + server.getIdentity() + ",
toString=" + server);
try
{
- server.addHaBackup(channelX, replicationChannel);
+ server.addHaBackup(rc);
}
catch (Exception e)
{
Modified:
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/HornetQPacketHandler.java
===================================================================
---
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/HornetQPacketHandler.java 2011-07-01
13:31:52 UTC (rev 10910)
+++
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/protocol/core/impl/HornetQPacketHandler.java 2011-07-01
16:00:03 UTC (rev 10911)
@@ -25,12 +25,10 @@
import org.hornetq.core.protocol.core.CoreRemotingConnection;
import org.hornetq.core.protocol.core.Packet;
import org.hornetq.core.protocol.core.ServerSessionPacketHandler;
-import org.hornetq.core.protocol.core.impl.ChannelImpl.CHANNEL_ID;
import org.hornetq.core.protocol.core.impl.wireformat.CreateQueueMessage;
import org.hornetq.core.protocol.core.impl.wireformat.CreateReplicationSessionMessage;
import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionMessage;
import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionResponseMessage;
-import org.hornetq.core.protocol.core.impl.wireformat.HaBackupRegistrationMessage;
import org.hornetq.core.protocol.core.impl.wireformat.HornetQExceptionMessage;
import org.hornetq.core.protocol.core.impl.wireformat.NullResponseMessage;
import org.hornetq.core.protocol.core.impl.wireformat.ReattachSessionMessage;
@@ -115,25 +113,6 @@
break;
}
- case PacketImpl.HA_BACKUP_REGISTRATION:
- {
- HaBackupRegistrationMessage msg = (HaBackupRegistrationMessage)packet;
- System.out.println("HA_BACKUP_REGISTRATION: " + msg + "
connector=" + msg.getConnector());
- long channelID = msg.getChannelID();
- Channel channelX = connection.getChannel(CHANNEL_ID.SESSION.id, -1);
- Channel replicationChannel = connection.getChannel(CHANNEL_ID.REPLICATION.id,
-1);
- System.out.println("HA_BR: " + server.getIdentity() + ",
toString=" + server);
- try
- {
- server.addHaBackup(channelX, replicationChannel);
- }
- catch (Exception e)
- {
- // XXX This is not what we want
- e.printStackTrace();
- throw new RuntimeException(e);
- }
- }
default:
{
HornetQPacketHandler.log.error("Invalid packet " + packet);
Modified:
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java
===================================================================
---
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2011-07-01
13:31:52 UTC (rev 10910)
+++
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2011-07-01
16:00:03 UTC (rev 10911)
@@ -83,7 +83,7 @@
private SessionFailureListener failureListener;
- private final Channel systemChannel;
+ private CoreRemotingConnection remotingConnection;
// Static --------------------------------------------------------
@@ -95,21 +95,16 @@
this.executorFactory = executorFactory;
CoreRemotingConnection conn = sessionFactory.getConnection();
- systemChannel = conn.getChannel(CHANNEL_ID.SESSION.id, -1);
replicatingChannel = conn.getChannel(CHANNEL_ID.REPLICATION.id, -1);
}
- // Public --------------------------------------------------------
-
/**
- * @param systemChannel
- * @param replicatingChannel
+ * @param remotingConnection
*/
- public ReplicationManagerImpl(Channel systemChannel, Channel replicatingChannel)
+ public ReplicationManagerImpl(CoreRemotingConnection remotingConnection)
{
- super();
- this.systemChannel = systemChannel;
- this.replicatingChannel = replicatingChannel;
+ replicatingChannel = remotingConnection.getChannel(CHANNEL_ID.REPLICATION.id, -1);
+ this.remotingConnection = remotingConnection;
}
public void appendAddRecord(final byte journalID, final long id, final byte
recordType, final EncodingSupport record)
@@ -314,20 +309,13 @@
throw new IllegalStateException("ReplicationManager is already
started");
}
-// replicatingConnection = sessionFactory.getConnection();
-//
-// if (replicatingConnection == null)
-// {
-// ReplicationManagerImpl.log.warn("Backup server MUST be started before
live server. Initialisation will not proceed.");
-// throw new HornetQException(HornetQException.ILLEGAL_STATE,
-// "Backup server MUST be started before live
server. Initialisation will not proceed.");
-// }
-
replicatingChannel.setHandler(responseHandler);
CreateReplicationSessionMessage replicationStartPackage =
new CreateReplicationSessionMessage(replicatingChannel.getID());
+ Channel systemChannel = remotingConnection.getChannel(CHANNEL_ID.SESSION.id, -1);
+
systemChannel.sendBlocking(replicationStartPackage);
failureListener = new SessionFailureListener()
@@ -358,10 +346,9 @@
{
}
};
- // sessionFactory.addFailureListener(failureListener);
+ remotingConnection.addFailureListener(failureListener);
started = true;
-
enabled = true;
}
@@ -397,15 +384,14 @@
replicatingChannel.close();
}
+ remotingConnection.removeFailureListener(failureListener);
// sessionFactory.causeExit();
-// sessionFactory.removeFailureListener(failureListener);
// if (replicatingConnection != null)
// {
// replicatingConnection.destroy();
// }
//
-// replicatingConnection = null;
-
+ remotingConnection = null;
started = false;
}
Modified:
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/server/HornetQServer.java
===================================================================
---
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/server/HornetQServer.java 2011-07-01
13:31:52 UTC (rev 10910)
+++
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/server/HornetQServer.java 2011-07-01
16:00:03 UTC (rev 10911)
@@ -28,6 +28,7 @@
import org.hornetq.core.persistence.StorageManager;
import org.hornetq.core.postoffice.PostOffice;
import org.hornetq.core.protocol.core.Channel;
+import org.hornetq.core.protocol.core.CoreRemotingConnection;
import org.hornetq.core.remoting.server.RemotingService;
import org.hornetq.core.replication.ReplicationEndpoint;
import org.hornetq.core.replication.ReplicationManager;
@@ -176,5 +177,5 @@
void stop(boolean failoverOnServerShutdown) throws Exception;
- void addHaBackup(Channel channelX, Channel replicationChannel) throws Exception;
+ void addHaBackup(CoreRemotingConnection rc) throws Exception;
}
Modified:
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java
===================================================================
---
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-07-01
13:31:52 UTC (rev 10910)
+++
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/core/server/impl/HornetQServerImpl.java 2011-07-01
16:00:03 UTC (rev 10911)
@@ -81,6 +81,7 @@
import org.hornetq.core.postoffice.impl.LocalQueueBinding;
import org.hornetq.core.postoffice.impl.PostOfficeImpl;
import org.hornetq.core.protocol.core.Channel;
+import org.hornetq.core.protocol.core.CoreRemotingConnection;
import org.hornetq.core.protocol.core.impl.ChannelImpl.CHANNEL_ID;
import org.hornetq.core.protocol.core.impl.wireformat.HaBackupRegistrationMessage;
import org.hornetq.core.remoting.server.RemotingService;
@@ -547,7 +548,7 @@
}
log.info("announce backup to live-server (id=" + liveConnectorName
+ ")");
liveServerSessionFactory.getConnection()
- .getChannel(CHANNEL_ID.SESSION.id, -1)
+ .getChannel(CHANNEL_ID.PING.id, -1)
.send(new
HaBackupRegistrationMessage(getNodeID().toString(), config));
log.info("backup registered");
@@ -1945,19 +1946,15 @@
}
@Override
- public void addHaBackup(Channel systemChannel, Channel replicatingChannel) throws
Exception
+ public void addHaBackup(CoreRemotingConnection rc) throws Exception
{
if (!(storageManager instanceof JournalStorageManager))
return;
JournalStorageManager journalStorageManager =
(JournalStorageManager)storageManager;
System.out.println(HornetQServerImpl.class.getName() + " " +
this.getIdentity() +
- ": create a ReplicationManagerImpl. Using ChannelID=" +
systemChannel);
- // XXX not sure this is the right call to use
-// final ServerLocatorInternal serverLocator =
-//
(ServerLocatorInternal)HornetQClient.createServerLocatorWithHA(connector);
-// ClientSessionFactoryInternal sessionFactory =
(ClientSessionFactoryInternal)serverLocator.createSessionFactory();
- replicationManager = new ReplicationManagerImpl(systemChannel,
replicatingChannel);
+ ": create a ReplicationManagerImpl");
+ replicationManager = new ReplicationManagerImpl(rc);
System.out.println("rep.start()");
replicationManager.start();
Modified:
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/spi/core/protocol/RemotingConnection.java
===================================================================
---
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/spi/core/protocol/RemotingConnection.java 2011-07-01
13:31:52 UTC (rev 10910)
+++
branches/HORNETQ-720_Replication/hornetq-core/src/main/java/org/hornetq/spi/core/protocol/RemotingConnection.java 2011-07-01
16:00:03 UTC (rev 10911)
@@ -43,17 +43,15 @@
long getCreationTime();
/**
- * returns a string representation of the remote address of this connection
- *
+ * Returns a string representation of the remote address of this connection.
* @return the remote address
*/
String getRemoteAddress();
/**
- * add a failure listener.
+ * Adds a failure listener.
* <p/>
* The listener will be called in the event of connection failure.
- *
* @param listener the listener
*/
void addFailureListener(FailureListener listener);
@@ -82,27 +80,27 @@
* @return true if removed
*/
boolean removeCloseListener(CloseListener listener);
-
+
List<CloseListener> removeCloseListeners();
-
+
void setCloseListeners(List<CloseListener> listeners);
-
-
+
+
/**
* return all the failure listeners
*
* @return the listeners
*/
List<FailureListener> getFailureListeners();
-
+
List<FailureListener> removeFailureListeners();
/**
- * set the failure listeners.
+ * Sets the failure listeners.
* <p/>
- * These will be called in the event of the connection being closed. Any previosuly
added listeners will be removed.
- *
+ * These will be called in the event of the connection being closed. Any previously
added
+ * listeners will be removed.
* @param listeners the listeners to add.
*/
void setFailureListeners(List<FailureListener> listeners);
@@ -146,20 +144,20 @@
*
* @return true if destroyed, otherwise false
*/
- boolean isDestroyed();
-
+ boolean isDestroyed();
+
/**
* Disconnect the connection, closing all channels
*/
void disconnect();
-
+
/**
* returns true if any data has been received since the last time this method was
called.
*
* @return true if data has been received.
*/
boolean checkDataReceived();
-
+
/**
* flush all outstanding data from the connection.
*/