Persistent streams

pom.xml

@@ -20,7 +20,7 @@
 
     <groupId>io.axoniq</groupId>
     <artifactId>axonserver-connector-java</artifactId>
-    <version>2023.2.1-SNAPSHOT</version>
+    <version>2024.0.0-SNAPSHOT</version>
 
     <name>AxonServer Connector</name>
     <description>
@@ -44,7 +44,7 @@
         <sonar.organization>axoniq</sonar.organization>
         <sonar.host.url>https://sonarcloud.io</sonar.host.url>
 
-        <axonserver.api.version>2023.0.1</axonserver.api.version>
+        <axonserver.api.version>2024.0.0-SNAPSHOT</axonserver.api.version>
 
         <grpc.version>1.59.1</grpc.version>
 

src/main/java/io/axoniq/axonserver/connector/event/EventChannel.java

@@ -20,9 +20,12 @@
 import io.axoniq.axonserver.grpc.InstructionAck;
 import io.axoniq.axonserver.grpc.event.Confirmation;
 import io.axoniq.axonserver.grpc.event.Event;
+import io.axoniq.axonserver.grpc.streams.StreamConnections;
+import io.axoniq.axonserver.grpc.streams.StreamStatus;
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -387,4 +390,51 @@ default ResultStream<EventQueryResultEntry> querySnapshotEvents(String queryExpr
     default ResultStream<EventQueryResultEntry> querySnapshotEvents(String queryExpression, boolean liveStream, String contextName) {
         throw new UnsupportedOperationException();
     }
+
+    /**
+     * Opens the persistent stream identified by {@code streamId}. Fails if the stream does not exist.
+     * @param streamId the unique identification of a persistent stream
+     * @return a PersistedStream streaming events per segment
+     */
+    PersistentStream openPersistentStream(String streamId);
+
+    /**
+     * Opens the persistent stream identified by {@code streamId}. If the stream does not exist it will be
+     * created with the properties specified in {@code creationProperties}.
+     * @param streamId the unique identification of a persistent stream
+     * @param creationProperties properties to initialize the persistent stream if it does not exist yet
+     * @return a PersistedStream streaming events per segment
+     */
+    PersistentStream openPersistentStream(String streamId, PersistedStreamProperties creationProperties);
+
+
+    /**
+     * Deletes a persistent stream.
+     * @param streamId the unique identification of a persistent stream
+     * @return a CompletableFuture that completes when the persistent stream is deleted
+     */
+    CompletableFuture<Void> deletePersistentStream(String streamId);
+
+    /**
+     * Updates properties for a persistent stream. If {@code segments} parameter is set to a positive value, it changes
+     * the number of segments for the persistent stream.
+     * @param streamId the unique identification of a persistent stream
+     * @param segments the requested number of segments for the persistent stream
+     * @param streamName the new logical name for the stream
+     * @return a CompletableFuture that completes when the persistent stream is updated
+     */
+    CompletableFuture<Void> updatePersistentStream(String streamId, Integer segments, String streamName);
+
+    /**
+     * Returns a list of persistent streams with the last confirmed token per segment.
+     * @return a list of persistent streams
+     */
+    CompletableFuture<List<StreamStatus>> persistentStreams();
+
+    /**
+     * Returns a list of persistent streams with the connected client per segment.
+     * @return a list of persistent streams
+     */
+    CompletableFuture<List<StreamConnections>> persistentStreamConnections();
+
 }

src/main/java/io/axoniq/axonserver/connector/event/PersistedStreamProperties.java

@@ -0,0 +1,46 @@
+package io.axoniq.axonserver.connector.event;
+
+import java.util.List;
+
+public class PersistedStreamProperties {
+    private final String streamName;
+    private final int segments;
+    private final String sequencingPolicyName;
+    private final List<String> sequencingPolicyParameters;
+    private final int initialPosition;
+    private final String filter;
+
+    public PersistedStreamProperties(String streamName, int segments, String sequencingPolicyName,
+                                     List<String> sequencingPolicyParameters, int initialPosition, String filter) {
+        this.streamName = streamName;
+        this.segments = segments;
+        this.sequencingPolicyName = sequencingPolicyName;
+        this.sequencingPolicyParameters = sequencingPolicyParameters;
+        this.initialPosition = initialPosition;
+        this.filter = filter;
+    }
+
+    public String streamName() {
+        return streamName;
+    }
+
+    public int segments() {
+        return segments;
+    }
+
+    public String sequencingPolicyName() {
+        return sequencingPolicyName;
+    }
+
+    public List<String> sequencingPolicyParameters() {
+        return sequencingPolicyParameters;
+    }
+
+    public int initialPosition() {
+        return initialPosition;
+    }
+
+    public String filter() {
+        return filter;
+    }
+}

src/main/java/io/axoniq/axonserver/connector/event/PersistentStream.java

@@ -0,0 +1,26 @@
+package io.axoniq.axonserver.connector.event;
+
+import java.util.function.Consumer;
+
+/**
+ * A connection to a persistent stream. Axon Server can assign zero or more segments to this connection.
+ */
+public interface PersistentStream {
+
+    /**
+     * Registers a callback to invoke when Axon Server assigns a segment to this connection.
+     * @param callback the callback to invoke when a segment is opened
+     */
+    void onSegmentOpened(Consumer<PersistentStreamSegment> callback);
+
+    /**
+     * Closes the persistent stream.
+     */
+    void close();
+
+    /**
+     * Registers a callback to invoke when Axon Server closes this connection (or the connection to Axon Server is lost).
+     * @param closedCallback the callback to invoke when the persistent stream is closed
+     */
+    void onClosed(Consumer<Throwable> closedCallback);
+}

src/main/java/io/axoniq/axonserver/connector/event/PersistentStreamSegment.java

@@ -0,0 +1,31 @@
+package io.axoniq.axonserver.connector.event;
+
+import io.axoniq.axonserver.connector.ResultStream;
+import io.axoniq.axonserver.grpc.event.EventWithToken;
+
+/**
+ * An event stream producing events for one segment of a persistent stream.
+ */
+public interface PersistentStreamSegment extends ResultStream<EventWithToken> {
+
+    /**
+     * Registers a callback that will be invoked when Axon Server closes the segment within the persistent
+     * stream connection. This happens when the number of segments in the persistent stream has changed or when
+     * Axon Server assigns a segment to another client.
+     * @param callback the callback to register
+     */
+    void onSegmentClosed(Runnable callback);
+
+    /**
+     * Sends the last processed token for this segment to Axon Server. Clients may choose to notify each processed event,
+     * or to only sent progress information after a number of processed events.
+     * @param token the last processed token for this segment
+     */
+    void acknowledge(long token);
+
+    /**
+     * Returns the segment number of the stream.
+     * @return the segment number of this stream
+     */
+    int segment();
+}

src/main/java/io/axoniq/axonserver/connector/event/impl/BufferedPersistentStreamSegment.java

@@ -0,0 +1,128 @@
+package io.axoniq.axonserver.connector.event.impl;
+
+import io.axoniq.axonserver.connector.event.PersistentStreamSegment;
+import io.axoniq.axonserver.connector.impl.StreamClosedException;
+import io.axoniq.axonserver.grpc.event.EventWithToken;
+
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.LongConsumer;
+
+public class BufferedPersistentStreamSegment
+        implements PersistentStreamSegment {
+
+    private static final Runnable NO_OP = () -> {
+    };
+
+    private final Set<Runnable> onSegmentClosedCallbacks = new CopyOnWriteArraySet<>();
+    private final BlockingQueue<EventWithToken> buffer = new LinkedBlockingQueue<>();
+    private final AtomicReference<Throwable> errorResult = new AtomicReference<>();
+    private final AtomicBoolean closed = new AtomicBoolean();
+
+    private final AtomicReference<Runnable> onAvailableCallback = new AtomicReference<>(NO_OP);
+
+    private final int segment;
+    private final LongConsumer progressCallback;
+
+    public BufferedPersistentStreamSegment(int segment, LongConsumer progressCallback) {
+        this.segment = segment;
+        this.progressCallback = progressCallback;
+    }
+
+    @Override
+    public EventWithToken peek() {
+        checkClosed();
+        return buffer.peek();
+    }
+
+    @Override
+    public EventWithToken nextIfAvailable() {
+        checkClosed();
+        return buffer.poll();
+    }
+
+    @Override
+    public EventWithToken nextIfAvailable(long timeout, TimeUnit unit) throws InterruptedException {
+        checkClosed();
+        return buffer.poll(timeout, unit);
+    }
+
+    @Override
+    public EventWithToken next() throws InterruptedException {
+        checkClosed();
+        return buffer.take();
+    }
+
+    @Override
+    public void onAvailable(Runnable callback) {
+        checkClosed();
+        if (callback == null) {
+            onAvailableCallback.set(NO_OP);
+        } else {
+            onAvailableCallback.set(callback);
+            if (isClosed() || peek() != null) {
+                callback.run();
+            }
+        }
+    }
+
+    private void checkClosed() {
+        if (closed.get()) {
+            throw new StreamClosedException(errorResult.get());
+        }
+    }
+
+    @Override
+    public void close() {
+        // No-op
+    }
+
+    @Override
+    public boolean isClosed() {
+        return closed.get();
+    }
+
+    @Override
+    public Optional<Throwable> getError() {
+        return Optional.ofNullable(errorResult.get());
+    }
+
+    @Override
+    public void onSegmentClosed(Runnable callback) {
+        onSegmentClosedCallbacks.add(callback);
+    }
+
+    public void onNext(EventWithToken streamSignal) {
+        buffer.add(streamSignal);
+        onAvailableCallback.get().run();
+    }
+
+    public void onError(Throwable throwable) {
+        errorResult.set(throwable);
+        onCompleted();
+    }
+
+    public void onCompleted() {
+        closed.set(true);
+        onSegmentClosedCallbacks.forEach(Runnable::run);
+        onAvailableCallback.get().run();
+    }
+
+    @Override
+    public void acknowledge(long token) {
+        if (!closed.get()) {
+            progressCallback.accept(token);
+        }
+    }
+
+    @Override
+    public int segment() {
+        return segment;
+    }
+}