package com.ankurm.kafkabasics;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

/**
 * Three ways to send, and only two of them tell you when the send failed.
 *
 * @see <a href="../../../../../docs/02-producing.md">docs/02-producing.md</a>
 */
@Component
public class OrderProducer {

	private static final Logger log = LoggerFactory.getLogger(OrderProducer.class);

	static final String TOPIC = "orders";

	private final KafkaTemplate<String, OrderEvent> template;

	public OrderProducer(KafkaTemplate<String, OrderEvent> template) {
		this.template = template;
	}

	/**
	 * Fire and forget. {@code send} returns a {@code CompletableFuture} and this method throws
	 * it away, so a broker-side rejection is invisible here: the method returns normally, the
	 * message never lands, and nothing in your logs says so unless you have the producer's own
	 * logger turned up. This is the single most common way to lose messages in Spring Kafka.
	 */
	public void sendAndForget(OrderEvent event) {
		this.template.send(TOPIC, event.orderId(), event);
	}

	/** Asynchronous, but the outcome is handled. This is the shape you want by default. */
	public CompletableFuture<SendResult<String, OrderEvent>> send(OrderEvent event) {
		CompletableFuture<SendResult<String, OrderEvent>> future =
				this.template.send(TOPIC, event.orderId(), event);
		future.whenComplete((result, ex) -> {
			if (ex != null) {
				log.error("send failed for orderId={}", event.orderId(), ex);
			}
			else {
				log.info("sent orderId={} to {}-{}@{}", event.orderId(),
						result.getRecordMetadata().topic(), result.getRecordMetadata().partition(),
						result.getRecordMetadata().offset());
			}
		});
		return future;
	}

	/**
	 * Synchronous. Correct when the caller must not proceed unless the write is durable, and
	 * expensive for exactly that reason: it blocks a thread for a network round trip plus the
	 * replication acknowledgement.
	 */
	public SendResult<String, OrderEvent> sendAndWait(OrderEvent event) throws Exception {
		return this.template.send(new ProducerRecord<>(TOPIC, event.orderId(), event)).get();
	}

}