Skip to content

Commit 442848f

Browse files
MGathierMGathier
authored
Merge pull request #441 from AxonIQ/feature/dcb-event-scheduler
add DCB event scheduler
2 parents 9bf1c08 + 2fddea9 commit 442848f

6 files changed

Lines changed: 187 additions & 20 deletions

File tree

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@
4444
<sonar.organization>axoniq</sonar.organization>
4545
<sonar.host.url>https://sonarcloud.io</sonar.host.url>
4646

47-
<axonserver.api.version>2025.1.1</axonserver.api.version>
47+
<axonserver.api.version>2025.1.2-SNAPSHOT</axonserver.api.version>
4848

4949
<grpc.version>1.69.0</grpc.version>
5050
<protobuf.version>4.31.1</protobuf.version>

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

Lines changed: 61 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,11 @@
1717
package io.axoniq.axonserver.connector.event;
1818

1919
import io.axoniq.axonserver.connector.ResultStream;
20+
import io.axoniq.axonserver.grpc.InstructionAck;
2021
import io.axoniq.axonserver.grpc.event.dcb.AddTagsResponse;
2122
import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse;
2223
import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition;
24+
import io.axoniq.axonserver.grpc.event.dcb.Event;
2325
import io.axoniq.axonserver.grpc.event.dcb.GetHeadResponse;
2426
import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtResponse;
2527
import io.axoniq.axonserver.grpc.event.dcb.GetTagsResponse;
@@ -32,9 +34,11 @@
3234
import io.axoniq.axonserver.grpc.event.dcb.Tag;
3335
import io.axoniq.axonserver.grpc.event.dcb.TaggedEvent;
3436

37+
import java.time.Duration;
3538
import java.time.Instant;
3639
import java.util.Collection;
3740
import java.util.concurrent.CompletableFuture;
41+
import javax.annotation.Nullable;
3842

3943
/**
4044
* 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
181185
*/
182186
CompletableFuture<GetSequenceAtResponse> getSequenceAt(Instant timestamp);
183187

188+
189+
/**
190+
* Schedule the given {@code event} to be published after given {@code triggerDuration}. The returned value can be
191+
* used to cancel the schedule, or to reschedule the event to another time.
192+
*
193+
* @param triggerDuration the amount of time to wait to publish the event
194+
* @param event the event to publish
195+
* @return a token used to cancel the schedule
196+
*/
197+
default CompletableFuture<String> scheduleEvent(Duration triggerDuration, Event event) {
198+
return scheduleEvent(Instant.now().plus(triggerDuration), event);
199+
}
200+
201+
/**
202+
* Schedule the given {@code event} to be published at given {@code scheduleTime}. The returned value can be used to
203+
* cancel the schedule, or to reschedule the event to another time.
204+
*
205+
* @param scheduleTime The scheduleTime at which to publish the event
206+
* @param event The event to publish
207+
* @return a token used to cancel the schedule
208+
*/
209+
CompletableFuture<String> scheduleEvent(Instant scheduleTime, Event event);
210+
211+
/**
212+
* Cancels the scheduled publication of an event for which the given {@code scheduleToken} was returned.
213+
*
214+
* @param scheduleToken the token provided when scheduling the event to be cancelled
215+
* @return a future reference to the result of the instruction
216+
*/
217+
CompletableFuture<InstructionAck> cancelSchedule(String scheduleToken);
218+
219+
/**
220+
* Reschedules the scheduled event with the given {@code scheduleToken} and reschedule it
221+
* to be published after given {@code triggerDuration}. If no event is provided, the original event will be
222+
* rescheduled.
223+
*
224+
* @param scheduleToken the token of the event to cancel
225+
* @param triggerDuration the point amount of time to wait to publish the event
226+
* @param event an optional new event to publish
227+
* @return a future reference to the token for the new schedule
228+
*/
229+
default CompletableFuture<String> reschedule(String scheduleToken, Duration triggerDuration, Event event) {
230+
return reschedule(scheduleToken, Instant.now().plus(triggerDuration), event);
231+
}
232+
233+
/**
234+
* Reschedules the scheduled event with the given {@code scheduleToken} and reschedule it
235+
* to be published at given {@code scheduleTime}. If no event is provided, the original event will be
236+
* rescheduled.
237+
*
238+
* @param scheduleToken the token of the event to cancel
239+
* @param scheduleTime the point in time to publish the event
240+
* @param event an optional new event to publish
241+
* @return a future reference to the token for the new schedule
242+
*/
243+
CompletableFuture<String> reschedule(String scheduleToken, Instant scheduleTime, @Nullable Event event);
244+
184245
/**
185246
* Provides operations to interact with a Transaction to append events and a condition onto the Event Store.
186247
*/

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

Lines changed: 59 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,20 +17,26 @@
1717
package io.axoniq.axonserver.connector.event.impl;
1818

1919
import com.google.protobuf.Empty;
20+
import io.axoniq.axonserver.connector.AxonServerException;
21+
import io.axoniq.axonserver.connector.ErrorCategory;
2022
import io.axoniq.axonserver.connector.ResultStream;
2123
import io.axoniq.axonserver.connector.event.DcbEventChannel;
2224
import io.axoniq.axonserver.connector.impl.AbstractAxonServerChannel;
2325
import io.axoniq.axonserver.connector.impl.AbstractBufferedStream;
2426
import io.axoniq.axonserver.connector.impl.AxonServerManagedChannel;
2527
import io.axoniq.axonserver.connector.impl.FutureStreamObserver;
2628
import io.axoniq.axonserver.grpc.FlowControl;
29+
import io.axoniq.axonserver.grpc.InstructionAck;
2730
import io.axoniq.axonserver.grpc.control.ClientIdentification;
2831
import io.axoniq.axonserver.grpc.event.dcb.AddTagsRequest;
2932
import io.axoniq.axonserver.grpc.event.dcb.AddTagsResponse;
3033
import io.axoniq.axonserver.grpc.event.dcb.AppendEventsRequest;
3134
import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse;
35+
import io.axoniq.axonserver.grpc.event.dcb.CancelScheduledEventRequest;
3236
import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition;
37+
import io.axoniq.axonserver.grpc.event.dcb.DcbEventSchedulerGrpc;
3338
import io.axoniq.axonserver.grpc.event.dcb.DcbEventStoreGrpc;
39+
import io.axoniq.axonserver.grpc.event.dcb.Event;
3440
import io.axoniq.axonserver.grpc.event.dcb.GetHeadRequest;
3541
import io.axoniq.axonserver.grpc.event.dcb.GetHeadResponse;
3642
import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtRequest;
@@ -41,6 +47,9 @@
4147
import io.axoniq.axonserver.grpc.event.dcb.GetTailResponse;
4248
import io.axoniq.axonserver.grpc.event.dcb.RemoveTagsRequest;
4349
import io.axoniq.axonserver.grpc.event.dcb.RemoveTagsResponse;
50+
import io.axoniq.axonserver.grpc.event.dcb.RescheduleEventRequest;
51+
import io.axoniq.axonserver.grpc.event.dcb.ScheduleEventRequest;
52+
import io.axoniq.axonserver.grpc.event.dcb.ScheduleToken;
4453
import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest;
4554
import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse;
4655
import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest;
@@ -50,8 +59,6 @@
5059
import io.grpc.Status;
5160
import io.grpc.StatusRuntimeException;
5261
import io.grpc.stub.StreamObserver;
53-
import org.slf4j.Logger;
54-
import org.slf4j.LoggerFactory;
5562

5663
import java.time.Instant;
5764
import java.util.Collection;
@@ -60,6 +67,7 @@
6067
import java.util.concurrent.ConcurrentHashMap;
6168
import java.util.concurrent.ScheduledExecutorService;
6269
import java.util.concurrent.atomic.AtomicBoolean;
70+
import javax.annotation.Nullable;
6371

6472
/**
6573
* {@link DcbEventChannel} implementation, serving as the event connection between Axon Server and a client
@@ -70,11 +78,10 @@
7078
*/
7179
public class DcbEventChannelImpl extends AbstractAxonServerChannel<Void> implements DcbEventChannel {
7280

73-
private static final Logger LOGGER = LoggerFactory.getLogger(DcbEventChannelImpl.class);
74-
7581
private static final int BUFFER_SIZE = 512;
7682
private static final int REFILL_BATCH = 16;
7783
private final DcbEventStoreGrpc.DcbEventStoreStub eventStore;
84+
private final DcbEventSchedulerGrpc.DcbEventSchedulerStub eventScheduler;
7885
private final ClientIdentification clientIdentification;
7986
private final Set<ResultStream<StreamEventsResponse>> buffers = ConcurrentHashMap.newKeySet();
8087

@@ -90,6 +97,7 @@ public DcbEventChannelImpl(ClientIdentification clientIdentification,
9097
AxonServerManagedChannel axonServerManagedChannel) {
9198
super(clientIdentification, executor, axonServerManagedChannel);
9299
this.eventStore = DcbEventStoreGrpc.newStub(axonServerManagedChannel);
100+
this.eventScheduler = DcbEventSchedulerGrpc.newStub(axonServerManagedChannel);
93101
this.clientIdentification = clientIdentification;
94102
}
95103

@@ -236,6 +244,53 @@ public CompletableFuture<RemoveTagsResponse> removeTags(long sequence, Collectio
236244
return future;
237245
}
238246

247+
@Override
248+
public CompletableFuture<String> scheduleEvent(Instant scheduleTime, Event event) {
249+
FutureStreamObserver<ScheduleToken>responseObserver = new FutureStreamObserver<>(new AxonServerException(
250+
ErrorCategory.INSTRUCTION_ACK_ERROR,
251+
"An unknown error occurred while scheduling an Event. No response received from Server.",
252+
""
253+
));
254+
255+
eventScheduler.scheduleEvent(ScheduleEventRequest.newBuilder()
256+
.setEvent(event)
257+
.setInstant(scheduleTime.toEpochMilli())
258+
.build(), responseObserver);
259+
return responseObserver.thenApply(ScheduleToken::getToken);
260+
}
261+
262+
@Override
263+
public CompletableFuture<InstructionAck> cancelSchedule(String token) {
264+
FutureStreamObserver<InstructionAck> responseObserver = new FutureStreamObserver<>(new AxonServerException(
265+
ErrorCategory.INSTRUCTION_ACK_ERROR,
266+
"An unknown error occurred while cancelling a scheduled Event. No response received from Server.",
267+
""
268+
));
269+
270+
eventScheduler.cancelScheduledEvent(CancelScheduledEventRequest.newBuilder().setToken(token).build(),
271+
responseObserver);
272+
return responseObserver;
273+
}
274+
275+
@Override
276+
public CompletableFuture<String> reschedule(String scheduleToken, Instant scheduleTime, @Nullable Event event) {
277+
FutureStreamObserver<ScheduleToken>responseObserver = new FutureStreamObserver<>(new AxonServerException(
278+
ErrorCategory.INSTRUCTION_ACK_ERROR,
279+
"An unknown error occurred while rescheduling Event. No response received from Server.",
280+
""
281+
));
282+
283+
RescheduleEventRequest.Builder builder = RescheduleEventRequest.newBuilder()
284+
.setToken(scheduleToken)
285+
.setInstant(scheduleTime.toEpochMilli());
286+
if (event != null) {
287+
builder.setEvent(event);
288+
}
289+
eventScheduler.rescheduleEvent(builder.build(), responseObserver);
290+
return responseObserver.thenApply(ScheduleToken::getToken);
291+
}
292+
293+
239294
private static class AppendEventsTransactionImpl implements AppendEventsTransaction {
240295

241296
private final StreamObserver<AppendEventsRequest> stream;

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ private static String getDockerImageName() {
8585
protected static Proxy axonServerProxy;
8686

8787
protected static ServerAddress axonServerAddress;
88-
private static ServerAddress axonServerHttpPort;
88+
protected static ServerAddress axonServerHttpPort;
8989
protected String axonServerVersion;
9090

9191
@BeforeAll

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

Lines changed: 56 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -7,21 +7,15 @@
77
import io.axoniq.axonserver.connector.AxonServerConnectionFactory;
88
import io.axoniq.axonserver.connector.ResultStream;
99
import io.axoniq.axonserver.connector.ResultStreamPublisher;
10-
import io.axoniq.axonserver.connector.impl.ServerAddress;
1110
import io.axoniq.axonserver.connector.event.DcbEventChannel;
11+
import io.axoniq.axonserver.connector.impl.ServerAddress;
1212
import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse;
1313
import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition;
1414
import io.axoniq.axonserver.grpc.event.dcb.Criterion;
1515
import io.axoniq.axonserver.grpc.event.dcb.Event;
16-
import io.axoniq.axonserver.grpc.event.dcb.GetHeadRequest;
17-
import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtRequest;
1816
import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtResponse;
19-
import io.axoniq.axonserver.grpc.event.dcb.GetTagsRequest;
2017
import io.axoniq.axonserver.grpc.event.dcb.GetTagsResponse;
21-
import io.axoniq.axonserver.grpc.event.dcb.GetTailRequest;
2218
import io.axoniq.axonserver.grpc.event.dcb.GetTailResponse;
23-
import io.axoniq.axonserver.grpc.event.dcb.AddTagsResponse;
24-
import io.axoniq.axonserver.grpc.event.dcb.RemoveTagsResponse;
2519
import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest;
2620
import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse;
2721
import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest;
@@ -36,11 +30,9 @@
3630
import reactor.core.publisher.Flux;
3731
import reactor.test.StepVerifier;
3832

39-
import java.io.IOException;
4033
import java.time.Duration;
4134
import java.time.Instant;
4235
import java.util.ArrayList;
43-
import java.util.Arrays;
4436
import java.util.List;
4537
import java.util.Set;
4638
import java.util.UUID;
@@ -71,11 +63,6 @@ class DcbEndToEndTest extends AbstractAxonServerIntegrationTest {
7163
*/
7264
private static final boolean LOCAL = false;
7365

74-
/**
75-
* HTTP port for local Axon Server instance
76-
*/
77-
private static ServerAddress axonServerHttpPort;
78-
7966
/**
8067
* Static initialization for local mode
8168
*/
@@ -917,6 +904,61 @@ void getSequenceAtBetweenEvents() {
917904
}
918905
}
919906

907+
@Test
908+
@Disabled("Enable when scheduling is supported in AxonServer")
909+
void scheduleEvent() {
910+
DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
911+
long head = dcbEventChannel.head().join().getSequence();
912+
Event event = anEvent(aString(), "scheduledEvent");
913+
dcbEventChannel.scheduleEvent(Duration.ofSeconds(5), event)
914+
.join();
915+
StepVerifier.create(streamFlux(head).take(1))
916+
.expectNextMatches(r -> r.getEvent().getEvent().getName().equals("scheduledEvent"))
917+
.verifyComplete();
918+
}
919+
920+
@Test
921+
@Disabled("Enable when scheduling is supported in AxonServer")
922+
void cancelScheduledEvent() {
923+
DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
924+
long head = dcbEventChannel.head().join().getSequence();
925+
Event event = anEvent(aString(), "scheduledEvent");
926+
String token = dcbEventChannel.scheduleEvent(Duration.ofSeconds(1), event)
927+
.join();
928+
dcbEventChannel.cancelSchedule(token).join();
929+
StepVerifier.create(streamFlux(head).take(Duration.ofSeconds(2)))
930+
.verifyComplete();
931+
}
932+
933+
@Test
934+
@Disabled("Enable when scheduling is supported in AxonServer")
935+
void rescheduleEvent() {
936+
DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
937+
long head = dcbEventChannel.head().join().getSequence();
938+
Event event = anEvent(aString(), "rescheduledEvent");
939+
String token = dcbEventChannel.scheduleEvent(Duration.ofSeconds(30), event)
940+
.join();
941+
dcbEventChannel.reschedule(token, Duration.ofSeconds(2), null).join();
942+
StepVerifier.create(streamFlux(head).take(Duration.ofSeconds(3)))
943+
.expectNextMatches(r -> r.getEvent().getEvent().getName().equals("rescheduledEvent"))
944+
.verifyComplete();
945+
}
946+
947+
@Test
948+
@Disabled("Enable when scheduling is supported in AxonServer")
949+
void rescheduleAndReplaceEvent() {
950+
DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
951+
long head = dcbEventChannel.head().join().getSequence();
952+
Event event = anEvent(aString(), "scheduledEvent");
953+
String token = dcbEventChannel.scheduleEvent(Duration.ofSeconds(5), event)
954+
.join();
955+
Event replaceEvent = anEvent(aString(), "rescheduledEvent");
956+
dcbEventChannel.reschedule(token, Duration.ofSeconds(2), replaceEvent).join();
957+
StepVerifier.create(streamFlux(head).take(Duration.ofSeconds(6)))
958+
.expectNextMatches(r -> r.getEvent().getEvent().getName().equals("rescheduledEvent"))
959+
.verifyComplete();
960+
}
961+
920962
private Flux<SourceEventsResponse> sourceFlux(long start) {
921963
DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
922964
return fluxStream(() -> dcbEventChannel.source(sourceEventsRequest(start)));

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,7 @@ public static void purgeEventsFromAxonServer(String context, String hostname, in
4747
throws IOException {
4848
deleteContext(context, hostname, port);
4949
createContext(context, hostname, port, dcbContext);
50+
waitFor(1000);
5051
}
5152

5253
public static void deleteContext(String context, String hostname, int port) throws IOException {
@@ -182,4 +183,12 @@ private static void checkContextsCondition(String hostname, int port,
182183
throw new RuntimeException(e);
183184
}
184185
}
186+
187+
private static void waitFor(long millis) {
188+
try {
189+
Thread.sleep(millis);
190+
} catch (InterruptedException e) {
191+
throw new RuntimeException(e);
192+
}
193+
}
185194
}

0 commit comments

Comments
 (0)