| /* |
| * 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 flex.messaging.FlexContext; |
| import flex.messaging.FlexSession; |
| import flex.messaging.client.FlexClient; |
| import flex.messaging.client.FlushResult; |
| import flex.messaging.client.PollFlushResult; |
| import flex.messaging.client.PollWaitListener; |
| import flex.messaging.client.UserAgentSettings; |
| import flex.messaging.config.ConfigMap; |
| import flex.messaging.config.ConfigurationConstants; |
| import flex.messaging.log.Log; |
| import flex.messaging.messages.CommandMessage; |
| import flex.messaging.util.UserAgentManager; |
| |
| import java.util.concurrent.ConcurrentHashMap; |
| |
| /** |
| * Base class for HTTP-based endpoints that support regular polling and long polling, |
| * which means placing request threads that are polling for messages into a wait |
| * state until messages are available for delivery or the configurable wait interval |
| * is reached. |
| */ |
| public abstract class BasePollingHTTPEndpoint extends BaseHTTPEndpoint implements PollWaitListener |
| { |
| //-------------------------------------------------------------------------- |
| // |
| // Private Static Constants |
| // |
| //-------------------------------------------------------------------------- |
| |
| private static final String POLLING_ENABLED = "polling-enabled"; |
| private static final String POLLING_INTERVAL_MILLIS = "polling-interval-millis"; |
| private static final String POLLING_INTERVAL_SECONDS = "polling-interval-seconds"; // Deprecated configuration option. |
| private static final String MAX_WAITING_POLL_REQUESTS = "max-waiting-poll-requests"; |
| private static final String WAIT_INTERVAL_MILLIS = "wait-interval-millis"; |
| private static final String CLIENT_WAIT_INTERVAL_MILLIS = "client-wait-interval-millis"; |
| // Force clients that exceed the long-poll limit to wait at least this long between poll requests. |
| // This matches the default polling interval defined in the client PollingChannel. |
| private static final int DEFAULT_WAIT_FOR_EXCESS_POLL_WAIT_CLIENTS = 3000; |
| |
| // User Agent based settings manager |
| private UserAgentManager userAgentManager = new UserAgentManager(); |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Constructor |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Constructs an unmanaged <code>BasePollingHTTPEndpoint</code>. |
| */ |
| public BasePollingHTTPEndpoint() |
| { |
| this(false); |
| } |
| |
| /** |
| * Constructs an <code>BasePollingHTTPEndpoint</code> with the indicated management. |
| * |
| * @param enableManagement <code>true</code> if the <code>BasePollingHTTPEndpoint</code> |
| * is manageable; <code>false</code> otherwise. |
| */ |
| public BasePollingHTTPEndpoint(boolean enableManagement) |
| { |
| super(enableManagement); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // 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) |
| { |
| // Initialize default user agent manager settings. |
| UserAgentManager.setupUserAgentManager(null, userAgentManager); |
| |
| return; // Nothing else to initialize. |
| } |
| |
| // General poll props. |
| pollingEnabled = properties.getPropertyAsBoolean(POLLING_ENABLED, false); |
| pollingIntervalMillis = properties.getPropertyAsLong(POLLING_INTERVAL_MILLIS, -1); |
| long pollingIntervalSeconds = properties.getPropertyAsLong(POLLING_INTERVAL_SECONDS, -1); // Deprecated |
| if (pollingIntervalSeconds > -1) |
| pollingIntervalMillis = pollingIntervalSeconds * 1000; |
| |
| // Piggybacking props. |
| piggybackingEnabled = properties.getPropertyAsBoolean(ConfigurationConstants.PIGGYBACKING_ENABLED_ELEMENT, false); |
| |
| // HTTP poll wait props. |
| maxWaitingPollRequests = properties.getPropertyAsInt(MAX_WAITING_POLL_REQUESTS, 0); |
| waitInterval = properties.getPropertyAsLong(WAIT_INTERVAL_MILLIS, 0); |
| clientWaitInterval = properties.getPropertyAsInt(CLIENT_WAIT_INTERVAL_MILLIS, 0); |
| |
| // User Agent props. |
| UserAgentManager.setupUserAgentManager(properties, userAgentManager); |
| |
| // Set initial state for the canWait flag based on whether we allow waits or not. |
| if (maxWaitingPollRequests > 0 && (waitInterval == -1 || waitInterval > 0)) |
| { |
| waitEnabled = true; |
| canWait = true; |
| } |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Variables |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * This flag is volatile to allow for consistent reads across thread without |
| * needing to pay the cost for a synchronized lock for each read. |
| */ |
| private volatile boolean canWait; |
| |
| /** |
| * Used to synchronize sets and gets to the number of waiting clients. |
| */ |
| protected final Object lock = new Object(); |
| |
| /** |
| * Set when properties are handled; used as a shortcut for logging to determine whether this |
| * instance attempts to put request threads in a wait state or not. |
| */ |
| private boolean waitEnabled; |
| |
| /** |
| * A count of the number of request threads that are currently in the wait state (including |
| * those on their way into or out of it). |
| */ |
| protected int waitingPollRequestsCount; |
| |
| /** |
| * A Map(notification Object for a waited request thread, Boolean.TRUE). |
| */ |
| private ConcurrentHashMap currentWaitedRequests; |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Properties |
| // |
| //-------------------------------------------------------------------------- |
| |
| //---------------------------------- |
| // clientWaitInterval |
| //---------------------------------- |
| |
| protected int clientWaitInterval = 0; |
| |
| /** |
| * Retrieves the number of milliseconds the client will wait after receiving a response for |
| * a poll with server wait before it issues its next poll request. |
| * A value of zero or less causes the client to use its default polling interval (based on the |
| * channel's polling-interval-millis configuration) and this value is ignored. |
| * A value greater than zero will cause the client to wait for the specified interval before |
| * issuing its next poll request with a value of 1 triggering an immediate poll from the client |
| * as soon as a waited poll response is received. |
| * @return The client wait interval. |
| */ |
| public int getClientWaitInterval() |
| { |
| return clientWaitInterval; |
| } |
| |
| /** |
| * Sets the number of milliseconds a client will wait after receiving a response for a poll |
| * with server wait before it issues its next poll request. |
| * A value of zero or less causes the client to use its default polling interval (based on the |
| * channel's polling-interval-millis configuration) and this value is ignored. |
| * A value greater than zero will cause the client to wait for the specified interval before |
| * issuing its next poll request with a value of 1 triggering an immediate poll from the client |
| * as soon as a waited poll response is received. |
| * This property does not effect polling clients that poll the server without a server wait. |
| * |
| * @param value The number of milliseconds a client will wait before issuing its next poll when the |
| * server is configured to wait. |
| */ |
| public void setClientWaitInterval(int value) |
| { |
| clientWaitInterval = value; |
| } |
| |
| //---------------------------------- |
| // maxWaitingPollRequests |
| //---------------------------------- |
| |
| protected int maxWaitingPollRequests = 0; |
| |
| /** |
| * Retrieves the maximum number of server poll response threads that will be |
| * waiting for messages to arrive for clients. |
| * @return The maximum number of waiting poll requests. |
| */ |
| public int getMaxWaitingPollRequests() |
| { |
| return maxWaitingPollRequests; |
| } |
| |
| /** |
| * Sets the maximum number of server poll response threads that will be |
| * waiting for messages to arrive for clients. |
| * |
| * @param maxWaitingPollRequests The maximum number of server poll response threads |
| * that will be waiting for messages to arrive for the client. |
| */ |
| public void setMaxWaitingPollRequests(int maxWaitingPollRequests) |
| { |
| this.maxWaitingPollRequests = maxWaitingPollRequests; |
| if (maxWaitingPollRequests > 0 && (waitInterval == -1 || waitInterval > 0)) |
| { |
| waitEnabled = true; |
| canWait = (waitingPollRequestsCount < maxWaitingPollRequests); |
| } |
| } |
| |
| //---------------------------------- |
| // pollingEnabled |
| //---------------------------------- |
| |
| /** |
| * |
| * This is a property used on the client. |
| */ |
| protected boolean piggybackingEnabled; |
| |
| //---------------------------------- |
| // pollingEnabled |
| //---------------------------------- |
| |
| /** |
| * |
| * This is a property used on the client. |
| */ |
| protected boolean pollingEnabled; |
| |
| //---------------------------------- |
| // pollingIntervalMillis |
| //---------------------------------- |
| |
| /** |
| * |
| * This is a property used on the client. |
| */ |
| protected long pollingIntervalMillis = -1; |
| |
| //---------------------------------- |
| // waitInterval |
| //---------------------------------- |
| |
| protected long waitInterval = 0; |
| |
| /** |
| * Retrieves the number of milliseconds the server poll response thread will be |
| * waiting for messages to arrive for the client. |
| * @return The wait interval. |
| */ |
| public long getWaitInterval() |
| { |
| return waitInterval; |
| } |
| |
| /** |
| * Sets the number of milliseconds the server poll response thread will be |
| * waiting for messages to arrive for the client. |
| * |
| * @param waitInterval The number of milliseconds the server poll response thread will be |
| * waiting for messages to arrive for the client. |
| */ |
| public void setWaitInterval(long waitInterval) |
| { |
| this.waitInterval = waitInterval; |
| if (maxWaitingPollRequests > 0 && (waitInterval == -1 || waitInterval > 0)) |
| { |
| waitEnabled = true; |
| canWait = (waitingPollRequestsCount < maxWaitingPollRequests); |
| } |
| } |
| |
| //---------------------------------- |
| // waitingPollRequestsCount |
| //---------------------------------- |
| |
| /** |
| * Retrieves the count of the number of request threads that are currently in the wait state |
| * (including those on their way into or out of it). |
| * |
| * @return The count of the number of request threads that are currently in the wait state. |
| */ |
| public int getWaitingPollRequestsCount() |
| { |
| return waitingPollRequestsCount; |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Public Methods |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * |
| * Returns a <code>ConfigMap</code> of endpoint properties that the client |
| * needs. This includes properties from <code>super.describeEndpoint</code> |
| * and additional <code>BaseHTTPEndpoint</code> specific properties under |
| * "properties" key. |
| */ |
| @Override |
| public ConfigMap describeEndpoint() |
| { |
| ConfigMap endpointConfig = super.describeEndpoint(); |
| |
| boolean createdProperties = false; |
| ConfigMap properties = endpointConfig.getPropertyAsMap(PROPERTIES_ELEMENT, null); |
| if (properties == null) |
| { |
| properties = new ConfigMap(); |
| createdProperties = true; |
| } |
| |
| if (pollingEnabled) |
| { |
| ConfigMap pollingEnabled = new ConfigMap(); |
| // Adding as a value rather than attribute to the parent |
| pollingEnabled.addProperty(EMPTY_STRING, TRUE_STRING); |
| properties.addProperty(POLLING_ENABLED, pollingEnabled); |
| } |
| |
| if (pollingIntervalMillis > -1) |
| { |
| ConfigMap pollingInterval = new ConfigMap(); |
| // Adding as a value rather than attribute to the parent |
| pollingInterval.addProperty(EMPTY_STRING, String.valueOf(pollingIntervalMillis)); |
| properties.addProperty(POLLING_INTERVAL_MILLIS, pollingInterval); |
| } |
| |
| if (piggybackingEnabled) |
| { |
| ConfigMap piggybackingEnabled = new ConfigMap(); |
| // Adding as a value rather than attribute to the parent |
| piggybackingEnabled.addProperty(EMPTY_STRING, String.valueOf(piggybackingEnabled)); |
| properties.addProperty(ConfigurationConstants.PIGGYBACKING_ENABLED_ELEMENT, piggybackingEnabled); |
| } |
| |
| if (createdProperties && properties.size() > 0) |
| endpointConfig.addProperty(ConfigurationConstants.PROPERTIES_ELEMENT, properties); |
| |
| return endpointConfig; |
| } |
| |
| /** |
| * Sets up monitoring of waited poll requests so they can be notified and exit when the |
| * endpoint stops. |
| * |
| * @see flex.messaging.endpoints.AbstractEndpoint#start() |
| */ |
| @Override |
| public void start() |
| { |
| if (isStarted()) |
| return; |
| |
| super.start(); |
| |
| currentWaitedRequests = new ConcurrentHashMap(); |
| } |
| |
| /** |
| * Ensures that no poll requests in a wait state are left un-notified when the endpoint stops. |
| * |
| * @see flex.messaging.endpoints.AbstractEndpoint#stop() |
| */ |
| @Override |
| public void stop() |
| { |
| if (!isStarted()) |
| return; |
| |
| // Notify any currently waiting polls. |
| for (Object notifier : currentWaitedRequests.keySet()) |
| { |
| synchronized (notifier) |
| { |
| notifier.notifyAll(); // Break any current waits. |
| } |
| } |
| currentWaitedRequests = null; |
| |
| super.stop(); |
| } |
| |
| /** |
| * (non-Javaodc) |
| * @see flex.messaging.client.PollWaitListener#waitStart(Object) |
| */ |
| public void waitStart(Object notifier) |
| { |
| currentWaitedRequests.put(notifier, Boolean.TRUE); |
| } |
| |
| /** |
| * (non-Javaodc) |
| * @see flex.messaging.client.PollWaitListener#waitEnd(Object) |
| */ |
| public void waitEnd(Object notifier) |
| { |
| if (currentWaitedRequests != null) |
| currentWaitedRequests.remove(notifier); |
| } |
| |
| //-------------------------------------------------------------------------- |
| // |
| // Protected Methods |
| // |
| //-------------------------------------------------------------------------- |
| |
| /** |
| * Overrides the base poll handling to support optionally putting Http request handling threads |
| * into a wait state until messages are available to be delivered in the poll response or a timeout is reached. |
| * The number of threads that may be put in a wait state is bounded by <code>max-waiting-poll-requests</code> |
| * and waits will only be attempted if the canWait flag that is based on the <code>max-waiting-poll-requests</code> |
| * and the specified <code>wait-interval</code> is true. |
| * |
| * @param flexClient The FlexClient that issued the poll request. |
| * @param pollCommand The poll command from the client. |
| * @return The flush info used to build the poll response. |
| */ |
| @Override |
| protected FlushResult handleFlexClientPoll(FlexClient flexClient, CommandMessage pollCommand) |
| { |
| FlushResult flushResult = null; |
| if (canWait && !pollCommand.headerExists(CommandMessage.SUPPRESS_POLL_WAIT_HEADER)) |
| { |
| FlexSession session = FlexContext.getFlexSession(); |
| // If canWait is true it means we currently have less than the max number of allowed waiting threads. |
| |
| // We need to protect writes/reads to the wait count with the endpoint's lock. |
| // Also, we have to be careful to handle the case where two threads get to this point when only |
| // one wait spot remains; one thread will win and the other needs to revert to a non-waitable poll. |
| boolean thisThreadCanWait; |
| synchronized (lock) |
| { |
| ++waitingPollRequestsCount; |
| if (waitingPollRequestsCount == maxWaitingPollRequests) |
| { |
| thisThreadCanWait = true; // This thread got the last wait spot. |
| canWait = false; |
| } |
| else if (waitingPollRequestsCount > maxWaitingPollRequests) |
| { |
| thisThreadCanWait = false; // This thread was beaten out for the last spot. |
| --waitingPollRequestsCount; // Decrement the count because we're not going to try a poll with wait. |
| canWait = false; // All the wait spots are currently occupied so prevent further attempts for now. |
| } |
| else |
| { |
| // We haven't hit the limit yet, allow this thread to wait. |
| thisThreadCanWait = true; |
| } |
| } |
| |
| // Check the max waiting connections per session count |
| if (thisThreadCanWait) |
| { |
| String userAgentValue = FlexContext.getHttpRequest().getHeader(UserAgentManager.USER_AGENT_HEADER_NAME); |
| UserAgentSettings agentSettings = userAgentManager.match(userAgentValue); |
| synchronized(session) |
| { |
| if (agentSettings != null) |
| session.maxConnectionsPerSession = agentSettings.getMaxPersistentConnectionsPerSession(); |
| |
| ++session.streamingConnectionsCount; |
| if (session.maxConnectionsPerSession == FlexSession.MAX_CONNECTIONS_PER_SESSION_UNLIMITED |
| || session.streamingConnectionsCount <= session.maxConnectionsPerSession) |
| { |
| thisThreadCanWait = true; // We haven't hit the limit yet, allow the wait. |
| } |
| else // (session.streamingConnectionsCount > session.maxConnectionsPerSession) |
| { |
| thisThreadCanWait = false; // no more from this client |
| --session.streamingConnectionsCount; |
| } |
| } |
| |
| if (!thisThreadCanWait) |
| { |
| // Decrement the waiting poll count, since this poll isn't going to wait. |
| synchronized (lock) |
| { |
| --waitingPollRequestsCount; |
| if (waitingPollRequestsCount < maxWaitingPollRequests) |
| canWait = true; |
| } |
| if (Log.isDebug()) |
| { |
| log.debug("Max long-polling requests per session limit (" + session.maxConnectionsPerSession + ") has been reached, this poll won't wait."); |
| } |
| } |
| |
| } |
| |
| if (thisThreadCanWait) |
| { |
| if (Log.isDebug()) |
| log.debug("Number of waiting threads for endpoint with id '"+ getId() +"' is " + waitingPollRequestsCount + "."); |
| |
| try |
| { |
| flushResult = flexClient.pollWithWait(getId(), FlexContext.getFlexSession(), this, waitInterval); |
| if (flushResult != null) |
| { |
| // Prevent busy-polling due to multiple clients sharing a session and swapping each other out too quickly. |
| if ((flushResult instanceof PollFlushResult) && ((PollFlushResult)flushResult).isAvoidBusyPolling() && (flushResult.getNextFlushWaitTimeMillis() < DEFAULT_WAIT_FOR_EXCESS_POLL_WAIT_CLIENTS)) |
| { |
| // Force the client polling interval to match the default defined in the client PollingChannel. |
| flushResult.setNextFlushWaitTimeMillis(DEFAULT_WAIT_FOR_EXCESS_POLL_WAIT_CLIENTS); |
| } |
| else if ((clientWaitInterval > 0) && (flushResult.getNextFlushWaitTimeMillis() == 0)) |
| { |
| // If the FlushResult doesn't specify it's own flush wait time, use the configured clientWaitInterval if defined. |
| flushResult.setNextFlushWaitTimeMillis(clientWaitInterval); |
| } |
| } |
| } |
| finally |
| { |
| // We're done waiting so decrement the count of waiting threads and update the canWait flag if necessary |
| synchronized (lock) |
| { |
| --waitingPollRequestsCount; |
| if (waitingPollRequestsCount < maxWaitingPollRequests) |
| canWait = true; |
| } |
| synchronized (session) |
| { |
| --session.streamingConnectionsCount; |
| } |
| |
| if (Log.isDebug()) |
| log.debug("Number of waiting threads for endpoint with id '"+ getId() +"' is " + waitingPollRequestsCount + "."); |
| } |
| } |
| } |
| else if (Log.isDebug() && waitEnabled) |
| { |
| if (pollCommand.headerExists(CommandMessage.SUPPRESS_POLL_WAIT_HEADER)) |
| log.debug("Suppressing poll wait for this request because it is part of a batch of messages to process."); |
| else |
| log.debug("Max waiting poll requests limit '" + maxWaitingPollRequests + "' has been reached for endpoint '" + getId() + "'. FlexClient with id '"+ flexClient.getId() + "' will poll with no wait."); |
| } |
| |
| // If we weren't able to do a poll with wait above for any reason just run the base poll handling logic. |
| if (flushResult == null) |
| { |
| flushResult = super.handleFlexClientPoll(flexClient, pollCommand); |
| // If this is an excess poll request that we couldn't wait on, make sure the client doesn't poll the endpoint too aggressively. |
| // In this case, force a client wait to match the default polling interval defined in the client PollingChannel. |
| if ( waitEnabled && (pollingIntervalMillis < DEFAULT_WAIT_FOR_EXCESS_POLL_WAIT_CLIENTS)) |
| { |
| if (flushResult == null) |
| { |
| flushResult = new FlushResult(); |
| } |
| flushResult.setNextFlushWaitTimeMillis(DEFAULT_WAIT_FOR_EXCESS_POLL_WAIT_CLIENTS); |
| } |
| } |
| |
| return flushResult; |
| } |
| } |
