[hornetq-commits] JBoss hornetq SVN: r8005 - in branches/Replication_Clebert: src/main/org/hornetq/core/remoting/impl/wireformat and 7 other directories.

Author: clebert.suconic(a)jboss.com Date: 2009-09-28 18:26:08 -0400 (Mon, 28 Sep 2009) New Revision: 8005 Added: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/CreateReplicationSessionMessage.java branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationEndpoint.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java Removed: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/server/HandlerFactory.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/BackupListener.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationPacketHandler.java Modified: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java branches/Replication_Clebert/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java branches/Replication_Clebert/src/main/org/hornetq/core/server/HornetQServer.java branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java Log: Backup Modified: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/PacketDecoder.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -13,6 +13,8 @@ package org.hornetq.core.remoting.impl; + +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATE_REPLICATION; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATESESSION; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATESESSION_RESP; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATE_QUEUE; @@ -62,6 +64,7 @@ import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.impl.wireformat.CreateQueueMessage; +import org.hornetq.core.remoting.impl.wireformat.CreateReplicationSessionMessage; import org.hornetq.core.remoting.impl.wireformat.CreateSessionMessage; import org.hornetq.core.remoting.impl.wireformat.CreateSessionResponseMessage; import org.hornetq.core.remoting.impl.wireformat.HornetQExceptionMessage; @@ -349,7 +352,13 @@ { packet = new SessionSendContinuationMessage(); break; - } + } + + case CREATE_REPLICATION: + { + packet = new CreateReplicationSessionMessage(); + break; + } default: { throw new IllegalArgumentException("Invalid type: " + packetType); Added: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/CreateReplicationSessionMessage.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/CreateReplicationSessionMessage.java (rev 0) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/CreateReplicationSessionMessage.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -0,0 +1,98 @@ +/* + * 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.remoting.impl.wireformat; + +import org.hornetq.core.remoting.spi.HornetQBuffer; +import org.hornetq.utils.DataConstants; + +/** + * @author <a href="mailto:tim.fox@jboss.com">Clebert Suconic</a> + */ +public class CreateReplicationSessionMessage extends PacketImpl +{ + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + private long sessionChannelID; + + private int windowSize; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public CreateReplicationSessionMessage(final long sessionChannelID, final int windowSize) + { + super(CREATE_REPLICATION); + + this.sessionChannelID = sessionChannelID; + + this.windowSize = windowSize; + } + + public CreateReplicationSessionMessage() + { + super(CREATE_REPLICATION); + } + + // Public -------------------------------------------------------- + public int getRequiredBufferSize() + { + return BASIC_PACKET_SIZE + + // buffer.writeLong(sessionChannelID); + DataConstants.SIZE_LONG + + // buffer.writeInt(windowSize); + DataConstants.SIZE_INT; + + } + + @Override + public void encodeBody(final HornetQBuffer buffer) + { + buffer.writeLong(sessionChannelID); + buffer.writeInt(windowSize); + } + + @Override + public void decodeBody(final HornetQBuffer buffer) + { + sessionChannelID = buffer.readLong(); + windowSize = buffer.readInt(); + } + + /** + * @return the sessionChannelID + */ + public long getSessionChannelID() + { + return sessionChannelID; + } + + /** + * @return the windowSize + */ + public int getWindowSize() + { + return windowSize; + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- +} Modified: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/PacketImpl.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -64,6 +64,8 @@ public static final byte CREATE_QUEUE = 34; public static final byte DELETE_QUEUE = 35; + + public static final byte CREATE_REPLICATION = 36; // Session @@ -136,6 +138,10 @@ public static final byte SESS_RECEIVE_MSG = 75; public static final byte SESS_RECEIVE_CONTINUATION = 76; + + // Replication + + public static final byte REPLICATION_SEND_REPLICATION = 77; // Static -------------------------------------------------------- Added: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java (rev 0) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/impl/wireformat/ReplicationPacket.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -0,0 +1,53 @@ +/* + * 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.remoting.impl.wireformat; + +/** + * A ReplicationPacket + * + * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class ReplicationPacket extends PacketImpl +{ + + /** + * @param type + */ + public ReplicationPacket(byte type) + { + super(type); + // TODO Auto-generated constructor stub + } + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + // Public -------------------------------------------------------- + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Deleted: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/server/HandlerFactory.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/server/HandlerFactory.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/server/HandlerFactory.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -1,32 +0,0 @@ -/* - * 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.remoting.server; - -import org.hornetq.core.remoting.Channel; -import org.hornetq.core.remoting.ChannelHandler; -import org.hornetq.core.remoting.RemotingConnection; - -/** - * The RemotingService could be used by either the Replication Manager or by the Server. - * - * Each will need to use a different Handler, so this factory may be used to pass what handler needs to be created - * - * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public interface HandlerFactory -{ - ChannelHandler getHandler(RemotingConnection conn, Channel channel); -} Modified: branches/Replication_Clebert/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/remoting/server/impl/RemotingServiceImpl.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -42,7 +42,6 @@ import org.hornetq.core.remoting.impl.invm.TransportConstants; import org.hornetq.core.remoting.impl.wireformat.PacketImpl; import org.hornetq.core.remoting.impl.wireformat.Ping; -import org.hornetq.core.remoting.server.HandlerFactory; import org.hornetq.core.remoting.server.RemotingService; import org.hornetq.core.remoting.spi.Acceptor; import org.hornetq.core.remoting.spi.AcceptorFactory; @@ -86,7 +85,7 @@ private final ExecutorFactory executorFactory; - private final HandlerFactory handlerFactory; + private final HornetQServer server; private ManagementService managementService; @@ -105,7 +104,7 @@ // Constructors -------------------------------------------------- public RemotingServiceImpl(final Configuration config, - final HandlerFactory handlerFactory, + final HornetQServer server, final ExecutorFactory executorFactory, final ManagementService managementService, final Executor threadPool, @@ -116,7 +115,7 @@ this.executorFactory = executorFactory; - this.handlerFactory = handlerFactory; + this.server = server; ClassLoader loader = Thread.currentThread().getContextClassLoader(); for (String interceptorClass : config.getInterceptorClassNames()) @@ -306,8 +305,19 @@ public void connectionCreated(final Connection connection) { - RemotingConnection rc = createChannel(connection); + RemotingConnection rc = new RemotingConnectionImpl(connection, + interceptors, + executorFactory != null ? executorFactory.getExecutor() : null); + Channel channel1 = rc.getChannel(1, -1, false); + + ChannelHandler handler = createHandler(rc, channel1); + + channel1.setHandler(handler); + + + + long ttl = ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL; if (config.getConnectionTTLOverride() != -1) { @@ -400,19 +410,9 @@ /** * Subclasses (on tests) may use this to create a different channel. */ - protected RemotingConnection createChannel(final Connection connection) + protected ChannelHandler createHandler(final RemotingConnection rc, Channel channel) { - RemotingConnection rc = new RemotingConnectionImpl(connection, - interceptors, - executorFactory != null ? executorFactory.getExecutor() : null); - - Channel channel1 = rc.getChannel(1, -1, false); - - ChannelHandler handler = handlerFactory.getHandler(rc, channel1); - - channel1.setHandler(handler); - - return rc; + return new HornetQPacketHandler(server, channel, rc); } // Private ------------------------------------------------------- Deleted: branches/Replication_Clebert/src/main/org/hornetq/core/replication/BackupListener.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/BackupListener.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/BackupListener.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -1,46 +0,0 @@ -/* - * 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.replication; - -/** - * A ReplicationListener - * - * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public interface BackupListener -{ - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - void onReplication(byte data[]); - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - -} Added: branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationEndpoint.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationEndpoint.java (rev 0) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationEndpoint.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -0,0 +1,28 @@ +/* + * 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.replication; + +import org.hornetq.core.remoting.ChannelHandler; + +/** + * A ReplicationEndpoint + * + * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public interface ReplicationEndpoint extends ChannelHandler +{ + +} Modified: branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/ReplicationManager.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -13,17 +13,15 @@ package org.hornetq.core.replication; +import org.hornetq.core.server.HornetQComponent; + /** * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> * * */ -public interface ReplicationManager +public interface ReplicationManager extends HornetQComponent { void replicate(byte[] bytes, ReplicationToken token); - - - /** to be used on the backup node only */ - void addBackupListener(BackupListener listener); } Copied: branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java (from rev 7996, branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationPacketHandler.java) =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java (rev 0) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationEndpointImpl.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -0,0 +1,64 @@ +/* + * 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.replication.impl; + +import org.hornetq.core.remoting.ChannelHandler; +import org.hornetq.core.remoting.Packet; +import org.hornetq.core.remoting.impl.wireformat.ReplicationPacket; +import org.hornetq.core.replication.ReplicationEndpoint; +import org.hornetq.core.replication.ReplicationManager; +import org.hornetq.core.server.HornetQServer; + +/** + * A ReplicationPacketHandler + * + * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * + * + */ +public class ReplicationEndpointImpl implements ReplicationEndpoint +{ + + // Constants ----------------------------------------------------- + + // Attributes ---------------------------------------------------- + + private final HornetQServer server; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + public ReplicationEndpointImpl(HornetQServer server) + { + this.server = server ; + } + + // Public -------------------------------------------------------- + /* (non-Javadoc) + * @see org.hornetq.core.remoting.ChannelHandler#handlePacket(org.hornetq.core.remoting.Packet) + */ + public void handlePacket(Packet packet) + { + + } + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- + +} Modified: branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationManagerImpl.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -13,7 +13,10 @@ package org.hornetq.core.replication.impl; -import org.hornetq.core.replication.BackupListener; +import org.hornetq.core.client.impl.ConnectionManager; +import org.hornetq.core.remoting.Channel; +import org.hornetq.core.remoting.RemotingConnection; +import org.hornetq.core.remoting.impl.wireformat.CreateReplicationSessionMessage; import org.hornetq.core.replication.ReplicationManager; import org.hornetq.core.replication.ReplicationToken; @@ -31,30 +34,86 @@ // Attributes ---------------------------------------------------- + // TODO: Should this be configurable or not? + private static final int WINDOW_SIZE = 100 * 1024; + + private final ConnectionManager connectionManager; + + private RemotingConnection connection; + + private Channel replicatingChannel; + + private boolean started; + // Static -------------------------------------------------------- // Constructors -------------------------------------------------- + /** + * @param replicationConnectionManager + */ + public ReplicationManagerImpl(ConnectionManager connectionManager) + { + super(); + this.connectionManager = connectionManager; + } + // Public -------------------------------------------------------- /* (non-Javadoc) - * @see org.hornetq.core.replication.ReplicationManager#addListener(org.hornetq.core.replication.ReplicationListener) + * @see org.hornetq.core.replication.ReplicationManager#replicate(byte[], org.hornetq.core.replication.ReplicationToken) */ - public void addBackupListener(BackupListener listener) + public void replicate(byte[] bytes, ReplicationToken token) { // TODO Auto-generated method stub - + } /* (non-Javadoc) - * @see org.hornetq.core.replication.ReplicationManager#replicate(byte[], org.hornetq.core.replication.ReplicationToken) + * @see org.hornetq.core.server.HornetQComponent#isStarted() */ - public void replicate(byte[] bytes, ReplicationToken token) + public synchronized boolean isStarted() { - // TODO Auto-generated method stub - + return this.started; } + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQComponent#start() + */ + public synchronized void start() throws Exception + { + this.started = true; + + connection = connectionManager.getConnection(1); + + long channelID = connection.generateChannelID(); + + Channel mainChannel = connection.getChannel(1, -1, false); + + Channel tempChannel = connection.getChannel(channelID, WINDOW_SIZE, false); + + CreateReplicationSessionMessage replicationStartPackage = new CreateReplicationSessionMessage(channelID, + WINDOW_SIZE); + + mainChannel.sendBlocking(replicationStartPackage); + + this.replicatingChannel = tempChannel; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQComponent#stop() + */ + public void stop() throws Exception + { + replicatingChannel.close(); + + this.started = false; + + connection.destroy(); + + connection = null; + } + // Package protected --------------------------------------------- // Protected ----------------------------------------------------- Deleted: branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationPacketHandler.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationPacketHandler.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/replication/impl/ReplicationPacketHandler.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -1,54 +0,0 @@ -/* - * 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.replication.impl; - -import org.hornetq.core.remoting.ChannelHandler; -import org.hornetq.core.remoting.Packet; - -/** - * A ReplicationPacketHandler - * - * @author <mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * - * - */ -public class ReplicationPacketHandler implements ChannelHandler -{ - - /* (non-Javadoc) - * @see org.hornetq.core.remoting.ChannelHandler#handlePacket(org.hornetq.core.remoting.Packet) - */ - public void handlePacket(Packet packet) - { - } - - // Constants ----------------------------------------------------- - - // Attributes ---------------------------------------------------- - - // Static -------------------------------------------------------- - - // Constructors -------------------------------------------------- - - // Public -------------------------------------------------------- - - // Package protected --------------------------------------------- - - // Protected ----------------------------------------------------- - - // Private ------------------------------------------------------- - - // Inner classes ------------------------------------------------- - -} Modified: branches/Replication_Clebert/src/main/org/hornetq/core/server/HornetQServer.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/server/HornetQServer.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/server/HornetQServer.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -27,6 +27,7 @@ import org.hornetq.core.remoting.impl.wireformat.CreateSessionResponseMessage; import org.hornetq.core.remoting.impl.wireformat.ReattachSessionResponseMessage; import org.hornetq.core.remoting.server.RemotingService; +import org.hornetq.core.replication.ReplicationEndpoint; import org.hornetq.core.security.HornetQSecurityManager; import org.hornetq.core.security.Role; import org.hornetq.core.server.cluster.ClusterManager; @@ -36,7 +37,6 @@ import org.hornetq.core.version.Version; import org.hornetq.utils.ExecutorFactory; import org.hornetq.utils.SimpleString; -import org.hornetq.utils.UUID; /** * This interface defines the internal interface of the HornetQ Server exposed to other components of the server. The @@ -69,6 +69,8 @@ void unregisterActivateCallback(ActivateCallback callback); ReattachSessionResponseMessage reattachSession(RemotingConnection connection, String name, int lastReceivedCommandID) throws Exception; + + ReplicationEndpoint createReplicationEndpoint(); CreateSessionResponseMessage createSession(String name, long channelID, Modified: branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQPacketHandler.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -16,6 +16,7 @@ import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATESESSION; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATE_QUEUE; import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.REATTACH_SESSION; +import static org.hornetq.core.remoting.impl.wireformat.PacketImpl.CREATE_REPLICATION; import org.hornetq.core.exception.HornetQException; import org.hornetq.core.logging.Logger; @@ -24,9 +25,12 @@ import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.wireformat.CreateQueueMessage; +import org.hornetq.core.remoting.impl.wireformat.CreateReplicationSessionMessage; import org.hornetq.core.remoting.impl.wireformat.CreateSessionMessage; import org.hornetq.core.remoting.impl.wireformat.HornetQExceptionMessage; +import org.hornetq.core.remoting.impl.wireformat.NullResponseMessage; import org.hornetq.core.remoting.impl.wireformat.ReattachSessionMessage; +import org.hornetq.core.replication.ReplicationEndpoint; import org.hornetq.core.server.HornetQServer; /** @@ -91,6 +95,16 @@ break; } + case CREATE_REPLICATION: + { + // Create queue can also be fielded here in the case of a replicated store and forward queue creation + + CreateReplicationSessionMessage request = (CreateReplicationSessionMessage)packet; + + handleCreateReplication(request); + + break; + } default: { log.error("Invalid packet " + packet); @@ -174,6 +188,38 @@ log.error("Failed to handle create queue", e); } } + + private void handleCreateReplication(CreateReplicationSessionMessage request) + { + Packet response; + try + { + Channel channel = connection.getChannel(request.getSessionChannelID(), request.getWindowSize(), false); + ReplicationEndpoint endpoint = server.createReplicationEndpoint(); + channel.setHandler(endpoint); + response = new NullResponseMessage(); + + } + catch (Exception e) + { + log.warn(e.getMessage(), e); + + if (e instanceof HornetQException) + { + response = new HornetQExceptionMessage((HornetQException)e); + } + else + { + response = new HornetQExceptionMessage(new HornetQException(HornetQException.INTERNAL_ERROR)); + } + } + + channel1.send(response); + } + + + + } \ No newline at end of file Modified: branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -32,8 +32,11 @@ import javax.management.MBeanServer; +import org.hornetq.core.client.impl.ClientSessionFactoryImpl; import org.hornetq.core.client.impl.ConnectionManager; +import org.hornetq.core.client.impl.ConnectionManagerImpl; import org.hornetq.core.config.Configuration; +import org.hornetq.core.config.TransportConfiguration; import org.hornetq.core.config.cluster.DivertConfiguration; import org.hornetq.core.config.cluster.QueueConfiguration; import org.hornetq.core.config.impl.ConfigurationImpl; @@ -65,13 +68,15 @@ import org.hornetq.core.postoffice.impl.LocalQueueBinding; import org.hornetq.core.postoffice.impl.PostOfficeImpl; import org.hornetq.core.remoting.Channel; -import org.hornetq.core.remoting.ChannelHandler; import org.hornetq.core.remoting.RemotingConnection; import org.hornetq.core.remoting.impl.wireformat.CreateSessionResponseMessage; import org.hornetq.core.remoting.impl.wireformat.ReattachSessionResponseMessage; -import org.hornetq.core.remoting.server.HandlerFactory; import org.hornetq.core.remoting.server.RemotingService; import org.hornetq.core.remoting.server.impl.RemotingServiceImpl; +import org.hornetq.core.replication.ReplicationEndpoint; +import org.hornetq.core.replication.ReplicationManager; +import org.hornetq.core.replication.impl.ReplicationEndpointImpl; +import org.hornetq.core.replication.impl.ReplicationManagerImpl; import org.hornetq.core.security.CheckType; import org.hornetq.core.security.HornetQSecurityManager; import org.hornetq.core.security.Role; @@ -192,6 +197,8 @@ private static AtomicInteger managementConnectorSequence = new AtomicInteger(0); private ConnectionManager replicatingConnectionManager; + + private ReplicationManager replicationManager; private final Set<ActivateCallback> activateCallbacks = new HashSet<ActivateCallback>(); @@ -583,6 +590,11 @@ return new CreateSessionResponseMessage(true, version.getIncrementingVersion()); } + + public synchronized ReplicationEndpoint createReplicationEndpoint() + { + return new ReplicationEndpointImpl(this); + } public void removeSession(final String name) throws Exception { @@ -658,83 +670,45 @@ // } // } - // private boolean setupReplicatingConnection() throws Exception - // { - // String backupConnectorName = configuration.getBackupConnectorName(); - // - // if (backupConnectorName != null) - // { - // TransportConfiguration backupConnector = configuration.getConnectorConfigurations().get(backupConnectorName); - // - // if (backupConnector == null) - // { - // log.warn("connector with name '" + backupConnectorName + "' is not defined in the configuration."); - // } - // else - // { - // replicatingConnectionManager = new ConnectionManagerImpl(null, - // backupConnector, - // null, - // false, - // 1, - // ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT, - // ClientSessionFactoryImpl.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, - // ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL, - // 0, - // 1.0d, - // 0, - // threadPool, - // scheduledPool); - // - // replicatingConnection = replicatingConnectionManager.getConnection(1); - // - // if (replicatingConnection != null) - // { - // replicatingChannel = replicatingConnection.getChannel(2, -1, false); - // - // replicatingConnection.addFailureListener(new FailureListener() - // { - // public void connectionFailed(HornetQException me) - // { - // replicatingChannel.executeOutstandingDelayedResults(); - // } - // }); - // - // // First time we get channel we send a message down it informing the backup of our node id - - // // backup and live must have the same node id - // - // Packet packet = new ReplicateStartupInfoMessage(uuid, storageManager.getCurrentUniqueID()); - // - // final Future future = new Future(); - // - // replicatingChannel.replicatePacket(packet, 1, new Runnable() - // { - // public void run() - // { - // future.run(); - // } - // }); - // - // // This may take a while especially if the journal is large - // boolean ok = future.await(60000); - // - // if (!ok) - // { - // throw new IllegalStateException("Timed out waiting for response from backup for initialisation"); - // } - // } - // else - // { - // log.warn("Backup server MUST be started before live server. Initialisation will not proceed."); - // - // return false; - // } - // } - // } - // - // return true; - // } + private boolean startReplication() throws Exception + { + String backupConnectorName = configuration.getBackupConnectorName(); + if (backupConnectorName != null) + { + TransportConfiguration backupConnector = configuration.getConnectorConfigurations().get(backupConnectorName); + + if (backupConnector == null) + { + log.warn("connector with name '" + backupConnectorName + "' is not defined in the configuration."); + } + else + { + + replicatingConnectionManager = new ConnectionManagerImpl(null, + backupConnector, + null, + false, + 1, + ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT, + ClientSessionFactoryImpl.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, + ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL, + 0, + 1.0d, + 0, + false, + threadPool, + scheduledPool, + null); + + this.replicationManager = new ReplicationManagerImpl(replicatingConnectionManager); + replicationManager.start(); + } + } + + return true; + } + public HornetQServerControlImpl getHornetQServerControl() { return messagingServerControl; @@ -883,7 +857,7 @@ { return threadPool; } - + /** * This method is protected as it may be used as a hook for creating a custom storage manager (on tests for instance) * @return @@ -900,8 +874,6 @@ } } - - // Private // -------------------------------------------------------------------------------------- @@ -961,25 +933,17 @@ managementService = new ManagementServiceImpl(mbeanServer, configuration, managementConnectorID); - final HandlerFactory handlerFactory = new HandlerFactory() - { - - public ChannelHandler getHandler(RemotingConnection conn, Channel channel) - { - return new HornetQPacketHandler(HornetQServerImpl.this, channel, conn); - } - - }; - remotingService = new RemotingServiceImpl(configuration, - handlerFactory, - (configuration.isAsyncConnectionExecutionEnabled() ? this.executorFactory : null), + this, + (configuration.isAsyncConnectionExecutionEnabled() ? this.executorFactory + : null), managementService, threadPool, scheduledPool, managementConnectorID); - memoryManager = new MemoryManagerImpl(configuration.getMemoryWarningThreshold(), configuration.getMemoryMeasureInterval()); + memoryManager = new MemoryManagerImpl(configuration.getMemoryWarningThreshold(), + configuration.getMemoryMeasureInterval()); memoryManager.start(); } @@ -1182,7 +1146,7 @@ queues.put(queueBindingInfo.getPersistenceID(), queue); postOffice.addBinding(binding); - + managementService.registerAddress(queueBindingInfo.getAddress()); managementService.registerQueue(queue, queueBindingInfo.getAddress(), storageManager); } @@ -1278,7 +1242,7 @@ } postOffice.addBinding(binding); - + managementService.registerAddress(address); managementService.registerQueue(queue, address, storageManager); Modified: branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java =================================================================== --- branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-09-28 20:30:13 UTC (rev 8004) +++ branches/Replication_Clebert/tests/src/org/hornetq/tests/integration/replication/ReplicationTest.java 2009-09-28 22:26:08 UTC (rev 8005) @@ -42,15 +42,15 @@ import org.hornetq.core.remoting.Interceptor; import org.hornetq.core.remoting.Packet; import org.hornetq.core.remoting.RemotingConnection; -import org.hornetq.core.remoting.impl.RemotingConnectionImpl; import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory; import org.hornetq.core.remoting.impl.wireformat.CreateSessionResponseMessage; import org.hornetq.core.remoting.impl.wireformat.NullResponseMessage; import org.hornetq.core.remoting.impl.wireformat.ReattachSessionResponseMessage; -import org.hornetq.core.remoting.server.HandlerFactory; import org.hornetq.core.remoting.server.RemotingService; import org.hornetq.core.remoting.server.impl.RemotingServiceImpl; -import org.hornetq.core.remoting.spi.Connection; +import org.hornetq.core.replication.ReplicationEndpoint; +import org.hornetq.core.replication.impl.ReplicationEndpointImpl; +import org.hornetq.core.replication.impl.ReplicationManagerImpl; import org.hornetq.core.security.HornetQSecurityManager; import org.hornetq.core.security.Role; import org.hornetq.core.server.ActivateCallback; @@ -59,7 +59,6 @@ import org.hornetq.core.server.QueueFactory; import org.hornetq.core.server.ServerSession; import org.hornetq.core.server.cluster.ClusterManager; -import org.hornetq.core.server.impl.HornetQPacketHandler; import org.hornetq.core.settings.HierarchicalRepository; import org.hornetq.core.settings.impl.AddressSettings; import org.hornetq.core.transaction.ResourceManager; @@ -89,7 +88,7 @@ private ExecutorService executor; - private ConnectionManager connectionManagerLive; + private ConnectionManager connectionManager; private ScheduledExecutorService scheduledExecutor; @@ -101,52 +100,39 @@ public void testBasicConnection() throws Exception { - - RemotingConnection conn = connectionManagerLive.getConnection(1); - - Channel chann = conn.getChannel(2, -1, false); - - chann.close(); - + ReplicationManagerImpl manager = new ReplicationManagerImpl(connectionManager); + manager.start(); + manager.stop(); } // Package protected --------------------------------------------- - - class LocalHandler implements ChannelHandler + class LocalRemotingServiceImpl extends RemotingServiceImpl { - final Channel channel; - - /** - * @param channel - */ - public LocalHandler(Channel channel) + public LocalRemotingServiceImpl(Configuration config, + HornetQServer server, + ExecutorFactory executorFactory, + ManagementService managementService, + Executor threadPool, + ScheduledExecutorService scheduledThreadPool, + int managementConnectorID) { - super(); - this.channel = channel; + super(config, + server, + executorFactory, + managementService, + threadPool, + scheduledThreadPool, + managementConnectorID); } - /* (non-Javadoc) - * @see org.hornetq.core.remoting.ChannelHandler#handlePacket(org.hornetq.core.remoting.Packet) - */ - public void handlePacket(Packet packet) + protected ChannelHandler createHandler(RemotingConnection conn, Channel channel) { - channel.send(new NullResponseMessage()); + return super.createHandler(conn, channel); } } - HandlerFactory handlerFactory = new HandlerFactory() - { - - public ChannelHandler getHandler(RemotingConnection conn, Channel channel) - { - System.out.println("Created a handler"); - return new LocalHandler(channel); - } - - }; - // Protected ----------------------------------------------------- protected void setUp() throws Exception @@ -159,7 +145,7 @@ scheduledExecutor = new ScheduledThreadPoolExecutor(10, tFactory); - remoting = new RemotingServiceImpl(config, handlerFactory, null, null, executor, scheduledExecutor, 0); + remoting = new LocalRemotingServiceImpl(config, new FakeServer(), null, null, executor, scheduledExecutor, 0); remoting.start(); @@ -169,21 +155,21 @@ List<Interceptor> interceptors = new ArrayList<Interceptor>(); - connectionManagerLive = new ConnectionManagerImpl(null, - connectorConfig, - null, - false, - 1, - ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT, - ClientSessionFactoryImpl.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, - ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL, - 0, - 1.0d, - 0, - false, - executor, - scheduledExecutor, - interceptors); + connectionManager = new ConnectionManagerImpl(null, + connectorConfig, + null, + false, + 1, + ClientSessionFactoryImpl.DEFAULT_CALL_TIMEOUT, + ClientSessionFactoryImpl.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, + ClientSessionFactoryImpl.DEFAULT_CONNECTION_TTL, + 0, + 1.0d, + 0, + false, + executor, + scheduledExecutor, + interceptors); } @@ -206,4 +192,312 @@ // Inner classes ------------------------------------------------- + static class FakeServer implements HornetQServer + { + + public Queue createQueue(SimpleString address, + SimpleString queueName, + SimpleString filter, + boolean durable, + boolean temporary) throws Exception + { + return null; + } + + public CreateSessionResponseMessage createSession(String name, + long channelID, + String username, + String password, + int minLargeMessageSize, + int incrementingVersion, + RemotingConnection remotingConnection, + boolean autoCommitSends, + boolean autoCommitAcks, + boolean preAcknowledge, + boolean xa, + int producerWindowSize) throws Exception + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#deployQueue(org.hornetq.utils.SimpleString, org.hornetq.utils.SimpleString, org.hornetq.utils.SimpleString, boolean, boolean) + */ + public Queue deployQueue(SimpleString address, + SimpleString queueName, + SimpleString filterString, + boolean durable, + boolean temporary) throws Exception + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#destroyQueue(org.hornetq.utils.SimpleString, org.hornetq.core.server.ServerSession) + */ + public void destroyQueue(SimpleString queueName, ServerSession session) throws Exception + { + + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getAddressSettingsRepository() + */ + public HierarchicalRepository<AddressSettings> getAddressSettingsRepository() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getClusterManager() + */ + public ClusterManager getClusterManager() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getConfiguration() + */ + public Configuration getConfiguration() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getConnectionCount() + */ + public int getConnectionCount() + { + + return 0; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getExecutorFactory() + */ + public ExecutorFactory getExecutorFactory() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getHornetQServerControl() + */ + public HornetQServerControlImpl getHornetQServerControl() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getMBeanServer() + */ + public MBeanServer getMBeanServer() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getManagementService() + */ + public ManagementService getManagementService() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getNodeID() + */ + public SimpleString getNodeID() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getPostOffice() + */ + public PostOffice getPostOffice() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getQueueFactory() + */ + public QueueFactory getQueueFactory() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getRemotingService() + */ + public RemotingService getRemotingService() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getResourceManager() + */ + public ResourceManager getResourceManager() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getSecurityManager() + */ + public HornetQSecurityManager getSecurityManager() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getSecurityRepository() + */ + public HierarchicalRepository<Set<Role>> getSecurityRepository() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getSession(java.lang.String) + */ + public ServerSession getSession(String name) + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getSessions() + */ + public Set<ServerSession> getSessions() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getSessions(java.lang.String) + */ + public List<ServerSession> getSessions(String connectionID) + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getStorageManager() + */ + public StorageManager getStorageManager() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#getVersion() + */ + public Version getVersion() + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#isInitialised() + */ + public boolean isInitialised() + { + + return false; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#isStarted() + */ + public boolean isStarted() + { + + return false; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#reattachSession(org.hornetq.core.remoting.RemotingConnection, java.lang.String, int) + */ + public ReattachSessionResponseMessage reattachSession(RemotingConnection connection, + String name, + int lastReceivedCommandID) throws Exception + { + + return null; + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#registerActivateCallback(org.hornetq.core.server.ActivateCallback) + */ + public void registerActivateCallback(ActivateCallback callback) + { + + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#removeSession(java.lang.String) + */ + public void removeSession(String name) throws Exception + { + + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#unregisterActivateCallback(org.hornetq.core.server.ActivateCallback) + */ + public void unregisterActivateCallback(ActivateCallback callback) + { + + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQComponent#start() + */ + public void start() throws Exception + { + + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQComponent#stop() + */ + public void stop() throws Exception + { + + } + + /* (non-Javadoc) + * @see org.hornetq.core.server.HornetQServer#createReplicationEndpoint() + */ + public ReplicationEndpoint createReplicationEndpoint() + { + return new ReplicationEndpointImpl(this); + } + + } }