/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2010 - 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.Properties;

import org.objectweb.joram.shared.DestinationConstants;
import org.objectweb.joram.shared.excepts.MessageValueException;
import org.objectweb.joram.shared.messages.ConversionHelper;
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.common.Debug;

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

  /** Holds the distribution logic. */
  private DistributionHandler distributionHandler;
  
  private boolean isAsyncDistribution = false;

  public DistributionModule(String className, Properties properties, boolean firstTime) {
    try {
      distributionHandler = (DistributionHandler) Class.forName(className).newInstance();
    } catch (Exception exc) {
      logger.log(BasicLevel.ERROR, "DistributionModule: can't create distribution handler.", exc);
    }
    setProperties(properties, firstTime);
  }

  /**
   * Resets the distribution properties.
   */
  public void setProperties(Properties properties, boolean firstTime) {
    if (distributionHandler != null) {
      distributionHandler.init(properties, firstTime);
    }
    
    if (properties.containsKey(DestinationConstants.ASYNC_DISTRIBUTION_OPTION)) {
    	try {
    		isAsyncDistribution = ConversionHelper.toBoolean(properties.get(DestinationConstants.ASYNC_DISTRIBUTION_OPTION));
    	} catch (MessageValueException exc) {	}
    }
  }
  
  /**
   * Close the DistributionHandler.
   * Closes the specific distribution module and all associated resources.
   */
  public void close() {
    distributionHandler.close();
  }

  /**
   * 
   * Be careful, the message must be completely loaded (body).
   * 
   * @param fullMessage
   * @throws Exception
   */
  public void processMessage(Message fullMessage) throws Exception {
  	if (DEBUG)
  		logger.log(BasicLevel.DEBUG, "DistributionModule.processMessage(" + fullMessage.id + ") isAsyncDistribution = " + isAsyncDistribution);
  	
  	// Note (AF): this method is called even from preProcess (message coming from ClientMessage => full), or from the
  	// handling of WakeUpNot. In this last case getFullMessage is called).
  	if (!isAsyncDistribution) {
  		distributionHandler.distribute(fullMessage);
  	} else {
  		// the message are send by distributionDaemon.
  		throw new Exception("async distribution is on.");
  	}
  }

  public DistributionHandler getDistributionHandler() {
  	return distributionHandler;
  }
}
