Integration API producer.java

import dev.restate.integration.Invocation;
import dev.restate.integration.ProducerNotReadyException;

import java.util.concurrent.CompletableFuture;

/**
 * An at-least-once producer: the client assigns a monotonically increasing offset to each
 * invocation. Deduplication is disabled (empty producer id); add an idempotency key on the
 * invocations if you need handler-level dedup.
 *
 * <h2>Sending in order</h2>
 *
 * Await each {@link #send} future before the next one — that is what keeps ordering and applies
 * backpressure, so you never need {@link #waitReady()} here.
 *
 * <pre>{@code
 * try (IntegrationClient client = IntegrationClient.builder("http://localhost:8080").build();
 *     Producer producer = client.newProducer()) {
 *   for (byte[] payload : payloads) {
 *     long offset = producer.send(producer.newInvocation().setBody(payload)).get();
 *   }
 *   producer.waitAcknowledged().get(); // block until durably acknowledged by Restate
 * }
 * }</pre>
 *
 * <h2>Stream defaults</h2>
 *
 * Pass an {@link InvocationMetadata} to {@link IntegrationClient#newProducer(InvocationMetadata)}
 * to set fields shared by every record (e.g. the target service/handler) once; per-invocation
 * fields override them.
 *
 * <pre>{@code
 * Producer producer =
 *     client.newProducer(
 *         client.newInvocationMetadata().setServiceName("Greeter").setHandlerName("greet"));
 * }</pre>
 *
 * <h2>Non-blocking sends</h2>
 *
 * {@link #trySend} attempts a send and throws {@link ProducerNotReadyException} when the send
 * window is full; use {@link #waitReady()} to await capacity, then retry.
 *
 * <pre>{@code
 * try {
 *   producer.trySend(invocation);
 * } catch (ProducerNotReadyException notReady) {
 *   producer.waitReady().get();
 *   producer.trySend(invocation);
 * }
 * }</pre>
 *
 * <h2>Thread safety</h2>
 *
 * A producer is <b>not thread-safe</b> and, like a {@code KafkaProducer}, fails fast (throws {@link
 * java.util.ConcurrentModificationException}) if used from more than one thread at once. Awaiting a
 * returned future from any thread is fine.
 */
public interface Producer extends AutoCloseable {

    /**
     * Send an invocation, returning its assigned offset once it has been written to the stream
     * respecting flow control. Await the returned future before sending the next invocation: that is
     * what keeps ordering and applies backpressure.
     */
    CompletableFuture<Long> send(Invocation invocation);

    /**
     * Try to send an invocation without waiting, returning its assigned offset. Throws {@link
     * ProducerNotReadyException} if the producer is not ready right now — check with {@link
     * #waitReady()} first.
     */
    long trySend(Invocation invocation) throws ProducerNotReadyException;

    /** The highest offset handed to {@code send}/{@code trySend} so far, or {@code -1} if none. */
    long lastSentOffset();

    /**
     * Complete once the producer can accept another record right now (there is send window and the
     * transport is writable). When using {@code send} you don't need this: awaiting the {@code send}
     * future already waits for readiness.
     */
    CompletableFuture<Void> waitReady();

    /**
     * Complete once every record sent so far has been durably acknowledged by Restate, returning the
     * highest acknowledged offset.
     */
    default CompletableFuture<Long> waitAcknowledged() {
        return waitAcknowledged(lastSentOffset());
    }

    /**
     * Complete once all records up to and including {@code offset} have been durably acknowledged by
     * Restate, returning the highest acknowledged offset.
     */
    CompletableFuture<Long> waitAcknowledged(long offset);
}

/**
 * Like {@link Producer}, but with exactly-once semantics: the caller supplies a deterministic,
 * strictly increasing offset for each invocation.
 * Combined with the stream's producer id, Restate deduplicates on {@code (producerId, offset)},
 * so replays after a restart are dropped.
 *
 * <h2>Exactly once</h2>
 *
 * Pick a producer id that is <b>stable across restarts</b> and <b>distinct per independent offset
 * sequence</b> — for a Kafka consumer one id per partition (e.g. {@code groupId/topic/partition}),
 * for Postgres logical replication the slot name. Because dedup is on {@code (producerId, offset)},
 * it is then safe to replay from your last checkpoint after a crash: already-committed offsets are
 * dropped, and {@link #waitAcknowledged(long)} reports how far Restate has durably caught up so you
 * can advance the checkpoint.
 *
 * <pre>{@code
 * producer.send(lsn, producer.newInvocation().setBody(payload)).get();
 * long committed = producer.waitAcknowledged(lsn).get();
 * checkpoint.store(committed);
 * }</pre>
 *
 * <h2>Thread safety</h2>
 *
 * A producer is <b>not thread-safe</b> and, like a {@code KafkaProducer}, fails fast (throws {@link
 * java.util.ConcurrentModificationException}) if used from more than one thread at once. Awaiting a
 * returned future from any thread is fine.
 */
public interface ExactlyOnceProducer extends AutoCloseable {

    /**
     * Send an invocation at {@code offset}, completing once it has been written to the stream
     * respecting flow control. Await the returned future before sending the next invocation: that is
     * what keeps ordering and applies backpressure. Offsets must be strictly increasing.
     */
    CompletableFuture<Void> send(long offset, Invocation invocation);

    /**
     * Try to send an invocation at {@code offset} without waiting. Throws {@link
     * ProducerNotReadyException} if the producer is not ready right now — check with {@link
     * #waitReady()} first. Offsets must be strictly increasing.
     */
    void trySend(long offset, Invocation invocation) throws ProducerNotReadyException;

    /** The highest offset handed to {@code send}/{@code trySend} so far, or {@code -1} if none. */
    long lastSentOffset();

    /**
     * Complete once the producer can accept another record right now (there is send window and the
     * transport is writable). When using {@code send} you don't need this: awaiting the {@code send}
     * future already waits for readiness.
     */
    CompletableFuture<Void> waitReady();

    /**
     * Complete once every record sent so far has been durably acknowledged by Restate, returning the
     * highest acknowledged offset.
     */
    default CompletableFuture<Long> waitAcknowledged() {
        return waitAcknowledged(lastSentOffset());
    }

    /**
     * Complete once all records up to and including {@code offset} have been durably acknowledged by
     * Restate, returning the highest acknowledged offset.
     */
    CompletableFuture<Long> waitAcknowledged(long offset);
}