/*
 * Copyright 2016 Red Hat Inc.
 *
 * Licensed 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 io.vertx.kafka.client.producer;

import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.streams.WriteStream;
import io.vertx.kafka.client.producer.impl.KafkaWriteStreamImpl;
import io.vertx.kafka.client.serialization.VertxSerdes;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.serialization.Serializer;

import java.util.List;
import java.util.Map;
import java.util.Properties;

A WriteStream for writing to Kafka ProducerRecord. 
 The WriteStream.write(Object) provides global control over writing a record. 

/**
 * A {@link WriteStream} for writing to Kafka {@link ProducerRecord}.
 * <p>
 * The {@link #write(Object)} provides global control over writing a record.
 * <p>
 */
public interface KafkaWriteStream<K, V> extends WriteStream<ProducerRecord<K, V>> {

  int DEFAULT_MAX_SIZE = 1024 * 1024;

  
Create a new KafkaWriteStream instance
Params:
vertx – Vert.x instance to use
config –  Kafka producer configuration
Returns:
 an instance of the KafkaWriteStream/**
   * Create a new KafkaWriteStream instance
   *
   * @param vertx Vert.x instance to use
   * @param config  Kafka producer configuration
   * @return  an instance of the KafkaWriteStream
   */
  static <K, V> KafkaWriteStream<K, V> create(Vertx vertx, Properties config) {
    return KafkaWriteStreamImpl.create(vertx, config);
  }

  
Create a new KafkaWriteStream instance
Params:
vertx – Vert.x instance to use
config –  Kafka producer configuration
keyType – class type for the key serialization
valueType – class type for the value serialization
Returns:
 an instance of the KafkaWriteStream/**
   * Create a new KafkaWriteStream instance
   *
   * @param vertx Vert.x instance to use
   * @param config  Kafka producer configuration
   * @param keyType class type for the key serialization
   * @param valueType class type for the value serialization
   * @return  an instance of the KafkaWriteStream
   */
  static <K, V> KafkaWriteStream<K, V> create(Vertx vertx, Properties config, Class<K> keyType, Class<V> valueType) {
    Serializer<K> keySerializer = VertxSerdes.serdeFrom(keyType).serializer();
    Serializer<V> valueSerializer = VertxSerdes.serdeFrom(valueType).serializer();
    return KafkaWriteStreamImpl.create(vertx, config, keySerializer, valueSerializer);
  }

  
Create a new KafkaWriteStream instance
Params:
vertx – Vert.x instance to use
config –  Kafka producer configuration
keySerializer – key serializer
valueSerializer – value serializer
Returns:
 an instance of the KafkaWriteStream/**
   * Create a new KafkaWriteStream instance
   *
   * @param vertx Vert.x instance to use
   * @param config  Kafka producer configuration
   * @param keySerializer key serializer
   * @param valueSerializer value serializer
   * @return  an instance of the KafkaWriteStream
   */
  static <K, V> KafkaWriteStream<K, V> create(Vertx vertx, Properties config, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
    return KafkaWriteStreamImpl.create(vertx, config, keySerializer, valueSerializer);
  }

  
Create a new KafkaWriteStream instance
Params:
vertx – Vert.x instance to use
config –  Kafka producer configuration
Returns:
 an instance of the KafkaWriteStream/**
   * Create a new KafkaWriteStream instance
   *
   * @param vertx Vert.x instance to use
   * @param config  Kafka producer configuration
   * @return  an instance of the KafkaWriteStream
   */
  static <K, V> KafkaWriteStream<K, V> create(Vertx vertx, Map<String, Object> config) {
    return KafkaWriteStreamImpl.create(vertx, config);
  }

  
Create a new KafkaWriteStream instance
Params:
vertx – Vert.x instance to use
config –  Kafka producer configuration
keyType – class type for the key serialization
valueType – class type for the value serialization
Returns:
 an instance of the KafkaWriteStream/**
   * Create a new KafkaWriteStream instance
   *
   * @param vertx Vert.x instance to use
   * @param config  Kafka producer configuration
   * @param keyType class type for the key serialization
   * @param valueType class type for the value serialization
   * @return  an instance of the KafkaWriteStream
   */
  static <K, V> KafkaWriteStream<K, V> create(Vertx vertx, Map<String, Object> config, Class<K> keyType, Class<V> valueType) {
    Serializer<K> keySerializer = VertxSerdes.serdeFrom(keyType).serializer();
    Serializer<V> valueSerializer = VertxSerdes.serdeFrom(valueType).serializer();
    return KafkaWriteStreamImpl.create(vertx, config, keySerializer, valueSerializer);
  }

  
Create a new KafkaWriteStream instance
Params:
vertx – Vert.x instance to use
config –  Kafka producer configuration
keySerializer – key serializer
valueSerializer – value serializer
Returns:
 an instance of the KafkaWriteStream/**
   * Create a new KafkaWriteStream instance
   *
   * @param vertx Vert.x instance to use
   * @param config  Kafka producer configuration
   * @param keySerializer key serializer
   * @param valueSerializer value serializer
   * @return  an instance of the KafkaWriteStream
   */
  static <K, V> KafkaWriteStream<K, V> create(Vertx vertx, Map<String, Object> config, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
    return KafkaWriteStreamImpl.create(vertx, config, keySerializer, valueSerializer);
  }

  
Create a new KafkaWriteStream instance
Params:
vertx – Vert.x instance to use
producer –  native Kafka producer instance/**
   * Create a new KafkaWriteStream instance
   *
   * @param vertx Vert.x instance to use
   * @param producer  native Kafka producer instance
   */
  static <K, V> KafkaWriteStream<K, V> create(Vertx vertx, Producer<K, V> producer) {
    return new KafkaWriteStreamImpl<>(vertx.getOrCreateContext(), producer);
  }

  
Asynchronously write a record to a topic
Params:
record –  record to write
Returns:
 current KafkaWriteStream instance/**
   * Asynchronously write a record to a topic
   *
   * @param record  record to write
   * @return  current KafkaWriteStream instance
   */
  KafkaWriteStream<K, V> send(ProducerRecord<K, V> record);

  
Asynchronously write a record to a topic
Params:
record –  record to write
handler – handler called on operation completed
Returns:
 current KafkaWriteStream instance/**
   * Asynchronously write a record to a topic
   *
   * @param record  record to write
   * @param handler handler called on operation completed
   * @return  current KafkaWriteStream instance
   */
  KafkaWriteStream<K, V> send(ProducerRecord<K, V> record, Handler<AsyncResult<RecordMetadata>> handler);

  
Get the partition metadata for the give topic.
Params:
topic – topic partition for which getting partitions info
handler – handler called on operation completed
Returns:
 current KafkaWriteStream instance/**
   * Get the partition metadata for the give topic.
   *
   * @param topic topic partition for which getting partitions info
   * @param handler handler called on operation completed
   * @return  current KafkaWriteStream instance
   */
  KafkaWriteStream<K, V> partitionsFor(String topic, Handler<AsyncResult<List<PartitionInfo>>> handler);

  
Invoking this method makes all buffered records immediately available to write
Params:
completionHandler – handler called on operation completed
Returns:
 current KafkaWriteStream instance/**
   * Invoking this method makes all buffered records immediately available to write
   *
   * @param completionHandler handler called on operation completed
   * @return  current KafkaWriteStream instance
   */
  KafkaWriteStream<K, V> flush(Handler<Void> completionHandler);

  
Close the stream
/**
   * Close the stream
   */
  void close();

  
Close the stream
Params:
completionHandler – handler called on operation completed/**
   * Close the stream
   *
   * @param completionHandler handler called on operation completed
   */
  void close(Handler<AsyncResult<Void>> completionHandler);

  
Close the stream
Params:
timeout – timeout to wait for closing
completionHandler – handler called on operation completed/**
   * Close the stream
   *
   * @param timeout timeout to wait for closing
   * @param completionHandler handler called on operation completed
   */
  void close(long timeout, Handler<AsyncResult<Void>> completionHandler);

  
Returns:
the underlying producer/**
   * @return the underlying producer
   */
  Producer<K, V> unwrap();
}