Producer.java

package com.danubemessaging.client;

import com.danubemessaging.client.errors.DanubeClientException;
import com.danubemessaging.client.internal.producer.TopicProducer;
import com.danubemessaging.client.internal.retry.RetryManager;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Sends messages to a Danube topic.
 *
 * <p>Obtain an instance via {@link DanubeClient#newProducer()}. Call {@link #create()} before
 * sending messages. For partitioned topics, partitions are discovered automatically or set
 * explicitly via {@link ProducerBuilder#withPartitions(int)}.
 *
 * <p>This class is thread-safe.
 */
public final class Producer {
    private enum LifecycleState {
        NEW,
        CREATED,
        CLOSED
    }

    private final DanubeClient client;
    private final ProducerOptions options;
    private final List<TopicProducer> topicProducers = new ArrayList<>();
    private final AtomicInteger nextPartition = new AtomicInteger();
    private final AtomicReference<LifecycleState> lifecycleState = new AtomicReference<>(LifecycleState.NEW);

    Producer(DanubeClient client, ProducerOptions options) {
        this.client = Objects.requireNonNull(client, "client");
        this.options = Objects.requireNonNull(options, "options");
    }

    /** Returns the options this producer was built with. */
    public ProducerOptions options() {
        return options;
    }

    /**
     * Registers the producer on the broker asynchronously.
     * Equivalent to calling {@link #create()} on the IO executor.
     *
     * @return a future that completes when the producer is registered
     */
    public CompletableFuture<Void> createAsync() {
        return CompletableFuture.runAsync(this::create, client.ioExecutor());
    }

    /**
     * Registers the producer on the broker.
     * Must be called before {@link #send(byte[], java.util.Map)}.
     * Idempotent — calling twice on an already-created producer is a no-op.
     *
     * @throws com.danubemessaging.client.errors.DanubeClientException if registration fails
     */
    public synchronized void create() {
        ensureOpen();

        if (lifecycleState.get() == LifecycleState.CREATED) {
            return;
        }

        List<String> targets;
        if (options.partitions() > 0) {
            targets = new ArrayList<>(options.partitions());
            for (int i = 0; i < options.partitions(); i++) {
                targets.add(options.topic() + "-part-" + i);
            }
        } else {
            List<String> partitions = client.lookupService().topicPartitions(client.serviceUri(), options.topic());
            targets = partitions.isEmpty() ? List.of(options.topic()) : partitions;
        }

        SchemaRegistryClient schemaRegistry = options.schemaReference() != null
                ? client.newSchemaRegistry()
                : null;

        RetryManager retryManager = hasCustomRetryOptions()
                ? new RetryManager(options.maxRetries(), options.baseBackoffMs(), options.maxBackoffMs())
                : client.retryManager();

        try {
            for (int i = 0; i < targets.size(); i++) {
                String partitionTopic = targets.get(i);
                String partitionProducerName = targets.size() == 1
                        ? options.producerName()
                        : options.producerName() + "-" + i;

                TopicProducer topicProducer = new TopicProducer(
                        client.serviceUri(),
                        client.connectionManager(),
                        client.lookupService(),
                        client.authService(),
                        client.healthCheckService(),
                        schemaRegistry,
                        retryManager,
                        options,
                        partitionTopic,
                        partitionProducerName);
                try {
                    topicProducer.create();
                } catch (RuntimeException error) {
                    notifyProducerError(topicProducer, error, false);
                    throw error;
                }
                topicProducers.add(topicProducer);
            }
        } catch (RuntimeException error) {
            topicProducers.forEach(TopicProducer::close);
            topicProducers.clear();
            throw error;
        }

        lifecycleState.set(LifecycleState.CREATED);
    }

    /**
     * Sends a message asynchronously.
     *
     * @param payload    message body (may be empty but not null)
     * @param attributes optional key-value metadata attached to the message
     * @return a future resolving to the broker-assigned message sequence ID
     */
    public CompletableFuture<Long> sendAsync(byte[] payload, Map<String, String> attributes) {
        byte[] payloadCopy = payload == null ? new byte[0] : payload.clone();
        Map<String, String> attr = attributes == null ? Map.of() : Map.copyOf(attributes);
        return CompletableFuture.supplyAsync(() -> send(payloadCopy, attr), client.ioExecutor());
    }

    /**
     * Sends a message and blocks until the broker acknowledges receipt.
     *
     * @param payload    message body (may be empty but not null)
     * @param attributes optional key-value metadata; pass {@code Map.of()} for none
     * @return the broker-assigned message sequence ID
     * @throws com.danubemessaging.client.errors.DanubeClientException on unrecoverable error
     */
    public long send(byte[] payload, Map<String, String> attributes) {
        return sendInternal(payload, attributes, null, selectTopicProducer());
    }

    /**
     * Sends a message with a routing key asynchronously for KEY_SHARED subscriptions.
     *
     * @param payload    message body
     * @param attributes optional metadata
     * @param routingKey the routing key; all messages with the same key go to the same consumer
     * @return a future resolving to the broker-assigned message sequence ID
     */
    public CompletableFuture<Long> sendWithKeyAsync(byte[] payload, Map<String, String> attributes,
            String routingKey) {
        byte[] payloadCopy = payload == null ? new byte[0] : payload.clone();
        Map<String, String> attr = attributes == null ? Map.of() : Map.copyOf(attributes);
        return CompletableFuture.supplyAsync(() -> sendWithKey(payloadCopy, attr, routingKey), client.ioExecutor());
    }

    /**
     * Sends a message with a routing key for KEY_SHARED subscriptions.
     *
     * <p>For partitioned topics, hashes the routing key to a specific partition ensuring
     * all messages with the same key always go to the same partition. For non-partitioned
     * topics, simply tags the routing key on the message.
     *
     * @param payload    message body
     * @param attributes optional metadata; pass {@code Map.of()} for none
     * @param routingKey the routing key; must not be null
     * @return the broker-assigned message sequence ID
     * @throws com.danubemessaging.client.errors.DanubeClientException on unrecoverable error
     */
    public long sendWithKey(byte[] payload, Map<String, String> attributes, String routingKey) {
        ensureOpen();
        TopicProducer topicProducer = selectTopicProducerForKey(routingKey);
        return sendInternal(payload, attributes, routingKey, topicProducer);
    }

    private long sendInternal(byte[] payload, Map<String, String> attributes,
            String routingKey, TopicProducer topicProducer) {
        ensureOpen();

        if (lifecycleState.get() != LifecycleState.CREATED) {
            create();
        }

        int attempts = 0;

        while (true) {
            try {
                return topicProducer.send(payload, attributes, routingKey);
            } catch (RuntimeException error) {
                boolean unrecoverable = client.retryManager().isUnrecoverable(error);
                if (unrecoverable) {
                    notifyProducerError(topicProducer, error, true);
                    topicProducer.relookupAndCreate();
                    attempts = 0;
                    continue;
                }

                boolean retryable = client.retryManager().isRetryable(error);
                if (!retryable) {
                    notifyProducerError(topicProducer, error, false);
                    throw error;
                }

                notifyProducerError(topicProducer, error, true);

                attempts++;
                if (attempts > client.retryManager().maxRetries()) {
                    topicProducer.relookupAndCreate();
                    attempts = 0;
                    continue;
                }

                sleepBackoff(client.retryManager().calculateBackoff(attempts - 1));
            }
        }
    }

    /**
     * Closes this producer and releases all underlying resources.
     * Idempotent — safe to call multiple times.
     */
    public synchronized void close() {
        if (lifecycleState.get() == LifecycleState.CLOSED) {
            return;
        }

        lifecycleState.set(LifecycleState.CLOSED);
        topicProducers.forEach(TopicProducer::close);
        topicProducers.clear();
    }

    private void ensureOpen() {
        if (lifecycleState.get() == LifecycleState.CLOSED) {
            throw new DanubeClientException("Producer is closed");
        }
    }

    private void notifyProducerError(TopicProducer topicProducer, Throwable error, boolean retryable) {
        try {
            options.eventListener().onProducerError(topicProducer.topic(), topicProducer.producerName(), error,
                    retryable);
        } catch (RuntimeException ignore) {
            // Listener errors must never break client flow.
        }
    }

    private TopicProducer selectTopicProducer() {
        if (topicProducers.isEmpty()) {
            throw new DanubeClientException("Producer is not initialized");
        }

        if (topicProducers.size() == 1) {
            return topicProducers.get(0);
        }

        int index = Math.floorMod(nextPartition.getAndIncrement(), topicProducers.size());
        return topicProducers.get(index);
    }

    private TopicProducer selectTopicProducerForKey(String routingKey) {
        if (topicProducers.isEmpty()) {
            throw new DanubeClientException("Producer is not initialized");
        }

        if (topicProducers.size() == 1) {
            return topicProducers.get(0);
        }

        int index = Math.floorMod((int) fnv1aHash(routingKey), topicProducers.size());
        return topicProducers.get(index);
    }

    /**
     * FNV-1a 64-bit hash — must match Rust/Go/Python constants.
     */
    static long fnv1aHash(String key) {
        long hash = 0xcbf29ce484222325L;
        byte[] bytes = key.getBytes(java.nio.charset.StandardCharsets.UTF_8);
        for (byte b : bytes) {
            hash ^= (b & 0xFF);
            hash *= 0x100000001b3L;
        }
        return hash;
    }

    private boolean hasCustomRetryOptions() {
        return options.maxRetries() > 0 || options.baseBackoffMs() > 0 || options.maxBackoffMs() > 0;
    }

    private static void sleepBackoff(Duration backoff) {
        try {
            Thread.sleep(backoff);
        } catch (InterruptedException interrupted) {
            Thread.currentThread().interrupt();
            throw new DanubeClientException("Interrupted while backing off for retry", interrupted);
        }
    }
}