add DCB event scheduler

pom.xml

@@ -44,7 +44,7 @@
         <sonar.organization>axoniq</sonar.organization>
         <sonar.host.url>https://sonarcloud.io</sonar.host.url>
 
-        <axonserver.api.version>2025.1.1</axonserver.api.version>
+        <axonserver.api.version>2025.1.2-SNAPSHOT</axonserver.api.version>
 
         <grpc.version>1.69.0</grpc.version>
         <protobuf.version>4.31.1</protobuf.version>

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

@@ -17,9 +17,11 @@
 package io.axoniq.axonserver.connector.event;
 
 import io.axoniq.axonserver.connector.ResultStream;
+import io.axoniq.axonserver.grpc.InstructionAck;
 import io.axoniq.axonserver.grpc.event.dcb.AddTagsResponse;
 import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse;
 import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition;
+import io.axoniq.axonserver.grpc.event.dcb.Event;
 import io.axoniq.axonserver.grpc.event.dcb.GetHeadResponse;
 import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtResponse;
 import io.axoniq.axonserver.grpc.event.dcb.GetTagsResponse;
@@ -32,9 +34,11 @@
 import io.axoniq.axonserver.grpc.event.dcb.Tag;
 import io.axoniq.axonserver.grpc.event.dcb.TaggedEvent;
 
+import java.time.Duration;
 import java.time.Instant;
 import java.util.Collection;
 import java.util.concurrent.CompletableFuture;
+import javax.annotation.Nullable;
 
 /**
  * Communication channel for Event related interaction with Axon Server based on the Dynamic Consistency Boundaries
@@ -181,6 +185,63 @@ default ResultStream<StreamEventsResponse> stream(StreamEventsRequest request, i
      */
     CompletableFuture<GetSequenceAtResponse> getSequenceAt(Instant timestamp);
 
+
+    /**
+     * Schedule the given {@code event} to be published after given {@code triggerDuration}. The returned value can be
+     * used to cancel the schedule, or to reschedule the event to another time.
+     *
+     * @param triggerDuration the amount of time to wait to publish the event
+     * @param event           the event to publish
+     * @return a token used to cancel the schedule
+     */
+    default CompletableFuture<String> scheduleEvent(Duration triggerDuration, Event event) {
+        return scheduleEvent(Instant.now().plus(triggerDuration), event);
+    }
+
+    /**
+     * Schedule the given {@code event} to be published at given {@code scheduleTime}. The returned value can be used to
+     * cancel the schedule, or to reschedule the event to another time.
+     *
+     * @param scheduleTime The scheduleTime at which to publish the event
+     * @param event        The event to publish
+     * @return a token used to cancel the schedule
+     */
+    CompletableFuture<String> scheduleEvent(Instant scheduleTime, Event event);
+
+    /**
+     * Cancels the scheduled publication of an event for which the given {@code scheduleToken} was returned.
+     *
+     * @param scheduleToken the token provided when scheduling the event to be cancelled
+     * @return a future reference to the result of the instruction
+     */
+    CompletableFuture<InstructionAck> cancelSchedule(String scheduleToken);
+
+    /**
+     * Reschedules the scheduled event with the given {@code scheduleToken} and reschedule it
+     * to be published after given {@code triggerDuration}. If no event is provided, the original event will be
+     * rescheduled.
+     *
+     * @param scheduleToken   the token of the event to cancel
+     * @param triggerDuration the point amount of time to wait to publish the event
+     * @param event           an optional new event to publish
+     * @return a future reference to the token for the new schedule
+     */
+    default CompletableFuture<String> reschedule(String scheduleToken, Duration triggerDuration, Event event) {
+        return reschedule(scheduleToken, Instant.now().plus(triggerDuration), event);
+    }
+
+    /**
+     * Reschedules the scheduled event with the given {@code scheduleToken} and reschedule it
+     * to be published at given {@code scheduleTime}. If no event is provided, the original event will be
+     * rescheduled.
+     *
+     * @param scheduleToken the token of the event to cancel
+     * @param scheduleTime  the point in time to publish the event
+     * @param event         an optional new event to publish
+     * @return a future reference to the token for the new schedule
+     */
+    CompletableFuture<String> reschedule(String scheduleToken, Instant scheduleTime, @Nullable Event event);
+
     /**
      * Provides operations to interact with a Transaction to append events and a condition onto the Event Store.
      */

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

@@ -17,20 +17,26 @@
 package io.axoniq.axonserver.connector.event.impl;
 
 import com.google.protobuf.Empty;
+import io.axoniq.axonserver.connector.AxonServerException;
+import io.axoniq.axonserver.connector.ErrorCategory;
 import io.axoniq.axonserver.connector.ResultStream;
 import io.axoniq.axonserver.connector.event.DcbEventChannel;
 import io.axoniq.axonserver.connector.impl.AbstractAxonServerChannel;
 import io.axoniq.axonserver.connector.impl.AbstractBufferedStream;
 import io.axoniq.axonserver.connector.impl.AxonServerManagedChannel;
 import io.axoniq.axonserver.connector.impl.FutureStreamObserver;
 import io.axoniq.axonserver.grpc.FlowControl;
+import io.axoniq.axonserver.grpc.InstructionAck;
 import io.axoniq.axonserver.grpc.control.ClientIdentification;
 import io.axoniq.axonserver.grpc.event.dcb.AddTagsRequest;
 import io.axoniq.axonserver.grpc.event.dcb.AddTagsResponse;
 import io.axoniq.axonserver.grpc.event.dcb.AppendEventsRequest;
 import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse;
+import io.axoniq.axonserver.grpc.event.dcb.CancelScheduledEventRequest;
 import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition;
+import io.axoniq.axonserver.grpc.event.dcb.DcbEventSchedulerGrpc;
 import io.axoniq.axonserver.grpc.event.dcb.DcbEventStoreGrpc;
+import io.axoniq.axonserver.grpc.event.dcb.Event;
 import io.axoniq.axonserver.grpc.event.dcb.GetHeadRequest;
 import io.axoniq.axonserver.grpc.event.dcb.GetHeadResponse;
 import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtRequest;
@@ -41,6 +47,9 @@
 import io.axoniq.axonserver.grpc.event.dcb.GetTailResponse;
 import io.axoniq.axonserver.grpc.event.dcb.RemoveTagsRequest;
 import io.axoniq.axonserver.grpc.event.dcb.RemoveTagsResponse;
+import io.axoniq.axonserver.grpc.event.dcb.RescheduleEventRequest;
+import io.axoniq.axonserver.grpc.event.dcb.ScheduleEventRequest;
+import io.axoniq.axonserver.grpc.event.dcb.ScheduleToken;
 import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest;
 import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse;
 import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest;
@@ -50,8 +59,6 @@
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
 import io.grpc.stub.StreamObserver;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 import java.time.Instant;
 import java.util.Collection;
@@ -60,6 +67,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
+import javax.annotation.Nullable;
 
 /**
  * {@link DcbEventChannel} implementation, serving as the event connection between Axon Server and a client
@@ -70,11 +78,10 @@
  */
 public class DcbEventChannelImpl extends AbstractAxonServerChannel<Void> implements DcbEventChannel {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(DcbEventChannelImpl.class);
-
     private static final int BUFFER_SIZE = 512;
     private static final int REFILL_BATCH = 16;
     private final DcbEventStoreGrpc.DcbEventStoreStub eventStore;
+    private final DcbEventSchedulerGrpc.DcbEventSchedulerStub eventScheduler;
     private final ClientIdentification clientIdentification;
     private final Set<ResultStream<StreamEventsResponse>> buffers = ConcurrentHashMap.newKeySet();
 
@@ -90,6 +97,7 @@ public DcbEventChannelImpl(ClientIdentification clientIdentification,
                                AxonServerManagedChannel axonServerManagedChannel) {
         super(clientIdentification, executor, axonServerManagedChannel);
         this.eventStore = DcbEventStoreGrpc.newStub(axonServerManagedChannel);
+        this.eventScheduler = DcbEventSchedulerGrpc.newStub(axonServerManagedChannel);
         this.clientIdentification = clientIdentification;
     }
 
@@ -236,6 +244,53 @@ public CompletableFuture<RemoveTagsResponse> removeTags(long sequence, Collectio
         return future;
     }
 
+    @Override
+    public CompletableFuture<String> scheduleEvent(Instant scheduleTime, Event event) {
+        FutureStreamObserver<ScheduleToken>responseObserver = new FutureStreamObserver<>(new AxonServerException(
+                ErrorCategory.INSTRUCTION_ACK_ERROR,
+                "An unknown error occurred while scheduling an Event. No response received from Server.",
+                ""
+        ));
+
+        eventScheduler.scheduleEvent(ScheduleEventRequest.newBuilder()
+                                                                 .setEvent(event)
+                                                                 .setInstant(scheduleTime.toEpochMilli())
+                                                                 .build(), responseObserver);
+        return responseObserver.thenApply(ScheduleToken::getToken);
+    }
+
+    @Override
+    public CompletableFuture<InstructionAck> cancelSchedule(String token) {
+        FutureStreamObserver<InstructionAck> responseObserver = new FutureStreamObserver<>(new AxonServerException(
+                ErrorCategory.INSTRUCTION_ACK_ERROR,
+                "An unknown error occurred while cancelling a scheduled Event. No response received from Server.",
+                ""
+        ));
+
+        eventScheduler.cancelScheduledEvent(CancelScheduledEventRequest.newBuilder().setToken(token).build(),
+                                            responseObserver);
+        return responseObserver;
+    }
+
+    @Override
+    public CompletableFuture<String> reschedule(String scheduleToken, Instant scheduleTime, @Nullable Event event) {
+        FutureStreamObserver<ScheduleToken>responseObserver = new FutureStreamObserver<>(new AxonServerException(
+                ErrorCategory.INSTRUCTION_ACK_ERROR,
+                "An unknown error occurred while rescheduling Event. No response received from Server.",
+                ""
+        ));
+
+        RescheduleEventRequest.Builder builder = RescheduleEventRequest.newBuilder()
+                                                                       .setToken(scheduleToken)
+                                                                       .setInstant(scheduleTime.toEpochMilli());
+        if (event != null) {
+            builder.setEvent(event);
+        }
+        eventScheduler.rescheduleEvent(builder.build(), responseObserver);
+        return responseObserver.thenApply(ScheduleToken::getToken);
+    }
+
+
     private static class AppendEventsTransactionImpl implements AppendEventsTransaction {
 
         private final StreamObserver<AppendEventsRequest> stream;

src/test/java/io/axoniq/axonserver/connector/AbstractAxonServerIntegrationTest.java

@@ -85,7 +85,7 @@ private static String getDockerImageName() {
     protected static Proxy axonServerProxy;
 
     protected static ServerAddress axonServerAddress;
-    private static ServerAddress axonServerHttpPort;
+    protected static ServerAddress axonServerHttpPort;
     protected String axonServerVersion;
 
     @BeforeAll

src/test/java/io/axoniq/axonserver/connector/event/dcb/DcbEndToEndTest.java

@@ -7,21 +7,15 @@
 import io.axoniq.axonserver.connector.AxonServerConnectionFactory;
 import io.axoniq.axonserver.connector.ResultStream;
 import io.axoniq.axonserver.connector.ResultStreamPublisher;
-import io.axoniq.axonserver.connector.impl.ServerAddress;
 import io.axoniq.axonserver.connector.event.DcbEventChannel;
+import io.axoniq.axonserver.connector.impl.ServerAddress;
 import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse;
 import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition;
 import io.axoniq.axonserver.grpc.event.dcb.Criterion;
 import io.axoniq.axonserver.grpc.event.dcb.Event;
-import io.axoniq.axonserver.grpc.event.dcb.GetHeadRequest;
-import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtRequest;
 import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtResponse;
-import io.axoniq.axonserver.grpc.event.dcb.GetTagsRequest;
 import io.axoniq.axonserver.grpc.event.dcb.GetTagsResponse;
-import io.axoniq.axonserver.grpc.event.dcb.GetTailRequest;
 import io.axoniq.axonserver.grpc.event.dcb.GetTailResponse;
-import io.axoniq.axonserver.grpc.event.dcb.AddTagsResponse;
-import io.axoniq.axonserver.grpc.event.dcb.RemoveTagsResponse;
 import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest;
 import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse;
 import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest;
@@ -36,11 +30,9 @@
 import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
-import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.List;
 import java.util.Set;
 import java.util.UUID;
@@ -71,11 +63,6 @@ class DcbEndToEndTest extends AbstractAxonServerIntegrationTest {
      */
     private static final boolean LOCAL = false;
 
-    /**
-     * HTTP port for local Axon Server instance
-     */
-    private static ServerAddress axonServerHttpPort;
-
     /**
      * Static initialization for local mode
      */
@@ -917,6 +904,61 @@ void getSequenceAtBetweenEvents() {
         }
     }
 
+    @Test
+    @Disabled("Enable when scheduling is supported in AxonServer")
+    void scheduleEvent() {
+        DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
+        long head = dcbEventChannel.head().join().getSequence();
+        Event event = anEvent(aString(), "scheduledEvent");
+        dcbEventChannel.scheduleEvent(Duration.ofSeconds(5), event)
+                       .join();
+        StepVerifier.create(streamFlux(head).take(1))
+                    .expectNextMatches(r -> r.getEvent().getEvent().getName().equals("scheduledEvent"))
+                    .verifyComplete();
+    }
+
+    @Test
+    @Disabled("Enable when scheduling is supported in AxonServer")
+    void cancelScheduledEvent() {
+        DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
+        long head = dcbEventChannel.head().join().getSequence();
+        Event event = anEvent(aString(), "scheduledEvent");
+        String token = dcbEventChannel.scheduleEvent(Duration.ofSeconds(1), event)
+                       .join();
+        dcbEventChannel.cancelSchedule(token).join();
+        StepVerifier.create(streamFlux(head).take(Duration.ofSeconds(2)))
+                    .verifyComplete();
+    }
+
+    @Test
+    @Disabled("Enable when scheduling is supported in AxonServer")
+    void rescheduleEvent() {
+        DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
+        long head = dcbEventChannel.head().join().getSequence();
+        Event event = anEvent(aString(), "rescheduledEvent");
+        String token = dcbEventChannel.scheduleEvent(Duration.ofSeconds(30), event)
+                       .join();
+        dcbEventChannel.reschedule(token, Duration.ofSeconds(2), null).join();
+        StepVerifier.create(streamFlux(head).take(Duration.ofSeconds(3)))
+                    .expectNextMatches(r -> r.getEvent().getEvent().getName().equals("rescheduledEvent"))
+                    .verifyComplete();
+    }
+
+    @Test
+    @Disabled("Enable when scheduling is supported in AxonServer")
+    void rescheduleAndReplaceEvent() {
+        DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
+        long head = dcbEventChannel.head().join().getSequence();
+        Event event = anEvent(aString(), "scheduledEvent");
+        String token = dcbEventChannel.scheduleEvent(Duration.ofSeconds(5), event)
+                                      .join();
+        Event replaceEvent = anEvent(aString(), "rescheduledEvent");
+        dcbEventChannel.reschedule(token, Duration.ofSeconds(2), replaceEvent).join();
+        StepVerifier.create(streamFlux(head).take(Duration.ofSeconds(6)))
+                    .expectNextMatches(r -> r.getEvent().getEvent().getName().equals("rescheduledEvent"))
+                    .verifyComplete();
+    }
+
     private Flux<SourceEventsResponse> sourceFlux(long start) {
         DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
         return fluxStream(() -> dcbEventChannel.source(sourceEventsRequest(start)));

src/test/java/io/axoniq/axonserver/connector/testutils/AxonServerUtils.java

@@ -47,6 +47,7 @@ public static void purgeEventsFromAxonServer(String context, String hostname, in
             throws IOException {
         deleteContext(context, hostname, port);
         createContext(context, hostname, port, dcbContext);
+        waitFor(1000);
     }
 
     public static void deleteContext(String context, String hostname, int port) throws IOException {
@@ -182,4 +183,12 @@ private static void checkContextsCondition(String hostname, int port,
             throw new RuntimeException(e);
         }
     }
+
+    private static void waitFor(long millis) {
+        try {
+            Thread.sleep(millis);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
 }