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,11 @@
 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.StreamStatus;
 
 import java.time.Duration;
 import java.time.Instant;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 /**
@@ -387,4 +389,54 @@ 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, int bufferSize, int refillBatch,
+                                          PersistentStreamCallbacks callbacks);
+
+    /**
+     * 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 PersistentStream streaming events per segment
+     */
+    PersistentStream openPersistentStream(String streamId, int bufferSize, int refillBatch,
+                                          PersistentStreamCallbacks callbacks,
+                                          PersistentStreamProperties creationProperties);
+
+
+    /**
+     * Deletes a persistent stream. If the stream does not exist the operation completes successfully.
+     * @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 the name for a persistent stream.
+     * @param streamId the unique identification of a persistent stream
+     * @param streamName the new logical name for the stream
+     * @return a CompletableFuture that completes when the persistent stream is updated
+     */
+    CompletableFuture<Void> updatePersistentStreamName(String streamId, String streamName);
+
+    /**
+     * Updates the number of segments for a persistent stream.
+     * @param streamId the unique identification of a persistent stream
+     * @param segments the requested number of segments for the persistent stream
+     * @return a CompletableFuture that completes when the persistent stream is updated
+     */
+    CompletableFuture<Void> setPersistentStreamSegments(String streamId, int segments);
+
+    /**
+     * Returns a list of persistent streams with the last confirmed token per segment.
+     * @return a list of persistent streams
+     */
+    CompletableFuture<List<StreamStatus>> persistentStreams();
+
 }

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

@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2020-2024. AxonIQ
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.axoniq.axonserver.connector.event;
+
+/**
+ * A connection to a persistent stream. Axon Server can assign zero or more segments to this connection.
+ */
+public interface PersistentStream {
+
+    /**
+     * Closes the persistent stream.
+     */
+    void close();
+
+}

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

@@ -0,0 +1,86 @@
+/*
+ * Copyright (c) 2020-2024. AxonIQ
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.axoniq.axonserver.connector.event;
+
+import java.util.function.Consumer;
+import java.util.function.IntConsumer;
+
+/**
+ * Definitions of the callbacks to be added to a persistent stream.
+ */
+public class PersistentStreamCallbacks {
+
+    private final Consumer<PersistentStreamSegment> onSegmentOpened;
+
+    private final IntConsumer onSegmentClosed;
+    private final IntConsumer onAvailable;
+
+    private final Consumer<Throwable> onClosed;
+
+    /**
+     * Instantiates the callbacks object.
+     *
+     * @param onSegmentOpened callback that the connector invokes when a persistent stream segment is opened
+     * @param onSegmentClosed callback that the connector invokes when a persistent stream segment is closed
+     * @param onAvailable     callback that the connector invokes when an event is available on a persistent stream
+     *                        segment
+     * @param onClosed        callback that the connector invokes when the persistent stream is closed
+     */
+    public PersistentStreamCallbacks(Consumer<PersistentStreamSegment> onSegmentOpened, IntConsumer onSegmentClosed,
+                                     IntConsumer onAvailable,
+                                     Consumer<Throwable> onClosed) {
+        this.onSegmentOpened = onSegmentOpened;
+        this.onSegmentClosed = onSegmentClosed;
+        this.onAvailable = onAvailable;
+        this.onClosed = onClosed;
+    }
+
+    /**
+     * Returns the callback that the connector invokes when the persistent stream is closed.
+     *
+     * @return callback that the connector invokes when the persistent stream is closed
+     */
+    public Consumer<Throwable> onClosed() {
+        return onClosed;
+    }
+
+    /**
+     * Returns the callback that the connector invokes when a persistent stream segment is opened.
+     *
+     * @return callback that the connector invokes when a persistent stream segment is opened
+     */
+    public Consumer<PersistentStreamSegment> onSegmentOpened() {
+        return onSegmentOpened;
+    }
+
+    /**
+     * Returns the callback that the connector invokes when a persistent stream segment is closed.
+     *
+     * @return callback that the connector invokes when a persistent stream segment is closed
+     */
+    public IntConsumer onSegmentClosed() {
+        return onSegmentClosed;
+    }
+
+    /**
+     * Returns the callback that the connector invokes when an event is available on a persistent stream segment.
+     *
+     * @return callback that the connector invokes when an event is available on a persistent stream segment
+     */
+    public IntConsumer onAvailable() {
+        return onAvailable;
+    }
+}

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

@@ -0,0 +1,106 @@
+/*
+ * Copyright (c) 2020-2024. AxonIQ
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.axoniq.axonserver.connector.event;
+
+import io.axoniq.axonserver.connector.impl.AssertUtils;
+
+import java.util.List;
+
+/**
+ * Defines the properties for a new persistent stream.
+ */
+public class PersistentStreamProperties {
+    private final String streamName;
+    private final int segments;
+    private final String sequencingPolicyName;
+    private final List<String> sequencingPolicyParameters;
+    private final long initialPosition;
+    private final String filter;
+
+    /**
+     * Instantiates the persistent stream properties
+     *
+     * @param streamName                 a logical name for the persistent stream
+     * @param segments                   the number of segments, must be larger than 0
+     * @param sequencingPolicyName       the sequencing policy name, must be a name known in Axon Server
+     * @param sequencingPolicyParameters optional parameters for the sequencing policy
+     * @param initialPosition            first token to read
+     * @param filter                     an optional filter to filter events on Axon Server side
+     */
+    public PersistentStreamProperties(String streamName, int segments, String sequencingPolicyName,
+                                      List<String> sequencingPolicyParameters, long initialPosition, String filter) {
+        AssertUtils.assertParameter(segments > 0, "Segments must be > 0");
+        AssertUtils.assertParameter(initialPosition >= 0, "initialPosition must be >= 0");
+        AssertUtils.assertParameter(sequencingPolicyName != null,
+                                    "sequencingPolicyName must not be null");
+
+        this.streamName = streamName;
+        this.segments = segments;
+        this.sequencingPolicyName = sequencingPolicyName;
+        this.sequencingPolicyParameters = sequencingPolicyParameters;
+        this.initialPosition = initialPosition;
+        this.filter = filter;
+    }
+
+    /**
+     * Returns the logical name for the persistent stream.
+     *
+     * @return the logical name for the persistent stream
+     */
+    public String streamName() {
+        return streamName;
+    }
+
+    /**
+     * Returns the number of segments.
+     * @return the number of segments
+     */
+    public int segments() {
+        return segments;
+    }
+
+    /**
+     * Returns the sequencing policy name.
+     * @return the sequencing policy name
+     */
+    public String sequencingPolicyName() {
+        return sequencingPolicyName;
+    }
+
+    /**
+     * Returns the parameters for the sequencing policy.
+     * @return parameters for the sequencing policy
+     */
+    public List<String> sequencingPolicyParameters() {
+        return sequencingPolicyParameters;
+    }
+
+    /**
+     * Returns the first token to read.
+     * @return first token to read
+     */
+    public long initialPosition() {
+        return initialPosition;
+    }
+
+    /**
+     * Returns the filter to filter events on Axon Server side.
+     * @return filter to filter events on Axon Server side
+     */
+    public String filter() {
+        return filter;
+    }
+}

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2020-2024. AxonIQ
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+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();
+}