/*
 * 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.kafka.streams;

import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.streams.test.TestRecord;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.StringJoiner;

/**
 * {@code TestOutputTopic} is used to read records from a topic in {@link TopologyTestDriver}.
 * To use {@code TestOutputTopic} create a new instance via
 * {@link TopologyTestDriver#createOutputTopic(String, Deserializer, Deserializer)}.
 * In actual test code, you can read record values, keys, {@link KeyValue} or {@link TestRecord}
 * If you have multiple source topics, you need to create a {@code TestOutputTopic} for each.
 * <p>
 * If you need to test key, value and headers, use {@link #readRecord()} methods.
 * Using {@link #readKeyValue()} you get a {@link KeyValue} pair, and thus, don't get access to the record's
 * timestamp or headers.
 * Similarly using {@link #readValue()} you only get the value of a record.
 *
 * <h2>Processing records</h2>
 * <pre>{@code
 *     private TestOutputTopic<String, Long> outputTopic;
 *      ...
 *     outputTopic = testDriver.createOutputTopic(OUTPUT_TOPIC, stringDeserializer, longDeserializer);
 *     ...
 *     assertThat(outputTopic.readValue()).isEqual(1);
 * }</pre>
 *
 * @param <K> the type of the record key
 * @param <V> the type of the record value
 * @see TopologyTestDriver
 */
public class TestOutputTopic<K, V> {
    private final TopologyTestDriver driver;
    private final String topic;
    private final Deserializer<K> keyDeserializer;
    private final Deserializer<V> valueDeserializer;

    TestOutputTopic(final TopologyTestDriver driver,
                    final String topicName,
                    final Deserializer<K> keyDeserializer,
                    final Deserializer<V> valueDeserializer) {
        Objects.requireNonNull(driver, "TopologyTestDriver cannot be null");
        Objects.requireNonNull(topicName, "topicName cannot be null");
        Objects.requireNonNull(keyDeserializer, "keyDeserializer cannot be null");
        Objects.requireNonNull(valueDeserializer, "valueDeserializer cannot be null");
        this.driver = driver;
        this.topic = topicName;
        this.keyDeserializer = keyDeserializer;
        this.valueDeserializer = valueDeserializer;
    }

    /**
     * Read one record from the output topic and return record's value.
     *
     * @return Next value for output topic.
     */
    public V readValue() {
        final TestRecord<K, V> record = readRecord();
        return record.value();
    }

    /**
     * Read one record from the output topic and return its key and value as pair.
     *
     * @return Next output as {@link KeyValue}.
     */
    public KeyValue<K, V> readKeyValue() {
        final TestRecord<K, V> record = readRecord();
        return new KeyValue<>(record.key(), record.value());
    }

    /**
     * Read one Record from output topic.
     *
     * @return Next output as {@link TestRecord}.
     */
    public TestRecord<K, V> readRecord() {
        return driver.readRecord(topic, keyDeserializer, valueDeserializer);
    }

    /**
     * Read output to List.
     * This method can be used if the result is considered a stream.
     * If the result is considered a table, the list will contain all updated, ie, a key might be contained multiple times.
     * If you are only interested in the last table update (ie, the final table state),
     * you can use {@link #readKeyValuesToMap()} instead.
     *
     * @return List of output.
     */
    public List<TestRecord<K, V>> readRecordsToList() {
        final List<TestRecord<K, V>> output = new LinkedList<>();
        while (!isEmpty()) {
            output.add(readRecord());
        }
        return output;
    }


    /**
     * Read output to map.
     * This method can be used if the result is considered a table,
     * when you are only interested in the last table update (ie, the final table state).
     * If the result is considered a stream, you can use {@link #readRecordsToList()} instead.
     * The list will contain all updated, ie, a key might be contained multiple times.
     * If the last update to a key is a delete/tombstone, the key will still be in the map (with null-value).
     *
     * @return Map of output by key.
     */
    public Map<K, V> readKeyValuesToMap() {
        final Map<K, V> output = new HashMap<>();
        TestRecord<K, V> outputRow;
        while (!isEmpty()) {
            outputRow = readRecord();
            if (outputRow.key() == null) {
                throw new IllegalStateException("Null keys not allowed with readKeyValuesToMap method");
            }
            output.put(outputRow.key(), outputRow.value());
        }
        return output;
    }

    /**
     * Read all KeyValues from topic to List.
     *
     * @return List of output KeyValues.
     */
    public List<KeyValue<K, V>> readKeyValuesToList() {
        final List<KeyValue<K, V>> output = new LinkedList<>();
        KeyValue<K, V> outputRow;
        while (!isEmpty()) {
            outputRow = readKeyValue();
            output.add(outputRow);
        }
        return output;
    }

    /**
     * Read all values from topic to List.
     *
     * @return List of output values.
     */
    public List<V> readValuesToList() {
        final List<V> output = new LinkedList<>();
        V outputValue;
        while (!isEmpty()) {
            outputValue = readValue();
            output.add(outputValue);
        }
        return output;
    }

    /**
     * Get size of unread record in the topic queue.
     *
     * @return size of topic queue.
     */
    public final long getQueueSize() {
        return driver.getQueueSize(topic);
    }

    /**
     * Verify if the topic queue is empty.
     *
     * @return {@code true} if no more record in the topic queue.
     */
    public final boolean isEmpty() {
        return driver.isEmpty(topic);
    }

    @Override
    public String toString() {
        return new StringJoiner(", ", TestOutputTopic.class.getSimpleName() + "[", "]")
                .add("topic='" + topic + "'")
                .add("keyDeserializer=" + keyDeserializer.getClass().getSimpleName())
                .add("valueDeserializer=" + valueDeserializer.getClass().getSimpleName())
                .add("size=" + getQueueSize())
                .toString();
    }
}
