examples/mobile/src/test/java/org/apache/apex/examples/mobile/ApplicationTest.java

/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.apex.examples.mobile;

import java.net.URI;
import java.util.HashMap;
import java.util.Map;
import javax.servlet.Servlet;

import org.eclipse.jetty.server.Connector;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.servlet.ServletContextHandler;
import org.eclipse.jetty.servlet.ServletHolder;
import org.junit.Assert;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.apex.malhar.lib.utils.PubSubHelper;
import org.apache.hadoop.conf.Configuration;

import com.datatorrent.api.LocalMode;
import com.datatorrent.lib.helper.SamplePubSubWebSocketServlet;
import com.datatorrent.lib.io.PubSubWebSocketInputOperator;
import com.datatorrent.lib.io.PubSubWebSocketOutputOperator;
import com.datatorrent.lib.testbench.CollectorTestSink;

public class ApplicationTest
{
  private static final Logger LOG = LoggerFactory.getLogger(ApplicationTest.class);

  public ApplicationTest()
  {
  }

  /**
   * Test of getApplication method, of class Application.
   */
  @Test
  public void testGetApplication() throws Exception
  {
    Configuration conf = new Configuration(false);
    conf.addResource("dt-site-mobile.xml");
    Server server = new Server(0);
    Servlet servlet = new SamplePubSubWebSocketServlet();
    ServletHolder sh = new ServletHolder(servlet);
    ServletContextHandler contextHandler = new ServletContextHandler(server, "/", ServletContextHandler.SESSIONS);
    contextHandler.addServlet(sh, "/pubsub");
    contextHandler.addServlet(sh, "/*");
    server.start();
    Connector[] connector = server.getConnectors();
    conf.set("dt.attr.GATEWAY_CONNECT_ADDRESS", "localhost:" + connector[0].getLocalPort());
    URI uri = PubSubHelper.getURI("localhost:" + connector[0].getLocalPort());

    PubSubWebSocketOutputOperator<Object> outputOperator = new PubSubWebSocketOutputOperator<Object>();
    outputOperator.setUri(uri);
    outputOperator.setTopic(conf.get("dt.application.MobileExample.operator.QueryLocation.topic"));

    PubSubWebSocketInputOperator<Map<String, String>> inputOperator = new PubSubWebSocketInputOperator<Map<String, String>>();
    inputOperator.setUri(uri);
    inputOperator.setTopic(conf.get("dt.application.MobileExample.operator.LocationResults.topic"));

    CollectorTestSink<Object> sink = new CollectorTestSink<Object>();
    inputOperator.outputPort.setSink(sink);

    Map<String, String> data = new HashMap<String, String>();
    data.put("command", "add");
    data.put("phone", "5559990");

    Application app = new Application();
    LocalMode lma = LocalMode.newInstance();
    lma.prepareDAG(app, conf);
    LocalMode.Controller lc = lma.getController();
    lc.setHeartbeatMonitoringEnabled(false);
    lc.runAsync();
    Thread.sleep(5000);
    inputOperator.setup(null);
    outputOperator.setup(null);
    inputOperator.activate(null);
    outputOperator.beginWindow(0);
    outputOperator.input.process(data);
    outputOperator.endWindow();
    inputOperator.beginWindow(0);
    int timeoutMillis = 5000;
    while (sink.collectedTuples.size() < 5 && timeoutMillis > 0) {
      inputOperator.emitTuples();
      timeoutMillis -= 20;
      Thread.sleep(20);
    }
    inputOperator.endWindow();
    lc.shutdown();
    inputOperator.teardown();
    outputOperator.teardown();
    server.stop();
    Assert.assertTrue("size of output is 5 ", sink.collectedTuples.size() == 5);
    for (Object obj : sink.collectedTuples) {
      Assert.assertEquals("Expected phone number", "5559990", ((Map<String, String>)obj).get("phone"));
    }
  }
}