/*
 * JORAM: Java(TM) Open Reliable Asynchronous Messaging
 * Copyright (C) 2004 - 2022 ScalAgent Distributed Technologies
 * Copyright (C) 2004 France Telecom R&D
 *
 * 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.proxies.tcp;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;

import org.objectweb.joram.mom.dest.AdminTopic;
import org.objectweb.joram.mom.notifications.GetProxyIdNot;
import org.objectweb.joram.mom.proxies.GetConnectionNot;
import org.objectweb.joram.mom.proxies.OpenConnectionNot;
import org.objectweb.joram.mom.proxies.ReliableConnectionContext;
import org.objectweb.joram.mom.proxies.ConnectionContext.Type;
import org.objectweb.joram.shared.security.Identity;
import org.objectweb.joram.shared.stream.MetaData;
import org.objectweb.util.monolog.api.BasicLevel;
import org.objectweb.util.monolog.api.Logger;

import fr.dyade.aaa.agent.AgentId;
import fr.dyade.aaa.common.Configuration;
import fr.dyade.aaa.common.Daemon;
import fr.dyade.aaa.common.Debug;
import fr.dyade.aaa.common.stream.StreamUtil;

/**
 * Listens to the TCP connections from the JMS clients.
 * Creates a <code>TcpConnection</code> for each accepted TCP connection.
 * Opens the <code>UserConnection</code> with the right user's proxy.
 */
public class TcpConnectionListener extends Daemon {
  /** logger */
  private static final Logger logger = Debug.getLogger(TcpConnectionListener.class.getName());

  /**
   * The TCP proxy service 
   */
  private TcpProxyService proxyService;

  private int timeout;

  /**
   *  Name of the property allowing to change the threshold of warning for the
   * verification of the synchronization between the client and server clock.
   *  A warning is generated if there is more than this value in milliseconds
   * between the two clocks.
   * <p>
   *  By default the value is 1000 milliseconds. 
   */
  public static final String CLOCK_SYNCHRO_THRESHOLD = "org.objectweb.joram.TcpConnection.ClockSynchro.Threshold";

  /**
   *  Value of the threshold of warning for the verification of the synchronization
   * between the client and server clock.
   *  A warning is generated if there is more than this value in milliseconds between
   * the two clocks.
   * <p>
   *  By default the value is 1000 milliseconds. 
   */
  private long clockSynchroThreshold = 1000L;

  /**
   *  Name of the property allowing to change the default value of timeout during
   * TCP connection. This timeout is overloaded by value of cnxPendingTimer parameter
   * from client if defined.
   * <p>
   *  By default the value is 5000 milliseconds. 
   */
  public static final String CONNECTING_TIMER = "org.objectweb.joram.TcpConnection.connectingTimer";

  /**
   *  Value of timeout during TCP connection. This timeout is overloaded by value of
   * cnxPendingTimer parameter from client if defined.
   * <p>
   *  By default the value is 5000 milliseconds. 
   */
  private int connectingTimer = 5000;

  /**
   * Number of times this connection listener has encountered an erroneous
   * authentication.
   */
  private int failedLoginCount;

  /**
   * Number of connections started with this connection listener.
   */
  private int connectionCount;

  /**
   * Number of times this connection listener has encountered an erroneous magic
   * number or protocol version.
   */
  private int protocolErrorCount;
  
  /**
   * Creates a new connection listener
   *
   * @param proxyService  the TCP proxy service associated with this connection listener
   * @param timeout       the timeout
   */
  public TcpConnectionListener(TcpProxyService proxyService, int timeout) {
    super("TcpConnectionListener", logger);
    this.proxyService = proxyService;
    this.timeout = timeout;
    
    clockSynchroThreshold = Configuration.getLong(CLOCK_SYNCHRO_THRESHOLD, clockSynchroThreshold);
    connectingTimer = Configuration.getInteger(CONNECTING_TIMER, connectingTimer);
  }

  @Override
  public void run() {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "TcpConnectionListener.run()");

    // Wait for the administration topic deployment.
    // TODO (AF): a synchronization would be much better.
    try {
      Thread.sleep(500);
    } catch (InterruptedException exc) {
      // continue
    }

    while (running) {
      canStop = true;
      try {
        acceptConnection();
      } catch (Exception exc) {
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.INFO, "TcpConnectionListener.run()", exc);
        else
          logger.log(BasicLevel.INFO, "TcpConnectionListener.run()");
      }
    }
  }

  static class NetOutputStream extends ByteArrayOutputStream {
    private OutputStream os = null;

    NetOutputStream(Socket sock) throws IOException {
      super(1024);
      reset();
      os = sock.getOutputStream();
    }

    @Override
    public synchronized void reset() {
      count = 4;
    }

    public void send() throws IOException {
      try {
        buf[0] = (byte) ((count -4) >>>  24);
        buf[1] = (byte) ((count -4) >>>  16);
        buf[2] = (byte) ((count -4) >>>  8);
        buf[3] = (byte) ((count -4) >>>  0);

        writeTo(os);
        os.flush();
      } finally {
        reset();
      }
    }
  }

  /**
   * Accepts a TCP connection. Opens the <code>UserConnection</code> with the
   * right user's proxy, creates and starts the <code>TcpConnection</code>.
   */
  private void acceptConnection() throws Exception {
    if (logger.isLoggable(BasicLevel.DEBUG))
      logger.log(BasicLevel.DEBUG, "TcpConnectionListener.acceptConnection()");

    Socket sock = proxyService.getServerSocket().accept();
    String inaddr = sock.getInetAddress().getHostAddress();

    connectionCount++;

    if (logger.isLoggable(BasicLevel.INFO))
      logger.log(BasicLevel.INFO, " -> accept connection from " + inaddr);

    try {
      sock.setTcpNoDelay(true);

      // Fix bug when the client doesn't use the right protocol (e.g. Telnet)
      // and blocks this listener.
      sock.setSoTimeout(timeout);

      InputStream is = sock.getInputStream();
      NetOutputStream nos = new NetOutputStream(sock);

      byte[] magic = null;
      try {
        magic = StreamUtil.readByteArrayFrom(is, 8);
      } catch (IOException exc) {
        String errorMsg = "Connection from " + sock.getInetAddress() + ':' + sock.getPort() + " cannot read magic number. Client is not compatible with JORAM.";
        protocolErrorCount++;
        throw new IllegalAccessException(errorMsg);
      }
      
      for (int i = 0; i < 5; i++) {
        if (magic.length == i || magic[i] != MetaData.joramMagic[i] && magic[i] > 0) {
          String errorMsg = "Connection from " + sock.getInetAddress() + ':' + sock.getPort() + " bad magic number. Client is not compatible with JORAM.";
          protocolErrorCount++;
          throw new IllegalAccessException(errorMsg);
        }
      }
      if (magic[7] != MetaData.joramMagic[7]) {
        if (magic[7] > 0 && MetaData.joramMagic[7] > 0) {
          String errorMsg = "Connection from " + sock.getInetAddress() + ':' + sock.getPort() + " bad protocol version number " + magic[7] + " != " + MetaData.joramMagic[7];
          protocolErrorCount++;
          throw new IllegalAccessException(errorMsg);
        }
        
        logger.log(BasicLevel.WARN,
                   "Wildcard protocol version number: from stream = " + magic[7] + ", from MetaData = " + MetaData.joramMagic[7]);
      }

      //read the ack mode(noAckedQueue)
      boolean noAckedQueue = StreamUtil.readBooleanFrom(is);
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, " -> read noAckedQueue = " + noAckedQueue);
      if (noAckedQueue)
        TcpProxyService.createExecutors();
      
      long dt = Math.abs(StreamUtil.readLongFrom(is) - System.currentTimeMillis());
      if (dt > clockSynchroThreshold)
        logger.log(BasicLevel.WARN, " -> bad clock synchronization between client and server: " + dt);
      StreamUtil.writeTo(dt, nos);

      Identity identity = Identity.read(is);
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, " -> read identity = " + identity);

      int key = StreamUtil.readIntFrom(is);
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, " -> read key = " + key);

      int heartbeat = 0;
      if (key == -1) {
        heartbeat = StreamUtil.readIntFrom(is);
        if (logger.isLoggable(BasicLevel.DEBUG))
          logger.log(BasicLevel.DEBUG, " -> read heartBeat = " + heartbeat);
      }
      
      if (logger.isLoggable(BasicLevel.INFO))
        logger.log(BasicLevel.INFO, " -> open connection " + identity + "/" + key + " - " + heartbeat);
      
      GetProxyIdNot gpin = new GetProxyIdNot(identity, inaddr);
      AgentId proxyId;
      try {
        gpin.invoke(AdminTopic.getDefault(), connectingTimer);
        proxyId = gpin.getProxyId();
        if (proxyId == null) {
          String msg = "JMS connection cancelled after " + (connectingTimer /1000) + " secs.";
          IllegalStateException jmsExc = new IllegalStateException(msg);
          
          if (logger.isLoggable(BasicLevel.WARN))
            logger.log(BasicLevel.WARN,
                       " -> JMS connection cancelled after " + (connectingTimer /1000) + " secs.", jmsExc);
          
          throw jmsExc;
        }
      } catch (Exception exc) {
        if (logger.isLoggable(BasicLevel.WARN))
          logger.log(BasicLevel.WARN, " -> login failed", exc);
        failedLoginCount++;
        StreamUtil.writeTo(1, nos);
        StreamUtil.writeTo(exc.getMessage(), nos);
        nos.send();
        return;
      }

      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.DEBUG, " -> open connection, proxyId=" + proxyId);
      
      IOControl ioctrl;
      ReliableConnectionContext ctx;
      if (key == -1) {
        OpenConnectionNot ocn = new OpenConnectionNot(Type.RELIABLE, heartbeat, noAckedQueue);
        try {
          ocn.invoke(proxyId, connectingTimer);
          ctx = (ReliableConnectionContext) ocn.getConnectionContext();
          if (ctx == null) {
            String msg = "JMS connection cancelled after " + (connectingTimer /1000) + " secs.";
            IllegalStateException jmsExc = new IllegalStateException(msg);

            if (logger.isLoggable(BasicLevel.WARN))
              logger.log(BasicLevel.WARN,
                         "JMS connection cancelled after " + (connectingTimer /1000) + " secs.", jmsExc);
            
            // Note: The probability that the notification has been processed by the user proxy is
            // very small. In this case we have only to clean the context object in the proxy's connection
            // table. In the best case this item will be cleaned up if the heartbeat is activated. In the
            // worst case it will be cleaned at restart.
            // If we really want to treat this case it would be necessary to issue a CloseConnectionNot with
            // the connection key.

            // TODO (AF): We have to clean the pending connection if any.

            throw jmsExc;
          }
        } catch (Exception exc) {
          if (logger.isLoggable(BasicLevel.WARN))
            logger.log(BasicLevel.WARN, " -> login failed", exc);
          failedLoginCount++;
          StreamUtil.writeTo(1, nos);
          StreamUtil.writeTo(exc.getMessage(), nos);
          nos.send();
          return;
        }

        StreamUtil.writeTo(0, nos);
        key = ctx.getKey();
        StreamUtil.writeTo(key, nos);
        nos.send();
        ioctrl = new IOControl(sock);
      } else {
        GetConnectionNot gcn = new GetConnectionNot(key);
        try {
          gcn.invoke(proxyId);
        } catch (Exception exc) {
          if (logger.isLoggable(BasicLevel.WARN))
            logger.log(BasicLevel.WARN, "TcpConnectionListener: reconnection failed", exc);
          StreamUtil.writeTo(1, nos);
          StreamUtil.writeTo(exc.getMessage(), nos);
          nos.send();
          return;
        }
        ctx = (ReliableConnectionContext) gcn.getConnectionContext();
        StreamUtil.writeTo(0, nos);
        nos.send();
        ioctrl = new IOControl(sock, ctx.getInputCounter());

        // Close the remaining connection if it exists
        TcpConnection tcpConnection = proxyService.getConnection(proxyId, key);
        if (tcpConnection != null) {
          tcpConnection.close();
        }
      }

      // Reset the timeout in order to enable the server to indefinitely
      // wait for requests.
      sock.setSoTimeout(0);

      TcpConnection tcpConnection = new TcpConnection(ioctrl, ctx, proxyId, proxyService, identity);
      tcpConnection.start();
    } catch (IllegalAccessException exc) {
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.ERROR, "TcpConnectionListener: close connection", exc);
      else
        logger.log(BasicLevel.ERROR, "TcpConnectionListener: close connection, " + exc.getMessage());
      sock.close();
      throw exc;
    } catch (IOException exc) {
      if (logger.isLoggable(BasicLevel.DEBUG))
        logger.log(BasicLevel.WARN, "TcpConnectionListener: close connection", exc);
      else
        logger.log(BasicLevel.WARN, "TcpConnectionListener: close connection, " + exc.getMessage());
      sock.close();
      throw exc;
    }
  }

  @Override
  protected void shutdown() {
    close();
  }

  @Override
  protected void close() {
    proxyService.resetServerSocket();
  }

  public int getFailedLoginCount() {
    return failedLoginCount;
  }

  public int getInitiatedConnectionCount() {
    return connectionCount;
  }

  public int getProtocolErrorCount() {
    return protocolErrorCount;
  }
}
