/**
 * 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
  {

    if (false) {
      // this is not working yet - start server manually for this test
      System.out.println("Starting Daemon...");
      Daemon.setPort(9090);
      Daemon.setLocalMode(true);
      Daemon.setup();
      Daemon.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);

    // start publisher after subscriber listens to ensure we don't miss the message
    SampleWebSocketPublisher sp = new SampleWebSocketPublisher();
    sp.setChannelUrl("http://localhost:9090/channel/testChannel");
    Thread t = new Thread(sp);
    t.start();


    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();
  }

}

@@ -302,16 +302,41 @@
 Assert;%0A
+import org.junit.Ignore;%0A
 import o
@@ -393,16 +393,94 @@
 rTest%0A%7B%0A
+  @Ignore // ignore until daemon can be started automatically within the test%0A
   @Test%0A
@@ -722,36 +722,8 @@
 %22);%0A
-      Daemon.setPort(9090);%0A
     
@@ -770,16 +770,33 @@
 n.setup(
+%22localhost:19090%22
 );%0A     
@@ -844,32 +844,33 @@
 %22ws://localhost:
+1
 9090/channel/tes
@@ -1402,16 +1402,17 @@
 calhost:
+1
 9090/cha

load configuration appropriately, test port changed to 19090.