| /* |
| * 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.endpoints; |
| |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Set; |
| |
| import javax.servlet.http.HttpServletRequest; |
| import javax.servlet.http.HttpServletResponse; |
| |
| import flex.management.ManageableComponent; |
| import flex.management.runtime.messaging.MessageBrokerControl; |
| import flex.management.runtime.messaging.endpoints.EndpointControl; |
| import flex.messaging.FlexContext; |
| import flex.messaging.FlexSession; |
| import flex.messaging.MessageBroker; |
| import flex.messaging.MessageException; |
| import flex.messaging.Server; |
| import flex.messaging.client.FlexClient; |
| import flex.messaging.client.FlexClientOutboundQueueProcessor; |
| import flex.messaging.client.FlushResult; |
| import flex.messaging.client.PollFlushResult; |
| import flex.messaging.client.UserAgentSettings; |
| import flex.messaging.config.ChannelSettings; |
| import flex.messaging.config.ConfigMap; |
| import flex.messaging.config.ConfigurationConstants; |
| import flex.messaging.config.ConfigurationException; |
| import flex.messaging.config.SecurityConstraint; |
| import flex.messaging.io.ClassAliasRegistry; |
| import flex.messaging.io.SerializationContext; |
| import flex.messaging.io.TypeMarshaller; |
| import flex.messaging.io.TypeMarshallingContext; |
| import flex.messaging.io.amf.translator.ASTranslator; |
| import flex.messaging.log.Log; |
| import flex.messaging.log.LogCategories; |
| import flex.messaging.log.Logger; |
| import flex.messaging.messages.AcknowledgeMessage; |
| import flex.messaging.messages.AcknowledgeMessageExt; |
| import flex.messaging.messages.AsyncMessage; |
| import flex.messaging.messages.AsyncMessageExt; |
| import flex.messaging.messages.CommandMessage; |
| import flex.messaging.messages.CommandMessageExt; |
| import flex.messaging.messages.Message; |
| import flex.messaging.messages.SmallMessage; |
| import flex.messaging.security.SecurityException; |
| import flex.messaging.util.ClassUtil; |
| import flex.messaging.util.StringUtils; |
| import flex.messaging.util.UserAgentManager; |
| import flex.messaging.validators.DeserializationValidator; |
| |
| /** |
| * This is the default implementation of Endpoint, which provides a convenient |
| * base for behavior and associations common to all endpoints. |
| * |
| * These properties that appear in the endpoint configuration are only used by the |
| * client, therefore they have to be set on the appropriate client classes: connect-timeout-seconds set on Channel. |
| * |
| * @see flex.messaging.endpoints.Endpoint |
| */ |
| public abstract class AbstractEndpoint extends ManageableComponent |
| implements Endpoint2, ConfigurationConstants |
| { |
| /** Log category for <code>AbstractEndpoint</code>. */ |
| public static final String LOG_CATEGORY = LogCategories.ENDPOINT_GENERAL; |
| |
| /** |
| * HTTP header field names. |
| */ |
| public static final String HEADER_NAME_CACHE_CONTROL = "Cache-Control"; |
| public static final String HEADER_NAME_EXPIRES = "Expires"; |
| public static final String HEADER_NAME_PRAGMA = "Pragma"; |
| |
| // Errors |
| private static final int NONSECURE_PROTOCOL = 10066; |
| private static final int REQUIRES_FLEXCLIENT_SUPPORT = 10030; |
| private static final int ERR_MSG_INVALID_URL_SCHEME = 11100; |
| |
| // XML Configuration Properties |
| private static final String SERIALIZATION = "serialization"; |
| private static final String CREATE_ASOBJECT_FOR_MISSING_TYPE = "create-asobject-for-missing-type"; |
| private static final String CUSTOM_DESERIALIZER = "custom-deserializer"; |
| private static final String CUSTOM_SERIALIZER = "custom-serializer"; |
| private static final String ENABLE_SMALL_MESSAGES = "enable-small-messages"; |
| private static final String TYPE_MARSHALLER = "type-marshaller"; |
| private static final String RESTORE_REFERENCES = "restore-references"; |
| private static final String INSTANTIATE_TYPES = "instantiate-types"; |
| private static final String SUPPORT_REMOTE_CLASS = "support-remote-class"; |
| private static final String LEGACY_COLLECTION = "legacy-collection"; |
| private static final String LEGACY_DICTIONARY = "legacy-dictionary"; |
| private static final String LEGACY_MAP = "legacy-map"; |
| private static final String LEGACY_XML = "legacy-xml"; |
| private static final String LEGACY_XML_NAMESPACES = "legacy-xml-namespaces"; |
| private static final String LEGACY_THROWABLE = "legacy-throwable"; |
| private static final String LEGACY_BIG_NUMBERS = "legacy-big-numbers"; |
| private static final String LEGACY_EXTERNALIZABLE = "legacy-externalizable"; |
| private static final String ALLOW_XML_DOCTYPE_DECLARATION = "allow-xml-doctype-declaration"; |
| private static final String ALLOW_XML_EXTERNAL_ENTITY_EXPANSION = "allow-xml-external-entity-expansion"; |
| |
| private static final String LOG_PROPERTY_ERRORS = "log-property-errors"; |
| private static final String IGNORE_PROPERTY_ERRORS = "ignore-property-errors"; |
| private static final String INCLUDE_READ_ONLY = "include-read-only"; |
| private static final String GLOBAL_INCLUDE_READ_ONLY = "global-include-read-only"; |
| private static final String FLEX_CLIENT_OUTBOUND_QUEUE_PROCESSOR = "flex-client-outbound-queue-processor"; |
| private static final String SHOW_STACKTRACES = "show-stacktraces"; |
| private static final String MAX_OBJECT_NEST_LEVEL = "max-object-nest-level"; |
| private static final String MAX_COLLECTION_NEST_LEVEL = "max-collection-nest-level"; |
| private static final String PREFER_VECTORS = "prefer-vectors"; |
| |
| // Endpoint properties |
| protected Set<String> clientLoadBalancingUrls; |
| protected String clientType; |
| protected int connectTimeoutSeconds; |
| protected int requestTimeoutSeconds; |
| protected FlexClientOutboundQueueProcessor flexClientOutboundQueueProcessor; |
| protected SerializationContext serializationContext; |
| protected Class<?> deserializerClass; |
| protected Class<?> serializerClass; |
| protected TypeMarshaller typeMarshaller; |
| protected int port; |
| private SecurityConstraint securityConstraint; |
| protected String url; |
| protected boolean recordMessageSizes; |
| protected boolean recordMessageTimes; |
| protected boolean remote; |
| protected Server server; |
| protected boolean serverOnly; |
| |
| // Endpoint internal |
| protected String parsedUrl; |
| // Keeps track of what context path parsedUrl has been parsed for. If it is |
| // null, means parsedUrl has not been parsed already. |
| protected String parsedForContext; |
| protected boolean clientContextParsed; |
| protected String parsedClientUrl; |
| protected Logger log; |
| |
| protected Class<?> flexClientOutboundQueueProcessClass; |
| protected ConfigMap flexClientOutboundQueueProcessorConfig; |
| |
| // Supported messaging version |
| protected double messagingVersion = 1.0; |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Constructor |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Constructs an unmanaged <code>AbstractEndpoint</code>. |
| */ |
| public AbstractEndpoint() |
| { |
| this(false); |
| } |
| |
| /** |
| * Constructs an <code>AbstractEndpoint</code> with the indicated management. |
| * |
| * @param enableManagement <code>true</code> if the <code>AbstractEndpoint</code> |
| * is manageable; <code>false</code> otherwise. |
| */ |
| public AbstractEndpoint(boolean enableManagement) |
| { |
| super(enableManagement); |
| this.log = Log.getLogger(getLogCategory()); |
| serializationContext = new SerializationContext(); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Initialize, validate, start, and stop methods. |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Initializes the <code>Endpoint</code> with the properties. |
| * If subclasses override this method, they must call <code>super.initialize()</code>. |
| * |
| * @param id The ID of the <code>Endpoint</code>. |
| * @param properties Properties for the <code>Endpoint</code>. |
| */ |
| @Override |
| public void initialize(String id, ConfigMap properties) |
| { |
| super.initialize(id, properties); |
| |
| if (properties == null || properties.size() == 0) |
| return; |
| |
| // Client-targeted <client-load-balancing> |
| initializeClientLoadBalancing(id, properties); |
| |
| // Client-targeted <connect-timeout-seconds/> |
| connectTimeoutSeconds = properties.getPropertyAsInt(CONNECT_TIMEOUT_SECONDS_ELEMENT, 0); |
| |
| // Client-targeted <request-timeout-seconds/> |
| requestTimeoutSeconds = properties.getPropertyAsInt(REQUEST_TIMEOUT_SECONDS_ELEMENT, 0); |
| |
| // Check for a custom FlexClient outbound queue processor. |
| ConfigMap outboundQueueConfig = properties.getPropertyAsMap(FLEX_CLIENT_OUTBOUND_QUEUE_PROCESSOR, null); |
| if (outboundQueueConfig != null) |
| { |
| // Get nested props for the processor. |
| flexClientOutboundQueueProcessorConfig = outboundQueueConfig.getPropertyAsMap(PROPERTIES_ELEMENT, null); |
| |
| String pClassName = outboundQueueConfig.getPropertyAsString(CLASS_ATTR, null); |
| if (pClassName != null) |
| { |
| try |
| { |
| flexClientOutboundQueueProcessClass = createClass(pClassName); |
| // And now create an instance and initialize to make sure the properties are valid. |
| setFlexClientOutboundQueueProcessorConfig(flexClientOutboundQueueProcessorConfig); |
| } |
| catch (Throwable t) |
| { |
| if (Log.isWarn()) |
| log.warn("Cannot register custom FlexClient outbound queue processor class {0}", new Object[]{pClassName}, t); |
| } |
| } |
| } |
| |
| ConfigMap serialization = properties.getPropertyAsMap(SERIALIZATION, null); |
| if (serialization != null) |
| { |
| // Custom deserializers |
| List<?> deserializers = serialization.getPropertyAsList(CUSTOM_DESERIALIZER, null); |
| if (deserializers != null && Log.isWarn()) |
| log.warn("Endpoint <custom-deserializer> functionality is no longer available. Please remove this entry from your configuration."); |
| |
| // Custom serializers |
| List<?> serializers = serialization.getPropertyAsList(CUSTOM_SERIALIZER, null); |
| if (serializers != null && Log.isWarn()) |
| log.warn("Endpoint <custom-serializer> functionality is no longer available. Please remove this entry from your configuration."); |
| |
| // Type Marshaller implementation |
| String typeMarshallerClassName = serialization.getPropertyAsString(TYPE_MARSHALLER, null); |
| if (typeMarshallerClassName != null && typeMarshallerClassName.length() > 0) |
| { |
| try |
| { |
| Class<?> tmc = createClass(typeMarshallerClassName); |
| typeMarshaller = (TypeMarshaller)ClassUtil.createDefaultInstance(tmc, TypeMarshaller.class); |
| } |
| catch (Throwable t) |
| { |
| if (Log.isWarn()) |
| log.warn("Cannot register custom type marshaller for type {0}", new Object[]{typeMarshallerClassName}, t); |
| } |
| } |
| |
| // Boolean Serialization Flags |
| serializationContext.createASObjectForMissingType = serialization.getPropertyAsBoolean(CREATE_ASOBJECT_FOR_MISSING_TYPE, false); |
| serializationContext.enableSmallMessages = serialization.getPropertyAsBoolean(ENABLE_SMALL_MESSAGES, true); |
| serializationContext.instantiateTypes = serialization.getPropertyAsBoolean(INSTANTIATE_TYPES, true); |
| serializationContext.supportRemoteClass = serialization.getPropertyAsBoolean(SUPPORT_REMOTE_CLASS, false); |
| serializationContext.legacyCollection = serialization.getPropertyAsBoolean(LEGACY_COLLECTION, false); |
| serializationContext.legacyDictionary = serialization.getPropertyAsBoolean(LEGACY_DICTIONARY, false); |
| serializationContext.legacyMap = serialization.getPropertyAsBoolean(LEGACY_MAP, false); |
| serializationContext.legacyXMLDocument = serialization.getPropertyAsBoolean(LEGACY_XML, false); |
| serializationContext.legacyXMLNamespaces = serialization.getPropertyAsBoolean(LEGACY_XML_NAMESPACES, false); |
| serializationContext.legacyThrowable = serialization.getPropertyAsBoolean(LEGACY_THROWABLE, false); |
| serializationContext.legacyBigNumbers = serialization.getPropertyAsBoolean(LEGACY_BIG_NUMBERS, false); |
| serializationContext.legacyExternalizable = serialization.getPropertyAsBoolean(LEGACY_EXTERNALIZABLE, false); |
| serializationContext.allowXmlDoctypeDeclaration = serialization.getPropertyAsBoolean(ALLOW_XML_DOCTYPE_DECLARATION, false); |
| serializationContext.allowXmlExternalEntityExpansion = serialization.getPropertyAsBoolean(ALLOW_XML_EXTERNAL_ENTITY_EXPANSION, false); |
| serializationContext.maxObjectNestLevel = (int)serialization.getPropertyAsLong(MAX_OBJECT_NEST_LEVEL, 512); |
| serializationContext.maxCollectionNestLevel = (int)serialization.getPropertyAsLong(MAX_COLLECTION_NEST_LEVEL, 15); |
| serializationContext.preferVectors = serialization.getPropertyAsBoolean(PREFER_VECTORS, false); |
| |
| boolean showStacktraces = serialization.getPropertyAsBoolean(SHOW_STACKTRACES, false); |
| if (showStacktraces && Log.isWarn()) |
| log.warn("The " + SHOW_STACKTRACES + " configuration option is deprecated and non-functional. Please remove this from your configuration file."); |
| serializationContext.restoreReferences = serialization.getPropertyAsBoolean(RESTORE_REFERENCES, false); |
| serializationContext.logPropertyErrors = serialization.getPropertyAsBoolean(LOG_PROPERTY_ERRORS, false); |
| serializationContext.ignorePropertyErrors = serialization.getPropertyAsBoolean(IGNORE_PROPERTY_ERRORS, true); |
| serializationContext.includeReadOnly = serialization.getPropertyAsBoolean(INCLUDE_READ_ONLY, false); |
| } |
| |
| recordMessageSizes = properties.getPropertyAsBoolean(ConfigurationConstants.RECORD_MESSAGE_SIZES_ELEMENT, false); |
| |
| if (recordMessageSizes && Log.isWarn()) |
| log.warn("Setting <record-message-sizes> to true affects application performance and should only be used for debugging"); |
| |
| recordMessageTimes = properties.getPropertyAsBoolean(ConfigurationConstants.RECORD_MESSAGE_TIMES_ELEMENT, false); |
| } |
| |
| /** |
| * Starts the endpoint if its associated <code>MessageBroker</code> is started, |
| * and if the endpoint is not already running. If subclasses override this method, |
| * they must call <code>super.start()</code>. |
| */ |
| @Override |
| public void start() |
| { |
| if (isStarted()) |
| return; |
| |
| // Check if the MessageBroker is started |
| MessageBroker broker = getMessageBroker(); |
| if (!broker.isStarted()) |
| { |
| if (Log.isWarn()) |
| { |
| Log.getLogger(getLogCategory()).warn("Endpoint with id '{0}' cannot be started" + |
| " when the MessageBroker is not started.", |
| new Object[]{getId()}); |
| } |
| return; |
| } |
| |
| // Set up management |
| if (isManaged() && broker.isManaged()) |
| { |
| setupEndpointControl(broker); |
| MessageBrokerControl controller = (MessageBrokerControl)broker.getControl(); |
| if (getControl() != null) |
| controller.addEndpoint(this); |
| } |
| |
| // Setup Deserializer and Serializer for the SerializationContext |
| if (deserializerClass == null) |
| deserializerClass = createClass(getDeserializerClassName()); |
| |
| if (serializerClass == null) |
| serializerClass = createClass(getSerializerClassName()); |
| |
| serializationContext.setDeserializerClass(deserializerClass); |
| serializationContext.setSerializerClass(serializerClass); |
| |
| // Setup endpoint features |
| ClassAliasRegistry registry = ClassAliasRegistry.getRegistry(); |
| registry.registerAlias(AsyncMessageExt.CLASS_ALIAS, AsyncMessageExt.class.getName()); |
| registry.registerAlias(AcknowledgeMessageExt.CLASS_ALIAS, AcknowledgeMessageExt.class.getName()); |
| registry.registerAlias(CommandMessageExt.CLASS_ALIAS, CommandMessageExt.class.getName()); |
| super.start(); |
| } |
| |
| /** |
| * Stops the endpoint if it is running. If subclasses override this method, they must |
| * call <code>super.stop()</code>. |
| */ |
| @Override |
| public void stop() |
| { |
| if (!isStarted()) |
| return; |
| |
| super.stop(); |
| |
| // Remove management |
| if (isManaged() && getMessageBroker().isManaged()) |
| { |
| if (getControl() != null) |
| { |
| getControl().unregister(); |
| setControl(null); |
| } |
| setManaged(false); |
| } |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Public Getters and Setters for AbstractEndpoint properties |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Adds a client-load-balancing URL. |
| * |
| * @param url A client-load-balancing URL. |
| * @return <code>false</code> if the set already contains the URL, <code>true</code> otherwise. |
| * |
| */ |
| public boolean addClientLoadBalancingUrl(String url) |
| { |
| if (clientLoadBalancingUrls == null) |
| clientLoadBalancingUrls = new HashSet<String>(); |
| |
| if (url == null || url.length() == 0) |
| { |
| // Invalid {0} configuration for endpoint ''{1}''; cannot add empty url. |
| ConfigurationException ce = new ConfigurationException(); |
| ce.setMessage(ERR_MSG_EMTPY_CLIENT_LOAD_BALACNING_URL, new Object[]{CLIENT_LOAD_BALANCING_ELEMENT, getId()}); |
| throw ce; |
| } |
| |
| return clientLoadBalancingUrls.add(url); |
| } |
| |
| /** |
| * Removes the client-load-balancing URL. |
| * |
| * @param url The URL to remove. |
| * @return <code>true</code> if the set contained the URL, <code>false</code> otherwise. |
| */ |
| public boolean removeClientLoadBalancingUrl(String url) |
| { |
| if (clientLoadBalancingUrls != null) |
| return clientLoadBalancingUrls.remove(url); |
| return false; |
| } |
| |
| /** |
| * Retrieves a snapshot of the current client-load-balancing URLs. |
| * |
| * @return A snapshot of the current client-load-balancing URLs, or <code>null</code> if |
| * no URL exists. |
| */ |
| public Set<String> getClientLoadBalancingUrls() |
| { |
| return clientLoadBalancingUrls == null? null : new HashSet<String>(clientLoadBalancingUrls); |
| } |
| |
| /** |
| * Retrieves the corresponding client channel type for the endpoint. |
| * |
| * @return The corresponding client channel type for the endpoint. |
| */ |
| public String getClientType() |
| { |
| return clientType; |
| } |
| |
| /** |
| * Sets the corresponding client channel type for the endpoint. |
| * |
| * @param type The corresponding client channel type for the endpoint. |
| */ |
| public void setClientType(String type) |
| { |
| this.clientType = type; |
| } |
| |
| /** |
| * Retrieves the <code>FlexClientOutboundQueueProcessorClass</code> of the endpoint. |
| * |
| * @return The <code>FlexClientOutboundQueueProcessorClass</code> of the endpoint. |
| */ |
| public Class<?> getFlexClientOutboundQueueProcessorClass() |
| { |
| return flexClientOutboundQueueProcessClass; |
| } |
| |
| /** |
| * Sets the the <code>FlexClientOutboundQueueProcessor</code> of the endpoint. |
| * |
| * @param flexClientOutboundQueueProcessorClass the Class of the Flex client outbound queue processor. |
| */ |
| public void setFlexClientOutboundQueueProcessorClass(Class<?> flexClientOutboundQueueProcessorClass) |
| { |
| this.flexClientOutboundQueueProcessClass = flexClientOutboundQueueProcessorClass; |
| if (flexClientOutboundQueueProcessClass != null && flexClientOutboundQueueProcessorConfig != null) |
| { |
| FlexClientOutboundQueueProcessor processor = (FlexClientOutboundQueueProcessor)ClassUtil.createDefaultInstance(flexClientOutboundQueueProcessClass, null); |
| processor.initialize(flexClientOutboundQueueProcessorConfig); |
| } |
| } |
| |
| /** |
| * Retrieves the properties for the <code>FlexClientOutboundQueueProcessor</code> of the endpoint. |
| * |
| * @return The properties for the <code>FlexClientOutboundQueueProcessor</code> of the endpoint. |
| */ |
| public ConfigMap getFlexClientOutboundQueueProcessorConfig() |
| { |
| return flexClientOutboundQueueProcessorConfig; |
| } |
| |
| /** |
| * Sets the properties for the <code>FlexClientOutboundQueueProcessor</code> of the endpoint. |
| * |
| * @param flexClientOutboundQueueProcessorConfig The configuration map. |
| */ |
| public void setFlexClientOutboundQueueProcessorConfig(ConfigMap flexClientOutboundQueueProcessorConfig) |
| { |
| this.flexClientOutboundQueueProcessorConfig = flexClientOutboundQueueProcessorConfig; |
| if (flexClientOutboundQueueProcessorConfig != null && flexClientOutboundQueueProcessClass != null) |
| { |
| FlexClientOutboundQueueProcessor processor = (FlexClientOutboundQueueProcessor)ClassUtil.createDefaultInstance(flexClientOutboundQueueProcessClass, null); |
| processor.initialize(flexClientOutboundQueueProcessorConfig); |
| } |
| } |
| |
| /** |
| * Sets the ID of the <code>AbstractEndpoint</code>. If the <code>AbstractEndpoint</code> |
| * has a <code>MessageBroker</code> assigned, it also updates the ID in the |
| * <code>MessageBroker</code>. |
| * |
| * @param id The endpoint ID. |
| */ |
| @Override |
| public void setId(String id) |
| { |
| String oldId = getId(); |
| |
| if (oldId != null && oldId.equals(id)) |
| return; |
| |
| super.setId(id); |
| |
| // Update the endpoint id in the broker |
| MessageBroker broker = getMessageBroker(); |
| if (broker != null) |
| { |
| // broker must have the endpoint then |
| broker.removeEndpoint(oldId); |
| broker.addEndpoint(this); |
| } |
| } |
| |
| /** |
| * Retrieves the <code>MessageBroker</code> of the <code>AbstractEndpoint</code>. |
| * |
| * @return The <code>MessageBroker</code> of the <code>AbstractEndpoint</code>. |
| */ |
| public MessageBroker getMessageBroker() |
| { |
| return (MessageBroker)getParent(); |
| } |
| |
| /** |
| * Sets the <code>MessageBroker</code> of the <code>AbstractEndpoint</code>. |
| * Removes the <code>AbstractEndpoint</code> from the old broker |
| * (if there was one) and adds to the list of endpoints in the new broker. |
| * |
| * @param broker The <code>MessageBroker</code> of the <code>AbstractEndpoint</code>. |
| */ |
| public void setMessageBroker(MessageBroker broker) |
| { |
| MessageBroker oldBroker = getMessageBroker(); |
| |
| setParent(broker); |
| |
| if (oldBroker != null) |
| oldBroker.removeEndpoint(getId()); |
| |
| // Add endpoint to the new broker if needed |
| if (broker.getEndpoint(getId()) != this) |
| broker.addEndpoint(this); |
| } |
| |
| /** |
| * Return the highest messaging version currently available via this |
| * endpoint. |
| * @return double the messaging version |
| */ |
| public double getMessagingVersion() |
| { |
| return messagingVersion; |
| } |
| |
| /** |
| * Retrieves the port of the URL of the endpoint. |
| * A return value of 0 denotes no port in the channel URL. |
| * |
| * @return The port of the URL of the endpoint, or 0 if the URL does not contain |
| * a port number. |
| */ |
| public int getPort() |
| { |
| return port; |
| } |
| |
| /** |
| * Determines whether the endpoint is secure. |
| * |
| * @return <code>false</code> by default. |
| */ |
| public boolean isSecure() |
| { |
| return false; |
| } |
| |
| /** |
| * Determines if the endpoint clients connect to directly is mirrored and running |
| * on a remote host, in which case this local instance is not started and will service no direct |
| * client connections. |
| * |
| * @return <code>true</code> if this endpoint will not process direct client connections and is just |
| * a local representation of a symmetric endpoint on a remote host that will, <code>false</code> otherwise. |
| */ |
| public boolean isRemote() |
| { |
| return remote; |
| } |
| |
| /** |
| * Sets the remote status for this endpoint. |
| * |
| * @param value <code>true</code> if this endpoint will not process direct client connections and is just |
| * a local representation of a symmetric endpoint on a remote host that will, <code>false</code> otherwise. |
| */ |
| public void setRemote(boolean value) |
| { |
| remote = value; |
| } |
| |
| /** |
| * Retrieves the <tt>Server</tt> that the endpoint is using, or <code>null</code> if |
| * no server has been assigned. |
| * @return Server The Server object the endpoint is using. |
| */ |
| public Server getServer() |
| { |
| return server; |
| } |
| |
| /** |
| * Sets the <tt>Server</tt> that the endpoint will use. |
| * @param server The Server object. |
| */ |
| public void setServer(Server server) |
| { |
| this.server = server; |
| } |
| |
| /** |
| * Determines whether the endpoint is server only. |
| * |
| * @return <code>true</code> if the endpoint is server only, <code>false</code> otherwise. |
| */ |
| public boolean getServerOnly() |
| { |
| return serverOnly; |
| } |
| |
| /** |
| * Sets whether the endpoint is server only. |
| * |
| * @param serverOnly <code>true</code> if the endpoint is server only, <code>false</code> otherwise. |
| */ |
| public void setServerOnly(boolean serverOnly) |
| { |
| this.serverOnly = serverOnly; |
| } |
| |
| /** |
| * Retrieves the <code>SecurityConstraint</code> of the <code>Endpoint</code>. |
| * |
| * @return The <code>SecurityConstraint</code> of the <code>Endpoint</code>. |
| */ |
| public SecurityConstraint getSecurityConstraint() |
| { |
| return securityConstraint; |
| } |
| |
| /** |
| * Sets the <code>SecurityConstraint</code> of the <code>Endpoint</code>. |
| * |
| * @param securityConstraint The SecurityContraint object. |
| */ |
| public void setSecurityConstraint(SecurityConstraint securityConstraint) |
| { |
| this.securityConstraint = securityConstraint; |
| } |
| |
| /** |
| * Retrieves the <code>SerializationContext</code> of the endpoint. |
| * |
| * @return The <code>SerializationContext</code> of the endpoint. |
| */ |
| public SerializationContext getSerializationContext() |
| { |
| return serializationContext; |
| } |
| |
| /** |
| * Sets the <code>SerializationContext</code> of the endpoint. |
| * |
| * @param serializationContext The SerializationContext object. |
| */ |
| public void setSerializationContext(SerializationContext serializationContext) |
| { |
| this.serializationContext = serializationContext; |
| } |
| |
| /** |
| * Retrieves the <code>TypeMarshaller</code> of the endpoint. |
| * |
| * @return The <code>TypeMarshaller</code> of the endpoint. |
| */ |
| public TypeMarshaller getTypeMarshaller() |
| { |
| if (typeMarshaller == null) |
| typeMarshaller = new ASTranslator(); |
| |
| return typeMarshaller; |
| } |
| |
| /** |
| * Sets the <code>TypeMarshaller</code> of the endpoint. |
| * |
| * @param typeMarshaller The TypeMarshaller object. |
| */ |
| public void setTypeMarshaller(TypeMarshaller typeMarshaller) |
| { |
| this.typeMarshaller = typeMarshaller; |
| } |
| |
| /** |
| * Retrieves the URL of the endpoint. |
| * |
| * @return The URL of the endpoint. |
| */ |
| public String getUrl() |
| { |
| return url; |
| } |
| |
| /** |
| * Sets the URL of the endpoint. |
| * |
| * @param url The URL of the endpoint. |
| */ |
| public void setUrl(String url) |
| { |
| this.url = url; |
| port = internalParsePort(url); |
| parsedForContext = null; |
| clientContextParsed = false; |
| } |
| |
| /** |
| * |
| * Returns the url of the endpoint parsed for the client. |
| * |
| * @return The url of the endpoint parsed for the client. |
| */ |
| public String getUrlForClient() |
| { |
| if (!clientContextParsed) |
| { |
| HttpServletRequest req = FlexContext.getHttpRequest(); |
| if (req != null) |
| { |
| String contextPath = req.getContextPath(); |
| parseClientUrl(contextPath); |
| } |
| else |
| { |
| return url; |
| } |
| } |
| return parsedClientUrl; |
| } |
| |
| /** |
| * |
| * Returns the total throughput for the endpoint. |
| * |
| * @return The total throughput for the endpoint. |
| */ |
| public long getThroughput() |
| { |
| EndpointControl control = (EndpointControl)getControl(); |
| |
| return control.getBytesDeserialized().longValue() + control.getBytesSerialized().longValue(); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Other Public APIs |
| // |
| //-------------------------------------------------------------------------- |
| |
| |
| public static void addNoCacheHeaders(HttpServletRequest req, HttpServletResponse res) |
| { |
| String userAgent = req.getHeader(UserAgentManager.USER_AGENT_HEADER_NAME); |
| |
| // For MSIE over HTTPS, set additional Cache-Control values. |
| if (req.isSecure() && userAgent != null && userAgent.indexOf(UserAgentSettings.USER_AGENT_MSIE) != -1) |
| res.addHeader(HEADER_NAME_CACHE_CONTROL, "no-store, no-cache, must-revalidate, post-check=0, pre-check=0, no-transform, private"); |
| else // For the rest, set no-cache header value only. |
| res.addHeader(HEADER_NAME_CACHE_CONTROL, "no-cache"); |
| |
| // Set an expiration date in the past as well. |
| res.setDateHeader(HEADER_NAME_EXPIRES, 946080000000L); //Approx Jan 1, 2000 |
| |
| // Set Pragma no-cache header if we're not MSIE over HTTPS |
| if (!(req.isSecure() && userAgent != null && userAgent.indexOf(UserAgentSettings.USER_AGENT_MSIE) != -1)) |
| res.setHeader(HEADER_NAME_PRAGMA, "no-cache"); |
| } |
| |
| /** |
| * |
| */ |
| public Message convertToSmallMessage(Message message) |
| { |
| if (message instanceof SmallMessage) |
| { |
| Message smallMessage = ((SmallMessage)message).getSmallMessage(); |
| if (smallMessage != null) |
| message = smallMessage; |
| } |
| |
| return message; |
| } |
| |
| /** |
| * Retrieves a <code>ConfigMap</code> of the endpoint properties the client |
| * needs. Subclasses should add additional properties to <code>super.describeDestination</code>, |
| * or return <code>null</code> if they must not send their properties to the client. |
| * |
| * @return ConfigMap The ConfigMap object. |
| */ |
| public ConfigMap describeEndpoint() |
| { |
| ConfigMap channelConfig = new ConfigMap(); |
| |
| if (serverOnly) // Client does not need server only endpoints. |
| return channelConfig; |
| |
| channelConfig.addProperty(ID_ATTR, getId()); |
| channelConfig.addProperty(TYPE_ATTR, getClientType()); |
| |
| ConfigMap properties = new ConfigMap(); |
| |
| boolean containsClientLoadBalancing = clientLoadBalancingUrls != null && !clientLoadBalancingUrls.isEmpty(); |
| if (containsClientLoadBalancing) |
| { |
| ConfigMap clientLoadBalancing = new ConfigMap(); |
| for (Iterator<String> iterator = clientLoadBalancingUrls.iterator(); iterator.hasNext();) |
| { |
| ConfigMap url = new ConfigMap(); |
| // Adding as a value rather than attribute to the parent. |
| url.addProperty(EMPTY_STRING, iterator.next()); |
| clientLoadBalancing.addProperty(URL_ATTR, url); |
| } |
| properties.addProperty(CLIENT_LOAD_BALANCING_ELEMENT, clientLoadBalancing); |
| } |
| |
| // Add endpoint uri only if no client-load-balancing urls are defined. |
| if (!containsClientLoadBalancing) |
| { |
| ConfigMap endpointConfig = new ConfigMap(); |
| endpointConfig.addProperty(URI_ATTR, getUrlForClient()); |
| channelConfig.addProperty(ENDPOINT_ELEMENT, endpointConfig); |
| } |
| |
| if (connectTimeoutSeconds > 0) |
| { |
| ConfigMap connectTimeoutConfig = new ConfigMap(); |
| connectTimeoutConfig.addProperty(EMPTY_STRING, String.valueOf(connectTimeoutSeconds)); |
| properties.addProperty(CONNECT_TIMEOUT_SECONDS_ELEMENT, connectTimeoutConfig); |
| } |
| |
| if (requestTimeoutSeconds > 0) |
| { |
| ConfigMap requestTimeoutSeconds = new ConfigMap(); |
| requestTimeoutSeconds.addProperty(EMPTY_STRING, String.valueOf(requestTimeoutSeconds)); |
| properties.addProperty(REQUEST_TIMEOUT_SECONDS_ELEMENT, requestTimeoutSeconds); |
| } |
| |
| if (recordMessageTimes) |
| { |
| ConfigMap recordMessageTimesMap = new ConfigMap(); |
| // Adding as a value rather than attribute to the parent |
| recordMessageTimesMap.addProperty(EMPTY_STRING, TRUE_STRING); |
| properties.addProperty(RECORD_MESSAGE_TIMES_ELEMENT, recordMessageTimesMap); |
| } |
| |
| if (recordMessageSizes) |
| { |
| ConfigMap recordMessageSizesMap = new ConfigMap(); |
| // Adding as a value rather than attribute to the parent |
| recordMessageSizesMap.addProperty(EMPTY_STRING, TRUE_STRING); |
| properties.addProperty(RECORD_MESSAGE_SIZES_ELEMENT, recordMessageSizesMap); |
| } |
| |
| ConfigMap serialization = new ConfigMap(); |
| serialization.addProperty(ENABLE_SMALL_MESSAGES_ELEMENT, Boolean.toString(serializationContext.enableSmallMessages)); |
| properties.addProperty(SERIALIZATION_ELEMENT, serialization); |
| |
| if (properties.size() > 0) |
| channelConfig.addProperty(PROPERTIES_ELEMENT, properties); |
| |
| return channelConfig; |
| } |
| |
| /** |
| * |
| * Make sure this matches with ChannelSettings.getParsedUri. |
| */ |
| public String getParsedUrl(String contextPath) |
| { |
| parseUrl(contextPath); |
| return parsedUrl; |
| } |
| |
| /** |
| * |
| */ |
| public void handleClientMessagingVersion(Number version) |
| { |
| if (version != null) |
| { |
| boolean clientSupportsSmallMessages = version.doubleValue() >= messagingVersion; |
| if (clientSupportsSmallMessages && getSerializationContext().enableSmallMessages) |
| { |
| FlexSession session = FlexContext.getFlexSession(); |
| if (session != null) |
| session.setUseSmallMessages(true); |
| } |
| } |
| } |
| |
| /** |
| * Default implementation of the Endpoint <code>service</code> method. |
| * Subclasses should call <code>super.service</code> before their custom |
| * code. |
| * |
| * @param req The HttpServletRequest object. |
| * @param res The HttpServletResponse object. |
| */ |
| public void service(HttpServletRequest req, HttpServletResponse res) |
| { |
| validateRequestProtocol(req); |
| } |
| |
| /** |
| * Typically invoked by subclasses, this method transforms decoded message data |
| * into the appropriate Message object and routes the Message to the endpoint's broker. |
| * |
| * @param message The decoded message data. |
| * @return Message The transformed message. |
| */ |
| public Message serviceMessage(Message message) |
| { |
| if (isManaged()) |
| { |
| ((EndpointControl) getControl()).incrementServiceMessageCount(); |
| } |
| |
| try |
| { |
| FlexContext.setThreadLocalEndpoint(this); |
| Message ack = null; |
| |
| // Make sure this message is timestamped. |
| if (message.getTimestamp() == 0) |
| { |
| message.setTimestamp(System.currentTimeMillis()); |
| } |
| |
| // Reset the endpoint header for inbound messages to the id for this endpoint |
| // to guarantee that it's correct. Don't allow clients to spoof this. |
| // However, if the endpoint id is passed as null we need to tag the message to |
| // skip channel/endpoint validation at the destination level (MessageBroker.inspectChannel()). |
| if (message.getHeader(Message.ENDPOINT_HEADER) != null) |
| message.setHeader(Message.VALIDATE_ENDPOINT_HEADER, Boolean.TRUE); |
| message.setHeader(Message.ENDPOINT_HEADER, getId()); |
| |
| if (message instanceof CommandMessage) |
| { |
| CommandMessage command = (CommandMessage)message; |
| |
| // Apply channel endpoint level constraint; always allow login commands through. |
| int operation = command.getOperation(); |
| if (operation != CommandMessage.LOGIN_OPERATION) |
| checkSecurityConstraint(message); |
| |
| // Handle general (not Consumer specific) poll requests here. |
| // We need to fetch all outbound messages for client subscriptions over this endpoint. |
| // We identify these general poll messages by their operation and a null clientId. |
| if (operation == CommandMessage.POLL_OPERATION && message.getClientId() == null) |
| { |
| verifyFlexClientSupport(command); |
| |
| |
| FlexClient flexClient = FlexContext.getFlexClient(); |
| ack = handleFlexClientPollCommand(flexClient, command); |
| } |
| else if (operation == CommandMessage.DISCONNECT_OPERATION) |
| { |
| ack = handleChannelDisconnect(command); |
| } |
| else if (operation == CommandMessage.TRIGGER_CONNECT_OPERATION) |
| { |
| ack = new AcknowledgeMessage(); |
| ((AcknowledgeMessage)ack).setCorrelationId(message.getMessageId()); |
| |
| boolean needsConfig = false; |
| if (command.getHeader(CommandMessage.NEEDS_CONFIG_HEADER) != null) |
| needsConfig = ((Boolean)(command.getHeader(CommandMessage.NEEDS_CONFIG_HEADER))); |
| |
| // Send configuration information only if the client requested. |
| if (needsConfig) |
| { |
| ConfigMap serverConfig = getMessageBroker().describeServices(this); |
| if (serverConfig.size() > 0) |
| ack.setBody(serverConfig); |
| } |
| } |
| else |
| { |
| // Block a subset of commands for legacy clients that need to be recompiled to |
| // interop with a 2.5+ server. |
| if (operation == CommandMessage.SUBSCRIBE_OPERATION || operation == CommandMessage.POLL_OPERATION) |
| verifyFlexClientSupport(command); |
| |
| ack = getMessageBroker().routeCommandToService((CommandMessage) message, this); |
| |
| // Look for client advertised features on initial connect. |
| if (operation == CommandMessage.CLIENT_PING_OPERATION || operation == CommandMessage.LOGIN_OPERATION) |
| { |
| Number clientVersion = (Number)command.getHeader(CommandMessage.MESSAGING_VERSION); |
| handleClientMessagingVersion(clientVersion); |
| |
| // Also respond by advertising the messaging version on the |
| // acknowledgement. |
| ack.setHeader(CommandMessage.MESSAGING_VERSION, new Double(messagingVersion)); |
| } |
| } |
| } |
| else |
| { |
| // Block any AsyncMessages from a legacy client. |
| if (message instanceof AsyncMessage) |
| verifyFlexClientSupport(message); |
| |
| // Apply channel endpoint level constraint. |
| checkSecurityConstraint(message); |
| |
| ack = getMessageBroker().routeMessageToService(message, this); |
| } |
| |
| return ack; |
| } |
| finally |
| { |
| FlexContext.setThreadLocalEndpoint(null); |
| } |
| } |
| |
| /** |
| * Utility method that endpoint implementations (or associated classes) |
| * should invoke when they receive an incoming message from a client but before |
| * servicing it. This method looks up or creates the proper FlexClient instance |
| * based upon the client the message came from and places it in the FlexContext. |
| * |
| * @param message The incoming message to process. |
| * |
| * @return The FlexClient, or <code>null</code> if the message did not contain a FlexClient ID value. |
| */ |
| public FlexClient setupFlexClient(Message message) |
| { |
| FlexClient flexClient = null; |
| if (message.getHeaders().containsKey(Message.FLEX_CLIENT_ID_HEADER)) |
| { |
| String id = (String)message.getHeaders().get(Message.FLEX_CLIENT_ID_HEADER); |
| // If the id is null, reset to the special token value that let's us differentiate |
| // between legacy clients and 2.5+ clients. |
| if (id == null) |
| id = FlexClient.NULL_FLEXCLIENT_ID; |
| flexClient = setupFlexClient(id); |
| } |
| return flexClient; |
| } |
| |
| /** |
| * Utility method that endpoint implementations (or associated classes) |
| * should invoke when they receive an incoming message from a client but before |
| * servicing it. This method looks up or creates the proper FlexClient instance |
| * based upon the FlexClient ID value received from the client. |
| * It also associates this FlexClient instance with the current FlexSession. |
| * |
| * @param id The FlexClient ID value from the client. |
| * |
| * @return The FlexClient or null if the provided ID was <code>null</code>. |
| */ |
| public FlexClient setupFlexClient(String id) |
| { |
| FlexClient flexClient = null; |
| if (id != null) |
| { |
| // This indicates that we're dealing with a non-legacy client that hasn't been |
| // assigned a FlexClient Id yet. Reset to null to generate a fresh Id. |
| if (id.equals(FlexClient.NULL_FLEXCLIENT_ID)) |
| id = null; |
| |
| flexClient = getMessageBroker().getFlexClientManager().getFlexClient(id); |
| // Make sure the FlexClient and FlexSession are associated. |
| FlexSession session = FlexContext.getFlexSession(); |
| flexClient.registerFlexSession(session); |
| // And place the FlexClient in FlexContext for this request. |
| FlexContext.setThreadLocalFlexClient(flexClient); |
| } |
| return flexClient; |
| } |
| |
| /** |
| * |
| * Performance metrics gathering property |
| */ |
| public boolean isRecordMessageSizes() |
| { |
| return recordMessageSizes; |
| } |
| |
| /** |
| * |
| * Performance metrics gathering property |
| */ |
| public boolean isRecordMessageTimes() |
| { |
| return recordMessageTimes; |
| } |
| |
| /** |
| * |
| */ |
| public void setThreadLocals() |
| { |
| if (serializationContext != null) |
| { |
| SerializationContext context = (SerializationContext)serializationContext.clone(); |
| // Get the latest deserialization validator from the broker. |
| MessageBroker broker = getMessageBroker(); |
| DeserializationValidator validator = broker == null? null : broker.getDeserializationValidator(); |
| context.setDeserializationValidator(validator); |
| SerializationContext.setSerializationContext(context); |
| } |
| |
| TypeMarshallingContext.setTypeMarshaller(getTypeMarshaller()); |
| } |
| |
| /** |
| * |
| */ |
| public void clearThreadLocals() |
| { |
| SerializationContext.clearThreadLocalObjects(); |
| TypeMarshallingContext.clearThreadLocalObjects(); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Protected/private methods. |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Returns the log category of the <code>AbstractEndpoint</code>. Subclasses |
| * can override to provide a more specific logging category. |
| * |
| * @return The log category. |
| */ |
| @Override |
| protected String getLogCategory() |
| { |
| return LOG_CATEGORY; |
| } |
| |
| /** |
| * Hook method invoked when a disconnect command is received from a client channel. |
| * The response returned by this method is not guaranteed to get to the client, which |
| * is free to terminate its physical connection at any point. |
| * |
| * @param disconnectCommand The disconnect command. |
| * @return The response; by default an empty <tt>AcknowledgeMessage</tt>. |
| */ |
| protected Message handleChannelDisconnect(CommandMessage disconnectCommand) |
| { |
| return new AcknowledgeMessage(); |
| } |
| |
| /** |
| * Hook method for varying poll reply strategies for synchronous endpoints. |
| * The default behavior performs a non-waited, synchronous poll for the FlexClient |
| * and if any messages are currently queued they are returned immediately. If no |
| * messages are queued an empty response is returned immediately. |
| * |
| * @param flexClient The FlexClient that issued the poll request. |
| * @param pollCommand The poll command from the client. |
| * @return The FlushResult response. |
| */ |
| protected FlushResult handleFlexClientPoll(FlexClient flexClient, CommandMessage pollCommand) |
| { |
| return flexClient.poll(getId()); |
| } |
| |
| /** |
| * Handles a general poll request from a FlexClient to this endpoint. |
| * Subclasses may override to implement different poll handling strategies. |
| * |
| * @param flexClient The FlexClient that issued the poll request. |
| * @param pollCommand The poll command from the client. |
| * @return The poll response message; either for success or fault. |
| */ |
| protected Message handleFlexClientPollCommand(FlexClient flexClient, CommandMessage pollCommand) |
| { |
| if (Log.isDebug()) |
| Log.getLogger(getMessageBroker().getLogCategory(pollCommand)).debug( |
| "Before handling general client poll request. " + StringUtils.NEWLINE + |
| " incomingMessage: " + pollCommand + StringUtils.NEWLINE); |
| |
| FlushResult flushResult = handleFlexClientPoll(flexClient, pollCommand); |
| Message pollResponse = null; |
| |
| // Generate a no-op poll response if necessary; prevents a single client from busy polling when the server |
| // is doing wait()-based long-polls. |
| if ((flushResult instanceof PollFlushResult) && ((PollFlushResult)flushResult).isClientProcessingSuppressed()) |
| { |
| pollResponse = new CommandMessage(CommandMessage.CLIENT_SYNC_OPERATION); |
| pollResponse.setHeader(CommandMessage.NO_OP_POLL_HEADER, Boolean.TRUE); |
| } |
| |
| if (pollResponse == null) |
| { |
| List<Message> messagesToReturn = (flushResult != null) ? flushResult.getMessages() : null; |
| if (messagesToReturn != null && !messagesToReturn.isEmpty()) |
| { |
| pollResponse = new CommandMessage(CommandMessage.CLIENT_SYNC_OPERATION); |
| pollResponse.setBody(messagesToReturn.toArray()); |
| } |
| else |
| { |
| pollResponse = new AcknowledgeMessage(); |
| } |
| } |
| |
| // Set the adaptive poll wait time if necessary. |
| if (flushResult != null) |
| { |
| int nextFlushWaitTime = flushResult.getNextFlushWaitTimeMillis(); |
| if (nextFlushWaitTime > 0) |
| pollResponse.setHeader(CommandMessage.POLL_WAIT_HEADER, new Integer(nextFlushWaitTime)); |
| } |
| |
| if (Log.isDebug()) |
| { |
| String debugPollResult = Log.getPrettyPrinter().prettify(pollResponse); |
| Log.getLogger(getMessageBroker().getLogCategory(pollCommand)).debug( |
| "After handling general client poll request. " + StringUtils.NEWLINE + |
| " reply: " + debugPollResult + StringUtils.NEWLINE); |
| } |
| |
| return pollResponse; |
| } |
| |
| /** |
| * Initializes the <code>Endpoint</code> with the client-load-balancing urls. |
| * |
| * @param id Id of the <code>Endpoint</code>. |
| * @param properties Properties for the <code>Endpoint</code>. |
| */ |
| |
| protected void initializeClientLoadBalancing(String id, ConfigMap properties) |
| { |
| if (!properties.containsKey(CLIENT_LOAD_BALANCING_ELEMENT)) |
| return; |
| |
| ConfigMap clientLoadBalancing = properties.getPropertyAsMap(CLIENT_LOAD_BALANCING_ELEMENT, null); |
| if (clientLoadBalancing == null) |
| { |
| // Invalid {0} configuration for endpoint ''{1}''; no urls defined. |
| ConfigurationException ce = new ConfigurationException(); |
| ce.setMessage(ERR_MSG_EMPTY_CLIENT_LOAD_BALANCING_ELEMENT, new Object[]{CLIENT_LOAD_BALANCING_ELEMENT, getId()}); |
| throw ce; |
| } |
| |
| @SuppressWarnings("unchecked") |
| List<String> urls = clientLoadBalancing.getPropertyAsList(URL_ATTR, null); |
| if (urls == null || urls.isEmpty()) |
| { |
| // Invalid {0} configuration for endpoint ''{1}''; no urls defined. |
| ConfigurationException ce = new ConfigurationException(); |
| ce.setMessage(ERR_MSG_EMPTY_CLIENT_LOAD_BALANCING_ELEMENT, new Object[]{CLIENT_LOAD_BALANCING_ELEMENT, getId()}); |
| throw ce; |
| } |
| |
| for (Iterator<String> iterator = urls.iterator(); iterator.hasNext();) |
| { |
| String url = iterator.next(); |
| if (!addClientLoadBalancingUrl(url) && Log.isWarn()) |
| log.warn("Endpoint '{0}' is ignoring the url '{1}' as it's already in the set of client-load-balancing urls.", new Object[]{id, url}); |
| } |
| } |
| |
| protected void checkSecurityConstraint(Message message) |
| { |
| if (securityConstraint != null) |
| { |
| getMessageBroker().getLoginManager().checkConstraint(securityConstraint); |
| } |
| } |
| |
| /** |
| * Returns the deserializer class name used by the endpoint. |
| * |
| * @return The deserializer class name used by the endpoint. |
| */ |
| protected abstract String getDeserializerClassName(); |
| |
| /** |
| * Returns the serializer class name used by the endpoint. |
| * |
| * @return The serializer class name used by the endpoint. |
| */ |
| protected abstract String getSerializerClassName(); |
| |
| /** |
| * Returns the secure protocol scheme for the endpoint. |
| * |
| * @return The secure protocol scheme for the endpoint. |
| */ |
| protected abstract String getSecureProtocolScheme(); |
| |
| /** |
| * Returns the insecure protocol scheme for the endpoint. |
| * |
| * @return The insecure protocol scheme for the endpoint. |
| */ |
| protected abstract String getInsecureProtocolScheme(); |
| |
| /** |
| * Invoked automatically to allow the <code>AbstractEndpoint</code> to setup |
| * its corresponding MBean control. Subclasses should override to setup and |
| * register their MBean control. Manageable subclasses should override this |
| * template method. |
| * |
| * @param broker The <code>MessageBroker</code> that manages this |
| * <code>AbstractEndpoint</code>. |
| */ |
| protected abstract void setupEndpointControl(MessageBroker broker); |
| |
| /** |
| * Validates the endpoint url scheme. |
| */ |
| protected void validateEndpointProtocol() |
| { |
| String scheme = isSecure()? getSecureProtocolScheme() : getInsecureProtocolScheme(); |
| if (!url.startsWith(scheme)) |
| { |
| ConfigurationException ce = new ConfigurationException(); |
| ce.setMessage(ERR_MSG_INVALID_URL_SCHEME, new Object[] {url, scheme}); |
| throw ce; |
| } |
| } |
| |
| protected void validateRequestProtocol(HttpServletRequest req) |
| { |
| // Secure url can talk to secure or non-secure endpoint. |
| // Non-secure url can only talk to non-secure endpoint. |
| boolean secure = req.isSecure(); |
| if (!secure && isSecure()) |
| { |
| // Secure endpoints must be contacted via a secure protocol. |
| String endpointPath = req.getServletPath() + req.getPathInfo(); |
| SecurityException se = new SecurityException(); |
| se.setMessage(NONSECURE_PROTOCOL, new Object[]{endpointPath}); |
| throw se; |
| } |
| } |
| |
| /** |
| * |
| * Verifies that the remote client supports the FlexClient API. |
| * Legacy clients that do not support this receive a message fault for any messages they send. |
| * |
| * @param message The message to verify. |
| */ |
| protected void verifyFlexClientSupport(Message message) |
| { |
| if (FlexContext.getFlexClient() == null) |
| { |
| MessageException me = new MessageException(); |
| me.setMessage(REQUIRES_FLEXCLIENT_SUPPORT, new Object[] {message.getDestination()}); |
| throw me; |
| } |
| } |
| |
| /** |
| * |
| */ |
| protected Class<?> createClass(String className) |
| { |
| return ClassUtil.createClass(className, FlexContext.getMessageBroker() == null ? null : |
| FlexContext.getMessageBroker().getClassLoader()); |
| } |
| |
| // This should match with ChannelSetting.parseClientUri |
| private void parseClientUrl(String contextPath) |
| { |
| if (!clientContextParsed) |
| { |
| String channelEndpoint = url.trim(); |
| |
| // either {context-root} or {context.root} is legal |
| channelEndpoint = StringUtils.substitute(channelEndpoint, "{context-root}", ConfigurationConstants.CONTEXT_PATH_TOKEN); |
| |
| if ((contextPath == null) && (channelEndpoint.indexOf(ConfigurationConstants.CONTEXT_PATH_TOKEN) != -1)) |
| { |
| // context root must be specified before it is used |
| ConfigurationException e = new ConfigurationException(); |
| e.setMessage(ConfigurationConstants.UNDEFINED_CONTEXT_ROOT, new Object[]{getId()}); |
| throw e; |
| } |
| |
| // simplify the number of combinations to test by ensuring our |
| // context path always starts with a slash |
| if (contextPath != null && !contextPath.startsWith("/")) |
| { |
| contextPath = "/" + contextPath; |
| } |
| |
| // avoid double-slashes from context root by replacing /{context.root} |
| // in a single replacement step |
| if (channelEndpoint.indexOf(ConfigurationConstants.SLASH_CONTEXT_PATH_TOKEN) != -1) |
| { |
| // but avoid double-slash for /{context.root}/etc when we have |
| // the default context root |
| if ("/".equals(contextPath) && !ConfigurationConstants.SLASH_CONTEXT_PATH_TOKEN.equals(channelEndpoint)) |
| contextPath = ""; |
| |
| channelEndpoint = StringUtils.substitute(channelEndpoint, ConfigurationConstants.SLASH_CONTEXT_PATH_TOKEN, contextPath); |
| } |
| // otherwise we have something like {server.name}:{server.port}{context.root}... |
| else |
| { |
| // but avoid double-slash for {context.root}/etc when we have |
| // the default context root |
| if ("/".equals(contextPath) && !ConfigurationConstants.CONTEXT_PATH_TOKEN.equals(channelEndpoint)) |
| contextPath = ""; |
| |
| channelEndpoint = StringUtils.substitute(channelEndpoint, ConfigurationConstants.CONTEXT_PATH_TOKEN, contextPath); |
| } |
| |
| parsedClientUrl = channelEndpoint; |
| clientContextParsed = true; |
| } |
| } |
| |
| private int internalParsePort(String url) |
| { |
| int port = ChannelSettings.parsePort(url); |
| // If there is no specified port, log an info message as urls without ports are supported |
| if (port == 0 && Log.isInfo()) |
| log.info("No port specified in channel URL: {0}", new Object[]{url}); |
| |
| return port == -1? 0 : port; // Replace -1 with 0. |
| } |
| |
| private void parseUrl(String contextPath) |
| { |
| // Parse again only if never parsed before or parsed for a different contextPath. |
| if (parsedForContext == null || !parsedForContext.equals(contextPath)) |
| { |
| String channelEndpoint = url.toLowerCase().trim(); |
| |
| // Remove protocol and host info |
| String insecureProtocol = getInsecureProtocolScheme() + "://"; |
| String secureProtocol = getSecureProtocolScheme() + "://"; |
| if (channelEndpoint.startsWith(secureProtocol) || channelEndpoint.startsWith(insecureProtocol)) |
| { |
| int nextSlash = channelEndpoint.indexOf('/', 8); |
| if (nextSlash > 0) |
| { |
| channelEndpoint = channelEndpoint.substring(nextSlash); |
| } |
| } |
| |
| // either {context-root} or {context.root} is legal |
| channelEndpoint = StringUtils.substitute(channelEndpoint, "{context-root}", ConfigurationConstants.CONTEXT_PATH_TOKEN); |
| |
| // Remove context path info |
| if (channelEndpoint.startsWith(ConfigurationConstants.CONTEXT_PATH_TOKEN)) |
| { |
| channelEndpoint = channelEndpoint.substring(ConfigurationConstants.CONTEXT_PATH_TOKEN.length()); |
| } |
| else if (channelEndpoint.startsWith(ConfigurationConstants.SLASH_CONTEXT_PATH_TOKEN)) |
| { |
| channelEndpoint = channelEndpoint.substring(ConfigurationConstants.SLASH_CONTEXT_PATH_TOKEN.length()); |
| } |
| else if (contextPath.length() > 0) |
| { |
| if (channelEndpoint.startsWith(contextPath.toLowerCase())) |
| { |
| channelEndpoint = channelEndpoint.substring(contextPath.length()); |
| } |
| } |
| |
| // We also don't match on trailing slashes |
| if (channelEndpoint.endsWith("/")) |
| { |
| channelEndpoint = channelEndpoint.substring(0, channelEndpoint.length() - 1); |
| } |
| |
| parsedUrl = channelEndpoint; |
| parsedForContext = contextPath; |
| } |
| } |
| } |
