flex-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aha...@apache.org
Subject [03/51] [partial] BlazeDS Donation from Adobe Systems Inc
Date Fri, 25 Apr 2014 05:34:01 GMT
http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/FlexClientAttributeListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/FlexClientAttributeListener.java b/modules/core/src/flex/messaging/client/FlexClientAttributeListener.java
new file mode 100755
index 0000000..9481bc2
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/FlexClientAttributeListener.java
@@ -0,0 +1,47 @@
+/*
+ * 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.client;
+
+/**
+ * Interface for FlexClient attribute listeners.
+ */
+public interface FlexClientAttributeListener 
+{
+    /**
+     * Callback invoked after an attribute is added to the FlexClient.
+     * 
+     * @param event The event containing the associated FlexClient and attribute
+     *              information.
+     */
+    void attributeAdded(FlexClientBindingEvent event);
+    
+    /**
+     * Callback invoked after an attribute is removed from the FlexClient.
+     * 
+     * @param event The event containing the associated FlexClient and attribute
+     *              information.
+     */
+    void attributeReplaced(FlexClientBindingEvent event);
+    
+    /**
+     * Callback invoked after an attribute has been replaced with a new value.
+     * 
+     * @param event The event containing the associated FlexClient and attribute
+     *              information.
+     */
+    void attributeRemoved(FlexClientBindingEvent event);
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/FlexClientBindingEvent.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/FlexClientBindingEvent.java b/modules/core/src/flex/messaging/client/FlexClientBindingEvent.java
new file mode 100755
index 0000000..236f70a
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/FlexClientBindingEvent.java
@@ -0,0 +1,115 @@
+/*
+ * 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.client;
+
+/**
+ * Event used to notify FlexClientAttributeListeners of changes to FlexClient
+ * attributes.
+ */
+public class FlexClientBindingEvent 
+{
+    //--------------------------------------------------------------------------
+    //
+    // Constructor
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * Constructs an event for an attribute that is bound or unbound from a FlexClient.
+     * 
+     * @param client The FlexClient.
+     * @param name The attribute name.
+     */
+    public FlexClientBindingEvent(FlexClient client, String name)
+    {
+        this.client = client;
+        this.name = name;
+    }    
+
+    
+    /**
+     * Constructs an event for an attribute that is added to a FlexClient or 
+     * replaced by a new value.
+     * 
+     * @param client The FlexClient.
+     * @param name The attribute name.
+     * @param value The attribute value.
+     */
+    public FlexClientBindingEvent(FlexClient client, String name, Object value)
+    {
+        this.client = client;
+        this.name = name;
+        this.value = value;
+    }
+
+    //--------------------------------------------------------------------------
+    //
+    // Variables
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * The FlexClient that generated the event.
+     */
+    private FlexClient client;
+    
+    /**
+     * The name of the attribute associated with the event.
+     */
+    private String name;
+    
+    /**
+     * The value of the attribute associated with the event.
+     */
+    private Object value;
+    
+    //--------------------------------------------------------------------------
+    //
+    // Methods
+    //
+    //--------------------------------------------------------------------------
+    
+    /**
+     * Returns the FlexClient that generated the event.
+     * 
+     * @return The FlexClient that generated the event.
+     */
+    public FlexClient getClient()
+    {
+        return client;
+    }
+    
+    /**
+     * Returns the name of the attribute associated with the event.
+     * 
+     * @return The name of the attribute associated with the event.
+     */
+    public String getName()
+    {
+        return name;
+    }
+    
+    /**
+     * Returns the value of the attribute associated with the event.
+     * 
+     * @return The value of the attribute associated with the event.
+     */
+    public Object getValue()
+    {
+        return value;
+    }
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/FlexClientBindingListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/FlexClientBindingListener.java b/modules/core/src/flex/messaging/client/FlexClientBindingListener.java
new file mode 100755
index 0000000..d1fb36b
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/FlexClientBindingListener.java
@@ -0,0 +1,39 @@
+/*
+ * 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.client;
+
+/**
+ * Interface to be notified when the implementing object is bound or unbound from the FlexClient.
+ */
+public interface FlexClientBindingListener 
+{
+    /**
+     * Callback invoked when the object is bound to a FlexClient.
+     * 
+     * @param event The event containing the FlexClient and attribute
+     *              information.
+     */
+    void valueBound(FlexClientBindingEvent event);
+    
+    /**
+     * Callback invoked when the object is unbound from a FlexClient.
+     * 
+     * @param event The event containing the FlexClient and attribute
+     *              information.
+     */
+    void valueUnbound(FlexClientBindingEvent event);
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/FlexClientListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/FlexClientListener.java b/modules/core/src/flex/messaging/client/FlexClientListener.java
new file mode 100755
index 0000000..49dc24a
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/FlexClientListener.java
@@ -0,0 +1,40 @@
+/*
+ * 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.client;
+
+/**
+ * Interface to be notified when a FlexClient is created or destroyed. Implementations of this interface
+ * may add themselves as created listeners statically via <code>FlexClient.addClientCreatedListener()</code>.
+ * To listen for FlexClient destruction, the implementation instance must add itself as a listener to
+ * a specific FlexClient instance via the <code>addClientDestroyedListener()</code> method.
+ */
+public interface FlexClientListener 
+{
+    /**
+     * Notification that a FlexClient was created.
+     * 
+     * @param client The FlexClient that was created.
+     */
+    void clientCreated(FlexClient client);
+    
+    /**
+     * Notification that a FlexClient is about to be destroyed.
+     * 
+     * @param client The FlexClient that will be destroyed.
+     */
+    void clientDestroyed(FlexClient client);
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/FlexClientManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/FlexClientManager.java b/modules/core/src/flex/messaging/client/FlexClientManager.java
new file mode 100755
index 0000000..d1e7a1e
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/FlexClientManager.java
@@ -0,0 +1,519 @@
+/*
+ * 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.client;
+
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadFactory;
+
+import flex.management.ManageableComponent;
+import flex.management.runtime.messaging.client.FlexClientManagerControl;
+import flex.messaging.FlexContext;
+import flex.messaging.MessageBroker;
+import flex.messaging.MessageException;
+import flex.messaging.config.FlexClientSettings;
+import flex.messaging.endpoints.AbstractEndpoint;
+import flex.messaging.endpoints.Endpoint;
+import flex.messaging.log.Log;
+import flex.messaging.log.LogCategories;
+import flex.messaging.util.ClassUtil;
+import flex.messaging.util.TimeoutAbstractObject;
+import flex.messaging.util.TimeoutManager;
+
+/**
+ * @exclude
+ * Manages FlexClient instances for a MessageBroker.
+ */
+public class FlexClientManager extends ManageableComponent
+{
+    public static final String TYPE = "FlexClientManager";
+
+    //--------------------------------------------------------------------------
+    //
+    // Constructor
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * @exclude
+     */
+    public FlexClientManager()
+    {
+        this(MessageBroker.getMessageBroker(null));
+    }
+    /**
+     * Constructs a FlexClientManager for the passed MessageBroker.
+     *
+     * @param broker The MessageBroker that the Flex client manager is associated with.
+     */
+    public FlexClientManager(MessageBroker broker)
+    {
+        this(broker.isManaged(), broker);
+    }
+
+    /**
+     * @exclude
+     */
+    public FlexClientManager(boolean enableManagement, MessageBroker mbroker)
+    {
+        super(enableManagement);
+
+        super.setId(TYPE);
+
+        // Ensure that we have a message broker:
+        broker = (mbroker != null) ? mbroker : MessageBroker.getMessageBroker(null);
+
+        FlexClientSettings flexClientSettings = broker.getFlexClientSettings();
+        if (flexClientSettings != null && flexClientSettings.getTimeoutMinutes() != -1)
+        {
+            // Convert from minutes to millis.
+            setFlexClientTimeoutMillis(flexClientSettings.getTimeoutMinutes()*60*1000);
+        }
+
+        this.setParent(broker);
+    }
+
+    //--------------------------------------------------------------------------
+    //
+    // Variables
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * The MessageBroker that owns this manager.
+     */
+    private final MessageBroker broker;
+
+    /**
+     * The Mbean controller for this manager.
+     */
+    private FlexClientManagerControl controller;
+
+    /**
+     * Table to store FlexClients by id.
+     */
+    private final Map<String,FlexClient> flexClients = new ConcurrentHashMap<String,FlexClient>();
+
+
+    /**
+     * Manages time outs for FlexClients.
+     * This currently includes timeout of FlexClient instances, timeouts for async
+     * long-poll handling, and scheduling delayed flushes of outbound messages.
+     */
+    private volatile TimeoutManager flexClientTimeoutManager;
+
+    //--------------------------------------------------------------------------
+    //
+    // Properties
+    //
+    //--------------------------------------------------------------------------
+
+    //----------------------------------
+    //  clientIds
+    //----------------------------------
+
+    /**
+     * Returns a string array of the client IDs.
+     *
+     * @return A string array of the client IDs.
+     */
+    public String[] getClientIds()
+    {
+        String[] ids = new String[flexClients.size()];
+        ArrayList<String> idList = new ArrayList<String>(flexClients.keySet());
+
+        for (int i = 0; i < flexClients.size(); i++)
+            ids[i] = idList.get(i);
+
+        return ids;
+    }
+
+    //----------------------------------
+    //  flexClientCount
+    //----------------------------------
+
+    /**
+     * Returns the number of FlexClients in use.
+     *
+     * @return The number of FlexClients in use.
+     */
+    public int getFlexClientCount()
+    {
+        return flexClients.size();
+    }
+
+    //----------------------------------
+    //  flexClientTimeoutMillis
+    //----------------------------------
+
+    private volatile long flexClientTimeoutMillis;
+
+    /**
+     * Returns the idle timeout in milliseconds to apply to new FlexClient instances.
+     *
+     * @return The idle timeout in milliseconds to apply to new FlexClient instances.
+     */
+    public long getFlexClientTimeoutMillis()
+    {
+        return flexClientTimeoutMillis;
+    }
+
+    /**
+     * Sets the idle timeout in milliseconds to apply to new FlexClient instances.
+     *
+     * @param value The idle timeout in milliseconds to apply to new FlexClient instances.
+     */
+    public void setFlexClientTimeoutMillis(long value)
+    {
+        if (value < 1)
+            value = 0;
+
+        synchronized (this)
+        {
+            flexClientTimeoutMillis = value;
+        }
+    }
+
+    //----------------------------------
+    //  messageBroker
+    //----------------------------------
+
+    /**
+     * Returns the MessageBroker instance that owns this FlexClientManager.
+     *
+     * @return The parent MessageBroker instance.
+     */
+    public MessageBroker getMessageBroker()
+    {
+        return broker;
+    }
+
+    //--------------------------------------------------------------------------
+    //
+    // Public Methods
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * Get FlexClient with the specified id or a new one will be created.
+     * This method will return a valid existing FlexClient for the specific Id, 
+     * or a new FlexClient will created  
+     * @param id The id of the Flex client.
+     * @return FlexClient the FlexClient with the specified id
+     */
+    public FlexClient getFlexClient(String id)
+    {
+        return getFlexClient(id, true);
+    }
+    
+    /**
+     * Get the FlexClient with the specified id.
+     *
+     * @param id The id of the Flex client.
+     * @param createNewIfNotExist if true, a new FlexClient will be created if not exist
+     * @return FlexClient the FlexClient with the specified id
+     */
+    public FlexClient getFlexClient(String id, boolean createNewIfNotExist)
+    {
+        FlexClient flexClient = null;
+        // Try to lookup an existing instance if we receive an id.
+        if (id != null)
+        {
+            flexClient = flexClients.get(id);
+            if (flexClient != null)
+            {
+                if (flexClient.isValid() && !flexClient.invalidating)
+                {
+                    flexClient.updateLastUse();
+                    return flexClient;
+                }
+                // Invalid, remove it - it will be replaced below.
+                flexClients.remove(id);
+            }
+        }
+        // Use a manager-level lock (this) when creating/recreating a new FlexClient.
+        synchronized (this)
+        {
+            if (id != null)
+            {
+                flexClient = flexClients.get(id);
+                if (flexClient != null)
+                {
+                    flexClient.updateLastUse();
+                    return flexClient;
+                }
+                else
+                {
+                    if (!createNewIfNotExist)
+                    {
+                        return null;
+                    }
+                }
+            }
+            
+            flexClient = createFlexClient(id);
+            checkForNullAndDuplicateId(flexClient.getId());
+            flexClients.put(flexClient.getId(), flexClient);
+            if (flexClientTimeoutMillis > 0)
+                flexClientTimeoutManager.scheduleTimeout(flexClient);
+            flexClient.notifyCreated();
+            return flexClient;
+        }
+    }
+
+    /**
+     * Creates a FlexClientOutboundQueueProcessor instance and hooks it up to the passed
+     * FlexClient.
+     *
+     * @param flexClient The FlexClient to equip with a queue processor.
+     * @param endpointId The Id of the endpoint the queue processor is used for.
+     * @return The FlexClient with a configured queue processor.
+     */
+    public FlexClientOutboundQueueProcessor createOutboundQueueProcessor(FlexClient flexClient, String endpointId)
+    {
+        // First, try to create a custom outbound queue processor, if one exists.
+        FlexClientOutboundQueueProcessor processor = createCustomOutboundQueueProcessor(flexClient, endpointId);
+
+        // If no custom processor, then try to create default queue processor.
+        if (processor == null)
+            processor = createDefaultOutboundQueueProcessor(flexClient, endpointId);
+
+        // If MessageBroker's default queue processor fails, use the default processor.
+        if (processor == null)
+        {
+            processor = new FlexClientOutboundQueueProcessor();
+            processor.setFlexClient(flexClient);
+            processor.setEndpointId(endpointId);
+        }
+
+        return processor;
+    }
+
+    /**
+     * @exclude
+     * Monitors an async poll for a FlexClient for timeout.
+     *
+     * @param asyncPollTimeout The async poll task to monitor for timeout.
+     */
+    public void monitorAsyncPollTimeout(TimeoutAbstractObject asyncPollTimeout)
+    {
+        flexClientTimeoutManager.scheduleTimeout(asyncPollTimeout);
+    }
+
+    /**
+     * @exclude
+     * Monitors a scheduled flush for a FlexClient for timeout.
+     *
+     * @param scheduledFlushTimeout The schedule flush task to monitor for timeout.
+     */
+    public void monitorScheduledFlush(TimeoutAbstractObject scheduledFlushTimeout)
+    {
+        flexClientTimeoutManager.scheduleTimeout(scheduledFlushTimeout);
+    }
+
+    /**
+     * Starts the Flex client manager.
+     *
+     * @see flex.management.ManageableComponent#start()
+     */
+    @Override
+    public void start()
+    {
+        if (isManaged())
+        {
+            controller = new FlexClientManagerControl(getParent().getControl(), this);
+            setControl(controller);
+            controller.register();
+        }
+
+        final String baseId = getId();
+        flexClientTimeoutManager = new TimeoutManager(new ThreadFactory()
+                                                        {
+                                                            int counter = 1;
+                                                            public synchronized Thread newThread(Runnable runnable)
+                                                            {
+                                                                Thread t = new Thread(runnable);
+                                                                t.setName(baseId + "-FlexClientTimeoutThread-" + counter++);
+                                                                return t;
+                                                            }
+                                                        });
+    }
+
+    /**
+     * @see flex.management.ManageableComponent#stop()
+     */
+    public void stop()
+    {
+        if (controller != null)
+        {
+            controller.unregister();
+        }
+
+        if (flexClientTimeoutManager != null)
+            flexClientTimeoutManager.shutdown();
+    }
+
+    //--------------------------------------------------------------------------
+    //
+    // Protected Methods
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * Hook method invoked when a new <tt>FlexClient</tt> instance is created.
+     * 
+     * @param id The id the client provided, which was previously assigned by this server, 
+     *           or another server in a cluster. New clients will pass a <code>null</code>
+     *           value in which case this server must generate a unique id.
+     */
+    protected FlexClient createFlexClient(String id)
+    {
+        return (id == null) ? new FlexClient(this) : new FlexClient(this, id);
+    }
+    
+    /* (non-Javadoc)
+     * @see flex.management.ManageableComponent#getLogCategory()
+     */
+    protected String getLogCategory()
+    {
+        return LogCategories.CLIENT_FLEXCLIENT;
+    }
+
+    /**
+     * @exclude
+     * Removes a FlexClient from being managed by this manager.
+     * This method is invoked by FlexClients when they are invalidated.
+     *
+     * @param flexClient The id of the FlexClient being invalidated.
+     */
+    protected void removeFlexClient(FlexClient flexClient)
+    {
+        if (flexClient != null)
+        {
+            String id = flexClient.getId();
+            synchronized (id)
+            {
+                FlexClient storedClient = flexClients.get(id);
+                // If the stored instance is the same as the invalidating instance based upon identity,
+                // remove it.
+                if (storedClient == flexClient)
+                    flexClients.remove(id);
+            }
+        }
+    }
+
+    //--------------------------------------------------------------------------
+    //
+    // Private Methods
+    //
+    //--------------------------------------------------------------------------
+
+    private void checkForNullAndDuplicateId(String id)
+    {
+        if (id == null)
+        {
+            // Cannot create ''{0}'' with null id.
+            MessageException me = new MessageException();
+            me.setMessage(10039, new Object[]{"FlexClient"});
+            me.setCode("Server.Processing.NullId");
+            throw me;
+        }
+
+        if (flexClients.containsKey(id))
+        {
+            // Cannot create ''{0}'' with id ''{1}''; another ''{0}'' is already registered with the same id.
+            MessageException me = new MessageException();
+            me.setMessage(10040, new Object[]{"FlexClient", id});
+            me.setCode("Server.Processing.DuplicateId");
+            throw me;
+        }
+    }
+
+    private FlexClientOutboundQueueProcessor createDefaultOutboundQueueProcessor(
+            FlexClient flexClient, String endpointId)
+    {
+        FlexClientSettings flexClientSettings = broker.getFlexClientSettings();
+        if (flexClientSettings == null)
+            return null;
+
+        String queueProcessorClassName = flexClientSettings.getFlexClientOutboundQueueProcessorClassName();
+        if (queueProcessorClassName == null)
+            return null;
+
+        FlexClientOutboundQueueProcessor processor = null;
+        try
+        {
+            Class queueProcessorClass = createClass(queueProcessorClassName);
+            Object instance = ClassUtil.createDefaultInstance(queueProcessorClass, null);
+            processor = (FlexClientOutboundQueueProcessor)instance;
+            processor.setFlexClient(flexClient);
+            processor.setEndpointId(endpointId);
+            processor.initialize(flexClientSettings.getFlexClientOutboundQueueProcessorProperties());
+        }
+        catch (Throwable t)
+        {
+            String message = "Failed to create MessageBroker's outbound queue processor for FlexClient with id '" + flexClient.getId() + "'.";
+            if (Log.isWarn())
+                Log.getLogger(FlexClient.FLEX_CLIENT_LOG_CATEGORY).warn(message, t);
+
+            MessageException me = new MessageException(message, t);
+            throw me;
+        }
+
+        return processor;
+    }
+
+    private FlexClientOutboundQueueProcessor createCustomOutboundQueueProcessor(
+            FlexClient flexClient, String endpointId)
+    {
+        FlexClientOutboundQueueProcessor processor = null;
+        Endpoint endpoint = broker.getEndpoint(endpointId);
+        if (endpoint instanceof AbstractEndpoint)
+        {
+            Class processorClass = ((AbstractEndpoint)endpoint).getFlexClientOutboundQueueProcessorClass();
+            if (processorClass != null)
+            {
+                try
+                {
+                    Object instance = ClassUtil.createDefaultInstance(processorClass, null);
+                    if (instance instanceof FlexClientOutboundQueueProcessor)
+                    {
+                        processor = (FlexClientOutboundQueueProcessor)instance;
+                        processor.setFlexClient(flexClient);
+                        processor.setEndpointId(endpointId);
+                        processor.initialize(((AbstractEndpoint)endpoint).getFlexClientOutboundQueueProcessorConfig());
+                    }
+                }
+                catch (Throwable t)
+                {
+                    if (Log.isWarn())
+                        Log.getLogger(FlexClient.FLEX_CLIENT_LOG_CATEGORY).warn("Failed to create custom outbound queue processor for FlexClient with id '" + flexClient.getId() + "'. Using MessageBroker's default queue processor.", t);
+                }
+            }
+        }
+        return processor;
+    }
+
+    private Class createClass(String className)
+    {
+        Class c = ClassUtil.createClass(className, FlexContext.getMessageBroker() == null ? null :
+                    FlexContext.getMessageBroker().getClassLoader());
+
+        return c;
+    }
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/FlexClientNotSubscribedException.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/FlexClientNotSubscribedException.java b/modules/core/src/flex/messaging/client/FlexClientNotSubscribedException.java
new file mode 100755
index 0000000..629405d
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/FlexClientNotSubscribedException.java
@@ -0,0 +1,73 @@
+/*
+ * 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.client;
+
+import flex.messaging.MessageException;
+import flex.messaging.log.LogEvent;
+
+/**
+ * @exclude
+ */
+public class FlexClientNotSubscribedException extends MessageException
+{
+    /**
+     * @exclude
+     */
+    private static final long serialVersionUID = 773524927178340950L;
+
+    //--------------------------------------------------------------------------
+    //
+    // Properties
+    //
+    //--------------------------------------------------------------------------        
+    
+    //----------------------------------
+    //  defaultLogMessageIntro
+    //----------------------------------            
+
+    /**
+     * Overrides the intro text for the log message. 
+     */
+    public String getDefaultLogMessageIntro()
+    {
+        return "FlexClient not subscribed: ";        
+    }
+    
+    //----------------------------------
+    //  logStackTraceEnabled
+    //----------------------------------            
+    
+    /**
+     * Override to disable stack trace logging.
+     */
+    public boolean isLogStackTraceEnabled()
+    {
+        return false;        
+    }    
+    
+    //----------------------------------
+    //  peferredLogLevel
+    //----------------------------------            
+    
+    /**
+     * Override to lower the preferred log level to debug. 
+     */
+    public short getPreferredLogLevel()
+    {
+        return LogEvent.DEBUG;        
+    }
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/FlexClientOutboundQueueProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/FlexClientOutboundQueueProcessor.java b/modules/core/src/flex/messaging/client/FlexClientOutboundQueueProcessor.java
new file mode 100755
index 0000000..059c3ef
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/FlexClientOutboundQueueProcessor.java
@@ -0,0 +1,347 @@
+/*
+ * 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.client;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import flex.messaging.Destination;
+import flex.messaging.MessageClient;
+import flex.messaging.MessageDestination;
+import flex.messaging.config.ConfigMap;
+import flex.messaging.messages.Message;
+import flex.messaging.services.messaging.ThrottleManager;
+import flex.messaging.services.messaging.ThrottleManager.ThrottleResult;
+import flex.messaging.services.messaging.ThrottleManager.ThrottleResult.Result;
+
+/**
+ * The base FlexClientOutboundQueueProcessor implementation used if a custom implementation is not
+ * specified. Its behavior is very simple. It adds all new messages in order to the tail
+ * of the outbound queue and flushes all queued messages to the network as quickly as possible.
+ * It also handles the outbound client-level throttling specified at the destination level.
+ *
+ * @author shodgson
+ */
+public class FlexClientOutboundQueueProcessor
+{
+    //--------------------------------------------------------------------------
+    //
+    // Variables
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * The associated FlexClient.
+     */
+    private FlexClient client;
+
+    /**
+     * The last MessageClient messages were flushed to. This is mainly for faster
+     * lookup.
+     */
+    private MessageClient lastMessageClient;
+
+    /**
+     * The associated endpoint's Id.
+     */
+    private String endpointId;
+
+    /**
+     * Manages throttling of outbound client level messages.
+     */
+    protected OutboundQueueThrottleManager outboundQueueThrottleManager;
+
+    //--------------------------------------------------------------------------
+    //
+    // Public Methods
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * @exclude
+     * Stores the Id for the outbound queue's endpoint.
+     *
+     * @param value The Id for the outbound queue's endpoint.
+     */
+    public void setEndpointId(String value)
+    {
+        endpointId = value;
+    }
+
+    /**
+     * Returns the Id for the outbound queue's endpoint.
+     *
+     * @return The Id for the outbound queue's endpoint.
+     */
+    public String getEndpointId()
+    {
+        return endpointId;
+    }
+
+    /**
+     * @exclude
+     * Sets the associated FlexClient.
+     *
+     * @param value The associated FlexClient.
+     */
+    public void setFlexClient(FlexClient value)
+    {
+        client = value;
+    }
+
+    /**
+     * Returns the associated FlexClient.
+     *
+     * @return The associated FlexClient.
+     */
+    public FlexClient getFlexClient()
+    {
+        return client;
+    }
+
+    /**
+     * Returns the outbound queue throttle manager, or null if one does not exist.
+     *
+     * @return The outbound queue throttle manager.
+     */
+    public OutboundQueueThrottleManager getOutboundQueueThrottleManager()
+    {
+       return outboundQueueThrottleManager;
+    }
+
+    /**
+     * Utility method to initialize (if necessary) and return an outbound queue
+     * throttle manager.
+     *
+     * @return The outbound queue throttle manager.
+     */
+    public OutboundQueueThrottleManager getOrCreateOutboundQueueThrottleManager()
+    {
+        if (outboundQueueThrottleManager == null)
+            outboundQueueThrottleManager = new OutboundQueueThrottleManager(this);
+        return outboundQueueThrottleManager;
+    }
+
+    /**
+     * No-op; this default implementation doesn't require custom initialization.
+     * Subclasses may override to process any custom initialization properties that have been
+     * defined in the server configuration.
+     *
+     * @param properties A ConfigMap containing any custom initialization properties.
+     */
+    public void initialize(ConfigMap properties) {}
+
+    /**
+     * Always adds a new message to the tail of the queue.
+     *
+     * @param outboundQueue The queue of outbound messages.
+     * @param message The new message to add to the queue.
+     */
+    public void add(List<Message> outboundQueue, Message message)
+    {
+        outboundQueue.add(message);
+    }
+
+    /**
+     * Always empties the queue and returns all messages to be sent to the client.
+     *
+     * @param outboundQueue The queue of outbound messages.
+     * @return A FlushResult containing the messages that have been removed from the outbound queue
+     *         to be written to the network and a wait time for the next flush of the outbound queue
+     *         that is the default for the underlying Channel/Endpoint.
+     */
+    public FlushResult flush(List<Message> outboundQueue)
+    {
+        return flush(null /* no client distinction */, outboundQueue);
+    }
+
+    /**
+     * Removes all messages in the queue targeted to this specific MessageClient subscription(s) and
+     * returns them to be sent to the client.
+     * Overrides should be careful to only return messages for the specified MessageClient.
+     *
+     * @param messageClient The specific MessageClient to return messages for.
+     * @param outboundQueue The queue of outbound messages.
+     * @return A FlushResult containing the messages that have been removed from the outbound queue
+     *         to be written to the network for this MessageClient.
+     */
+    public FlushResult flush(MessageClient messageClient, List<Message> outboundQueue)
+    {
+        FlushResult flushResult = new FlushResult();
+        List<Message> messagesToFlush = null;
+
+        for (Iterator<Message> iter = outboundQueue.iterator(); iter.hasNext();)
+        {
+            Message message = iter.next();
+            if (messageClient == null || (message.getClientId().equals(messageClient.getClientId())))
+            {
+                if (isMessageExpired(message)) // Don't flush expired messages.
+                {
+                    iter.remove();
+                    continue;
+                }
+
+                messageClient = messageClient == null? getMessageClient(message) : messageClient;
+
+                // First, apply the destination level outbound throttling.
+                ThrottleResult throttleResult = throttleOutgoingDestinationLevel(messageClient, message, false);
+                Result result = throttleResult.getResult();
+
+                // No destination level throttling; check destination-client level throttling.
+                if (Result.OK == result)
+                {
+                    throttleResult = throttleOutgoingClientLevel(messageClient, message, false);
+                    result = throttleResult.getResult();
+                    // If no throttling, simply add the message to the list.
+                    if (Result.OK == result)
+                    {
+                        updateMessageFrequencyOutgoing(messageClient, message);
+                        if (messagesToFlush == null)
+                            messagesToFlush = new ArrayList<Message>();
+                        messagesToFlush.add(message);
+                    }
+                    // In rest of the policies (which is NONE), simply don't
+                    // add the message to the list.
+                }
+                iter.remove();
+            }
+        }
+
+        flushResult.setMessages(messagesToFlush);
+        return flushResult;
+    }
+
+    /**
+     * Utility method to test whether a message has expired or not.
+     * Messages with a timeToLive value that is shorter than the timespan from the message's
+     * timestamp up to the current system time will cause this method to return true.
+     * If there are expired messages in the outbound queue, flush implementations
+     * should use this helper method to only process and return messages that have
+     * not yet expired.
+     *
+     * @param message The message to test for expiration.
+     *
+     * @return true if the message has a timeToLive value that has expired; otherwise false.
+     */
+    public boolean isMessageExpired(Message message)
+    {
+        return (message.getTimeToLive() > 0 && (System.currentTimeMillis() - message.getTimestamp()) >= message.getTimeToLive());
+    }
+
+    /**
+     * Attempts to throttle the outgoing message at the destination level.
+     *
+     * @param msgClient The client the message is intended for.
+     * @param message The message to consider to throttle.
+     * @param buffered Whether the message has already been buffered. In that case,
+     * parts of regular throttling code is skipped.
+     * @return The result of throttling attempt.
+     */
+    protected ThrottleResult throttleOutgoingDestinationLevel(MessageClient msgClient, Message message, boolean buffered)
+    {
+        ThrottleManager throttleManager = getThrottleManager(msgClient);
+        if (throttleManager != null)
+        {
+            // In already buffered messages, don't use ThrottleManager#throttleOutgoingMessage
+            // to avoid regular throttling handling as the message has already been buffered.
+            if (buffered)
+                return throttleManager.throttleDestinationLevel(message, false /*incoming*/);
+
+            // Otherwise, regular throttling.
+            return throttleManager.throttleOutgoingMessage(message);
+        }
+        return new ThrottleResult(); // Otherwise, return OK result.
+    }
+
+    /**
+     * Attempts to throttle the outgoing message at the destination-client level.
+     *
+     * @param msgClient The client the message is intended for.
+     * @param message The message to consider to throttle.
+     * @param buffered Whether the message has already been buffered. In that case,
+     * parts of regular throttling code is skipped.
+     * @return The result of throttling attempt.
+     */
+    protected ThrottleResult throttleOutgoingClientLevel(MessageClient msgClient, Message message, boolean buffered)
+    {
+        if (outboundQueueThrottleManager != null) // Means client level throttling enabled.
+        {
+            ThrottleResult throttleResult = outboundQueueThrottleManager.throttleOutgoingClientLevel(message);
+            if (!buffered)
+            {
+                ThrottleManager throttleManager = getThrottleManager(msgClient);
+                if (throttleManager != null)
+                    throttleManager.handleOutgoingThrottleResult(message, throttleResult, true /*isClientLevel*/);
+            }
+            return throttleResult;
+        }
+        return new ThrottleResult(); // Otherwise, return OK result.
+    }
+
+    /**
+     * Returns the message client that the message is intended to.
+     *
+     * @param message The message.
+     * @return The message client that the message is intended to.
+     */
+    protected MessageClient getMessageClient(Message message)
+    {
+        // First try using the cached message client.
+        if (lastMessageClient != null && message.getClientId().equals(lastMessageClient.getClientId()))
+        {
+            return lastMessageClient;
+        }
+        else // Go ahead with the lookup.
+        {
+            lastMessageClient = client.getMessageClient((String)message.getClientId());
+            return lastMessageClient;
+        } 
+    }
+
+    /**
+     * Returns the throttle manager associated with the destination the message
+     * is intended to.
+     *
+     * @param msgClient The message client; it can be null.
+     * @return The throttle manager.
+     */
+    protected ThrottleManager getThrottleManager(MessageClient msgClient)
+    {
+        Destination destination = msgClient != null? msgClient.getDestination() : null;
+        return (destination != null && destination instanceof MessageDestination)? 
+                ((MessageDestination)destination).getThrottleManager() : null;
+    }
+
+    /**
+     * Updates the outgoing message's message frequency.
+     *
+     * @param msgClient The MessageClient that might have been passed to the flush; it can be null.
+     * @param message The message.
+     */
+    protected void updateMessageFrequencyOutgoing(MessageClient msgClient, Message message)
+    {
+        // Update the destination level message frequency.
+        ThrottleManager throttleManager = getThrottleManager(msgClient);
+        if (throttleManager != null)
+            throttleManager.updateMessageFrequencyDestinationLevel(false /*incoming*/);
+
+        // Update the client level message frequency.
+        if (outboundQueueThrottleManager != null)
+            outboundQueueThrottleManager.updateMessageFrequencyOutgoingClientLevel(message);
+    }
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/FlushResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/FlushResult.java b/modules/core/src/flex/messaging/client/FlushResult.java
new file mode 100755
index 0000000..6dd8cad
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/FlushResult.java
@@ -0,0 +1,117 @@
+/*
+ * 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.client;
+
+import java.util.List;
+
+import flex.messaging.messages.Message;
+
+/**
+ * Stores the messages that should be written to the network as a result of a flush
+ * invocation on a FlexClient's outbound queue.
+ */
+public class FlushResult
+{
+    //--------------------------------------------------------------------------
+    //
+    // Constructor
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * Constructs a <tt>FlushResult</tt> instance to return from a
+     * flush invocation on a FlexClient's outbound queue.
+     * This instance stores the list of messages to write over the network to
+     * the client as well as an optional wait time in milliseconds for when the
+     * next flush should be invoked.
+     */
+    public FlushResult() {}
+
+    //--------------------------------------------------------------------------
+    //
+    // Properties
+    //
+    //--------------------------------------------------------------------------
+
+    //----------------------------------
+    //  messages
+    //----------------------------------
+
+    private List<Message> messages;
+
+    /**
+     * Returns the messages to write to the network for this flush invocation.
+     * This list may be null, in which case no messages are written.
+     *
+     * @return The messages to write to the network for this flush invocation.
+     */
+    public List<Message> getMessages()
+    {
+        return messages;
+    }
+
+    /**
+     * Sets the messages to write to the network for this flush invocation.
+     *
+     * @param value The messages to write to the network for this flush invocation.
+     */
+    public void setMessages(List<Message> value)
+    {
+        messages = value;
+    }
+
+    //----------------------------------
+    //  nextFlushWaitTimeMillis
+    //----------------------------------
+
+    private int nextFlushWaitTimeMillis = 0;
+
+    /**
+     * Returns the wait time in milliseconds for when the next flush invocation should occur.
+     * If this value is 0, the default, a delayed flush is not scheduled and the next flush will
+     * depend upon the underlying Channel/Endpoint.
+     * For client-side polling Channels the next flush invocation will happen when the client sends
+     * its next poll request at its regular interval.
+     * For client-side Channels that support direct writes to the client a flush invocation is triggered
+     * when the next message is added to the outbound queue.
+     *
+     * @return The wait time in milliseconds before flush is next invoked. A value of 0, the default,
+     *         indicates that the default flush behavior for the underlying Channel/Endpoint should be
+     *         used.
+     */
+    public int getNextFlushWaitTimeMillis()
+    {
+        return nextFlushWaitTimeMillis;
+    }
+
+    /**
+     * Sets the wait time in milliseconds for when the next flush invocation should occur.
+     * If this value is 0, the default, a delayed flush is not scheduled and the next flush will
+     * depend upon the underlying Channel/Endpoint.
+     * For client-side polling Channels the next flush invocation will happen when the client sends
+     * its next poll request at its regular interval.
+     * For client-side Channels that support direct writes to the client a flush invocation is triggered
+     * when the next message is added to the outbound queue.
+     * Negative value assignments are treated as 0.
+     *
+     * @param value The wait time in milliseconds before flush will be invoked.
+     */
+    public void setNextFlushWaitTimeMillis(int value)
+    {
+        nextFlushWaitTimeMillis = (value < 1) ? 0 : value;
+    }
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/OutboundQueueThrottleManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/OutboundQueueThrottleManager.java b/modules/core/src/flex/messaging/client/OutboundQueueThrottleManager.java
new file mode 100755
index 0000000..e81a249
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/OutboundQueueThrottleManager.java
@@ -0,0 +1,269 @@
+/*
+ * 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.client;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+import flex.messaging.MessageClient.SubscriptionInfo;
+import flex.messaging.config.ThrottleSettings.Policy;
+import flex.messaging.log.Log;
+import flex.messaging.messages.Message;
+import flex.messaging.services.messaging.MessageFrequency;
+import flex.messaging.services.messaging.ThrottleManager;
+import flex.messaging.services.messaging.ThrottleManager.ThrottleResult;
+import flex.messaging.util.StringUtils;
+
+
+/**
+ * Used to keep track of and limit outbound message rates of a single FlexClient queue.
+ * An outbound FlexClient queue can contain messages from multiple MessageClients
+ * across multiple destinations. It can also contain messages for multiple
+ * subscriptions (for each subtopic/selector) across the same destination for
+ * the same MessageClient.
+ */
+public class OutboundQueueThrottleManager
+{
+    //--------------------------------------------------------------------------
+    //
+    // Constructor
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * Constructs a default outbound queue throttle manager.
+     *
+     * @param processor The outbound queue processor that is using this throttle manager.
+     */
+    public OutboundQueueThrottleManager(FlexClientOutboundQueueProcessor processor)
+    {
+        destinationFrequencies = new ConcurrentHashMap<String, DestinationFrequency>();
+        this.processor = processor;
+    }
+
+    //--------------------------------------------------------------------------
+    //
+    // Variables
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * Map of destination id and destination message frequencies.
+     */
+    protected final ConcurrentHashMap<String, DestinationFrequency> destinationFrequencies;
+
+    /**
+     * The parent queue processor of the throttle manager.
+     */
+    protected final FlexClientOutboundQueueProcessor processor;
+
+    //--------------------------------------------------------------------------
+    //
+    // Public Methods
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * Registers the destination with the outbound throttle manager.
+     *
+     * @param destinationId The id of the destination.
+     * @param outboundMaxClientFrequency The outbound max-client-frequency specified
+     * at the destination.
+     * @param outboundPolicy The outbound throttle policy specified at the destination.
+     */
+    public void registerDestination(String destinationId, int outboundMaxClientFrequency, Policy outboundPolicy)
+    {
+        DestinationFrequency frequency = destinationFrequencies.get(destinationId);
+        if (frequency == null)
+        {
+            frequency = new DestinationFrequency(outboundMaxClientFrequency, outboundPolicy);
+            destinationFrequencies.putIfAbsent(destinationId, frequency);
+        }
+    }
+
+    /**
+     * Registers the subscription of a client talking to a destination with the
+     * specified subscription info.
+     *
+     * @param destinationId The destination id.
+     * @param si The subscription information.
+     */
+    public void registerSubscription(String destinationId, SubscriptionInfo si)
+    {
+        DestinationFrequency frequency = destinationFrequencies.get(destinationId);
+        frequency.logMaxFrequencyDuringRegistration(frequency.outboundMaxClientFrequency, si);
+    }
+
+    /**
+     * Unregisters the subscription.
+     *
+     * @param destinationId The destination id.
+     * @param si The subscription information.
+     */
+    public void unregisterSubscription(String destinationId, SubscriptionInfo si)
+    {
+        unregisterDestination(destinationId);
+    }
+
+    /**
+     * Unregisters all subscriptions of the client under the specified destination.
+     *
+     * @param destinationId The destination id.
+     */
+    public void unregisterAllSubscriptions(String destinationId)
+    {
+        unregisterDestination(destinationId);
+    }
+
+    /**
+     * Attempts to throttle the outgoing message.
+     *
+     * @param message The message to consider to throttle.
+     * @return True if the message was throttled; otherwise false.
+     */
+    public ThrottleResult throttleOutgoingClientLevel(Message message)
+    {
+        String destinationId = message.getDestination();
+        if (isDestinationRegistered(destinationId))
+        {
+            DestinationFrequency frequency = destinationFrequencies.get(message.getDestination());
+            int maxFrequency = frequency.getMaxFrequency(message); // Limit to check against.
+            MessageFrequency messageFrequency = frequency.getMessageFrequency(message); // Message rate of the client.
+            if (messageFrequency != null)
+            {
+                ThrottleResult result = messageFrequency.checkLimit(maxFrequency, frequency.outboundPolicy);
+                return result;
+            }
+        }
+        return new ThrottleResult(); // Otherwise, return OK result.
+    }
+
+    /**
+     * Updates the outgoing client level message frequency of the message.
+     *
+     * @param message The message.
+     */
+    public void updateMessageFrequencyOutgoingClientLevel(Message message)
+    {
+        String destinationId = message.getDestination();
+        if (isDestinationRegistered(destinationId))
+        {
+            DestinationFrequency frequency = destinationFrequencies.get(message.getDestination());
+            MessageFrequency messageFrequency = frequency.getMessageFrequency(message);
+            if (messageFrequency != null)
+                messageFrequency.updateMessageFrequency();
+        }
+    }
+
+    //--------------------------------------------------------------------------
+    //
+    // Protected Methods
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * Determines whether the destination has been registered or not.
+     *
+     * @param destinationId The destination id.
+     * @return True if the destination with the specified id has been registered.
+     */
+    protected boolean isDestinationRegistered(String destinationId)
+    {
+        return destinationFrequencies.containsKey(destinationId);
+    }
+
+    /**
+     * Unregisters the destination from the outbound throttle manager.
+     *
+     * @param destinationId The id of the destination.
+     */
+    protected void unregisterDestination(String destinationId)
+    {
+        if (isDestinationRegistered(destinationId))
+            destinationFrequencies.remove(destinationId);
+    }
+
+    //--------------------------------------------------------------------------
+    //
+    // Inner Classes
+    //
+    //--------------------------------------------------------------------------
+
+    /**
+     * Used to keep track of max-client-frequency and outgoing throttle policy
+     * specified at the destination. It also keeps track of outbound message
+     * rates of all MessageClient subscriptions across the destination.
+     */
+    class DestinationFrequency
+    {
+        protected final int outboundMaxClientFrequency; // destination specified client limit.
+        protected final MessageFrequency outboundClientFrequency;
+        protected final Policy outboundPolicy; // destination specified policy.
+
+        /**
+         * Default constructor.
+         *
+         * @param outboundMaxClientFrequency The outbound throttling max-client-frequency of the destination.
+         * @param outboundPolicy The outbound throttling policy of the destination.
+         */
+        DestinationFrequency(int outboundMaxClientFrequency, Policy outboundPolicy)
+        {
+            this.outboundMaxClientFrequency = outboundMaxClientFrequency;
+            this.outboundPolicy = outboundPolicy;
+            outboundClientFrequency = new MessageFrequency(outboundMaxClientFrequency);
+        }
+
+        /**
+         * Returns the max-client-frequency for the subscription the message is
+         * intended for (which is simply the max-client-frequency specified at 
+         * the destination).
+         *
+         * @param message The message.
+         *
+         * @return The max-frequency for the subscription.
+         */
+        int getMaxFrequency(Message message)
+        {
+            return outboundMaxClientFrequency;
+        }
+
+        /**
+         * Given a subscription the message is intended to, returns the message
+         * rate frequency for that subscription.
+         *
+         * @param message The message.
+         * @return The message frequency for the subscription, if it exists; otherwise null.
+         */
+        MessageFrequency getMessageFrequency(Message message)
+        {
+            return outboundClientFrequency;
+        }
+
+        /**
+         * Utility function to log the maxFrequency being used during subscription.
+         *
+         * @param maxFrequency The maxFrequency to log.
+         */
+        void logMaxFrequencyDuringRegistration(int maxFrequency, SubscriptionInfo si)
+        {
+            if (Log.isDebug())
+                Log.getLogger(ThrottleManager.LOG_CATEGORY).debug("Outbound queue throttle manager for FlexClient '"
+                        + processor.getFlexClient().getId() + "' is using '" + maxFrequency
+                        + "' as the throttling limit for its subscription: "
+                        +  StringUtils.NEWLINE + si);
+        }
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/PollFlushResult.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/PollFlushResult.java b/modules/core/src/flex/messaging/client/PollFlushResult.java
new file mode 100755
index 0000000..fe4a76d
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/PollFlushResult.java
@@ -0,0 +1,97 @@
+/*
+ * 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.client;
+
+/**
+ * Extends <tt>FlushResult</tt> and adds additional properties for controlling 
+ * client polling behavior.
+ */
+public class PollFlushResult extends FlushResult
+{
+    //--------------------------------------------------------------------------
+    //
+    // Properties
+    //
+    //--------------------------------------------------------------------------
+
+    //----------------------------------
+    //  avoidBusyPolling
+    //----------------------------------
+
+    private boolean avoidBusyPolling;
+    
+    /**
+     * Indicates whether the handling of this result should attempt to avoid
+     * potential busy-polling cycles.
+     * This will be set to <code>true</code> in the case of two clients that are both
+     * long-polling the server over the same session.
+     * 
+     * @return <code>true</code> if the handling of this result should attempt to avoid potential
+     *         busy-polling cycles.
+     */
+    public boolean isAvoidBusyPolling()
+    {
+        return avoidBusyPolling;
+    }
+    
+    /**
+     * Set to <code>true</code> to signal that handling for this result should attempt to avoid
+     * potential busy-polling cycles.
+     * 
+     * @param value <code>true</code> to signal that handling for this result should attempt to 
+     *        avoid potential busy-polling cycles.
+     */
+    public void setAvoidBusyPolling(boolean value)
+    {
+        avoidBusyPolling = value;
+    }
+    
+    //----------------------------------
+    //  clientProcessingSuppressed
+    //----------------------------------
+
+    private boolean clientProcessingSuppressed;
+    
+    /**
+     * Indicates whether client processing of this result should be
+     * suppressed.
+     * This should be <code>true</code> for results generated for poll requests
+     * that arrive while a long-poll request from the same client is being serviced
+     * to avoid a busy polling cycle.
+     * 
+     * @return <code>true</code> if client processing of this result should be suppressed;
+     *         otherwise <code>false</code>.
+     */
+    public boolean isClientProcessingSuppressed()
+    {
+        return clientProcessingSuppressed;
+    }
+    
+    /**
+     * Set to <code>true</code> to suppress client processing of this result.
+     * Default is <code>false</code>.
+     * This should be set to <code>true</code> for results generated for poll requests
+     * that arrive while a long-poll request from the same client is being serviced
+     * to avoid a busy polling cycle.
+     * 
+     * @param value <code>true</code> to suppress client processing of the result.
+     */
+    public void setClientProcessingSuppressed(boolean value)
+    {
+        clientProcessingSuppressed = value;
+    }
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/PollWaitListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/PollWaitListener.java b/modules/core/src/flex/messaging/client/PollWaitListener.java
new file mode 100755
index 0000000..ae4b820
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/PollWaitListener.java
@@ -0,0 +1,40 @@
+/*
+ * 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.client;
+
+/**
+ * Used alongside invocations of <code>FlexClient.pollWithWait()</code> to allow calling code to
+ * maintain a record of the Objects being used to place waited poll requests into a wait
+ * state. This can be used to break the threads out of their wait state separately from the
+ * internal waited poll handling within <code>FlexClient</code>.
+ */
+public interface PollWaitListener
+{
+    /**
+     * Hook method invoked directly before a wait begins.
+     * 
+     * @param notifier The <tt>Object</tt> being used to <code>wait()/notify()</code>.
+     */
+    void waitStart(Object notifier);
+    
+    /**
+     * Hook method invoked directly after a wait completes.
+     * 
+     * @param notifier The <tt>Object</tt> being used to <code>wait()/notify()</code>.
+     */
+    void waitEnd(Object notifier);
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/UserAgentSettings.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/UserAgentSettings.java b/modules/core/src/flex/messaging/client/UserAgentSettings.java
new file mode 100755
index 0000000..e04ef88
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/UserAgentSettings.java
@@ -0,0 +1,258 @@
+/*
+ * 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.client;
+
+/**
+ * A class to hold user agent specific properties. For example, in streaming
+ * endpoints, a certain number of bytes need to be written before the
+ * streaming connection can be used and this value is specific to user agents.
+ * Similarly, the number of simultaneous connections a session can have is user
+ * agent specific.
+ */
+public class UserAgentSettings
+{
+    /**
+     * The prefixes of the version token used by various browsers.
+     */
+    public static final String USER_AGENT_ANDROID = "Android";
+    public static final String USER_AGENT_CHROME = "Chrome";
+    public static final String USER_AGENT_FIREFOX = "Firefox";
+    public static final String USER_AGENT_FIREFOX_1 = "Firefox/1";
+    public static final String USER_AGENT_FIREFOX_2 = "Firefox/2";
+    public static final String USER_AGENT_MSIE = "MSIE";
+    public static final String USER_AGENT_MSIE_5 = "MSIE 5";
+    public static final String USER_AGENT_MSIE_6 = "MSIE 6";
+    public static final String USER_AGENT_MSIE_7 = "MSIE 7";
+    public static final String USER_AGENT_OPERA = "Opera";
+    public static final String USER_AGENT_OPERA_8 = "Opera 8";
+    // Opera 10,11 ship as User Agent Opera/9.8.
+    public static final String USER_AGENT_OPERA_10 = "Opera/9.8"; 
+    public static final String USER_AGENT_SAFARI = "Safari";
+
+    /**
+     * Bytes needed to kickstart the streaming connections for IE.
+     */
+    public static final int KICKSTART_BYTES_MSIE = 2048;
+    /**
+     * Bytes needed to kickstart the streaming connections for SAFARI.
+     */
+    public static final int KICKSTART_BYTES_SAFARI = 512;
+    /**
+     * Bytes needs to kicksart the streaming connections for Android.
+     */
+    public static final int KICKSTART_BYTES_ANDROID = 4010;
+
+    /**
+     * The default number of persistent connections per session for various browsers.
+     */
+    private static final int MAX_PERSISTENT_CONNECTIONS_LEGACY =  1;
+    public static final int MAX_PERSISTENT_CONNECTIONS_DEFAULT = 5;
+    private static final int MAX_PERSISTENT_CONNECTIONS_OPERA_LEGACY = 3;
+    private static final int MAX_PERSISTENT_CONNECTIONS_CHROME = MAX_PERSISTENT_CONNECTIONS_DEFAULT;
+    private static final int MAX_PERSISTENT_CONNECTIONS_FIREFOX = MAX_PERSISTENT_CONNECTIONS_DEFAULT;
+    private static final int MAX_PERSISTENT_CONNECTIONS_MSIE = MAX_PERSISTENT_CONNECTIONS_DEFAULT;
+    private static final int MAX_PERSISTENT_CONNECTIONS_OPERA = 7;
+    private static final int MAX_PERSISTENT_CONNECTIONS_SAFARI = 3;
+
+    private String matchOn;
+    private int kickstartBytes;
+    private int maxPersistentConnectionsPerSession = MAX_PERSISTENT_CONNECTIONS_DEFAULT;
+
+    /**
+     *  Static method to retrieve pre-initialized user agents which are as follows:
+     *
+     *  In Chrome 0, 1, 2, the limit is 6:
+     *      match-on="Chrome" max-persistent-connections-per-session="5"
+     *
+     *  In Firefox 1, 2, the limit is 2:
+     *      match-on="Firefox" max-persistent-connections-per-session="1"
+     *
+     *  In Firefox 3, the limit is 6:
+     *      match-on="Firefox/3" max-persistent-connections-per-session="5"
+     *
+     *  In MSIE 5, 6, 7, the limit is 2 with kickstart bytes of 2K:
+     *      match-on="MSIE" max-persistent-connections-per-session="1" kickstart-bytes="2048"
+     *
+     *  In MSIE 8, the limit is 6 with kickstart bytes of 2K:
+     *      match-on="MSIE 8" max-persistent-connections-per-session="5" kickstart-bytes="2048"
+     *
+     *  In Opera 7, 9, the limit is 4:
+     *      match-on="Opera" max-persistent-connections-per-session="3"
+     *
+     *  In Opera 8, the limit is 8:
+     *      match-on="Opera 8" max-persistent-connections-per-session="7"
+     *
+     *  In Opera 10, the limit is 8.
+     *      match-on="Opera 10" max-persistent-connections-per-session="7"
+     *
+     *  In Safari 3, 4, the limit is 4.
+     *      match-on="Safari" max-persistent-connections-per-session="3"
+     *
+     * @param matchOn String to use match the agent.
+     */
+    public static UserAgentSettings getAgent(String matchOn)
+    {
+        UserAgentSettings userAgent = new UserAgentSettings();
+        userAgent.setMatchOn(matchOn);
+
+        if (USER_AGENT_ANDROID.equals(matchOn))
+        {
+            userAgent.setKickstartBytes(KICKSTART_BYTES_ANDROID);
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_SAFARI);
+        }
+        if (USER_AGENT_CHROME.equals(matchOn))
+        {
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_CHROME);
+        }
+        else if (USER_AGENT_FIREFOX.equals(matchOn))
+        {
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_FIREFOX);
+        }
+        else if (USER_AGENT_FIREFOX_1.equals(matchOn))
+        {
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_LEGACY);
+        }
+        else if (USER_AGENT_FIREFOX_2.equals(matchOn))
+        {
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_LEGACY);
+        }
+        else if (USER_AGENT_MSIE.equals(matchOn))
+        {
+            userAgent.setKickstartBytes(KICKSTART_BYTES_MSIE);
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_MSIE);
+        }
+        else if (USER_AGENT_MSIE_5.equals(matchOn))
+        {
+            userAgent.setKickstartBytes(KICKSTART_BYTES_MSIE);
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_LEGACY);
+        }
+        else if (USER_AGENT_MSIE_6.equals(matchOn))
+        {
+            userAgent.setKickstartBytes(KICKSTART_BYTES_MSIE);
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_LEGACY);
+        }
+        else if (USER_AGENT_MSIE_7.equals(matchOn))
+        {
+            userAgent.setKickstartBytes(KICKSTART_BYTES_MSIE);
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_LEGACY);
+        }
+        else if (USER_AGENT_OPERA.equals(matchOn))
+        {
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_OPERA_LEGACY);
+        }
+        else if (USER_AGENT_OPERA_8.equals(matchOn))
+        {
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_OPERA);
+        }
+        else if (USER_AGENT_OPERA_10.equals(matchOn))
+        {
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_OPERA);
+        }
+        else if (USER_AGENT_SAFARI.equals(matchOn))
+        {
+            userAgent.setKickstartBytes(KICKSTART_BYTES_SAFARI);
+            userAgent.setMaxPersistentConnectionsPerSession(MAX_PERSISTENT_CONNECTIONS_SAFARI);
+        }
+        return userAgent;
+    }
+
+    /**
+     * Returns the String to use to match the agent.
+     *
+     * @return The String to use to match the agent.
+     */
+    public String getMatchOn()
+    {
+        return matchOn;
+    }
+
+    /**
+     * Sets the String to use to match the agent.
+     *
+     * @param matchOn The String to use to match the agent.
+     */
+    public void setMatchOn(String matchOn)
+    {
+        this.matchOn = matchOn;
+    }
+
+    /**
+     * Returns the number of bytes needed to kickstart the streaming connections
+     * for the user agent.
+     *
+     * @return The number of bytes needed to kickstart the streaming connections
+     * for the user agent.
+     */
+    public int getKickstartBytes()
+    {
+        return kickstartBytes;
+    }
+
+    /**
+     * Sets the number of bytes needed to kickstart the streaming connections
+     * for the user agent.
+     *
+     * @param kickstartBytes The number of bytes needed to kickstart the streaming
+     * connections for the user agent.
+     */
+    public void setKickstartBytes(int kickstartBytes)
+    {
+        if (kickstartBytes < 0)
+            kickstartBytes = 0;
+        this.kickstartBytes = kickstartBytes;
+    }
+
+    /**
+     * @deprecated Use {@link UserAgentSettings#getMaxPersistentConnectionsPerSession()} instead.
+     */
+    public int getMaxStreamingConnectionsPerSession()
+    {
+        return getMaxPersistentConnectionsPerSession();
+    }
+
+    /**
+     * @deprecated Use {@link UserAgentSettings#setMaxPersistentConnectionsPerSession(int)} instead.
+     */
+    public void setMaxStreamingConnectionsPerSession(int maxStreamingConnectionsPerSession)
+    {
+        setMaxPersistentConnectionsPerSession(maxStreamingConnectionsPerSession);
+    }
+
+    /**
+     * Returns the number of simultaneous streaming connections per session
+     * the user agent supports.
+     *
+     * @return The number of streaming connections per session the user agent supports.
+     */
+    public int getMaxPersistentConnectionsPerSession()
+    {
+        return maxPersistentConnectionsPerSession;
+    }
+
+    /**
+     * Sets the number of simultaneous streaming connections per session
+     * the user agent supports.
+     *
+     * @param maxStreamingConnectionsPerSession The number of simultaneous
+     * streaming connections per session the user agent supports.
+     */
+    public void setMaxPersistentConnectionsPerSession(int maxStreamingConnectionsPerSession)
+    {
+        this.maxPersistentConnectionsPerSession = maxStreamingConnectionsPerSession;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/client/package-info.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/client/package-info.java b/modules/core/src/flex/messaging/client/package-info.java
new file mode 100755
index 0000000..e9a5b98
--- /dev/null
+++ b/modules/core/src/flex/messaging/client/package-info.java
@@ -0,0 +1,18 @@
+/*
+ * 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.client;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/flex-blazeds/blob/7a58369c/modules/core/src/flex/messaging/cluster/BroadcastHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/flex/messaging/cluster/BroadcastHandler.java b/modules/core/src/flex/messaging/cluster/BroadcastHandler.java
new file mode 100755
index 0000000..7d6f365
--- /dev/null
+++ b/modules/core/src/flex/messaging/cluster/BroadcastHandler.java
@@ -0,0 +1,45 @@
+/*
+ * 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.cluster;
+
+import java.util.List;
+
+/**
+ * @exclude
+ * This interface represents a handler for a message broadcast by a Cluster.
+ * Clusters broadcast messages across their physical nodes, and when they
+ * receive those messages they locate a BroadcastHandler capable of handling
+ * the broadcast.
+ */
+public interface BroadcastHandler
+{
+    /**
+     * Handle the broadcast message.
+     *
+     * @param sender sender of the original message
+     * @param params any parameters need to handle the message
+     */
+    void handleBroadcast(Object sender, List<Object> params);
+
+    /**
+     * Determine whether this Handler supports a particular operation by name.
+     *
+     * @return whether or not this handler supports the named operation
+     * @param name name of the operation
+     */
+    boolean isSupportedOperation(String name);
+}


Mime
View raw message