Page History

/**
* Simple Test case using a collection as an input for a High level application
*/
 
public class MyStreamApplication implements StreamApplication {
  @Override
  public void init(StreamGraph graph, Config config) {
    MessageStream<Integer> pageViews = graph.getInputStream(“test-samza.page-views”);

    pageViews.map(s -> "processed " + s)
             .sendTo(graph.getOutputStream(“test-samza.Output”));
 }
}


 
// Initialize and run the test framework
TestRunner
	.of(new MyStreamApplication());
    .addInputStream(CollectionStream.of("test-samza.input", {1,2,3}))
    .addOutputStream(CollectionStream.empty("test-samza.output"))
    .run();
 
// Assertions on the outputs
StreamAssert.that("test-samza.inputOutput").contains({"processed 1", "processed 2", "processed 4"});