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

import com.malhartech.bufferserver.Buffer;
import com.malhartech.bufferserver.ClientHandler;
import com.malhartech.netty.ClientPipelineFactory;
import java.net.InetSocketAddress;
import java.util.Collections;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 */
public class SocketStreamTest
{

  private static Logger LOG = LoggerFactory.getLogger(SocketStreamTest.class);

  static {
    //   java.util.logging.Logger.getLogger("").setLevel(java.util.logging.Level.FINEST);
    //    java.util.logging.Logger.getLogger("").info("test");
  }

  public static class BufferServerInputSocketStream extends InputSocketStream
  {

    /**
     * Requires upstream node info to setup subscriber TODO: revisit context
     */
    public void setContext(StreamContext context, String upstreamNodeId, String upstreamNodeLogicalName, String downstreamNodeId)
    {
      super.setContext(context);
      String downstreamNodeLogicalName = "downstreamNodeLogicalName"; // TODO: why do we need this?
      ClientHandler.registerPartitions(channel, downstreamNodeId, downstreamNodeLogicalName, upstreamNodeId, upstreamNodeLogicalName, Collections.<String>emptyList());
    }
//    @Override
//    public ClientPipelineFactory getClientPipelineFactory()
//    {
//      return new ClientPipelineFactory(ClientHandler.class);
//    }
  }

  public static class BufferServerOutputSocketStream extends OutputSocketStream
  {

    public void setContext(com.malhartech.dag.StreamContext context, String upstreamNodeId, String upstreamNodeLogicalName)
    {
      super.setContext(context);

      // send publisher request
      LOG.info("registering publisher: {} {}", upstreamNodeId, upstreamNodeLogicalName);
      ClientHandler.publish(channel, upstreamNodeId, upstreamNodeLogicalName);
    }
//    @Override
//    public ClientPipelineFactory getClientPipelineFactory()
//    {
//      return new ClientPipelineFactory(ClientHandler.class);
//    }
  }

  /**
   * Send tuple on outputstream and receive tuple from inputstream
   *
   * @throws Exception
   */
  // @Ignore
  @Test
  public void test1() throws Exception
  {

    Sink sink = new Sink()
    {

      @Override
      public void doSomething(Tuple t)
      {
        System.out.println("received: " + t.getObject());
        synchronized (SocketStreamTest.this) {
          SocketStreamTest.this.notify();
        }
      }
    };

    SerDe serde = new InputActiveMQStreamTest.MySerDe();

    int port = 9080; //50001; // TODO: find random port
    //com.malhartech.bufferserver.Server s = new Server(port);
    //SocketAddress bindAddr  = s.run();
    //port = ((InetSocketAddress)bindAddr).getPort();


    StreamContext issContext = new StreamContext(sink);
    issContext.setSerde(serde);


    StreamConfiguration sconf = new StreamConfiguration();
    sconf.setSocketAddr(StreamConfiguration.SERVER_ADDRESS, InetSocketAddress.createUnresolved("localhost", port));

    BufferServerInputSocketStream iss = new BufferServerInputSocketStream();
    iss.setup(sconf);
    iss.setContext(issContext, "upstreamNodeId", "upstreamNodeLogicalId", "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", "upstreamNodeLogicalId");

    Tuple t = DataProcessingTest.generateTuple("hello", ossContext);
    LOG.info("Sending hello message");
    oss.doSomething(t);
    synchronized (this) {
      this.wait(2000);
    }
    System.out.println("exiting...");

  }
}

@@ -102,288 +102,314 @@
 ort 
-com.malhartech.bufferserver.Buffer;%0Aimport com.malhartech.bufferserver.ClientHandler;%0Aimport com.malhartech.netty.ClientPipelineFactory;%0Aimport java.net.InetSocketAddress;%0Aimport java.util.Collections;%0Aimport org.junit.Test;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory
+java.net.InetSocketAddress;%0Aimport java.util.Collections;%0Aimport java.util.concurrent.atomic.AtomicInteger;%0A%0Aimport org.junit.Assert;%0Aimport org.junit.Test;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport com.malhartech.bufferserver.ClientHandler;%0Aimport com.malhartech.bufferserver.Server
 ;%0A%0A/
@@ -1323,165 +1323,8 @@
   %7D%0A
-//    @Override%0A//    public ClientPipelineFactory getClientPipelineFactory()%0A//    %7B%0A//      return new ClientPipelineFactory(ClientHandler.class);%0A//    %7D%0A
   %7D%0A
@@ -1783,165 +1783,8 @@
   %7D%0A
-//    @Override%0A//    public ClientPipelineFactory getClientPipelineFactory()%0A//    %7B%0A//      return new ClientPipelineFactory(ClientHandler.class);%0A//    %7D%0A
   %7D%0A
@@ -1895,21 +1895,8 @@
  */%0A
-  // @Ignore%0A
   @T
@@ -1939,24 +1939,84 @@
 eption%0A  %7B%0A%0A
+    final AtomicInteger messageCount = new AtomicInteger();%0A
     Sink sin
@@ -2201,24 +2201,66 @@
 est.this) %7B%0A
+          messageCount.incrementAndGet();%0A
           So
@@ -2285,16 +2285,19 @@
 s.notify
+All
 ();%0A    
@@ -2395,31 +2395,13 @@
 t = 
-9080; //50001; // TODO:
+0; //
  fin
@@ -2418,18 +2418,16 @@
 ort%0A    
-//
 com.malh
@@ -2479,18 +2479,20 @@
 t);%0A    
-//
+Inet
 SocketAd
@@ -2509,16 +2509,35 @@
 Addr  = 
+(InetSocketAddress)
 s.run();
@@ -2545,45 +2545,23 @@
     
-//
 port = 
-((InetSocketAddress)
 bindAddr
 ).ge
@@ -2556,17 +2556,16 @@
 bindAddr
-)
 .getPort
@@ -2565,25 +2565,24 @@
 getPort();%0A%0A
-%0A
     StreamCo
@@ -3495,44 +3495,244 @@
 ed (
-this) %7B%0A      this.wait(2000);%0A    %7D
+SocketStreamTest.this) %7B%0A      if (messageCount.get() == 0) %7B // receiver could be done before we get here%0A        SocketStreamTest.this.wait(2000);%0A      %7D%0A    %7D%0A    %0A    Assert.assertEquals(%22Received messages%22, 1, messageCount.get());
 %0A   

Fix socket stream test.