package nxm.ucl.prim;

import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.Vector;

import nxm.sys.lib.BaseFile;
import nxm.sys.lib.Data;
import nxm.sys.lib.DataFile;
import nxm.sys.lib.MidasException;
import nxm.sys.lib.Primitive;
import nxm.sys.lib.Time;

/**
  zpipe - NeXtMidas version of ZPIPE

  @author Jay Hallam
*/
public class zpipe extends Primitive {

  private DataFile hcb;   // The input or output data file
  private Data data;      // The data read from file
  private int port;       // The socket port
  private String host;    // The hostname (client only)

  private Vector<SocketConnection> connections = new Vector<>(); // List of connections
  private ServerSocket server;               // The Server Socket
  private Socket client;                     // The Client Socket
  private InputStream inputStream = null;    // The input stream for reader
  private ConnectionThread connThread = null;// The Accept Connection Thread
  private boolean isServer     = false;      // Write to a socket
  private boolean isSocketWriter = true;     // Write to a socket
  private double ConnAttemptInterval = 1.0;  // Maximum time to wait for connections
  private int NumBytesRead = 0;              //
  private int totalBytesToRead = 0;          //
  private int sockLinger = 1;                // Socket Linger Time in sec
  private int sockBufSize = 0;               // Socket Buffer Size in Bytes
  private double socketTimeout = 2.0;        // SO_TIMEOUT in seconds
  private boolean warn = true;               // Turn on/off all warnings
  private boolean haveTLSwitch = false;      // Passed in /TL
  /** Turn on/off all debug */
  public int dbg = 0;

  /** default constructor */
  public zpipe () {}

  @Override
  public int open () {
    //M.warning("This is a BETA version of ZPIPE.  Expect changes in future releases.  "
    //         +"Not all original X-Midas capabilities are implemented yet");

    // Get parameters
    dbg = MA.getL("/DEBUG",0);
    warn  = MA.getState("/WARN", warn);

    boolean reader = MA.getState("/XM_IN",  false);
    boolean writer = MA.getState("/XM_OUT", false);
    if (reader && writer) {
      M.error("You can not be both a socket reader and writer");
    }
    if (!reader && !writer) {
      isSocketWriter = true; //Default
    }
    else {
      isSocketWriter = writer;
    }

    host = MA.getCS("HOST", host);
    port = MA.getL("PORT", port);
    info("writer? "+isSocketWriter+" host "+host+" port "+port);

    ConnAttemptInterval  = MA.getD("/ACCEPTTIME", ConnAttemptInterval);
    boolean reuseAddress = MA.getState("/REUSEADDR",true);
    socketTimeout        = MA.getD("/TIMEOUT",socketTimeout);
    sockLinger           = MA.getL("/LINGER",sockLinger);
    sockBufSize          = MA.getL("/SOCKBUF",sockBufSize); // This only gets set if it is greater than the initial socketBufferSize
    String format        = MA.getS("/FORMAT","SB");
    isServer             = isEmptyString(host);
    isServer             = MA.getState("/SERVER", isServer);
    haveTLSwitch         = MA.find("/TL");
    int xfer             = MA.getL("/TL",32768);

    // Open type-3000 file for INPUT
    hcb = MA.getDataFile("IN");
    hcb.setFormat(format);
    if (isSocketWriter) {
      M.info("Opening "+hcb.getURL()+" for INPUT");
      hcb.open(BaseFile.INPUT);
    }
    else {
      M.info("Opening "+hcb.getURL()+" for OUTPUT");
      hcb.open(BaseFile.OUTPUT);
    }
    if (!hcb.isOpen()) {
      M.warning ("'"+hcb.getFileName().getFullName()+"' does not exist");
      return FINISH;
    }
    // Set the transfer length if the user did not.
    if (!haveTLSwitch) xfer = Math.max(1,(int)(32768/hcb.dbpe));

    data = hcb.getDataBuffer(xfer);
    totalBytesToRead = data.buf.length;
    debug("xfer "+xfer+" format "+format+" dbpe "+hcb.dbpe+" buffer length "+totalBytesToRead);

    info("server? "+isServer);
    if  (isServer) {
      try {
        server = new ServerSocket(port);
        server.setReuseAddress(reuseAddress);
        server.setSoTimeout((int)(socketTimeout*1000.0)); // Timeout in millisecs
        if (sockBufSize > server.getReceiveBufferSize()) server.setReceiveBufferSize(sockBufSize);

        acceptConnections();
      }
      catch (Exception e) {
        throw new MidasException("Unable to open server socket on port "+port, e);
      }
    }
    else {
      // Client
      client = connectToSocket(host, port);
    }

    return NORMAL;
  }

  @Override
  public int process () {
    int status = NORMAL;
    if (isServer) {
      if (isSocketWriter) {
        status = processServerSocketWriter();
      }
      else {
        warn("Server socket reader is TBD");
        status = FINISH;
      }
    }
    else {
      //debug("process... client "+client);
      if (isSocketWriter) {
        M.warning("Client socket writer is TBD");
        status = FINISH;
      }
      else {
        status = processClientSocketReader();
        // RESTART means we read a -1 or client == null and we need to try reconnect
        if (status == RESTART) {
          if (client != null) closeConnection(client);  // Server disconnect close current client
          Time.sleep(ConnAttemptInterval);

          client = connectToSocket(host, port);
          status = NOOP;  // Set to NOOP for process()
        }
      }
    }
    return status;
  }

  @Override
  public int close () {
    info("close()::closing...");
    if (hcb != null) {
      hcb.close();
    }

    if (getNumConnections() > 0) closeAllConnections();
    if (connThread != null) connThread.interrupt();
    closeConnection(server);
    closeConnection(client);
    return NORMAL;
  }

  /** This is a client that reads a file/pipe and writes to a socket(s) */
  private int processClientSocketReader () {
    int status = NOOP;

    // If server connection not there try to reconnect
    if (client == null) {
      return RESTART;
    }

    try {
      int numRead = 0;
      int numToReadThisPass = totalBytesToRead - NumBytesRead;

      debug("reading from socket: have ("+NumBytesRead+") of ("+totalBytesToRead+") bytes, with ("+
            inputStream.available()+") bytes avail", 1);

      numRead = inputStream.read(data.buf, NumBytesRead, numToReadThisPass);
      NumBytesRead += numRead;
      debug("Num bytes read "+NumBytesRead, 1);

      if (numRead > 0) {
        status = NORMAL;
      }
      else if (numRead == -1) {
        // We use restart to indicate that the socket was closed (since we read -1 from InputStream)
        status = RESTART;
      }
      if (NumBytesRead == totalBytesToRead) {
        hcb.write(data);
        if (verbose) System.out.println("Writing to file NumBytesRead = " + NumBytesRead);
        NumBytesRead = 0;
      }
      else if (numRead < 0 && NumBytesRead != 0) {
        warn("End of socket: connection is in middle of read.");
        NumBytesRead = 0;
      }
    }
    catch (Exception e) {
      //jph -- TBD  -- count number of exceptions here and error out...
      warn("Exception "+e+" reading input stream");
      NumBytesRead = 0;
    }
    return status;
  }

  /** This is a server that reads a file/pipe and writes to a socket(s) */
  private int processServerSocketWriter () {
    int nelem = hcb.read(data);

    if (nelem > 0) {
      boolean wroteToSocket = false;
      int numBytes = (int)hcb.getBPE()*nelem;
      int offset   = data.boff;
      debug("Got "+nelem+" elements and "+numBytes+" bytes",2);

      //Iterate backwards so I can remove from connections vector
      for (int i = connections.size()-1; i >=0; i--) {
        SocketConnection sc = connections.get(i);
        OutputStream     os = sc.getOutputStream();
        Socket         sock = sc.getSocket();

        if (sock.isClosed()) {
          debug("------> SOCKET CLOSED removing a connection at "+i);
          connections.remove(i);
        }
        else if (!sock.isConnected()) {
          debug("------> SOCKET not CONNECTED removing a connection at "+i);
          connections.remove(i);
          closeConnection(sc);
        }
        else if (!sock.isBound()) {
          debug("------> SOCKET not BOUND removing a connection at "+i);
          connections.remove(i);
          closeConnection(sc);
        }
        else {
          try {
            os.write(data.buf, offset, numBytes);
            debug("Wrote "+nelem+" elements "+numBytes+" (bytes) to "+sc,1);
            wroteToSocket = true;
          }
          catch (IOException e) {
            M.warning("Unable to write to socket output stream number "+i+" for "+sc);
            if (sock != null) {
              debug("Connected? "+sock.isConnected()+" bound? "+sock.isBound()+" closed? "+sock.isClosed());
              connections.remove(i);
              closeConnection(sc);
            }
            if ("Connection reset".equalsIgnoreCase(e.getMessage()) ||
                "Broken pipe".equalsIgnoreCase(e.getMessage())) {
              debug("Client Side Broke connection",1);
            }
            else {
              e.printStackTrace();
            }
          }
        }
      }
      if (wroteToSocket) {
        //System.out.println("Wrote "+numBytes+" bytes to "+getNumConnections()+" socket(s)");
        return NORMAL;
      } else {
        return NOOP;    // Pause if nothing written to socket
      }
    } else {
      // no data
      debug("No data ... FINISHING");
      return FINISH;
    }
  }

  private Socket connectToSocket(String host, int port) {
    Socket sock = null;
    info("Attempting to connect to "+host+" port "+port);
    try {
      sock = new Socket(host, port);
      sock.setSoTimeout((int)(socketTimeout*1000.0));  // Timeout in millisecs
      sock.setSoLinger(true,sockLinger);
      if (sockBufSize > sock.getReceiveBufferSize()) sock.setReceiveBufferSize(sockBufSize);

      // No need to do bind..
      inputStream = new BufferedInputStream (sock.getInputStream());
      M.info("Created socket to remotehost '"+host+" port "+port+"', client "+sock);
    }
    catch (Exception e) {
      debug("Unable to open client socket to host "+host+" port "+port+", is server up?");
      Time.sleep(ConnAttemptInterval);
    }
    return sock;
  }

  private void openConnection(Socket sock) {
    try {
      SocketConnection sc = new SocketConnection(sock);
      connections.add(sc);
    }
    catch (Exception e) {
      M.warning("Unable to open a connection for socket"+sock);
    }
  }

  /** Closes a socket connection. Socket can be null. */
  private void closeConnection(Socket sock) {
    if (sock != null) {
      try {
        sock.close();
      }
      catch (Exception e) {
        warn("Unable to socket connection to "+sock);
      }
    }
  }

  /** Closes a server socket connection. Socket can be null. */
  private void closeConnection(ServerSocket sock) {
    if (sock != null) {
      try {
        sock.close();
      }
      catch (Exception e) {
        warn("Unable to socket connection to "+sock);
      }
    }
  }

  private void closeConnection(SocketConnection sc) {
    M.info("Closing SocketConnection "+sc);
    Socket sock = sc.getSocket();
    if (!sock.isClosed()) {
      try {
        sock.close();
      }
      catch (Exception e) {
        M.warning("Unable to close connection to socket "+sc);
      }
    }
  }

  private synchronized void closeAllConnections() {
    for (SocketConnection socketConnection : connections) {
      closeConnection(socketConnection);
    }
    connections.clear();
  }

  /* True if string is null, zero length or the string null or NULL. */
  private boolean isEmptyString(String str) {
    return (str == null || str.length() == 0 || str.equalsIgnoreCase("NULL"));
  }
  private void debug (String str) {
    debug(str,0);
  }
  private void debug (String str, int indbg) {
    if (dbg > indbg) System.out.println("ZPIPE: "+str);
  }

  private void info(String str) {
    if (verbose) M.info("ZPIPE: "+str);
  }
  private void warn(String str) {
    if (warn) M.warning("ZPIPE: "+str);
  }

  /** Open a new ConnectionThread and start accepting incoming connections */
  public void acceptConnections () {
    M.info("Starting ACCEPT connections thread, server closed? "+server.isClosed());
    try {
      connThread = new ConnectionThread();
      connThread.start();
    }
    catch (Exception e) {
      M.warning("acceptConnections() failed to start ConnectionThread");
    }
  }

/* -------------------------------------------- Set/Get controls -------------------------------- */
  /** Return the number of connections
   *  @return the number of connections */
  public int getNumConnections() {
    return connections.size();
  }

  /** Get the maximum time to wait for connections
   *  @return the maximum time to wait for connections */
  public double getAcceptTime() {
    return ConnAttemptInterval;
  }
  /** Set the maximum time to wait for connections
   *  @param t the maximum time to wait for connections */
  public void setAcceptTime(double t) {
    ConnAttemptInterval = t;
  }

/* -------------------------------------------- Set/Get controls -------------------------------- */


  private class ConnectionThread extends Thread {
    public ConnectionThread() {
      super("ConnectionThread"); // Give the thread a name
    }

    public void run () {
      M.info("ConnectionThread: running ...");
      while (!server.isClosed()) {
        try {
          if (verbose) M.info("Waiting for socket ...");
          Socket sock = server.accept();

          if (sock != null) {
            M.info("Accepted a socket "+sock);
            openConnection(sock);
          } else {
            M.info("No socket ... ");
          }
        }
        catch (java.net.SocketTimeoutException e) {
          // Not an error, happens a lot during connection wait
          debug("Accept timeout exception: " +e,2);

        }
        catch (java.net.SocketException e) {
          if (server != null) {  // Not really and error
            if ("socket closed".equalsIgnoreCase(e.getMessage())) debug("Server Socket closed",1);
          } else {
            System.out.println("Socket Exception trying to accept connection: " +e);
            e.printStackTrace();
          }

        }
        catch (IOException e) {
          System.out.println("Error trying to accept connection: " +e);
          e.printStackTrace();
        }
      }
    }
  }

  /** Inner class to hold the socket and output stream information */
  private class SocketConnection {
    private Socket       socket; // The socket to write on.
    private OutputStream os;   // The output stream to write to.

    public SocketConnection (Socket s) throws IOException{
      socket = s;
      // Does Not Work as expected, Seems to be a max
      //if (sockBufSize > socket.getSendBufferSize()) socket.setSendBufferSize(sockBufSize);
      os   = socket.getOutputStream();
    }

    /** Gets the output socket. */
    public Socket getSocket() {
      return socket;
    }

    /** Gets the output stream. */
    public OutputStream getOutputStream() {
      return os;
    }

    public String toString() {
      return "SocketConnection: socket="+socket+" address="+socket.getInetAddress();
    }
  }

}