package fr.kosmos.pulsar.multitenant.testapp;

import static fr.kosmos.pulsar.multitenant.testapp.Constants.FORMATTER;

import java.time.ZonedDateTime;
import java.util.concurrent.TimeUnit;

import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;

public class SimpleConsumerWithRetry {

	public static void main(String[] args) throws PulsarClientException {
		PulsarClient pulsarClient = Constants.initPulsarClient();

		DeadLetterPolicy deadLetterPolicy = DeadLetterPolicy.builder()
		                                                    .maxRedeliverCount(2)
		                                                    .retryLetterTopic("test-subs-RETRY")
		                                                    .deadLetterTopic("test-subs-DLQ")
		                                                    .initialSubscriptionName("dlq-sub")
		                                                    .build();

		final MessageListener<String> listener = (consumer, message) -> {
			MessageId msgId = message.getMessageId();
			String value = message.getValue();
			String formattedDate = ZonedDateTime.now().format(FORMATTER);
			System.out.println(formattedDate + ": receive msg " + msgId + ", value:" + value);
			try {
				consumer.reconsumeLater(message, 2, TimeUnit.SECONDS);
			} catch (PulsarClientException e) {
				throw new RuntimeException(e);
			}
		};

		final MessageListener<String> listener2 = (consumer, message) -> {
			MessageId msgId = message.getMessageId();
			String value = message.getValue();
			String formattedDate = ZonedDateTime.now().format(FORMATTER);
			System.out.println(formattedDate + ": receive msg from DLQ " + msgId + ", value:" + value);
			try {
				consumer.acknowledge(message);
			} catch (PulsarClientException e) {
				throw new RuntimeException(e);
			}
		};

		pulsarClient.newConsumer(Schema.STRING)
		            .topic("persistent://public/testns/test-subs-DLQ")
		            .subscriptionName("dlq-sub")
		            .subscriptionType(SubscriptionType.Shared)
		            .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
		            .messageListener(listener2)
		            .subscribe();

		System.out.println(">> pulsar DLQ consumer created");

		pulsarClient.newConsumer(Schema.STRING)
		            .topic("persistent://public/testns/test")
		            .subscriptionName("subs")
		            .subscriptionType(SubscriptionType.Shared)
		            .enableRetry(true)
		            .deadLetterPolicy(deadLetterPolicy)
		            .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
		            .messageListener(listener)
		            .subscribe();

		System.out.println(">> pulsar consumer created");
	}
}