stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStreamTest.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.malhar.stream.sample;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;

import org.joda.time.Duration;
import org.junit.Assert;
import org.junit.Test;

import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.util.KeyValPair;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
import org.apache.apex.malhar.stream.api.impl.ApexStreamImpl;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;

/**
 * A test class which test your own stream implementation build on default one
 */
@SuppressWarnings("unchecked")
public class MyStreamTest
{
  static Map<String, Long> expected = new HashMap<>();
  static String testId = null;
  static Callable<Boolean> exitCondition = null;
  static {
    expected.put("newword1", 4L);
    expected.put("newword2", 8L);
    expected.put("newword3", 4L);
    expected.put("newword4", 4L);
    expected.put("newword5", 4L);
    expected.put("newword7", 4L);
    expected.put("newword9", 6L);

    exitCondition = new Callable<Boolean>()
    {
      @Override
      public Boolean call() throws Exception
      {
        if (!TupleCollector.results.containsKey(testId) || TupleCollector.results.get(testId).isEmpty()) {
          return false;
        }
        Map<String, Long> dataMap = new HashMap<>();
        List<Tuple.TimestampedTuple<KeyValPair<String, Long>>> data = (List<Tuple.TimestampedTuple<KeyValPair<String, Long>>>)TupleCollector.results.get(testId);
        for (Tuple.TimestampedTuple<KeyValPair<String, Long>> entry : data) {
          dataMap.put(entry.getValue().getKey(), entry.getValue().getValue());
        }
        return (dataMap != null) && dataMap.size() >= 1 && expected.equals(dataMap);
      }
    };
  }

  @Test
  public void testMethodChainWordcount() throws Exception
  {

    testId = "testMethodChainWordcount";

    TupleCollector<Tuple.WindowedTuple<KeyValPair<String, Long>>> collector = new TupleCollector<>();
    collector.id = testId;
    new MyStream<>((ApexStreamImpl<String>)StreamFactory.fromFolder("./src/test/resources/data"))
        .<String, MyStream<String>>flatMap(new Function.FlatMapFunction<String, String>()
        {
          @Override
          public Iterable<String> f(String input)
          {
            return Arrays.asList(input.split(" "));
          }
        }).myFilterAndMap(new Function.MapFunction<String, String>()
        {
          @Override
          public String f(String input)
          {
            return input.replace("word", "newword");
          }
        }, new Function.FilterFunction<String>()
        {
          @Override
          public boolean f(String input)
          {
            return input.startsWith("word");
          }
        }).window(new WindowOption.GlobalWindow(), new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(Duration.millis(1000)))
        .countByKey(new Function.ToKeyValue<String, String, Long>()
        {
          @Override
          public Tuple<KeyValPair<String, Long>> f(String input)
          {
            return new Tuple.PlainTuple(new KeyValPair<>(input, 1L));
          }
        }).addOperator(collector, collector.inputPort, null)
        .runEmbedded(false, 30000, exitCondition);

    Map<String, Long> dataMap = new HashMap<>();
    for (Tuple.TimestampedTuple<KeyValPair<String, Long>> entry : (List<Tuple.TimestampedTuple<KeyValPair<String, Long>>>)TupleCollector.results.get(testId)) {
      dataMap.put(entry.getValue().getKey(), entry.getValue().getValue());
    }

    Assert.assertTrue(dataMap.size() > 1);
    Assert.assertEquals(expected, dataMap);
  }

  @Test
  public void testNonMethodChainWordcount() throws Exception
  {
    testId = "testNonMethodChainWordcount";

    TupleCollector<Tuple.WindowedTuple<KeyValPair<String, Long>>> collector = new TupleCollector<>();
    collector.id = testId;
    MyStream<String> mystream = new MyStream<>((ApexStreamImpl<String>)StreamFactory
        .fromFolder("./src/test/resources/data"))
        .flatMap(new Function.FlatMapFunction<String, String>()
        {
          @Override
          public Iterable<String> f(String input)
          {
            return Arrays.asList(input.split(" "));
          }
        });
    mystream.myFilterAndMap(new Function.MapFunction<String, String>()
    {
      @Override
      public String f(String input)
      {
        return input.replace("word", "newword");
      }
    }, new Function.FilterFunction<String>()
    {
      @Override
      public boolean f(String input)
      {
        return input.startsWith("word");
      }
    }).window(new WindowOption.GlobalWindow(), new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(Duration.millis(1000)))
    .countByKey(new Function.ToKeyValue<String, String, Long>()
    {
      @Override
      public Tuple<KeyValPair<String, Long>> f(String input)
      {
        return new Tuple.PlainTuple(new KeyValPair<>(input, 1L));
      }
    }).addOperator(collector, collector.inputPort, collector.outputPort).runEmbedded(false, 30000, exitCondition);


    Map<String, Long> dataMap = new HashMap<>();
    for (Tuple.TimestampedTuple<KeyValPair<String, Long>> entry : (List<Tuple.TimestampedTuple<KeyValPair<String, Long>>>)TupleCollector.results.get(testId)) {
      dataMap.put(entry.getValue().getKey(), entry.getValue().getValue());
    }

    Assert.assertTrue(dataMap.size() > 1);
    Assert.assertEquals(expected, dataMap);
  }

}