/**
 * Copyright (c) 2012-2012 Malhar, Inc. All rights reserved.
 */
package com.malhartech.dag;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 */
public class SocketStreamTest
{

  private static Logger LOG = LoggerFactory.getLogger(SocketStreamTest.class);
  private static int bufferServerPort = 0;
  private static Server bufferServer = null;
  
  @BeforeClass
  public static void setup() throws InterruptedException, IOException {
    //   java.util.logging.Logger.getLogger("").setLevel(java.util.logging.Level.FINEST);
    //    java.util.logging.Logger.getLogger("").info("test");
    bufferServer = new Server(0); // find random port
    InetSocketAddress bindAddr  = (InetSocketAddress)bufferServer.run();
    bufferServerPort = bindAddr.getPort();
  
  }

  @AfterClass
  public static void tearDown() throws IOException {
    if (bufferServer != null) {
      bufferServer.shutdown();
    }
  }
  
  /**
   * Send tuple on outputstream and receive tuple from inputstream
   *
   * @throws Exception
   */
  @Test
  public void testBufferServerStream() throws Exception
  {

    final AtomicInteger messageCount = new AtomicInteger();
    Sink sink = new Sink()
    {

      @Override
      public void doSomething(Tuple t)
      {
        switch (t.getData().getType()) {
          case BEGIN_WINDOW:
            System.out.println(t.getData().getBeginwindow().getNode() + " begin window for window " + t.getData().getWindowId());
            break;

          case END_WINDOW:
            System.out.println(t.getData().getEndwindow().getNode() + " end window for window " + t.getData().getWindowId());
            break;

          case SIMPLE_DATA:
            System.out.println("received: " + t.getObject());
            synchronized (SocketStreamTest.this) {
              messageCount.incrementAndGet();
            }
        }
      }
    };

    SerDe serde = new MySerDe();


    StreamContext issContext = new StreamContext(sink);
    issContext.setSerde(serde);
    
    String streamName = "streamName"; // AKA "type"
    String upstreamNodeId = "upstreamNodeId";
    String downstreamNodeId = "downStreamNodeId";
    
    StreamConfiguration sconf = new StreamConfiguration();
    sconf.setSocketAddr(StreamConfiguration.SERVER_ADDRESS, InetSocketAddress.createUnresolved("localhost", bufferServerPort));

    String upstreamNodeId = "upstreamId";
    String upstreamNodeType = "upstreamType";
    String downstreamNodeId = "downstreamId";
    String downstreamNodeType = "downstreamType";

    BufferServerInputSocketStream iss = new BufferServerInputSocketStream();
    iss.setup(sconf);
    iss.setContext(issContext, upstreamNodeId, streamName, downstreamNodeId);
    System.out.println("input stream ready");

    BufferServerOutputSocketStream oss = new BufferServerOutputSocketStream();
    StreamContext ossContext = new StreamContext(null);
    ossContext.setSerde(serde);
    oss.setup(sconf);
    oss.setContext(ossContext, upstreamNodeId, streamName);

    LOG.info("Sending hello message");
    oss.doSomething(StramTestSupport.generateBeginWindowTuple(upstreamNodeId, 0, ossContext));
    oss.doSomething(StramTestSupport.generateTuple("hello", 0, ossContext));
    oss.doSomething(StramTestSupport.generateEndWindowTuple(upstreamNodeId, 0, 1, ossContext));
    synchronized (SocketStreamTest.this) {
      if (messageCount.get() == 0) { // receiver could be done before we get here
        SocketStreamTest.this.wait(3000);
      }
    }

    Assert.assertEquals("Received messages", 1, messageCount.get());
    System.out.println("exiting...");

  }
}

@@ -91,16 +91,111 @@
 h.dag;%0A%0A
+import com.malhartech.bufferserver.Server;%0Aimport com.malhartech.dag.StramTestSupport.MySerDe;%0A
 import j
@@ -299,17 +299,16 @@
 nteger;%0A
-%0A
 import o
@@ -675,18 +675,16 @@
 = null;%0A
-  
 %0A  @Befo
@@ -756,24 +756,26 @@
  IOException
+%0A 
  %7B%0A    //   
@@ -1003,17 +1003,16 @@
 indAddr 
- 
 = (InetS
@@ -1024,16 +1024,17 @@
 Address)
+ 
 bufferSe
@@ -1088,18 +1088,16 @@
 Port();%0A
-  
 %0A  %7D%0A%0A  
@@ -1158,16 +1158,18 @@
 xception
+%0A 
  %7B%0A    i
@@ -1236,18 +1236,16 @@
   %7D%0A  %7D%0A
-  
 %0A  /**%0A 
@@ -1935,32 +1935,146 @@
 getWindowId());%0A
+            synchronized (SocketStreamTest.this) %7B%0A              SocketStreamTest.this.notifyAll();%0A            %7D%0A
             brea
@@ -2171,61 +2171,8 @@
 ));%0A
-            synchronized (SocketStreamTest.this) %7B%0A  
     
@@ -2207,38 +2207,24 @@
 ntAndGet();%0A
-            %7D%0A
         %7D%0A  
@@ -2360,20 +2360,16 @@
 serde);%0A
-    
 %0A    Str
@@ -2509,20 +2509,16 @@
 odeId%22;%0A
-    
 %0A    Str
@@ -2702,193 +2702,8 @@
 );%0A%0A
-    String upstreamNodeId = %22upstreamId%22;%0A    String upstreamNodeType = %22upstreamType%22;%0A    String downstreamNodeId = %22downstreamId%22;%0A    String downstreamNodeType = %22downstreamType%22;%0A%0A
     
@@ -3526,92 +3526,8 @@
 ) %7B%0A
-      if (messageCount.get() == 0) %7B // receiver could be done before we get here%0A  
     
@@ -3559,23 +3559,15 @@
 ait(
-3
+2
 000);%0A
-      %7D%0A
     

send begin window and end window packet along with the packet for testing. Signed-off-by:Chetan Narsude <chetan@malhar-inc.com>