Page History

/**
* TestApplication provides static factory for quick setup of Samza environment for testing High Level Api, users can configure input  
* streams and then apply various operators on streams and run the application
*/
 
public class TestApplication<T> {  
  // Static factory to config & create runner for High level api 
  public static TestApplication create(Map<String, String> config, Mode mode) {...}
  
  // Configure state for application
  public TestApplication addState(Serde<?> serde) {...}  
  // Configure any kind of input stream for samza system, and get the handle of message stream to apply operators on
  public <T> MessageStream<T> getInputStream(EventStream<T> stream) {...}
  public <T> MessageStream<T> getInputStream(CollectionStream<T> stream) {...}
  public <T> MessageStream<T> getInputStream(FileStream<T> stream) {...}
  
  // Run the app
  public void run() { ...};
}

/**
* CollectionStream provides utilities to build an in memory input stream of collections(list, map). It also supports initialization of 
* multiple partitions for an input stream 
*/
 
public class CollectionStream<T> { 
  // Create an empty stream that a Samza task can produce to
  public static <T> CollectionStream<T> empty(String steamId) {...}
  
  // Create a stream of messages from input list with single partition
  public static <T> CollectionStream<T> of(String steamId, Iterable<T> collection){...}
  
  // Create a stream of messages from input list with multiple partition, key of partitions map is partitionId
  public static <T> CollectionStream<T> of(String steamId, Map<Integer,Iterable> partitions){...}
  
  // Create a stream of messages from input list with multiple partition (REMOVE THIS ??????)
  public static <K, V> CollectionStream<KV<K, V>> of(String steamId, Map<K, V> elems) {...}
}