package org.example;

import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionMode;

public class Main {

  public static void main(String[] args) throws PulsarClientException, InterruptedException {
    AtomicBoolean running = new AtomicBoolean(true);

    String topic = "tbce0";

    PulsarClient client = PulsarClient.builder()
        .serviceUrl("pulsar://localhost:61723")
        .build();

    Producer<byte[]> producer = client.newProducer()
        .topic(topic)
        .create();

    String message = "Hello, Pulsar!";
    for (int i = 0; i < 1000; i++) {
      producer.send(message.getBytes());
    }
    System.out.println("Sent messages: " + message);
    producer.close();

    Thread.ofVirtual().start(() -> {
      try {
        Consumer<byte[]> consumer = client.newConsumer()
            .topic(topic)
            .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
            .subscriptionName("bce-subscription-0")
            .subscriptionMode(SubscriptionMode.NonDurable)
            //.readCompacted(true)
            .subscribe();

        int counter = 1;
        while (running.get()) {
          Message<byte[]> receivedMessage = consumer.receive();
          System.out.println(
              "Received " + counter++ + " message: " + new String(receivedMessage.getData()));
          consumer.acknowledge(receivedMessage);
        }
        consumer.close();
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    });

    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
      System.out.println("Exiting...");
      running.set(false);
      try {
        client.close();
      } catch (PulsarClientException e) {
        throw new RuntimeException(e);
      }
    }));

    Thread.sleep(600_000);
  }
}