/*
* JBoss, Home of Professional Open Source
* Copyright 2005, JBoss Inc., and individual contributors as indicated
* by the @authors tag. See the copyright.txt in the distribution for a
* full listing of individual contributors.
*
* This 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 (at your option) any later version.
*
* This software 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 software; if not, write to the Free
* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
* 02110-1301 USA, or see the FSF site: http://www.fsf.org.
*/

/*
 * Created on Aug 13, 2005
 */
package org.jboss.test.remoting.transport.multiplex;

import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.util.Date;

import org.apache.log4j.Category;
import org.apache.log4j.FileAppender;
import org.apache.log4j.PatternLayout;
import org.jboss.jrunit.extensions.ServerTestCase;
import org.jboss.logging.Logger;
import org.jboss.remoting.transport.multiplex.MasterServerSocket;
import org.jboss.remoting.transport.multiplex.VirtualServerSocket;
import org.jboss.remoting.transport.multiplex.VirtualSocket;


/**
 * A MultiplexServer.
 *
 * @author <a href="mailto:r.sigal@computer.org">Ron Sigal</a>
 * @version $Revision: 583 $
 *          <p/>
 *          Copyright (c) 2005
 *          </p>
 */

public class MultiplexServer extends ServerTestCase implements MultiplexConstants
{
   private static final Logger log = Logger.getLogger(MultiplexServer.class);

   private ServerSocket serverSocket;
   private Socket scriptSocket;
   private InputStream scriptInputStream;
   private Socket testSocket;
   private int nextRemotePort = clientServerSocketPort - 1;
   private int nextBindPort = masterServerSocketPort - 1;

   private ObjectOutputStream os;
   private ObjectInputStream is;

   private byte[] script = new byte[1000];
   private int scriptLength;
   private boolean shutdownServer = false;
   private boolean testOver = false;


   public void setUp()
   {
      
      String pattern = "<%d{ABSOLUTE}> %5p [%t] (%F:%L) - %m%n";
      PatternLayout layout = new PatternLayout(pattern);
      
      try
      {
         File logFile = new File("test_logs");
         logFile.mkdir();
         FileAppender fileAppender = new FileAppender(layout, "test_logs" + File.separator + "server.output.log");
         fileAppender.setAppend(false);
         //fileAppender.setThreshold(Level.toLevel(testLogLevel));
         Category.getRoot().addAppender(fileAppender);
      }
      catch(IOException e)
      {
         e.printStackTrace();
      }
      
      new Thread(new Runnable()
      {
         public void run()
         {
            runBehaviorTests();
         }
      }).start();
   }

   protected void runBehaviorTests()
   {
      log.info("starting behavior tests");

      try
      {
         serverSocket = new MasterServerSocket(basicBehaviorServerPort, 50, InetAddress.getByName(basicBehaviorServerHost));
         log.info("MultiplexServer: created MasterServerSocket");
         log.info("localAddress: " + serverSocket.getLocalSocketAddress() + ", local port: " + serverSocket.getLocalPort());
      }
      catch(Exception e)
      {
         log.error("cannot create MasterServerSocket", e);
         e.printStackTrace();
         System.exit(1);
      }

      try
      {
         scriptSocket = serverSocket.accept();
         log.info("created script socket");
         scriptInputStream = scriptSocket.getInputStream();
      }
      catch(Exception e)
      {
         log.error("cannot create script socket", e);
         e.printStackTrace();
         System.exit(1);
      }

      while(!shutdownServer)
      {
         try
         {
            log.info("MultiplexServer: server socket accepting");
            testSocket = serverSocket.accept();
            log.info("MultiplexServer: created VirtualSocket on port: " + ((VirtualSocket) testSocket).getLocalVirtualPort());
            log.info("localAddress: " + testSocket.getLocalAddress() + ", local port: " + testSocket.getLocalPort());
            log.info("remoteAddress: " + testSocket.getInetAddress() + ", remote port: " + testSocket.getPort());

            try
            {
               scriptLength = getScript();
            }
            catch(IOException e)
            {
               log.error("unable to read script");
               System.exit(1);
            }

            Thread thread = new BehaviorTestThread(testSocket, script, scriptLength);
            thread.start();
            thread.join();
         }
         catch(SocketException ignored)
         {
            // it's because the ServerSocket was closed
            log.info(ignored);
            log.info("shutdownServer: " + shutdownServer);
            log.info("serverSocket:   " + serverSocket.isClosed());
            log.info("testSocket:     " + testSocket.isClosed());
         }
         catch(Exception e)
         {
            log.error("ServerTest.main(): cannot create VirtualSocket", e);
            e.printStackTrace();
            System.exit(1);
         }
      }

      log.info("Ending behavior tests");
   }


   protected int getScript() throws IOException
   {
      try
      {
         scriptLength = scriptInputStream.read();

         for(int i = 0; i < scriptLength; i++)
         {
            script[i] = (byte) scriptInputStream.read();

            if(script[i] < 0)
            {
               log.error("end of file reading script");
               throw new IOException("end of file reading script");
            }
         }
      }
      catch(IOException e)
      {
         log.error("error reading script");
         throw e;
      }

      return scriptLength;
   }


   class BehaviorTestThread extends Thread
   {
      private Socket socket;
      private byte[] script;
      private int scriptLength;
      private InputStream is;
      private OutputStream os;
      private VirtualSocket socketToClient;
      private InputStream is_client;
      private OutputStream os_client;
      private MasterServerSocket mss;
      private VirtualServerSocket vss;


      public BehaviorTestThread(Socket testSocket, byte[] script, int scriptLength)
      {
         this.socket = testSocket;
         this.script = script;
         this.scriptLength = scriptLength;

         try
         {
            is = testSocket.getInputStream();
            os = testSocket.getOutputStream();
         }
         catch(IOException e)
         {
            log.error("i/o error creating InputStream or OutputStream", e);
            e.printStackTrace();
         }
      }


      public void run()
      {
         log.info("starting behavior test thread");
         log.info("script length: " + scriptLength);
         int b = 0;

         try
         {
            for(int i = 0; i < scriptLength; i++)
            {
               log.info("script command: " + script[i]);

               switch(script[i])
               {
                  case READ:
                     b = is.read();
                     log.debug("read: " + b);
                     break;

                  case WRITE:
                     os.write(b);
                     break;

                  case SHUTDOWN_INPUT:
                     socket.shutdownInput();
                     break;

                  case SHUTDOWN_OUTPUT:
                     socket.shutdownOutput();
                     break;

                  case SLEEP:
                     b = script[++i] << 8 | script[++i];
                     log.info(new Date().getTime() + ": sleeping for " + b + " milliseconds");
                     Thread.sleep(b);
                     break;

                  case CLOSE_TEST_SOCKET:
                     socket.close();
                     break;

                  case CLOSE_SCRIPT_SOCKET:
                     scriptSocket.close();
                     break;

                  case ACCEPT_SERVER_SOCKET:
                     log.info("accepting server socket on port: " + ++nextBindPort);
                     mss = new MasterServerSocket(nextBindPort);
                     os.write(i);
                     mss.acceptServerSocketConnection();
                     mss.close();
                     break;

                  case CONNECT_TO_CLIENT:

                     try
                     {
                        socketToClient = new VirtualSocket(clientServerSocketHost, socket.getPort());
                        is_client = socketToClient.getInputStream();
                        os_client = socketToClient.getOutputStream();
                     }
                     catch(IOException e)
                     {
                        log.error("unable to create VirtualSocket back to client", e);
                        socketToClient = null;
                     }

                     break;

                  case CONNECT_TO_CLIENT_VSS:

                     socketToClient = null;
                     nextRemotePort++;

                     // Since we're reusing ports, the ServerSocket might not be available yet.
                     for(int j = 0; j < 5; j++)
                     {
                        try
                        {
                           log.info("connecting to (" + clientServerSocketHost + ", " + nextRemotePort + ")");
                           socketToClient = new VirtualSocket(clientServerSocketHost, nextRemotePort);
                           is_client = socketToClient.getInputStream();
                           os_client = socketToClient.getOutputStream();
                           break;
                        }
                        catch(IOException e)
                        {
                           log.error("unable to create VirtualSocket back to client: trying again", e);
                           Thread.sleep(1000);
                        }
                     }

                     if(socketToClient == null)
                     {
                        log.error("giving up attempt to create VirtualSocket back to client");
                     }

                     break;


                  case CONNECT_TO_CLIENT_MSS:

                     socketToClient = null;
                     nextRemotePort++;
                     log.info("connecting to (" + clientServerSocketHost + ", " + nextRemotePort + ")");

                     // Since we're reusing ports, the ServerSocket might not be available yet.
                     for(int j = 0; j < 5; j++)
                     {
                        try
                        {
                           socketToClient = new VirtualSocket(clientServerSocketHost, nextRemotePort);
                           is_client = socketToClient.getInputStream();
                           os_client = socketToClient.getOutputStream();
                           break;
                        }
                        catch(IOException e)
                        {
                           log.info("unable to create VirtualSocket back to client: trying again", e);
                           Thread.sleep(1000);
                        }
                     }

                     if(socketToClient == null)
                     {
                        log.error("giving up attempt to create VirtualSocket back to client");
                     }

                     break;

                  case READ_FROM_CLIENT:

                     if(socketToClient == null)
                     {
                        log.error("unable to read from client: socket has not been opened");
                     }
                     else
                     {
                        b = is_client.read();
                     }

                     break;

                  case WRITE_TO_CLIENT:

                     if(socketToClient == null)
                     {
                        log.error("unable to write to client: socket has not been opened");
                     }
                     else
                     {
                        os_client.write(i);
                     }

                     break;

                  case CLOSE_CLIENT_SOCKET:

                     if(socketToClient == null)
                     {
                        log.error("unable to close socket to client: not open");
                     }
                     else
                     {
                        socketToClient.close();
                        socketToClient = null;
                     }

                     break;

                  case RUN_VIRTUALSERVERSOCKET:

                     mss = new MasterServerSocket(basicBehaviorServerPort + 1);
                     os.write(3);
                     Socket virtualSocket1 = mss.accept();
                     Socket virtualSocket2 = mss.accept();
                     int localPort = virtualSocket2.getLocalPort();
                     log.info("VirtualServerSocket binding to port: " + localPort);
                     vss = new VirtualServerSocket(localPort);
                     os.write(5);
                     Socket virtualSocket3 = vss.accept();
                     DataOutputStream vos = new DataOutputStream(virtualSocket3.getOutputStream());
                     vos.writeInt(localPort);
                     virtualSocket1.close();
                     virtualSocket2.close();
                     virtualSocket3.close();
                     mss.close();
                     vss.close();
                     break;

                  case RUN_SERVER_TIMEOUT_TEST:
                     Socket virtualSocket4 = null;
                     Socket virtualSocket5 = null;
                     Socket virtualSocket6 = null;
                     Socket virtualSocket7 = null;

                     mss = new MasterServerSocket(basicBehaviorServerPort + 100);
                     mss.setSoTimeout(10000);
                     os.write(3);
                     Thread.sleep(1000);

                     try
                     {
                        virtualSocket4 = mss.accept();
                        log.info("accepted virtualSocket4");
                     }
                     catch(SocketTimeoutException e)
                     {
                        log.info(e);
                     }

                     try
                     {
                        virtualSocket5 = mss.accept();
                        log.info("accepted virtualSocket5");
                     }
                     catch(SocketTimeoutException e)
                     {
                        log.info("timed out waiting to accept virtualSocket5");
                     }

                     is.read();
                     Thread.sleep(1000);
                     vss = new VirtualServerSocket(basicBehaviorServerPort + 101);
                     SocketAddress address1 = new InetSocketAddress(clientServerSocketHost, clientServerSocketPort + 101);
                     vss.connect(address1);
                     os.write(5);
                     Thread.sleep(1000);
                     vss.setSoTimeout(10000);

                     try
                     {
                        virtualSocket6 = vss.accept();
                        log.info("accepted virtualSocket6");
                     }
                     catch(SocketTimeoutException e)
                     {
                        log.info(e);
                     }

                     try
                     {
                        virtualSocket7 = vss.accept();
                        log.info("accepted virtualSocket7");
                     }
                     catch(SocketTimeoutException e)
                     {
                        log.info(e);
                     }

                     if(virtualSocket4 != null)
                     {
                        virtualSocket4.close();
                     }
                     if(virtualSocket5 != null)
                     {
                        virtualSocket5.close();
                     }
                     if(virtualSocket6 != null)
                     {
                        virtualSocket6.close();
                     }
                     if(virtualSocket7 != null)
                     {
                        virtualSocket7.close();
                     }
                     mss.close();
                     vss.close();
                     break;

                  case RUN_VSS_TO_VSS:
                     mss = new MasterServerSocket(++nextBindPort);
                     log.info("MasterServerSocket accepting connections on: " + nextBindPort);
                     os.write(3);
//                     int port = mss.acceptServerSocketConnection().getSocket().getLocalPort();
                     int port = mss.acceptServerSocketConnection();
                     vss = new VirtualServerSocket(port);
                     InetSocketAddress address = new InetSocketAddress(clientServerSocketHost, ++nextRemotePort);
                     is.read();
                     log.info("VirtualServerSocket connecting to port: " + nextRemotePort);
                     vss.connect(address);
                     log.info("VirtualServerSocket accepting connections on: " + nextBindPort);
                     os.write(7);
                     Socket virtualSocket8 = vss.accept();
                     log.info("first virtual socket created");
                     InputStream is8 = virtualSocket8.getInputStream();
                     OutputStream os8 = virtualSocket8.getOutputStream();
                     log.info("second virtual socket connecting to port: " + nextRemotePort);
                     Socket virtualSocket9 = new VirtualSocket(clientServerSocketHost, nextRemotePort);
                     log.info("second virtual socket created");
                     InputStream is9 = virtualSocket9.getInputStream();
                     OutputStream os9 = virtualSocket9.getOutputStream();
                     os8.write(is8.read());
                     os9.write(is9.read());
                     virtualSocket8.close();
                     virtualSocket9.close();
                     vss.close();
                     mss.close();
                     break;
                     
                  case RUN_LONG_MESSAGE_TEST:
                     vss = new VirtualServerSocket(((VirtualSocket)testSocket).getLocalPort());
                     log.info("VirtualServerSocket accepting connections on: " + nextBindPort);
                     os.write(3);
                     final Socket virtualSocket10 = vss.accept();
                     final Socket virtualSocket11 = vss.accept();
                     final Socket virtualSocket12 = vss.accept();
                     final InputStream is10 = virtualSocket10.getInputStream();
                     final InputStream is11 = virtualSocket11.getInputStream();
                     final InputStream is12 = virtualSocket12.getInputStream();
                     final OutputStream os10 = virtualSocket10.getOutputStream();
                     final OutputStream os11 = virtualSocket11.getOutputStream();
                     final OutputStream os12 = virtualSocket12.getOutputStream();
  
                     Thread t1 = new Thread()
                     {
                        public void run()
                        {
                           try
                           {
                              BasicSocketBehaviorClient.C.echo(is10, os10);
                              BasicSocketBehaviorClient.C.echo(is10, os10);
                              BasicSocketBehaviorClient.C.echo(is10, os10);
                              virtualSocket10.close();
                           }
                           catch (Exception e)
                           {
                              log.error(e);
                           }
                        }
                     };

                     Thread t2 = new Thread()
                     {
                        public void run()
                        {
                           try
                           {
                              BasicSocketBehaviorClient.C.echo(is11, os11);
                              BasicSocketBehaviorClient.C.echo(is11, os11);
                              BasicSocketBehaviorClient.C.echo(is11, os11);
                              virtualSocket11.close();
                           }
                           catch (Exception e)
                           {
                              log.error(e);
                           }
                        }
                     };
                     

                     Thread t3 = new Thread()
                     {
                        public void run()
                        {
                           try
                           {
                              BasicSocketBehaviorClient.C.echo(is12, os12);
                              BasicSocketBehaviorClient.C.echo(is12, os12);
                              BasicSocketBehaviorClient.C.echo(is12, os12);
                              virtualSocket12.close();
                           }
                           catch (Exception e)
                           {
                              log.error(e);
                           }
                        }
                     };
                     
                     t1.start();
                     t2.start();
                     t3.start();
                     t1.join();
                     t2.join();
                     t3.join();
                     vss.close();
                     break;
                     
                  case RUN_MULTIPLE_READERS_TEST:
                     for (int j = 0; j < 1000; j++)
                        os.write(j);
                     break;

                  case END_TESTS:
                     socket.close();
                     shutdownServer = true;
                     serverSocket.close();
                     break;

                  default:
                     log.error("unrecognized command: " + script[i]);
               }
            }
         }
         catch(Exception e)
         {
            log.error("error in behavior test thread", e);
            e.printStackTrace();
         }

         log.info("behavior test thread shutting down");
      }
   }

   public static void main(String[] args)
   {
      MultiplexServer multiplexServer = new MultiplexServer();
//      multiplexServer.runPrimeScenario();
      multiplexServer.runBehaviorTests();
   }

}

