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.1.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>2024.0.0</axonserver.api.version>
+        <axonserver.api.version>2024.1.0-SNAPSHOT</axonserver.api.version>
 
         <grpc.version>1.59.1</grpc.version>
 
@@ -60,7 +60,7 @@
         <junit.jupiter.version>5.10.2</junit.jupiter.version>
         <mockito.version>4.11.0</mockito.version>
         <testcontainers.version>1.19.7</testcontainers.version>
-        <testng.version>7.10.1</testng.version>
+        <testng.version>6.14.3</testng.version>
 
         <httpclient5.version>5.3.1</httpclient5.version>
         <toxiproxy.version>2.1.7</toxiproxy.version>

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020. AxonIQ
+ * Copyright (c) 2024. AxonIQ
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -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,85 @@ 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
+     * @param bufferSize  the number of events to buffer locally
+     * @param refillBatch the number of events to be consumed prior to refilling the buffer
+     * @param callbacks   the callbacks that are invoked on persistent stream events
+     * @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 bufferSize         the number of events to buffer locally
+     * @param refillBatch        the number of events to be consumed prior to refilling the buffer
+     * @param callbacks          the callbacks that are invoked on persistent stream events
+     * @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();
+
+    /**
+     * Resets the position of a persistent stream to the head (most recent) event in the event store.
+     * @param streamId the unique identification of a persistent stream
+     */
+    CompletableFuture<Void> resetPersistentStreamAtHead(String streamId);
+
+    /**
+     * Resets the position of a persistent stream to the first (oldest) event in the event store.
+     * @param streamId the unique identification of a persistent stream
+     */
+    CompletableFuture<Void> resetPersistentStreamAtTail(String streamId);
+
+    /**
+     * Resets the position of a persistent stream to the event at given position (global index) in the event store.
+     * @param streamId the unique identification of a persistent stream
+     * @param position the global index of the first event to be returned
+     */
+    CompletableFuture<Void> resetPersistentStreamAtPosition(String streamId, long position);
+
+    /**
+     * Resets the position of a persistent stream to the event at or after the given timestamp in the event store.
+     * @param streamId the unique identification of a persistent stream
+     * @param instant the timestamp of the first event to be included in the stream
+     */
+    CompletableFuture<Void> resetPersistentStreamAtTime(String streamId, long instant);
+
 }

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

@@ -0,0 +1,31 @@
+/*
+ * 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.
+ *
+ *  @author Marc Gathier
+ *  @since 2024.0.0
+ */
+public interface PersistentStream {
+
+    /**
+     * Closes the persistent stream.
+     */
+    void close();
+
+}

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

@@ -0,0 +1,89 @@
+/*
+ * 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.
+ *
+ * @author Marc Gathier
+ * @since 2024.0.0
+ */
+public class PersistentStreamCallbacks {
+
+    private final Consumer<PersistentStreamSegment> onSegmentOpened;
+
+    private final Consumer<PersistentStreamSegment> onSegmentClosed;
+    private final Consumer<PersistentStreamSegment> 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, Consumer<PersistentStreamSegment> onSegmentClosed,
+                                     Consumer<PersistentStreamSegment> 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 Consumer<PersistentStreamSegment> 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 Consumer<PersistentStreamSegment> onAvailable() {
+        return onAvailable;
+    }
+}

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

@@ -0,0 +1,109 @@
+/*
+ * 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.
+ *
+ * @author Marc Gathier
+ * @since 2024.0.0
+ */
+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;
+    }
+}