| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF 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 flex.messaging.services.messaging; |
| |
| import flex.messaging.MessageDestination; |
| import flex.messaging.MessageClient; |
| import flex.messaging.services.MessageService; |
| import flex.messaging.log.Log; |
| import flex.messaging.util.StringUtils; |
| import flex.messaging.cluster.RemoveNodeListener; |
| |
| import java.util.List; |
| |
| /** |
| * The RemoteSubscriptionManager monitors subscriptions from other |
| * servers, not other clients. One MessageClient instance is used for |
| * each remote server. It clientId is the address of the remote server. |
| * Using a separate instance of this class keeps the subscriptions |
| * of local clients separate from remote clients. |
| */ |
| public class RemoteSubscriptionManager extends SubscriptionManager implements RemoveNodeListener { |
| private Object syncLock = new Object(); |
| |
| /* |
| * A monitor lock used for synchronizing the attempt to request subscriptions |
| * across the cluster during startup. |
| */ |
| private static final Object initRemoteSubscriptionsLock = new Object(); |
| |
| public RemoteSubscriptionManager(MessageDestination destination) { |
| this(destination, false); |
| } |
| |
| public RemoteSubscriptionManager(MessageDestination destination, boolean enableManagement) { |
| super(destination, enableManagement); |
| } |
| |
| public void setSessionTimeout(long sessionConfigValue) { |
| } |
| |
| public long getSessionTimeout() { |
| return 0; // not used for remote subscriptions |
| } |
| |
| public void addSubscriber(String flexClientId, Object clientId, String selector, String subtopic) { |
| synchronized (syncLock) { |
| /* |
| * Only process subscriptions for servers whose subscription state we have received |
| * We may receive a subscribe/unsubscribe from a peer before we get their |
| * subscription state... we ignore these since they will be included in the |
| * state we receive later |
| */ |
| if (allSubscriptions.get(clientId) != null) |
| super.addSubscriber(clientId, selector, subtopic, null); |
| else if (Log.isDebug()) |
| Log.getLogger(MessageService.LOG_CATEGORY).debug("Ignoring new remote subscription for server: " + clientId + " whose subscription state we have not yet received. selector: " + selector + " subtopic: " + subtopic); |
| } |
| } |
| |
| public void removeSubscriber(String flexClientId, Object clientId, String selector, String subtopic, String endpoint) { |
| synchronized (syncLock) { |
| /* Only process subscriptions for servers whose subscription state we have received */ |
| if (allSubscriptions.get(clientId) != null) |
| super.removeSubscriber(clientId, selector, subtopic, null); |
| } |
| } |
| |
| protected void sendSubscriptionToPeer(boolean subscribe, String selector, String subtopic) { |
| // Don't do this for remote subscriptions |
| } |
| |
| protected MessageClient newMessageClient(Object clientId, String endpointId) { |
| return new RemoteMessageClient(clientId, destination, endpointId); |
| } |
| |
| /** |
| * Takes the selector and subtopic list from this address and |
| * for each one create a RemoteSubscription which gets |
| * registered in this table. We also register the remote |
| * subscription with a "per server" index so we can easily |
| * remove them later on. |
| * |
| * @param state the subscription state object |
| * @param address the remote cluster node address |
| */ |
| public void setSubscriptionState(Object state, Object address) { |
| MessageClient client = newMessageClient(address, null); |
| |
| if (Log.isDebug()) |
| Log.getLogger(MessageService.LOG_CATEGORY).debug("Received subscription state for destination: " + destination.getId() + " from server: " + address + StringUtils.NEWLINE + state); |
| |
| /* |
| * need to be sure we do not accept any remote sub/unsub messages |
| * from a given server until we have received its subscription state |
| * |
| * Also, we need to ensure we do not process any remote subscribe/unsubs |
| * from a remote server until we finish this list. |
| */ |
| synchronized (syncLock) { |
| allSubscriptions.put(address, client); |
| |
| List list = (List) state; |
| |
| for (int i = 0; i < list.size(); i += 2) { |
| addSubscriber(null, address, (String) list.get(i), (String) list.get(i + 1)); |
| } |
| } |
| synchronized (initRemoteSubscriptionsLock) { |
| initRemoteSubscriptionsLock.notifyAll(); |
| } |
| } |
| |
| /** |
| * This method waits for some time for the receipt of the subscription |
| * state for the server with the given address. If we fail to receive |
| * a message after waiting for the 5 seconds, a warning is printed. |
| * |
| * @param addr the remote cluster node address |
| */ |
| public void waitForSubscriptions(Object addr) { |
| /* If we have not gotten the response yet from this client... */ |
| if (getSubscriber(addr) == null) { |
| synchronized (initRemoteSubscriptionsLock) { |
| try { |
| if (Log.isDebug()) |
| Log.getLogger(MessageService.LOG_CATEGORY).debug("Waiting for subscriptions from cluster node: " + addr + " for destination: " + destination.getId()); |
| |
| initRemoteSubscriptionsLock.wait(5000); |
| |
| if (Log.isDebug()) |
| Log.getLogger(MessageService.LOG_CATEGORY).debug("Done waiting for subscriptions from cluster node: " + addr + " for destination: " + destination.getId()); |
| } catch (InterruptedException exc) { |
| } |
| } |
| if (getSubscriber(addr) == null && Log.isWarn()) |
| Log.getLogger(MessageService.LOG_CATEGORY).warn("No response yet from request subscriptions request for server: " + addr + " for destination: " + destination.getId()); |
| } else if (Log.isDebug()) |
| Log.getLogger(MessageService.LOG_CATEGORY).debug("Already have subscriptions from server: " + addr + " for destination: " + destination.getId()); |
| } |
| |
| /** |
| * Called when a cluster node gets removed. We need to make sure that all subscriptions |
| * for this node are removed. |
| * |
| * @param address the remote cluster node address |
| */ |
| public void removeClusterNode(Object address) { |
| if (Log.isDebug()) |
| Log.getLogger(MessageService.LOG_CATEGORY).debug("Cluster node: " + address + " subscriptions being removed for destination:" + destination.getId() + " before: " + StringUtils.NEWLINE + getDebugSubscriptionState()); |
| |
| MessageClient client = getSubscriber(address); |
| if (client != null) { |
| client.invalidate(); |
| } |
| |
| if (Log.isDebug()) |
| Log.getLogger(MessageService.LOG_CATEGORY).debug("Cluster node: " + address + " subscriptions being removed for destination:" + destination.getId() + " after: " + StringUtils.NEWLINE + getDebugSubscriptionState()); |
| } |
| |
| protected void monitorTimeout(MessageClient client) { |
| // Remote subscriptions do not timeout |
| } |
| |
| } |