package org.apache.kafka.clientsstreams;

//Newpublic Interfaceclass to make possible to use ProducereRecord in TopologyTestDriver
//ConsumerRecord not modified to implement this because timestamp() return long, not Long
public interface ClientRecord<K, V> {
    String topic(TopologyTestDriver {
    public TopologyTestDriver(Topology topology, Properties config); // existing constructor
    @Deprecate public TopologyTestDriver(Topology topology, Properties config, long initialWallClockTimeMs);
    Headerspublic headersTopologyTestDriver();
Topology topology, Properties config, KInstant key(initialWallClockTime);

    @Deprecate public Vvoid valueadvanceWallClockTime(long advanceMs);
 // can trigger wall-clock-time punctuation
	public Longvoid timestampadvanceWallClockTime(Duration advance);
}

package org.apache.kafka.streams.test;

public class TestInputTopic<K, V> {
    //Create by TopologyTestDriver, Constructors are package private

    //Timestamp handling 
    //Record timestamp can be provided when piping input or use internally tracked time initialized configured with configureTimingparameters:
    //startTimestampMsstartTimestamp the initial timestamp for generated records, if not provided uses current system time as start timestamp.
    //autoAdvanceMsautoAdvance the time increment per generated record, if not provided auto-advance is disabled.

    void configureTiming(final long startTimestampMs, final long autoAdvanceMs);
    //Advances the internally tracked time.
    void advanceTimeMsadvanceTime(final longDuration advanceMsadvance);

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

    // Use provided timestamp, does not auto advance internally tracked time.
    void pipeInput(final V value, final longInstant timestampMstimestamp);
    void pipeInput(final K key, final V value, final longInstant timestampMstimestamp);

    // Method with long provided to support easier migration of old tests
   void pipeInput(final K key, final V value, final Headerslong headerstimestampMs);

    // voidIf pipeInput(finalrecord Ktimestamp keyset, final V value, final Headers headers, final long timestampMs does not auto advance internally tracked time.
    void pipeInput(final TestRecord<K, V> record);

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

    // Use provided timestamp, does not auto advance internally tracked time.
    void pipeValueList(final List<V> values, final Instant startTimestamp, final Duration advanceMs);
    void pipeKeyValueList(final List<KeyValue<K, V>> keyValues, final longInstant startTimestamp, final longDuration advanceMs);

    void pipeValueList(final List<V> values, final long startTimestamp, final long advanceMs// If record timestamp set, does not auto advance internally tracked time.
    void pipeRecordList(final List<? extends TestRecord<K, V>> records);
}

package org.apache.kafka.streams.test;

public class TestOutputTopic<K, V> {
    //Create by TopologyTestDriver, Constructors are package private

	//Method to check queue size
    final long getQueueSize();
    final boolean isEmpty();

	//Methods to readOutput, throw NoSuchElement if no record in queue
    ClientRecord<K, V> readRecordV readValue();
    KeyValue<K, V> readKeyValue();
    V readValueTestRecord<K, V> readRecord();

    //Output as collection
    List<V> readValuesToList();
    List<KeyValue<K, V>> readKeyValuesToList();
    Map<K, V> readKeyValuesToMap();
    List<KeyValue<K, V>> readKeyValuesToList();
    List<V> readValuesToList();
}List<TestRecord<K, V>> readRecordsToList();
}

public class SimpleTopicTest {

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

  @Before
  public void setup() {
    testDriver = new TopologyTestDriver(TestStream.getTopology(), TestStream.getConfig());
    	inputTopic = new TestInputTopic<>(testDriver, testDriver.createInputTopic(TestStream.INPUT_TOPIC, new Serdes.StringSerdeStringDeserializer(), new Serdes.StringSerdeStringDeserializer());
    outputTopic = new TestOutputTopic<>(testDriver, testDriver.createOutputTopic(TestStream.OUTPUT_TOPIC, new Serdes.StringSerdeStringSerializer(), new Serdes.StringSerdeLongSerializer());
  }

  @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.readRecordisEmpty()).isNullisTrue();
  }
}