/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2001 - 2026 ScalAgent Distributed Technologies
 * 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
 * Contributor(s): Alex Porras (MediaOcean)
 */
package org.objectweb.joram.mom.proxies.tcp;

import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.util.StringTokenizer;
import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.objectweb.joram.mom.proxies.ConnectionManager;
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.common.Debug;

/**
 * Starts a TCP entry point for MOM clients.
 */
public class TcpProxyService implements TcpProxyServiceMBean {
  /** logger */
  private static final Logger logger = Debug.getLogger(TcpProxyService.class.getName());
  private static final boolean DEBUG = logger.isLoggable(BasicLevel.DEBUG);

  /**
   * Name the property that allow to fix the TCP SO_TIMEOUT property for the
   * client's connections.
   */
  public static final String SO_TIMEOUT_PROP = "org.objectweb.joram.mom.proxies.tcp.soTimeout";

  /**
   * Default value for the TCP SO_TIMEOUT property.
   */
  public static final int DEFAULT_SO_TIMEOUT = 5000;

  /**
   * Name the property that allow to fix the pool size for the
   * connection's listener.
   */
  public static final String POOL_SIZE_PROP = "org.objectweb.joram.mom.proxies.tcp.poolSize";

  /**
   * Default value for the pool size.
   */
  public static final int DEFAULT_POOL_SIZE = 1;

  /**
   * Name the property that allow to fix the TCP BACKLOG property for the
   * client's connections.
   */
  public static final String BACKLOG_PROP = "org.objectweb.joram.mom.proxies.tcp.backlog";

  /**
   * Default value for the TCP BACKLOG property.
   */
  public static final int DEFAULT_BACKLOG = 10;

  /**
   * Default value for the TCP port of the listen socket.
   */
  public static final int DEFAULT_PORT = 16010;

  /**
   * Default IP address for binding the listen socket.
   */
  public static final String DEFAULT_BINDADDRESS = "0.0.0.0"; // all

  private static final String MBEAN_NAME = "type=Connection,mode=tcp";

  /**
   * The proxy service reference (used to stop it).
   */
  private static TcpProxyService proxyService;
  
  private static ExecutorService executorService = null;
  
  public static boolean isExecutor() {
    return executorService != null;
  }
  
  static private boolean cnxWithNoAckedQueue = false;

  private int port;
  private int backlog;
  private String address;

  /**
   * The server socket listening to connections from the JMS clients.
   */
  private ServerSocket serverSocket = null;

  /**
   * Gets the listening socket.
   * 
   * @return the listening socket.
   */
  ServerSocket getServerSocket() {
    return serverSocket;
  }

  /**
   * Closes the listening socket and sets the variable to null.
   */
  protected void resetServerSocket() {
    if (serverSocket == null) return;

    try {
      serverSocket.close();
    } catch (IOException exc) {}
    serverSocket = null;
  }

  /**
   * Initialize the listening socket.
   *  
   * @param port
   * @param backlog
   * @param address
   * @return The initialized socket.
   * 
   * @throws Exception
   */
  protected ServerSocket createServerSocket(int port, int backlog, String address) throws Exception {
  	if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.createServerSocket(" + port + ',' + backlog + ',' + address + ')');
  	
  	ServerSocket serverSocket;
  	try {
  	  // TODO (AF): Use the ServerSocketFactory
  	  if (address.equals("0.0.0.0")) {
  	    serverSocket = new ServerSocket(port, backlog);
  	  } else {
  	    serverSocket = new ServerSocket(port, backlog, InetAddress.getByName(address));
  	  }
  	} catch (Exception exc) {
  	  if (DEBUG)
  	    logger.log(BasicLevel.ERROR,
  	               "TcpProxyService.createServerSocket(" + port + ',' + backlog + ',' + address + ')', exc);
  	  else
  	    logger.log(BasicLevel.ERROR,
  	               "TcpProxyService.createServerSocket(" + port + ',' + backlog + ',' + address + "): " + exc.getMessage());
  	  throw exc;
  	}
  	
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.createServerSocket serverSocket = " + serverSocket);
    
    return serverSocket;
  }
  
  public String getVersion() {
    return ConnectionManager.version;
  }

  /**
   * Initializes the TCP entry point by creating a server socket listening
   * to the specified port.
   * 
   * @param args stringified listening port
   * @param firstTime <code>true</code>  when the agent server starts.   
   */
  public static void init(String args, boolean firstTime) throws Exception {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.init(" + args + ',' + firstTime + ')');

    int port = DEFAULT_PORT;
    String address = DEFAULT_BINDADDRESS;
    if (args != null) {
      StringTokenizer st = new StringTokenizer(args);      
      port = Integer.parseInt(st.nextToken());
      if (st.hasMoreTokens()) {
        address = st.nextToken();
      }
    }
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.init() - binding to " + address + ", port " + port);

    int backlog = AgentServer.getInteger(BACKLOG_PROP, DEFAULT_BACKLOG).intValue();

    // Create the socket here in order to throw an exception
    // if the socket can't be created (even if firstTime is false).
    proxyService = new TcpProxyService(port, backlog, address);
    proxyService.start();
  }

  /**
   * Stops the service.
   */ 
  public static void stopService() {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.stop()");
    proxyService.stop();
  }

  /**
   * The list of opened connections
   */
  private Vector<TcpConnection> connections;

  /**
   * The thread listening to incoming
   * TCP connections.
   */
  private TcpConnectionListener[] connectionListeners;

  private boolean activated = false;

  public TcpProxyService(int port, int backlog, String address) throws Exception {
    // Memorizes values for future socket creation (see activate).
    this.port = port;
    this.backlog = backlog;
    this.address = address;

    this.serverSocket = createServerSocket(port, backlog, address);

    int poolSize = AgentServer.getInteger(POOL_SIZE_PROP, DEFAULT_POOL_SIZE).intValue();
    int timeout = AgentServer.getInteger(SO_TIMEOUT_PROP, DEFAULT_SO_TIMEOUT).intValue();
    
    this.connections = new Vector<TcpConnection>();
    connectionListeners = new TcpConnectionListener[poolSize];
    for (int i = 0; i < poolSize; i++) {
      connectionListeners[i] = new TcpConnectionListener(this, "TcpConnectionListener#" +i, timeout);
    }

    logger.log(BasicLevel.INFO,
                 "TcpProxy initialized, listening to " + address + ':' + port + ", backlog=" + backlog + ", SoTimeout=" + timeout + ", listener pool=" + poolSize);
  }

  protected void start() {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.start()");
    ConnectionManager.getCurrentInstance().addManager(this);
    activate();
  }

  public String getMBeanName() {
    return MBEAN_NAME;
  }

  void registerConnection(TcpConnection tcpConnection) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.registerConnection(" + tcpConnection + ')');
    connections.addElement(tcpConnection);
  }

  void unregisterConnection(TcpConnection tcpConnection) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.unregisterConnection(" + tcpConnection + ')');
    connections.removeElement(tcpConnection);
  }

  TcpConnection getConnection(AgentId proxyId, int key) {
    for (int i = 0; i < connections.size(); i++) {
      TcpConnection tc = connections.elementAt(i);
      if (tc.getProxyId() == proxyId &&
          tc.getKey() == key) {
        return tc;
      }
    }
    return null;
  }

  private void stop() {
    deactivate();
    closeAllConnections();
    ConnectionManager.getCurrentInstance().removeManager(this);
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.stoped.");
  }

  /**
   * Activates the TcpProxyService (TCP/JMS connector), creates the socket and starts listeners.
   */
  @Override
  public void activate() {
  	if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.activate()");
  	
  	if (activated) {
  	  logger.log(BasicLevel.WARN, "TcpProxyService.activate(): Already active, do nothing.");
  	  return;
  	}
  	
    try {
      if (cnxWithNoAckedQueue && executorService == null) {
        createExecutors();
      }
      
      if (serverSocket == null)
        serverSocket = createServerSocket(port, backlog, address);

      // Note: If there is a failure during socket initialization, an exception is thrown, serverSocket is
      // null and activated is false.
      
      for (int i = 0; i < connectionListeners.length; i++) {
        if (!connectionListeners[i].isRunning())
          connectionListeners[i].start();
      }
      activated = true;
      logger.log(BasicLevel.INFO, "TcpProxyService activated.");
    } catch (Exception exc) {
      if (logger.isLoggable(BasicLevel.ERROR))
        logger.log(BasicLevel.ERROR, "TcpProxyService.activate()", exc);
    }
  }

  public synchronized static void createExecutors() {
    if (executorService != null)
      return;
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.createExecutors()");
    int poolSize = Integer.getInteger("joram.QueueWorkerPoolSize", 5);
    executorService = Executors.newFixedThreadPool(poolSize);
    cnxWithNoAckedQueue = true;
  }
  
  public synchronized static void removeExecutors() {
    if (executorService == null)
      return;
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.removeExecutors()");
    executorService.shutdown();
    executorService = null;
  }
  
  public static void execute(Runnable command) {
    if (DEBUG)
      logger.log(BasicLevel.DEBUG, "TcpProxyService.execute(" + command + ')');
    try {
      executorService.execute(command);
    } catch (Exception e) {
      logger.log(BasicLevel.ERROR, "TcProxyService.execute:: EXCEPTION : " + e.getMessage());
    }
  }
  
  /**
   * Closes all active connections.
   */
  @Override
  public void closeAllConnections() {
    Vector<TcpConnection> stopList = (Vector<TcpConnection>) connections.clone();
    for (int i = 0; i < stopList.size(); i++) {
      TcpConnection tc = (TcpConnection) stopList.elementAt(i);
      tc.close();
    }
  }

  /**
   * Deactivates the TcpProxyService (TCP/JMS connector), stops all listeners.
   */
  @Override
  public void deactivate() {
    if (! activated) {
      logger.log(BasicLevel.WARN, "TcpProxyService already stopped, do nothing.");
      return;
    }

    for (int i = 0; i < connectionListeners.length; i++) {
      connectionListeners[i].stop();
    }
    
    // shutdown the executor service
    removeExecutors();
    
    activated = false;
    logger.log(BasicLevel.INFO, "TcpProxyService deactivated.");
  }

  /**
   * Returns true if the TcpProxyService (TCP/JMS connector) is started.
   * @return true if the TcpProxyService is started.
   */
  @Override
  public boolean isActivated() {
    return activated;
  }

  /**
   * Gets the number of active connections.
   * 
   * @return the number of active connections.
   */
  @Override
  public int getRunningConnectionsCount() {
    return connections.size();
  }

  /**
   * Gets the number of threads listening for incoming tcp connections.
   * 
   * @return the tcp listeners pool size.
   */
  @Override
  public int getTcpListenersPoolSize() {
    return connectionListeners.length;
  }

  /**
   * Gets the average time for incoming tcp connections.
   * 
   * @return the average time for incoming tcp connections.
   */
  @Override
  public long getAverageConnectionTime() {
    long average = 0;
    for (TcpConnectionListener listener : connectionListeners)
      average += listener.getAverage();
    return average / connectionListeners.length;
  }

  /**
   * Gets the listen port of the server.
   * 
   * @return the listen port of the server.
   */
  @Override
  public int getListenPort() {
    return port;
  }
  
  /**
   * Gets the socket address of the server.
   * 
   * @return the server socket address.
   */
  @Override
  public String getServerAddress() {
    return address;
  }

  /**
   * Gets the number of failed login.
   * 
   * @return the number of failed login.
   */
  @Override
  public int getFailedLoginCount() {
    int failedCount = 0;
    for (int i = 0; i < connectionListeners.length; i++) {
      failedCount += connectionListeners[i].getFailedLoginCount();
    }
    return failedCount;
  }

  /**
   * Gets the number of connections initiated since boot time.
   * 
   * @return the number of connections initiated since boot time.
   */
  @Override
  public int getInitiatedConnectionCount() {
    int initCount = 0;
    for (int i = 0; i < connectionListeners.length; i++) {
      initCount += connectionListeners[i].getInitiatedConnectionCount();
    }
    return initCount;
  }

  /**
   * Gets the number of connections rejected due to a wrong protocol header.
   * 
   * @return the number of connections rejected due to a wrong protocol header.
   */
  @Override
  public int getProtocolErrorCount() {
    int errorCount = 0;
    for (int i = 0; i < connectionListeners.length; i++) {
      errorCount += connectionListeners[i].getProtocolErrorCount();
    }
    return errorCount;
  }
}
