/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2011 - 2025 ScalAgent Distributed Technologies
 *
 * 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): ScalAgent Distributed Technologies
 * Contributor(s): 
 */
package org.objectweb.joram.mom.dest;

import java.util.Comparator;

import org.objectweb.joram.shared.messages.Message;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.agent.AgentServer;
import fr.dyade.aaa.common.Daemon;
import fr.dyade.aaa.common.Debug;
import fr.dyade.aaa.common.EmptyQueueException;
import fr.dyade.aaa.common.Queue;

public class DistributionDaemon extends Daemon {
  private static final Logger logger = Debug.getLogger(DistributionDaemon.class.getName());
  private static final boolean DEBUG = logger.isLoggable(BasicLevel.DEBUG);

  /** Holds the distribution logic. */
  private DistributionHandler distributionHandler;
  
  private Queue distributeQueue;
  private Queue ackQueue;
  
  private Destination dest;
  private String acklistTxName;

  public DistributionDaemon(DistributionHandler distributionHandler, String destinationId, String destinationName, Destination dest) {
  	super("DistributionDaemon_" + destinationName, logger);
  	this.distributionHandler = distributionHandler;
    this.acklistTxName = destinationId + "AL";
  	distributeQueue = new Queue();
  	ackQueue = new Queue();
  	this.dest = dest;
  	if (DEBUG)
      logger.log(BasicLevel.DEBUG, "DistributionDaemon<> distributionHandler = " + distributionHandler + ", txDest = " + dest);
  }
  
  class ComparatorMessage implements Comparator {
    @Override
    public int compare(Object o1, Object o2) {
      if (((Message) o1).id.equals(o2)) return 0;
      return (o1.hashCode() - o2.hashCode());
    }
  }
  
  class ComparatorString implements Comparator {
    @Override
    public int compare(Object o1, Object o2) {
      if (o1.equals(o2)) return 0;
      return (o1.hashCode() - o2.hashCode());
    }
  }
  
  synchronized boolean isHandling(String id) {
    if (distributeQueue.search(new ComparatorMessage(), id)) return true;
    if (ackQueue.search(new ComparatorString(), id)) return true;
    return false;
  }
  
  synchronized void ackMessage(String id) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "DistributionDaemon run: distributeQueue.pop = " + id);
    
    // delete the message from the distributeQueue
    distributeQueue.pop();
    // add message id to the ackQueue
    ackQueue.push(id);
    
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "DistributionDaemon run: ackQueue.push : " + id);    
  }
  
	public void run() {
		if (DEBUG)
      logger.log(BasicLevel.DEBUG, "DistributionDaemon run()");
		
		while (running) {
      canStop = true;
      Message fullMessage = null;
      try {
      	if (DEBUG)
          logger.log(BasicLevel.DEBUG, "DistributionDaemon run: distributeQueue.size() = " + distributeQueue.size());
      	// Get the first message
      	fullMessage = (Message) distributeQueue.get(5000L);
      	if (fullMessage == null) continue;
      	
        // Note (AF): messages from distributeQueue are completely loaded (message coming from ClientMessage => full), or from the
        // handling of WakeUpNot. In this last case getFullMessage is called).
      	
      	if (DEBUG)
          logger.log(BasicLevel.DEBUG, "DistributionDaemon run: distributeQueue.get() = " + fullMessage.id);
      	
      	// test if this message is deliverable
      	if (isUndeliverable(fullMessage)) {
      		if (DEBUG)
            logger.log(BasicLevel.DEBUG, "DistributionDaemon run: the message " +  fullMessage.id + " is undeliverable.");
      		// delete the message from the distributeQueue
      		distributeQueue.pop();
      		continue;
      	}
      } catch (InterruptedException exc) {
        if (DEBUG)
          logger.log(BasicLevel.DEBUG, "DistributionDaemon run()", exc);
        return;
      }

      canStop = false;
      // process
      try {
      	// distribute the message
      	distributionHandler.distribute(fullMessage);
      	ackMessage(fullMessage.id);

        // transaction save the ack list
        AgentServer.getTransaction().save(ackQueue.list(), acklistTxName);
        AgentServer.getTransaction().begin();
        AgentServer.getTransaction().commit(true);
      } catch (Exception e) {
      	if (DEBUG)
          logger.log(BasicLevel.WARN, "DistributionDaemon run()", e);
      	else
          logger.log(BasicLevel.WARN, "DistributionDaemon run(): " + e.getMessage());
      	
      	if (e instanceof EmptyQueueException) {
      		continue;
      	}
      	
      	// Increment the delivery count of failed message
      	incDeliveryCount(fullMessage);
      	
        if (!running) return;

      	canStop = true;
      	// the connection is down, wait a wakeup from DistributionQueue or DistributionTopic.
      	synchronized (this) {
      	  try {
      	    if (DEBUG)
      	      logger.log(BasicLevel.DEBUG, "DistributionDaemon run: wait.");
      	    wait();
      	    if (DEBUG)
      	      logger.log(BasicLevel.DEBUG, "DistributionDaemon run: wakeup.");
      	  } catch (InterruptedException e1) {
      	    if (logger.isLoggable(BasicLevel.WARN))
      	      logger.log(BasicLevel.WARN, "DistributionDaemon run wait InterruptedException.");
      	  }
      	}
      }
    }
  }

	@Override
  protected synchronized void shutdown() {
		if (DEBUG)
      logger.log(BasicLevel.DEBUG, "DistributionDaemon shutdown()");
		notify();
		close();
  }

	@Override
  protected void close() {
		if (DEBUG)
      logger.log(BasicLevel.DEBUG, "DistributionDaemon close()");
	  distributeQueue.clear();
	  ackQueue.clear();
	  distributeQueue.close();
	  ackQueue.close();
  }

  /**
   * 
   * Be careful, the message must be completely loaded (body).
   * 
   * @param fullMessage
   * @throws Exception
   */
	public void push(Message fullMessage) {
		if (DEBUG)
      logger.log(BasicLevel.DEBUG, "DistributionDaemon.push(" + fullMessage.id + ')');
		
    // Note (AF): this method is called even from DistributionQueue.preProcess or DistributionTopic.postProcess (message coming
		// from ClientMessage => full), or from the handling of WakeUpNot. In this last case getFullMessage is called).

		distributeQueue.push(fullMessage);
  }

  public synchronized String getNextAck() {
    if (!ackQueue.isEmpty()) {
      try {
        return (String) ackQueue.getAndPop();
      } catch (InterruptedException exc) {
      }
    }
    return null;
  }
  
  public synchronized void cleanAckList() {
    ackQueue.clear();
  }

	public boolean isEmpty() {
	  return distributeQueue.isEmpty();
  }

  private boolean isUndeliverable(Message message) {
  	if (! isValid(message))
  		return true;
  	
  	int threshold = 0;
  	if (dest instanceof DistributionQueue) {
      threshold =((DistributionQueue)dest).getThreshold();
  	}
  	
  	if (DEBUG)
      logger.log(BasicLevel.DEBUG, "DistributionDaemon isUndeliverable: deliveryCount = " + message.deliveryCount + ", threshold = " + threshold);
  	
    if (threshold == 0) return false;
    if (threshold > 0)
      return (message.deliveryCount >= threshold);
    else if (org.objectweb.joram.mom.dest.Queue.getDefaultThreshold() > 0)
      return (message.deliveryCount >= org.objectweb.joram.mom.dest.Queue.getDefaultThreshold());
    return false;
  }
	
  /**
   * Returns <code>true</code> if the message is valid. 
   * The message is valid if not expired.
   */
  private boolean isValid(Message message) {
  	return (message.expiration <= 0) || (message.expiration > System.currentTimeMillis());
  }
  
  /** Increments the message delivery count. */
  private void incDeliveryCount(Message message) {
    message.deliveryCount += 1;
  }
}
