/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2001 - 2026 ScalAgent Distributed Technologies
 * Copyright (C) 2003 - 2004 Bull SA
 * Copyright (C) 1996 - 2000 Dyade
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 * 
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA.
 *
 * Initial developer(s): Frederic Maistre (INRIA)
 * Contributor(s): ScalAgent Distributed Technologies
 */
package org.objectweb.joram.mom.dest;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Vector;

import org.objectweb.joram.mom.notifications.AbstractRequestNot;
import org.objectweb.joram.mom.notifications.ClientMessages;
import org.objectweb.joram.mom.notifications.ClusterJoinAck;
import org.objectweb.joram.mom.notifications.ClusterJoinNot;
import org.objectweb.joram.mom.notifications.ClusterRemoveNot;
import org.objectweb.joram.mom.notifications.ExceptionReply;
import org.objectweb.joram.mom.notifications.FwdAdminRequestNot;
import org.objectweb.joram.mom.notifications.ClientSubscriptionNot;
import org.objectweb.joram.mom.notifications.SubscribeReply;
import org.objectweb.joram.mom.notifications.SubscribeRequest;
import org.objectweb.joram.mom.notifications.TopicForwardNot;
import org.objectweb.joram.mom.notifications.TopicMsgsReply;
import org.objectweb.joram.mom.notifications.UnsubscribeRequest;
import org.objectweb.joram.mom.notifications.WakeUpNot;
import org.objectweb.joram.mom.util.DMQManager;
import org.objectweb.joram.shared.DestinationConstants;
import org.objectweb.joram.shared.MessageErrorConstants;
import org.objectweb.joram.shared.admin.AdminReply;
import org.objectweb.joram.shared.admin.AdminRequest;
import org.objectweb.joram.shared.admin.ClusterAdd;
import org.objectweb.joram.shared.admin.ClusterLeave;
import org.objectweb.joram.shared.admin.ClusterList;
import org.objectweb.joram.shared.admin.ClusterListReply;
import org.objectweb.joram.shared.admin.GetDMQSettingsReply;
import org.objectweb.joram.shared.admin.GetDMQSettingsRequest;
import org.objectweb.joram.shared.admin.GetFatherReply;
import org.objectweb.joram.shared.admin.GetFatherRequest;
import org.objectweb.joram.shared.admin.GetNumberReply;
import org.objectweb.joram.shared.admin.GetSubscriberIds;
import org.objectweb.joram.shared.admin.GetSubscriberIdsRep;
import org.objectweb.joram.shared.admin.GetSubscriptionsRequest;
import org.objectweb.joram.shared.admin.SetFather;
import org.objectweb.joram.shared.excepts.AccessException;
import org.objectweb.joram.shared.excepts.MomException;
import org.objectweb.joram.shared.messages.Message;
import org.objectweb.joram.shared.selectors.Selector;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.agent.AgentId;
import fr.dyade.aaa.agent.AgentServer;
import fr.dyade.aaa.agent.Channel;
import fr.dyade.aaa.agent.DeleteNot;
import fr.dyade.aaa.agent.Notification;
import fr.dyade.aaa.agent.UnknownAgent;
import fr.dyade.aaa.common.Debug;

/**
 * The <code>Topic</code> class implements the MOM topic behavior,
 * basically distributing the received messages to subscribers.
 * <p>
 * A Topic might be part of a hierarchy; if it is the case, and if the topic is
 * not on top of that hierarchy, it will have a father to forward messages to.
 * <p>
 * A topic might also be part of a cluster; if it is the case, it will have
 * friends to forward messages to.
 * <p>
 * A topic can be part of a hierarchy and of a cluster at the same time.
 */
public class Topic extends Destination implements TopicMBean {
  private static final Logger logger = Debug.getLogger(Topic.class.getName());
  private static final boolean DEBUG = logger.isLoggable(BasicLevel.DEBUG);

  /** define serialVersionUID for interoperability */
  private static final long serialVersionUID = 1L;

  /** Identifier of this topic's father, if any. */
  protected AgentId fatherId = null;

  /** Set of cluster elements (including itself), if any. */
  protected Set<AgentId> friends = null;

  /** Vector of subscribers' identifiers. */
  protected List<AgentId> subscribers = new Vector<AgentId>();

  /** Table of subscribers' selectors. */
  protected Map<AgentId, String> selectors = new Hashtable<AgentId, String>();

  protected long nbMsgsReceiveSinceCreation = 0;

  /**
   * Creates a Topic agent.
   */
  public Topic() {
    super();
  }

  /**
   * Creates a Topic agent with specific stamp.
   * 
   * @param name  Name of topic;
   * @param fixed If true, topic is fixed in memory.
   * @param stamp Specific stamp for resulting agent.
   */
  public Topic(String name, boolean fixed, int stamp) {
    super(name, fixed, stamp);
  }

  public final byte getType() {
    return DestinationConstants.TOPIC_TYPE;
  }

  /**
   * Distributes the received notifications to the appropriate reactions.
   * 
   * @throws Exception
   */
  public void react(AgentId from, Notification not) throws Exception {
    int reqId = -1;
    if (not instanceof AbstractRequestNot)
      reqId = ((AbstractRequestNot) not).getRequestId();

    if (DEBUG)
      logger.log(BasicLevel.DEBUG,
                 "--- " + this + ": got " + not.getClass().getName() + " with id: " + reqId + " from: " + from.toString());
    try {
      if (not instanceof ClusterJoinAck)
        clusterJoinAck((ClusterJoinAck) not);
      else if (not instanceof ClusterJoinNot)
        clusterJoin((ClusterJoinNot) not);
      else if (not instanceof ClusterRemoveNot)
        clusterRemove(from);
      else if (not instanceof SubscribeRequest)
        subscribeRequest(from, (SubscribeRequest) not);
      else if (not instanceof UnsubscribeRequest)
        unsubscribeRequest(from);
      else if (not instanceof TopicForwardNot)
        topicForwardNot(from, (TopicForwardNot) not);
      else if (not instanceof ClientSubscriptionNot) {
        // Do nothing, this notification is only handled by ElasticTopic.
      } else
        super.react(from, not);
    } catch (MomException exc) {
      // MOM exceptions are sent to the requester.
      if (DEBUG)
        logger.log(BasicLevel.WARN, "", exc);
      else
        logger.log(BasicLevel.WARN, exc.toString());

      AbstractRequestNot req = (AbstractRequestNot) not;
      Channel.sendTo(from, new ExceptionReply(req, exc));
    }
  }

  /**
   * Initializes the destination.
   * 
   * @param firstTime		true when first called by the factory
   */
  public void initialize(boolean firstTime) {}

  /**
   * Finalizes the destination before it is garbaged.
   * 
   * @param last true if the destination is deleted
   */
  protected void finalize(boolean last) {}

  /**
   * Returns a string representation of this destination.
   */
  public String toString() {
    return "Topic:" + getId().toString();
  }

  public void wakeUpNot(WakeUpNot not) {
    // nothing to do
  }

  /**
   * Reaction to the request of adding a new cluster element.
   */
  private void clusterAdd(FwdAdminRequestNot req, String joiningTopic) {
    AgentId newFriendId = AgentId.fromString(joiningTopic);

    // Adds the given topic to the cluster containing the current one.
    if (friends == null) {
      friends = new HashSet<AgentId>();
      friends.add(getId());
    }
    forward(newFriendId,
        new ClusterJoinNot(friends, req.getReplyTo(), req.getRequestMsgId(), req.getReplyMsgId()));
  }

  /**
   * Method implementing the reaction to a {@link ClusterJoinNot} notification,
   * sent by a fellow topic for notifying this topic to join the cluster, doing
   * a transitive closure of clusters, if any.
   */
  protected void clusterJoin(ClusterJoinNot not) {
    setSave(); // state change, so save.
    if (friends == null) {
      friends = new HashSet<AgentId>();
      friends.add(getId());
    }
    friends.addAll(not.getCluster());
    sendToCluster(new ClusterJoinAck(friends));
    replyToTopic(new AdminReply(true, null), not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
  }

  /**
   * Method implementing the reaction to a {@link ClusterJoinAck} notification,
   * doing a transitive closure with the current cluster and the one of the new
   * cluster element.
   */
  protected void clusterJoinAck(ClusterJoinAck not) {
    friends.addAll(not.getCluster());

    if (DEBUG)
      logger.log(BasicLevel.DEBUG,
                 "--- " + this + " Topic.clusterJoinAck(" + not + ")" + "\nfriends=" + friends);
  }

  /**
   * Returns the cluster list.
   * 
   * @return the cluster list.
   */
  private List<String> clusterList() {
    List<String> cluster = new ArrayList<String>();
    if (friends != null) {
      Iterator<AgentId> iterator = friends.iterator();
      while (iterator.hasNext()) {
        cluster.add(iterator.next().toString());
      }
    } else {
      cluster.add(getAgentId());
    }
    return cluster;
  }

  public String[] getClusterElements() {
    List<String> list = clusterList();
    return (String[]) list.toArray(new String[list.size()]);
  }

  /**
   * Ask this topic to leave the cluster.
   */
  private void clusterLeave() {
    // Removes this topic of its cluster
    if (friends != null) {
      // Sends a notification to all members asking to remove the topic
      sendToCluster(new ClusterRemoveNot());
      friends = null;
      setSave(); // state change, so save.
    }
  }

  /**
   * Remove the specified topic from current cluster.
   * 
   * @param topic The topic which left the cluster
   */
  private void clusterRemove(AgentId topic) {
    setSave(); // state change, so save.
    friends.remove(topic);
    if (friends.size() == 1)
      friends = null;
  }

  /**
   * Sends a notification to all topics in cluster.
   * 
   * @param not The notification to send.
   */
  protected void sendToCluster(Notification not) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "--- " + this + " Topic.sendToCluster(" + not + ")");

    if (friends == null || friends.size() < 2)
      return;

    for (Iterator<AgentId> e = friends.iterator(); e.hasNext();) {
      AgentId id = (AgentId) e.next();
      if (!id.equals(getId()))
        forward(id, not);
    }
  }

  /**
   * Method implementing the reaction to a <code>SubscribeRequest</code>
   * instance. 
   *
   * @exception AccessException  If the sender is not a READER.
   */
  protected void subscribeRequest(AgentId from, SubscribeRequest not) throws AccessException {
    if (! isReader(from))
      throw new AccessException("READ right not granted");

    setSave(); // state change, so save.
    
    // Adding new subscriber.
    if (! subscribers.contains(from)) {
      subscribers.add(from);
    }

    // The requester might either be a new subscriber, or an existing one;
    // setting the selector, possibly by removing or modifying an already set
    // expression.
    if (not.getSelector() != null && ! not.getSelector().equals(""))
      selectors.put(from, not.getSelector());
    else
      selectors.remove(from);

    if (!not.isAsyncSub())
      forward(from, new SubscribeReply(not));

    if (DEBUG)
      logger.log(BasicLevel.DEBUG, 
                 "Client " + from + " set as a subscriber with selector " + not.getSelector());
  }

  /**
   * Method implementing the reaction to an <code>UnsubscribeRequest</code>
   * instance, requesting to remove a subscriber.
   */
  protected void unsubscribeRequest(AgentId from) {
    setSave(); // state change, so save.
    subscribers.remove(from);
    selectors.remove(from);

    if (DEBUG)
      logger.log(BasicLevel.DEBUG,  "Client " + from + " removed from the subscribers.");
  }

  /**
   * Method implementing the reaction to a <code>TopicForwardNot</code>
   * instance, carrying messages forwarded by a cluster fellow or a
   * hierarchical son.
   */
  protected void topicForwardNot(AgentId from, TopicForwardNot not) {
    doClientMessages(from, not.getMessages(), not.isFromCluster(), false);
  }

  /**
   * 
   * @see org.objectweb.joram.mom.dest.Destination#handleAdminRequestNot(fr.dyade.aaa.agent.AgentId, org.objectweb.joram.mom.notifications.FwdAdminRequestNot)
   */
  public void handleAdminRequestNot(AgentId from, FwdAdminRequestNot not) {
    AdminRequest adminRequest = not.getRequest();
    
    if (adminRequest instanceof GetSubscriberIds) {
      replyToTopic(new GetSubscriberIdsRep(getSubscriberIds()),
                   not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
    } else if (adminRequest instanceof GetSubscriptionsRequest) {
      replyToTopic(new GetNumberReply(getNumberOfSubscribers()),
                   not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
    } else if (adminRequest instanceof GetDMQSettingsRequest) {
      replyToTopic(new GetDMQSettingsReply((dmqId != null)?dmqId.toString():null, 0),
                   not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
    } else if (adminRequest instanceof SetFather) {
      setSave(); // state change, so save.
      this.fatherId = AgentId.fromString(((SetFather) adminRequest).getFather());
      replyToTopic(new AdminReply(true, null),
                   not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
    } else if (adminRequest instanceof GetFatherRequest) {
      replyToTopic(new GetFatherReply((fatherId != null)?fatherId.toString():null),
                   not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
    } else if (adminRequest instanceof ClusterList) {
      List<String> clstr = clusterList();
      replyToTopic(new ClusterListReply(clstr), not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
    } else if (adminRequest instanceof ClusterAdd) {
      clusterAdd(not, ((ClusterAdd) adminRequest).getAddedDest());
    } else if (adminRequest instanceof ClusterLeave) {
      clusterLeave();
      replyToTopic(new AdminReply(true, null), not.getReplyTo(), not.getRequestMsgId(), not.getReplyMsgId());
    } else {
      super.handleAdminRequestNot(from, not);
    }
  }

  /**
   * Returns the number of subscribers.
   * Each user appears once even if there is multiples subscriptions, the different
   * subscriptions can be enumerate through the proxy MBean.
   * 
   * @return the number of subscribers.
   */
  public final int getNumberOfSubscribers() {
    return subscribers.size();
  }

  /**
   * Returns the list of unique identifiers of all subscribers. Each user
   * appears once even if there is multiples subscriptions, the different
   * subscriptions can be enumerate through the proxy MBean.
   *
   * @return the list of unique identifiers of all subscribers.
   */
  public final String[] getSubscriberIds() {
    String[] res = new String[subscribers.size()];
    for (int i = 0; i < res.length; i++) {
      AgentId aid = (AgentId) subscribers.get(i);
      res[i] = aid.toString();
    }
    return res;
  }

  /**
   * Method specifically processing a <code>SetRightRequest</code> instance.
   * <p>
   * When a reader is removed, deleting this reader's subscription if any,
   * and sending an <code>ExceptionReply</code> notification to the client.
   */
  protected void doRightRequest(AgentId user, int right) {
    // If the request does not unset a reader, doing nothing.
    if (right != -READ) return;

    if (user != null) {
      // Identified user: removing it.
      setSave(); // state change, so save.
      subscribers.remove(user);
      selectors.remove(user);
      forward(user, new ExceptionReply(new AccessException("READ right removed.")));
    } else {
      // Free reading right removed: removing all non readers.
      for (Iterator<AgentId> subs = subscribers.iterator(); subs.hasNext();) {
        user = (AgentId) subs.next();
        if (! isReader(user)) {
          setSave(); // state change, so save.
          subs.remove();
          selectors.remove(user);
          forward(user, new ExceptionReply(new AccessException("READ right removed.")));
        }
      }
    }
  }

  /**
   * Method specifically processing a <code>ClientMessages</code> instance.
   * <p>
   * This method may forward the messages to the topic father if any, or
   * to the cluster fellows if any.It may finally send
   * <code>TopicMsgsReply</code> instances to the valid subscribers.
   */
  protected void doClientMessages(AgentId from, ClientMessages not, boolean throwsExceptionOnFullDest) {
    doClientMessages(from, not, false, throwsExceptionOnFullDest);
  }

  private void doClientMessages(AgentId from, ClientMessages not, boolean fromCluster, boolean throwsExceptionOnFullDest) {
    ClientMessages clientMsgs = preProcess(from, not);
    if (clientMsgs != null) {
      // Forwarding the messages to the father or the cluster fellows, if any:
      forwardMessages(from, clientMsgs, fromCluster);

      // Processing the messages:
      processMessages(from, clientMsgs);

      postProcess(clientMsgs);
    }
  }

  /**
   * Method specifically processing an <code>UnknownAgent</code> instance.
   * <p>
   * This method notifies the administrator of the failing cluster or
   * hierarchy building request, if needed, or removes the subscriptions of
   * the deleted client, if any, or sets the father identifier to null if it
   * comes from a deleted father.
   */
  protected void doUnknownAgent(UnknownAgent uA) {
    AgentId agId = uA.agent;
    Notification not = uA.not;

    if (not instanceof ClusterJoinNot) {
      ClusterJoinNot cT = (ClusterJoinNot) not;
      logger.log(BasicLevel.ERROR, "Cluster join failed: " + uA.agent + " unknown.");
      String info = "Cluster join failed: Unknown destination.";
      replyToTopic(new AdminReply(AdminReply.BAD_CLUSTER_REQUEST, info), cT.getReplyTo(),
          cT.getRequestMsgId(), cT.getReplyMsgId());
    } else if (not instanceof ClusterJoinAck || not instanceof ClusterRemoveNot) {
      logger.log(BasicLevel.ERROR, "Cluster error: " + uA.agent + " unknown. "
          + "The topic has probably been removed in the meantime.");
      clusterRemove(agId);
    } else {
      setSave(); // state change, so save.
      // Removing the deleted client's subscriptions, if any.
      subscribers.remove(agId);
      selectors.remove(agId);

      // Removing the father identifier, if needed.
      if (agId.equals(fatherId))
        fatherId = null;
    }
  }

  /**
   * Method specifically processing a <code>fr.dyade.aaa.agent.DeleteNot</code>
   * instance.
   * <p>
   * <code>UnknownAgent</code> notifications are sent to each subscriber and
   * <code>UnclusterNot</code> notifications to the cluster fellows.
   */
  protected void doDeleteNot(DeleteNot not) {
    AgentId clientId;

    // For each subscriber...
    for (int i = 0; i < subscribers.size(); i++) {
      clientId = (AgentId) subscribers.get(i);
      forward(clientId, new UnknownAgent(getId(), null));
    }

    clusterLeave();
    setSave(); // state change, so save.
  }

  /**
   * Actually forwards a list of messages to the father or the cluster
   * fellows, if any.
   */
  protected void forwardMessages(AgentId from, ClientMessages messages) {
    forwardMessages(from, messages, false);
  }
  
  protected TopicForwardNot createTopicForward(AgentId destId, ClientMessages messages, boolean fromCluster) {
    TopicForwardNot topicForwardNot = new TopicForwardNot(messages, fromCluster);
    if (destId.getTo() == AgentServer.getServerId()) {
      // Local destination
      // The initial notification may be persistent (e.g. if remote).
      // In that case the forward needs to be persistent too.
      topicForwardNot.setPersistent(messages.isPersistent());
      
      // Pass the callback in any cases (transient or persistent)
      // in order to provide flow control.
      messages.passCallback(topicForwardNot);
    } else {
      // Remote destination
      // The initial notification may be persistent.
      // In that case the forward needs to be persistent too.
      boolean persistent = messages.isPersistent();
      
      if (!persistent) {
        // Check if there is a persistent message to transmit.
        // In that case, the forward needs to be persistent.
        List<Message> msgList = messages.getMessages();
        for (Message msg : msgList) {
          if (msg.persistent) {
            persistent = true;
            break;
          }
        }
      }
      topicForwardNot.setPersistent(persistent);
      
      // Pass the callback in any cases (transient or persistent)
      // in order to provide flow control with the network.
      messages.passCallback(topicForwardNot);
    }
    return topicForwardNot;
  }

  protected void forwardMessages(AgentId from, ClientMessages messages, boolean fromCluster) {
    if (!fromCluster) {
      if (friends != null && friends.size() > 1) {
        for (Iterator<AgentId> e = friends.iterator(); e.hasNext();) {
          AgentId id = (AgentId) e.next();
          if (!id.equals(getId()))
            forward(id, createTopicForward(id, messages, true));
        }
      }
    }
    
    if (fatherId != null && !fatherId.equals(from)) {
      forward(fatherId, createTopicForward(fatherId, messages, false));
      if (DEBUG)
        logger.log(BasicLevel.DEBUG, "Messages forwarded to father " + fatherId.toString());
    }
  }

  /**
   * Actually processes the distribution of the received messages to the
   * valid subscriptions by sending a <code>TopicMsgsReply</code> notification
   * to the valid subscribers.
   */
  protected void processMessages(AgentId from, ClientMessages not) {
    List<Message> messages = not.getMessages();
    Message message;

    setNoSave();
    
    // TODO (AF): strict management of values ​​for counters.
    nbMsgsReceiveSinceCreation += messages.size();
    
    // Processing of all messages by interceptors. Sends the filtered messages to the DMQ, the others are distributed
    // to any subscriptions.
    if (interceptorsAvailable()) {
    	DMQManager dmqManager = null;
    	List<Message> newMessages = new ArrayList<Message>();
    	Iterator<Message> it = messages.iterator();
    	while (it.hasNext()) {
    		Message m = it.next();
    		// set the destination name to the message
    		m.setProperty("JoramDestinationName", getName());
    		message = processInterceptors(m);
    		if (message != null) {
    			newMessages.add(message);
    		} else {
    			//send message to the DMQ
    			if (dmqManager == null)
    				dmqManager = new DMQManager(dmqId, getId());
          nbMsgsSentToDMQSinceCreation++;
          dmqManager.addDeadMessage(m, MessageErrorConstants.INTERCEPTORS);
    		}
    	}
    	
    	if (dmqManager != null)
    		dmqManager.sendToDMQ();
    	
    	if (newMessages.isEmpty()) 
    		return;
    	
    	messages = newMessages;
    }
    
    // Internal boolean used for tagging local sendings.
    // Local notifications are not cloned, so if a message is sent to more than one local subscription it must
    // be explicitly cloned to avoid unwanted side effects. 
    boolean alreadySentLocally = false;

    // Browses all subscribers and determines the messages to send to each (according to the filter).
    for (Iterator<AgentId> subs = subscribers.iterator(); subs.hasNext();) {
      AgentId subscriber = (AgentId) subs.next();
      boolean local = (subscriber.getTo() == AgentServer.getServerId());
      String selector = (String) selectors.get(subscriber);
      
      boolean messagesDeliveredLocally = false;
      
      List<Message> deliverables;
      if (selector == null || selector.equals("")) {
        // Current subscriber does not filter messages, all messages will be sent.
        if (! local) {
          // Subscriber is not local: The notification will be serialized, and never modified locally. There is no
          //need to clone included messages.
          deliverables = messages;
        } else if (! alreadySentLocally) {
          // The messages have not yet been sent to any local subscription, there is no need to clone them.
          deliverables = messages;
          // Messages sent to next local subscription will be cloned.
          messagesDeliveredLocally = true;
        } else {
          // A local sending has already taken place, the messages must be cloned to avoid any unwanted side effects.
          deliverables = new Vector<Message>();
          for (Iterator<Message> msgs = messages.iterator(); msgs.hasNext();)
            deliverables.add((Message) msgs.next().clone());
        }
      } else {
        // Only messages satisfying the subscription filter should be sent.
        deliverables = new Vector<Message>();
        for (int i = 0; i < messages.size(); i++) {
          message = messages.get(i);
          
          if (Selector.matches(message, selector)) {
            // The message satisfies the filter, it must be transmitted.
            if (! local) {
              // Subscriber is not local: The notification will be serialized, and never modified locally. There is no
              //need to clone included messages.
              deliverables.add(message);
            } else if (! alreadySentLocally) {
              // No messages have yet been sent locally, there is no need to clone it.
              deliverables.add(message);
              // Messages sent to next local subscription will be cloned.
              messagesDeliveredLocally = true;
            } else {
              // A local sending has already taken place, the messages must be coned to avoid any unwanted side effects.
              deliverables.add((Message) message.clone());
            }
          }
        }  
      } 
      // If messages were delivered locally to this subscription, messages sent to future local subscriptions will need to be cloned.
      if (messagesDeliveredLocally)
        alreadySentLocally = true;
      
      // There are messages to send.
      if (! deliverables.isEmpty()) {
        TopicMsgsReply topicMsgsReply = new TopicMsgsReply(deliverables);
        // Locally reliability is ensured via the callback, if the subscriber is remote the notification must be persistent.
        topicMsgsReply.setPersistent(!local);
        
        // Set the callback in any cases: local and remote.
        // Remotely, the callback provides flow control.
        // Locally, the callback provides flow control and reliability for persistent messages.
        not.passCallback(topicMsgsReply);

        setDmq(topicMsgsReply);
        forward(subscriber, topicMsgsReply);
        // TODO (AF): strict management of values ​​for counters.
        nbMsgsDeliverSinceCreation = nbMsgsDeliverSinceCreation + deliverables.size();
      }
    }
  }

  private void setDmq(TopicMsgsReply not) {
    // Setting the producer's DMQ identifier field:
    if (dmqId != null) {
      not.setDMQId(dmqId);
    } else {
      not.setDMQId(Queue.getDefaultDMQId());
    }
  }

  public long getNbMsgsReceiveSinceCreation() {
    return nbMsgsReceiveSinceCreation;
  }
	
	// Flow Control related fields
  protected fr.dyade.aaa.common.stream.Properties getStats() {
    return null;
  }
}
