/*
  * 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.
  */
package org.jboss.test.remoting.transport.multiplex.config;

import java.util.HashMap;
import java.util.Map;

import javax.management.MBeanServer;

import junit.framework.TestCase;

import org.jboss.logging.Logger;
import org.jboss.remoting.Client;
import org.jboss.remoting.InvocationRequest;
import org.jboss.remoting.InvokerLocator;
import org.jboss.remoting.ServerInvocationHandler;
import org.jboss.remoting.ServerInvoker;
import org.jboss.remoting.callback.Callback;
import org.jboss.remoting.callback.HandleCallbackException;
import org.jboss.remoting.callback.InvokerCallbackHandler;
import org.jboss.remoting.transport.Connector;
import org.jboss.remoting.transport.PortUtil;
import org.jboss.remoting.transport.multiplex.Multiplex;

/**
 * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a>
 *
 */
public class MultiplexPerfTest extends TestCase
{
   // Constants -----------------------------------------------------
   
   private static final Logger log = Logger.getLogger(MultiplexPerfTest.class);
   private Runtime runtime = Runtime.getRuntime();

   // Static --------------------------------------------------------
   
   public static void main(String[] args)
   {
      try
      {
         MultiplexPerfTest test = new MultiplexPerfTest("MultiplexTest");
         test.setUp();
         test.testMultiplex();
         test.tearDown();
      }
      catch (Throwable e)
      {
         log.error("Failed to run test", e);
      }
      
   }
   
   // Attributes ----------------------------------------------------
   
   // Constructors --------------------------------------------------
   
   public MultiplexPerfTest(String name)
   {
      super(name);
   }
   
   // TestCase overrides -------------------------------------------
   
   public void setUp() throws Exception
   {
      
      super.setUp();
      
   }
   
   public void tearDown() throws Exception
   {
      
      super.tearDown();
   }
   
   
   public void testNone() throws Exception
   {
      
   }
   
   
   public void testMultiplex() throws Throwable
   {
      //Start a server
      
      Connector serverConnector = new Connector();
      InvokerLocator serverLocator = new InvokerLocator("multiplex://localhost:9099");
      serverConnector.setInvokerLocator(serverLocator.getLocatorURI());
      serverConnector.create();
      SimpleServerInvocationHandler invocationHandler = new SimpleServerInvocationHandler();
      serverConnector.addInvocationHandler("JMS", invocationHandler);
      serverConnector.start(); 
      
      log.info("Server started");
      long maxUsedHeap = 0;
      long minUsedHeap = Integer.MAX_VALUE;
      
      for (int i = 0; i < 10000; i++)
      {         
         //Create a client for invoking on the server:         
         int localPort = PortUtil.findFreePort("localhost");
         
         log.info("Free port is:" + localPort);
         
         String locatorURI =      
            "multiplex://localhost:9099/?blah=" + localPort;
         
         InvokerLocator locatorClient = new InvokerLocator(locatorURI);
         Map configuration = new HashMap();
         configuration.put(Multiplex.CLIENT_MULTIPLEX_ID, "mytestid" + i);
         configuration.put(Multiplex.MULTIPLEX_BIND_HOST, "localhost");
         configuration.put(Multiplex.MULTIPLEX_BIND_PORT, String.valueOf(localPort));
         
         Client client = new Client(locatorClient, configuration);
         
         log.info("Created client");
         
         if ((i + 1) % 20 == 0)
         {
            System.gc();
         }
         if ((i + 1) % 100 == 0)
         {
            System.out.print("Created client: " + (i+1) + " (" + localPort + "): ");
            long usedHeap = runtime.totalMemory() - runtime.freeMemory();
            if (maxUsedHeap < usedHeap)
            {
               maxUsedHeap = usedHeap;
               log.error("new heap max: iteration " + i + "(" + maxUsedHeap + ")");
            }
            if (usedHeap < minUsedHeap)
               minUsedHeap = usedHeap;
            System.out.println("" + usedHeap);
         }
         
         client.connect();
         
         log.info("Invoking");
         
         String response = (String)client.invoke("Cheese");
         
         log.info("Invocation completed");
         
         assertNotNull(response);
         assertEquals("Sausages", response);
               
         Connector callbackServerConnector = new Connector();      
         
         InvokerLocator callbackServerLocator = new InvokerLocator("multiplex://localhost:" + localPort + "/?serverMultiplexId=mytestid" + i);
               
         callbackServerConnector.setInvokerLocator(callbackServerLocator.getLocatorURI());      
         callbackServerConnector.create();      
         callbackServerConnector.start();
         
         log.info("Created callback server");
         
     
         log.info("Invoking");
         
         response = (String)client.invoke("Cheese");
         
         log.info("Invocation completed");
         
         assertNotNull(response);
         assertEquals("Sausages", response);
         
         SimpleCallbackHandler callbackHandler = new SimpleCallbackHandler();
         
         client.addListener(callbackHandler, callbackServerLocator);
         
         assertNotNull(invocationHandler.handler);
         
         Callback cb = new Callback("Squirrel");
         
         invocationHandler.handler.handleCallback(cb);
         
         assertNotNull(callbackHandler.callback);
         assertEquals("Squirrel", callbackHandler.callback.getParameter());
                  
//         client.removeListener(callbackHandler);
         client.disconnect();
         
         callbackServerConnector.stop();
         
         callbackServerConnector.destroy();
                     
      }
      
      serverConnector.stop();
      
      System.out.println("");
      System.out.println("min heap: " + minUsedHeap);
      System.out.println("max heap: " + maxUsedHeap);
      //Thread.sleep(1000 * 60 * 30 * 10);
      
      //Now run netstat -a and see how many TCP connections are open!
      
   }
   
   
   public void testConnectionSetupTeardownSpeedSocket() throws Throwable
   {
      //Start a server
      
      Connector serverConnector = new Connector();
      InvokerLocator serverLocator = new InvokerLocator("socket://localhost:9098");
      serverConnector.setInvokerLocator(serverLocator.getLocatorURI());
      serverConnector.create();
      SimpleServerInvocationHandler invocationHandler = new SimpleServerInvocationHandler();
      serverConnector.addInvocationHandler("JMS", invocationHandler);
      serverConnector.start(); 
      
      log.info("Server started");
      
      long start = System.currentTimeMillis();
      
      final int NUM_CONNECTIONS = 200;
      
      for (int i = 0; i < NUM_CONNECTIONS; i++)
      {
         
         //Create a client for invoking on the server:         
         int localPort = PortUtil.findFreePort("localhost");
               
         log.info("Free port is:" + localPort);
                                 
         //dummy param "blah" prevents the local invoker from being used
         String locatorURI =      
            "socket://localhost:9098/?blah=123";
    
         InvokerLocator locatorClient = new InvokerLocator(locatorURI);
         
         Client client = new Client(locatorClient);
         
         log.info("Created client");
         
         client.connect();
                 
         Connector callbackServerConnector = new Connector();      
         
         InvokerLocator callbackServerLocator = new InvokerLocator("socket://localhost:" + localPort);
               
         callbackServerConnector.setInvokerLocator(callbackServerLocator.getLocatorURI());      
         callbackServerConnector.create();      
         callbackServerConnector.start();
         
         log.info("Created callback server");
              
         client.disconnect();
         
         callbackServerConnector.stop();
                      
      }
      
      serverConnector.stop();
      long end = System.currentTimeMillis();
      
      log.info("That took " + (end - start) + " ms, that's " + (double)(1000 * NUM_CONNECTIONS) / (end - start) + " connections/sec");
      System.out.println("socket:    that took " + (end - start) + " ms, that's " + (double)(1000 * NUM_CONNECTIONS) / (end - start) + " connections/sec");
              
   }
   
   public void testConnectionSetupTeardownSpeedMultiplex() throws Throwable
   {
      //Start a server
      
      Connector serverConnector = new Connector();
      InvokerLocator serverLocator = new InvokerLocator("multiplex://localhost:9099");
      serverConnector.setInvokerLocator(serverLocator.getLocatorURI());
      serverConnector.create();
      SimpleServerInvocationHandler invocationHandler = new SimpleServerInvocationHandler();
      serverConnector.addInvocationHandler("JMS", invocationHandler);
      serverConnector.start(); 
      
      log.info("Server started");
      
      final int NUM_CONNECTIONS = 150;
      
      long start = System.currentTimeMillis();
      
      for (int i = 0; i < NUM_CONNECTIONS; i++)
      {
         
         //Create a client for invoking on the server:         
         int localPort = PortUtil.findFreePort("localhost");
               
         //log.info("Free port is:" + localPort);
                                 
         //dummy param "blah" prevents the local invoker from being used
         String locatorURI =      
            "multiplex://localhost:9099/?blah=123";
    
         InvokerLocator locatorClient = new InvokerLocator(locatorURI);
         Map configuration = new HashMap();
         configuration.put(Multiplex.CLIENT_MULTIPLEX_ID, "mytestid" + i);
         configuration.put(Multiplex.MULTIPLEX_BIND_HOST, "localhost");
         configuration.put(Multiplex.MULTIPLEX_BIND_PORT, String.valueOf(localPort));
         
         Client client = new Client(locatorClient, configuration);
         
         //log.info("Created client");
         
         client.connect();
                 
         Connector callbackServerConnector = new Connector();      
         
         InvokerLocator callbackServerLocator = new InvokerLocator("multiplex://localhost:" + localPort + "/?serverMultiplexId=mytestid" + i);
               
         callbackServerConnector.setInvokerLocator(callbackServerLocator.getLocatorURI());      
         callbackServerConnector.create();      
         callbackServerConnector.start();
         
         //log.info("Created callback server");
              
         client.disconnect();
         
         callbackServerConnector.stop();
                      
      }
      
      long end = System.currentTimeMillis();
      
      log.info("That took " + (end - start) + " ms, that's " + ((double)(1000 * NUM_CONNECTIONS)) / (end - start) + " connections/sec");
      System.out.println("multiplex: took " + (end - start) + " ms, that's " + ((double)(1000 * NUM_CONNECTIONS)) / (end - start) + " connections/sec");
       
   }
   
   
   
   
   
   public void testInvocationSpeedSocket() throws Throwable
   {
      //Start a server
      
      Connector serverConnector = new Connector();
      InvokerLocator serverLocator = new InvokerLocator("socket://localhost:9099");
      serverConnector.setInvokerLocator(serverLocator.getLocatorURI());
      serverConnector.create();
      SimpleServerInvocationHandler invocationHandler = new SimpleServerInvocationHandler();
      serverConnector.addInvocationHandler("JMS", invocationHandler);
      serverConnector.start(); 
      
      log.info("Server started");
      
      //Create a client for invoking on the server:         
      int localPort = PortUtil.findFreePort("localhost");
            
      log.info("Free port is:" + localPort);
                              
      //dummy param "blah" prevents the local invoker from being used
      String locatorURI =      
         "socket://localhost:9099/?blah=123";
 
      InvokerLocator locatorClient = new InvokerLocator(locatorURI);
      
      Client client = new Client(locatorClient);
      
      log.info("Created client");
      
      client.connect();
              
      Connector callbackServerConnector = new Connector();      
      
      InvokerLocator callbackServerLocator = new InvokerLocator("socket://localhost:" + localPort);
            
      callbackServerConnector.setInvokerLocator(callbackServerLocator.getLocatorURI());      
      callbackServerConnector.create();      
      callbackServerConnector.start();
      
      log.info("Created callback server");
         
      SimpleCallbackHandler callbackHandler = new SimpleCallbackHandler();
      
      client.addListener(callbackHandler, callbackServerLocator);
      
      assertNotNull(invocationHandler.handler);
      
      final int NUM_INVOCATIONS = 10000;
      
      long start = System.currentTimeMillis();
      
      for (int i = 0; i < NUM_INVOCATIONS; i++)
      {         
         //log.info("Invoking");
         
         String response = (String)client.invoke("Cheese");
         
         //log.info("Invocation completed");
         
         assertNotNull(response);
         assertEquals("Sausages", response);
               
         Callback cb = new Callback("Squirrel");
         
         invocationHandler.handler.handleCallback(cb);
         
         assertNotNull(callbackHandler.callback);
         
         assertEquals("Squirrel", callbackHandler.callback.getParameter());
      }
      
      long end = System.currentTimeMillis();
      
      log.info("That took " + (end - start) + " ms, that's " + (double)(1000 * NUM_INVOCATIONS) / (end - start) + " invocations+callbacks/sec");
            
      client.disconnect();
      
      callbackServerConnector.stop();
                      
   }
   
   public void testInvocationSpeedMultiplex() throws Throwable
   {
      //Start a server
      
      Connector serverConnector = new Connector();
      InvokerLocator serverLocator = new InvokerLocator("multiplex://localhost:9099");
      serverConnector.setInvokerLocator(serverLocator.getLocatorURI());
      serverConnector.create();
      SimpleServerInvocationHandler invocationHandler = new SimpleServerInvocationHandler();
      serverConnector.addInvocationHandler("JMS", invocationHandler);
      serverConnector.start(); 
      
      log.info("Server started");
      
      //Create a client for invoking on the server:         
      int localPort = PortUtil.findFreePort("localhost");
            
      log.info("Free port is:" + localPort);
                              
      //dummy param "blah" prevents the local invoker from being used
      String locatorURI =      
         "multiplex://localhost:9099/?blah=123";
 
      InvokerLocator locatorClient = new InvokerLocator(locatorURI);
      Map configuration = new HashMap();
      configuration.put(Multiplex.CLIENT_MULTIPLEX_ID, "mytestid");
      configuration.put(Multiplex.MULTIPLEX_BIND_HOST, "localhost");
      configuration.put(Multiplex.MULTIPLEX_BIND_PORT, String.valueOf(localPort));
      
      Client client = new Client(locatorClient, configuration);
      
      log.info("Created client");
      
      client.connect();
              
      Connector callbackServerConnector = new Connector();      
      
      InvokerLocator callbackServerLocator = new InvokerLocator("multiplex://localhost:" + localPort + "/?serverMultiplexId=mytestid");
            
      callbackServerConnector.setInvokerLocator(callbackServerLocator.getLocatorURI());      
      callbackServerConnector.create();      
      callbackServerConnector.start();
      
      log.info("Created callback server");
         
      SimpleCallbackHandler callbackHandler = new SimpleCallbackHandler();
      
      client.addListener(callbackHandler, callbackServerLocator);
      
      assertNotNull(invocationHandler.handler);
      
      final int NUM_INVOCATIONS = 10000;
      
      long start = System.currentTimeMillis();
      
      for (int i = 0; i < NUM_INVOCATIONS; i++)
      {         
         //log.info("Invoking");
         
         String response = (String)client.invoke("Cheese");
         
         //log.info("Invocation completed");
         
         assertNotNull(response);
         assertEquals("Sausages", response);
               
         Callback cb = new Callback("Squirrel");
         
         invocationHandler.handler.handleCallback(cb);
         
         assertNotNull(callbackHandler.callback);
         
         assertEquals("Squirrel", callbackHandler.callback.getParameter());
      }
      
      long end = System.currentTimeMillis();
      
      log.info("That took " + (end - start) + " ms, that's " + (double)(1000 * NUM_INVOCATIONS) / (end - start) + " invocations+callbacks/sec");
            
      client.disconnect();
      
      callbackServerConnector.stop();
                      
   }
   
   
   public void testInvocationSpeedConcurrentSocket() throws Throwable
   {
      
      //Start a server
      
      Connector serverConnector = new Connector();
      InvokerLocator serverLocator = new InvokerLocator("socket://localhost:9099");
      serverConnector.setInvokerLocator(serverLocator.getLocatorURI());
      serverConnector.create();
      SimpleServerInvocationHandler invocationHandler = new SimpleServerInvocationHandler();
      serverConnector.addInvocationHandler("JMS", invocationHandler);
      serverConnector.start(); 
      
      log.info("Server started");
      
      //Create a client for invoking on the server:         
      int localPort = PortUtil.findFreePort("localhost");
            
      log.info("Free port is:" + localPort);
                              
      //dummy param "blah" prevents the local invoker from being used
      String locatorURI =      
         "socket://localhost:9099/?blah=123";
 
      InvokerLocator locatorClient = new InvokerLocator(locatorURI);
      Client client = new Client(locatorClient);
      
      log.info("Created client");
      
      client.connect();
              
      Connector callbackServerConnector = new Connector();      
      
      InvokerLocator callbackServerLocator = new InvokerLocator("socket://localhost:" + localPort);
            
      callbackServerConnector.setInvokerLocator(callbackServerLocator.getLocatorURI());      
      callbackServerConnector.create();      
      callbackServerConnector.start();
      
      log.info("Created callback server");
         
      SimpleCallbackHandler callbackHandler = new SimpleCallbackHandler();
      
      client.addListener(callbackHandler, callbackServerLocator);
      
      assertNotNull(invocationHandler.handler);
                  
      final int NUM_INVOCATIONS = 10000;
      
      long start = System.currentTimeMillis();
      
      CallbackSender sender = new CallbackSender(NUM_INVOCATIONS, invocationHandler, callbackHandler);
      Thread t = new Thread(sender);
      t.start();
      
      for (int i = 0; i < NUM_INVOCATIONS; i++)
      {         
         //log.info("Invoking");
         
         String response = (String)client.invoke("Cheese");
         
         //log.info("Invocation completed");
         
         assertNotNull(response);
         assertEquals("Sausages", response);
                        
      }
      
      t.join();
      
      assertFalse(sender.failed);
      
      long end = System.currentTimeMillis();
      
      log.info("That took " + (end - start) + " ms, that's " + (double)(1000 * NUM_INVOCATIONS) / (end - start) + " invocations+callbacks/sec");
            
      client.disconnect();
      
      callbackServerConnector.stop();
                      
   }
   
   
   public void testInvocationSpeedConcurrentMultiplex() throws Throwable
   {
      
      //Start a server
      
      Connector serverConnector = new Connector();
      InvokerLocator serverLocator = new InvokerLocator("multiplex://localhost:9099");
      serverConnector.setInvokerLocator(serverLocator.getLocatorURI());
      serverConnector.create();
      SimpleServerInvocationHandler invocationHandler = new SimpleServerInvocationHandler();
      serverConnector.addInvocationHandler("JMS", invocationHandler);
      serverConnector.start(); 
      
      log.info("Server started");
      
      //Create a client for invoking on the server:         
      int localPort = PortUtil.findFreePort("localhost");
            
      log.info("Free port is:" + localPort);
                              
      //dummy param "blah" prevents the local invoker from being used
      String locatorURI =      
         "multiplex://localhost:9099/?blah=123";
 
      InvokerLocator locatorClient = new InvokerLocator(locatorURI);
      Map configuration = new HashMap();
      configuration.put(Multiplex.CLIENT_MULTIPLEX_ID, "mytestid");
      configuration.put(Multiplex.MULTIPLEX_BIND_HOST, "localhost");
      configuration.put(Multiplex.MULTIPLEX_BIND_PORT, String.valueOf(localPort));
      
      Client client = new Client(locatorClient, configuration);
      
      log.info("Created client");
      
      client.connect();
              
      Connector callbackServerConnector = new Connector();      
      
      InvokerLocator callbackServerLocator = new InvokerLocator("multiplex://localhost:" + localPort + "/?serverMultiplexId=mytestid");
            
      callbackServerConnector.setInvokerLocator(callbackServerLocator.getLocatorURI());      
      callbackServerConnector.create();      
      callbackServerConnector.start();
      
      log.info("Created callback server");
         
      SimpleCallbackHandler callbackHandler = new SimpleCallbackHandler();
      
      client.addListener(callbackHandler, callbackServerLocator);
      
      assertNotNull(invocationHandler.handler);
                  
      final int NUM_INVOCATIONS = 10000;
      
      long start = System.currentTimeMillis();
      
      CallbackSender sender = new CallbackSender(NUM_INVOCATIONS, invocationHandler, callbackHandler);
      Thread t = new Thread(sender);
      t.start();
      
      for (int i = 0; i < NUM_INVOCATIONS; i++)
      {         
         //log.info("Invoking");
         
         String response = (String)client.invoke("Cheese");
         
         //log.info("Invocation completed");
         
         assertNotNull(response);
         assertEquals("Sausages", response);
                        
      }
      
      t.join();
      
      assertFalse(sender.failed);
      
      long end = System.currentTimeMillis();
      
      log.info("That took " + (end - start) + " ms, that's " + (double)(1000 * NUM_INVOCATIONS) / (end - start) + " invocations+callbacks/sec");
            
      client.disconnect();
      
      callbackServerConnector.stop();
                      
   }
 
   class SimpleCallbackHandler implements InvokerCallbackHandler
   {
      
      Callback callback;

      public void handleCallback(Callback callback) throws HandleCallbackException
      {
         this.callback = callback;         
      }
      
   }
   
   class SimpleServerInvocationHandler implements ServerInvocationHandler
   {
      InvokerCallbackHandler handler;
      
      
      public void addListener(InvokerCallbackHandler callbackHandler)
      {
         this.handler = callbackHandler;
         
      }

      public Object invoke(InvocationRequest invocation) throws Throwable
      {
         //log.info("Received invocation:" + invocation);
         
         return "Sausages";
      }

      public void removeListener(InvokerCallbackHandler callbackHandler)
      {
         // FIXME removeListener
         
      }

      public void setInvoker(ServerInvoker invoker)
      {
         // FIXME setInvoker
         
      }

      public void setMBeanServer(MBeanServer server)
      {
         // FIXME setMBeanServer
         
      }
      
   }
   
   
   
   class CallbackSender implements Runnable
   {
      int numCallbacks;
      boolean failed;
      SimpleServerInvocationHandler invocationHandler;
      SimpleCallbackHandler callbackHandler;
      CallbackSender(int numCallbacks, SimpleServerInvocationHandler invocationHandler,
            SimpleCallbackHandler callbackHandler)
      {
         this.numCallbacks = numCallbacks;
         this.invocationHandler = invocationHandler;
         this.callbackHandler = callbackHandler;
      }
      public void run()
      {
         try
         {
            for (int i = 0; i < numCallbacks; i++)
            {
               Callback cb = new Callback("Squirrel");
               
               invocationHandler.handler.handleCallback(cb);
               
               assertNotNull(callbackHandler.callback);
               
               assertEquals("Squirrel", callbackHandler.callback.getParameter());
            }
         }
         catch (Exception e)
         {
            log.error("Failed to send callback", e);
            failed = true;
         }
      }
   }
   
}



