Author: timfox Date: 2010-07-01 14:00:35 -0400 (Thu, 01 Jul 2010) New Revision: 9378 Added: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/SubscribeClusterTopologyUpdatesMessage.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java Log: HA improvements Added: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/SubscribeClusterTopologyUpdatesMessage.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/SubscribeClusterTopologyUpdatesMessage.java (rev 0) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/wireformat/SubscribeClusterTopologyUpdatesMessage.java 2010-07-01 18:00:35 UTC (rev 9378) @@ -0,0 +1,77 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.core.protocol.core.impl.wireformat; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.protocol.core.impl.PacketImpl; + +/** + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + */ +public class SubscribeClusterTopologyUpdatesMessage extends PacketImpl +{ + // Constants ----------------------------------------------------- + + private static final Logger log = Logger.getLogger(SubscribeClusterTopologyUpdatesMessage.class); + + // Attributes ---------------------------------------------------- + + private boolean clusterConnection; + + // Static -------------------------------------------------------- + + // Constructors -------------------------------------------------- + + public SubscribeClusterTopologyUpdatesMessage(final boolean clusterConnection) + { + super(PacketImpl.SUBSCRIBE_TOPOLOGY); + + this.clusterConnection = clusterConnection; + } + + public SubscribeClusterTopologyUpdatesMessage() + { + super(PacketImpl.SUBSCRIBE_TOPOLOGY); + } + + // Public -------------------------------------------------------- + + public boolean isClusterConnection() + { + return clusterConnection; + } + + @Override + public void encodeRest(final HornetQBuffer buffer) + { + buffer.writeBoolean(clusterConnection); + } + + @Override + public void decodeRest(final HornetQBuffer buffer) + { + clusterConnection = buffer.readBoolean(); + } + + + // Package protected --------------------------------------------- + + // Protected ----------------------------------------------------- + + // Private ------------------------------------------------------- + + // Inner classes ------------------------------------------------- +} Added: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java (rev 0) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterConnectionBridge.java 2010-07-01 18:00:35 UTC (rev 9378) @@ -0,0 +1,227 @@ +/* + * Copyright 2010 Red Hat, Inc. + * Red Hat licenses this file to you under the Apache License, version + * 2.0 (the "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + * implied. See the License for the specific language governing + * permissions and limitations under the License. + */ + +package org.hornetq.core.server.cluster.impl; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.Executor; +import java.util.concurrent.ScheduledExecutorService; + +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.SimpleString; +import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.api.core.client.ClientConsumer; +import org.hornetq.api.core.client.ClientMessage; +import org.hornetq.api.core.client.ClientProducer; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.ServerLocator; +import org.hornetq.api.core.management.ManagementHelper; +import org.hornetq.api.core.management.NotificationType; +import org.hornetq.api.core.management.ResourceNames; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.message.impl.MessageImpl; +import org.hornetq.core.persistence.StorageManager; +import org.hornetq.core.postoffice.BindingType; +import org.hornetq.core.server.Queue; +import org.hornetq.core.server.ServerMessage; +import org.hornetq.core.server.cluster.MessageFlowRecord; +import org.hornetq.core.server.cluster.Transformer; +import org.hornetq.utils.UUID; +import org.hornetq.utils.UUIDGenerator; + +/** + * A ClusterConnectionBridge + * + * @author tim + * + * + */ +public class ClusterConnectionBridge extends BridgeImpl +{ + private static final Logger log = Logger.getLogger(ClusterConnectionBridge.class); + + private final MessageFlowRecord flowRecord; + + private final SimpleString managementAddress; + + private final SimpleString managementNotificationAddress; + + private ClientConsumer notifConsumer; + + private final SimpleString idsHeaderName; + + private final TransportConfiguration connector; + + public ClusterConnectionBridge(final ServerLocator serverLocator, + final UUID nodeUUID, + final SimpleString name, + final Queue queue, + final Executor executor, + final SimpleString filterString, + final SimpleString forwardingAddress, + final ScheduledExecutorService scheduledExecutor, + final Transformer transformer, + final boolean useDuplicateDetection, + final String user, + final String password, + final boolean activated, + final StorageManager storageManager, + final SimpleString managementAddress, + final SimpleString managementNotificationAddress, + final MessageFlowRecord flowRecord, + final TransportConfiguration connector) throws Exception + { + super(serverLocator, + nodeUUID, + name, + queue, + executor, + filterString, + forwardingAddress, + scheduledExecutor, + transformer, + useDuplicateDetection, + user, + password, + activated, + storageManager); + + idsHeaderName = MessageImpl.HDR_ROUTE_TO_IDS.concat(name); + + this.managementAddress = managementAddress; + this.managementNotificationAddress = managementNotificationAddress; + this.flowRecord = flowRecord; + this.connector = connector; + } + + @Override + protected ServerMessage beforeForward(ServerMessage message) + { + // We make a copy of the message, then we strip out the unwanted routing id headers and leave + // only + // the one pertinent for the address node - this is important since different queues on different + // nodes could have same queue ids + // Note we must copy since same message may get routed to other nodes which require different headers + message = message.copy(); + + // TODO - we can optimise this + + Set<SimpleString> propNames = new HashSet<SimpleString>(message.getPropertyNames()); + + byte[] queueIds = message.getBytesProperty(idsHeaderName); + + for (SimpleString propName : propNames) + { + if (propName.startsWith(MessageImpl.HDR_ROUTE_TO_IDS)) + { + message.removeProperty(propName); + } + } + + message.putBytesProperty(MessageImpl.HDR_ROUTE_TO_IDS, queueIds); + + message = super.beforeForward(message); + + return message; + } + + private void setupNotificationConsumer() throws Exception + { + if (flowRecord != null) + { + flowRecord.reset(); + + if (notifConsumer != null) + { + try + { + notifConsumer.close(); + + notifConsumer = null; + } + catch (HornetQException e) + { + log.error("Failed to close consumer", e); + } + } + + // Get the queue data + + String qName = "notif." + UUIDGenerator.getInstance().generateStringUUID(); + + SimpleString notifQueueName = new SimpleString(qName); + + SimpleString filter = new SimpleString(ManagementHelper.HDR_BINDING_TYPE + "<>" + + BindingType.DIVERT.toInt() + + " AND " + + ManagementHelper.HDR_NOTIFICATION_TYPE + + " IN ('" + + NotificationType.BINDING_ADDED + + "','" + + NotificationType.BINDING_REMOVED + + "','" + + NotificationType.CONSUMER_CREATED + + "','" + + NotificationType.CONSUMER_CLOSED + + "','" + + NotificationType.PROPOSAL + + "','" + + NotificationType.PROPOSAL_RESPONSE + + "') AND " + + ManagementHelper.HDR_DISTANCE + + "<" + + flowRecord.getMaxHops() + + " AND (" + + ManagementHelper.HDR_ADDRESS + + " LIKE '" + + flowRecord.getAddress() + + "%')"); + + session.createQueue(managementNotificationAddress, notifQueueName, filter, false); + + notifConsumer = session.createConsumer(notifQueueName); + + notifConsumer.setMessageHandler(flowRecord); + + session.start(); + + ClientMessage message = session.createMessage(false); + + ManagementHelper.putOperationInvocation(message, + ResourceNames.CORE_SERVER, + "sendQueueInfoToQueue", + notifQueueName.toString(), + flowRecord.getAddress()); + + ClientProducer prod = session.createProducer(managementAddress); + + prod.send(message); + } + } + + @Override + protected void afterConnect() throws Exception + { + setupNotificationConsumer(); + } + + @Override + protected ClientSessionFactory createSessionFactory() throws Exception + { + //We create the session factory using the specified connector + + return serverLocator.createSessionFactory(connector); + } + +}