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

import com.malhartech.daemon.Daemon;
import com.malhartech.engine.TestSink;
import com.malhartech.stream.SampleWebSocketPublisher;
import java.net.URI;
import java.util.Map;
import junit.framework.Assert;
import org.junit.Test;

public class WebSocketInputOperatorTest
{
  @Test
  @SuppressWarnings("SleepWhileInLoop")
  public void testWebSocketInputModule() throws Exception
  {

    System.out.println("Starting Daemon...");
    Daemon.setPort(9090);
    Daemon.setLocalMode(true);
    Daemon.setup();
    Daemon.start();

    SampleWebSocketPublisher sp = new SampleWebSocketPublisher();
    sp.setChannelUrl("http://localhost:9090/channel/testChannel");
    Thread t = new Thread(sp);
    t.start();

    String url = "ws://localhost:9090/channel/testChannel";
    final WebSocketInputOperator operator = new WebSocketInputOperator();

    TestSink<Map<String, String>> sink = new TestSink<Map<String, String>>();

    operator.outputPort.setSink(sink);
    operator.setName("testWebSocketInputNode");
    operator.setUrl(new URI(url));

    operator.setup(null);
    operator.activate(null);

    int timeoutMillis = 2000;
    while (sink.collectedTuples.isEmpty() && timeoutMillis > 0) {
      operator.emitTuples();
      timeoutMillis -= 20;
      Thread.sleep(20);
    }

    Assert.assertTrue("tuple emitted", sink.collectedTuples.size() > 0);

    Map<String, String> tuple = sink.collectedTuples.get(0);
    Assert.assertEquals("", tuple.get("hello"), "world");

    operator.deactivate();
    operator.teardown();
    t.interrupt();
    t.join();
  }

}

@@ -479,16 +479,106 @@
 on%0A  %7B%0A%0A
+    if (false) %7B%0A      // this is not working yet - start server manually for this test%0A  
     Syst
@@ -615,24 +615,26 @@
 n...%22);%0A    
+ 
+ 
 Daemon.setPo
@@ -639,24 +639,26 @@
 Port(9090);%0A
+  
     Daemon.s
@@ -676,24 +676,26 @@
 (true);%0A    
+ 
+ 
 Daemon.setup
@@ -706,203 +706,31 @@
     
-Daemon.start();%0A%0A    SampleWebSocketPublisher sp = new SampleWebSocketPublisher();%0A    sp.setChannelUrl(%22http://localhost:9090/channel/testChannel%22);%0A    Thread t = new Thread(sp);%0A    t.start();
+  Daemon.start();%0A    %7D
 %0A%0A  
@@ -1090,16 +1090,16 @@
 (null);%0A
-
     oper
@@ -1116,24 +1116,289 @@
 ate(null);%0A%0A
+    // start publisher after subscriber listens to ensure we don't miss the message%0A    SampleWebSocketPublisher sp = new SampleWebSocketPublisher();%0A    sp.setChannelUrl(%22http://localhost:9090/channel/testChannel%22);%0A    Thread t = new Thread(sp);%0A    t.start();%0A%0A%0A
     int time

make test run clean (with external server)