package org.apache.kafka.streams.test;

public class TestInputTopic<K, V> {
    //Timestamp handling 
    //Record timestamp can be provided when piping input or use internally tracked time initialized with constructors:
    //startTimestampMs the initial timestamp for generated records, if not provided uses current system time as start timestamp.
    //autoAdvanceMs the time increment per generated record, if not provided auto-advance is disabled.

    //Constructors with serializers
    TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serializer<K> keySerializer, final Serializer<V> valueSerializer);
    TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serializer<K> keySerializer, final Serializer<V> valueSerializer, final long startTimestampMs);
    TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serializer<K> keySerializer, final Serializer<V> valueSerializer, final long startTimestampMs, final long autoAdvanceMs);
    //Constructors with serdes
    TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serde<K> keySerde, final Serde<V> valueSerde);
    TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serde<K> keySerde, final Serde<V> valueSerde, final long startTimestampMs);
    TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serde<K> keySerde, final Serde<V> valueSerde, final long startTimestampMs, final long autoAdvanceMs);

    //Advances the internally tracked time.
    void advanceTimeMs(final long advanceMs);

	//Methods to pipe single record
    void pipeInput(final V value);
    void pipeInput(final K key, final V value);
    void pipeInput(final V value, final long timestampMs);
    void pipeInput(final K key, final V value, final long timestampMs);
    void pipeInput(final K key, final V value, final Headers headers);
    void pipeInput(final K key, final V value, final Headers headers, final long timestampMs);

	//Methods to pipe list of records
    void pipeKeyValueList(final List<KeyValue<K, V>> keyValues);
    void pipeValueList(final List<V> values);
    void pipeKeyValueList(final List<KeyValue<K, V>> keyValues, final long startTimestamp, final long advanceMs);
    void pipeValueList(final List<V> values, final long startTimestamp, final long advanceMs);
}

public class SimpleTopicTest {

  private TopologyTestDriver testDriver;
  private TestInputTopic<String, String> inputTopic;
  private TestOutputTopic<String, String> outputTopic;

  @Before
  public void setup() {
    final StreamsBuildertestDriver builder = new StreamsBuilder();
    TestStream app = new TestStream();
    //Create Actual Stream Processing pipeline
    app.createStream(builder);
    testDriver = new TopologyTestDriver(builder.build(),app.configTopologyTestDriver(TestStream.getTopology(), TestStream.getConfig());
    inputTopic = new TestInputTopic<String, String>TestInputTopic<>(testDriver, TestStream.INPUT_TOPIC, new Serdes.StringSerde(), new Serdes.StringSerde());
    outputTopic = new TestOutputTopic<String, String>TestOutputTopic<>(testDriver, TestStream.OUTPUT_TOPIC, new Serdes.StringSerde(), new Serdes.StringSerde());
  }

  @After
  public void tearDown() {
      testDriver.close();
  }

  @Test
  public void testOneWord() {
    //Feed word "Hello" to inputTopic and no kafka key, timestamp is irrelevant in this case
    inputTopic.pipeInput("Hello");
    assertThat(outputTopic.readValue()).isEqualTo("Hello");
    //No more output in topic
    assertThat(outputTopic.readRecord()).isNull();
  }
}