/*
    * Created on Apr 16, 2004
    */
   package org.neo.swarm.util.network.tcp;
   
   import java.io.InputStream;
   import java.io.OutputStream;
   import java.net.InetAddress;
   import java.nio.ByteBuffer;
  
  import junit.framework.TestCase;
  
  import org.neo.swarm.util.io.ByteBufferInputStream;
  import org.neo.swarm.util.io.ByteBufferOutputStream;
  import org.neo.swarm.util.io.PacketInputStream;
  import org.neo.swarm.util.io.PacketOutputStream;
  import org.neo.swarm.util.network.tcp.ListenCallback;
  import org.neo.swarm.util.network.tcp.NIOTcpServer;
  import org.neo.swarm.util.network.tcp.TcpServiceAPI;
  import org.neo.swarm.util.network.tcp.client.SocketReader;
  import org.neo.swarm.util.network.tcp.client.SocketWriter;
  import org.neo.swarm.util.threads.OswegoThreadPool;
  
  import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
  
  /***
   * Tests the ability of SocketStreams that support ByteBuffer
   *  @author navery
   */
  public class SocketStreamsTest extends TestCase {
  	private static String PAD = "BytePaddingMessageThatDoesStuffXXXXYYYZZZAABBCCAADDEEFFQQWWEERRTTYY";
      TcpServiceAPI server;
      int dataSize;
      long startTime;
  	int numIterations;
  	int callbackCount;
  	
  	boolean verbose = false;
  	String host = "localhost";
  	int port = 8888;
  	int threadpoolsize = 1;
  	int timeout = 1000; // 0 blocks indefinately (othewise time in ms)
  
  
  	
      /* (non-Javadoc)
       * @see junit.framework.TestCase#setUp()
       */
      protected void setUp() throws Exception {
       	this.startTime = System.currentTimeMillis();
      }
      
      protected void tearDown() throws Exception {
  		long endTime = System.currentTimeMillis();
  		double timeSecs = (endTime - startTime) / 1000.0;
  		double rate = numIterations/ timeSecs;
  		double tp = rate * dataSize;
  		System.out.println("\nDone elapsed time:" 	+ timeSecs + " pktSize:" + dataSize + " rate:" + rate + " throughput:" + tp / 1000.0 + "kb/s");		
  		this.server.stopServer();
      }
      
  	public void _testSimpleClientServer() throws Exception {
  		verbose = true;
  		MyListenerCallbackOneWay cb = new MyListenerCallbackOneWay();
  		String expectedDataString = "0123456789";
  		byte[] expectedData = expectedDataString.getBytes();
  		// create the tcp server which will callback on our object.
  		 server = new NIOTcpServer(cb, new OswegoThreadPool(new PooledExecutor(), 1, 5),  1024, InetAddress.getByName(host), port, timeout);
  		server.startServer();
  		
  		// create an OutputSocketStream.
  		OutputStream os =  new PacketOutputStream(new ByteBufferOutputStream(ByteBuffer.allocate(100), new SocketWriter(InetAddress.getByName(host), port, true)));
  		os.write(expectedData);
  		os.flush();
  		// wait until the cb object has received the data
  		int wait = 0;
  		while (!cb.called) {
  			Thread.sleep(100);
  		}
  	}
  
  	public void _testSimpleClientServerwithReply() throws Exception {
  		verbose = true;
  		ListenCallback cb = new MyListenerCallbackWithReply();
  		String expectedDataString = "myTestMessagemyTestMessagemyTestMessagemyTestMessage";
  		byte[] expectedData = expectedDataString.getBytes();
  		// create the tcp server which will callback on our object.
  		this.server = new NIOTcpServer(cb, new OswegoThreadPool(new PooledExecutor(), 1, 5), 1024, InetAddress.getByName(host), port, timeout);
  		this.server.startServer();
  		
  		SocketWriter sw = new SocketWriter(InetAddress.getByName(host),	port, true);
  		OutputStream os =  new PacketOutputStream(new ByteBufferOutputStream(ByteBuffer.allocate(100), sw ));		
          InputStream is = new PacketInputStream (new ByteBufferInputStream(PacketInputStream.START_DATA.length + 4, PacketInputStream.END_DATA.length, ByteBuffer.allocate(100), new SocketReader(null, 8888, sw.getSocketChannel())));		
  		
  		os.write(expectedData);
  		os.flush();
  		
  		int available = is.available();
  		byte[] returnBuffer = new byte[available];
 		int count = is.read(returnBuffer);
 		System.out.println("reponse:" + new String(returnBuffer));
 		assertEquals(10, count);
 	}
 	
 	
 	public void testOneWayPerformance() throws Exception {
 		verbose = false;
 		ListenCallback cb = new MyListenerCallbackOneWay();
 
 		byte[] expectedData = makeMessage(1024);
 		// create the tcp server which will callback on our object.
 		this.server = new NIOTcpServer(cb, new OswegoThreadPool(new PooledExecutor(), 1, 5), 4096, InetAddress.getByName(host), port, timeout);
 		this.server.startServer();
 		
 		OutputStream os = new PacketOutputStream(new ByteBufferOutputStream(ByteBuffer.allocate(4096), new SocketWriter(InetAddress.getByName(host),	port, true)));		
 
 		this.numIterations = 100000;
 		this.callbackCount = 0;
 		this.startTime = System.currentTimeMillis();
 		this.dataSize = expectedData.length;
 		
 		// pump through the messages 
 		for (int i = 0; i < numIterations; i++) {
 			os.write(expectedData);
 			os.flush();
 		}
 			
 		while (this.callbackCount < this.numIterations) {
 			Thread.sleep(1000);
 		}
 	}
 		
 	public void testTwoWayPerformance() throws Exception {
 		verbose = false;
 		int dataSize = 1024;
 		int bufferSize = 4096;
 		ListenCallback cb = new MyListenerCallbackWithReply();
 		byte[] expectedData = makeMessage(dataSize);
 
 		// create the tcp server which will callback on our object.
 		this.server = new NIOTcpServer(cb, new OswegoThreadPool(new PooledExecutor(), 1, 5), bufferSize, InetAddress.getByName(host), port, timeout);
 		this.server.startServer();
 		
 		SocketWriter sw = new SocketWriter(InetAddress.getByName(host),	port, true);
 		OutputStream os = new PacketOutputStream(new ByteBufferOutputStream(ByteBuffer.allocate(bufferSize), sw));		
 		InputStream is = new PacketInputStream (new ByteBufferInputStream(PacketInputStream.START_DATA.length + 4, PacketInputStream.END_DATA.length, ByteBuffer.allocate(bufferSize), new SocketReader(null, 8888, sw.getSocketChannel())));		
 
 
 		//setup stats
 		this.numIterations = 100000;
 		this.callbackCount = 0;	
 		this.startTime = System.currentTimeMillis();
 		this.dataSize = expectedData.length;
 		
 		int avail;
 		for (int i = 0; i < numIterations; i++) {
 			os.write(expectedData);
 			os.flush();
 			avail = is.available();
 			byte[] reply = new byte[avail];
 			is.read(reply);
 		}
 
 		while (this.callbackCount < this.numIterations) {
 			Thread.sleep(500);
 		}
 	}
 
 	/***
 	 * Simple callback class that allows the server to call out onto user app.
 	 */
 	class MyListenerCallbackOneWay implements ListenCallback {
 		public boolean called;
 		public byte[] messageDataReceived(byte[] data) {
 			called = true;
 			callbackCount++;
 			if (verbose) System.out.println(callbackCount + " datasize: "+ data.length);
 			if (callbackCount % 5000 == 0) 
 				System.out.println(callbackCount + "[" + data.length + "]");
 			return new byte[0];
 		}
 	}
 	/***
 	 * Simple callback class that allows the server to call out onto user app.
 	 */
 	class MyListenerCallbackWithReply implements ListenCallback {
 		
 		public byte[] messageDataReceived(byte[] data) {
 			if (verbose) System.out.println(callbackCount + " MyListenerCallbackWithReply responding..:" + new String(data));
 			if (callbackCount % 5000 == 0) {
 				System.out.print(callbackCount + "v");
 			}
 
 			callbackCount++;
 			return "0123456789".getBytes();
 		}
 	}
 
 	private byte[] makeMessage(int size) throws Exception {
 		byte[] sendMessage = new byte[size];
 		
 		byte[] padding = PAD.getBytes();
 
 		// fill the buffer
 		int pos = 0;
 		while (pos < size) {
 			sendMessage[pos] = padding[pos++ % padding.length];
 		}
 		return sendMessage;
 	}
 
 }