Skip to content

Commit 037e7e5

Browse files
smcvbsmcvb
authored
Merge pull request #512 from AxonIQ/feature/dcb-snapshot-event-streaming-integration
Implement `DcbEventChannel#source(SnapshottedSourceRequest)`
2 parents c39624e + 41e7ceb commit 037e7e5

3 files changed

Lines changed: 136 additions & 4 deletions

File tree

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

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,8 @@
2727
import io.axoniq.axonserver.grpc.event.dcb.GetTagsResponse;
2828
import io.axoniq.axonserver.grpc.event.dcb.GetTailResponse;
2929
import io.axoniq.axonserver.grpc.event.dcb.RemoveTagsResponse;
30+
import io.axoniq.axonserver.grpc.event.dcb.SnapshottedSourceEventsResponse;
31+
import io.axoniq.axonserver.grpc.event.dcb.SnapshottedSourceRequest;
3032
import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest;
3133
import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse;
3234
import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest;
@@ -208,6 +210,18 @@ default ResultStream<StreamEventsResponse> stream(StreamEventsRequest request, i
208210
*/
209211
ResultStream<SourceEventsResponse> source(SourceEventsRequest request);
210212

213+
/**
214+
* Provides a finite stream of events used to source a model, optionally preceded by the latest snapshot stored
215+
* under the snapshot key in the given {@code request}. When a snapshot is present, it is emitted first, followed by
216+
* events with a sequence greater than the snapshot's sequence. When no snapshot is present, this behaves like
217+
* {@link #source(SourceEventsRequest)}, sourcing from the beginning.
218+
*
219+
* @param request the query used to filter events for sourcing and identify the snapshot to prefix the stream with
220+
* @return the response containing an optional snapshot, events to source a model, and a consistency marker to be
221+
* used when trying to append new events to the event store
222+
*/
223+
ResultStream<SnapshottedSourceEventsResponse> source(SnapshottedSourceRequest request);
224+
211225
/**
212226
* Provides tags for an event at the given global sequence.
213227
*

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

Lines changed: 30 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,9 @@
5050
import io.axoniq.axonserver.grpc.event.dcb.RescheduleEventRequest;
5151
import io.axoniq.axonserver.grpc.event.dcb.ScheduleEventRequest;
5252
import io.axoniq.axonserver.grpc.event.dcb.ScheduleToken;
53+
import io.axoniq.axonserver.grpc.event.dcb.SnapshottedDcbEventStoreGrpc;
54+
import io.axoniq.axonserver.grpc.event.dcb.SnapshottedSourceEventsResponse;
55+
import io.axoniq.axonserver.grpc.event.dcb.SnapshottedSourceRequest;
5356
import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest;
5457
import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse;
5558
import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest;
@@ -61,7 +64,6 @@
6164
import io.grpc.stub.StreamObserver;
6265

6366
import java.time.Instant;
64-
import java.util.Arrays;
6567
import java.util.Collection;
6668
import java.util.Set;
6769
import java.util.concurrent.CompletableFuture;
@@ -81,7 +83,9 @@ public class DcbEventChannelImpl extends AbstractAxonServerChannel<Void> impleme
8183

8284
private static final int BUFFER_SIZE = 512;
8385
private static final int REFILL_BATCH = 16;
86+
8487
private final DcbEventStoreGrpc.DcbEventStoreStub eventStore;
88+
private final SnapshottedDcbEventStoreGrpc.SnapshottedDcbEventStoreStub snapshottedEventStore;
8589
private final DcbEventSchedulerGrpc.DcbEventSchedulerStub eventScheduler;
8690
private final ClientIdentification clientIdentification;
8791
private final Set<ResultStream<StreamEventsResponse>> buffers = ConcurrentHashMap.newKeySet();
@@ -98,6 +102,7 @@ public DcbEventChannelImpl(ClientIdentification clientIdentification,
98102
AxonServerManagedChannel axonServerManagedChannel) {
99103
super(clientIdentification, executor, axonServerManagedChannel);
100104
this.eventStore = DcbEventStoreGrpc.newStub(axonServerManagedChannel);
105+
this.snapshottedEventStore = SnapshottedDcbEventStoreGrpc.newStub(axonServerManagedChannel);
101106
this.eventScheduler = DcbEventSchedulerGrpc.newStub(axonServerManagedChannel);
102107
this.clientIdentification = clientIdentification;
103108
}
@@ -171,9 +176,9 @@ protected StreamEventsResponse terminalMessage() {
171176
@Override
172177
public ResultStream<SourceEventsResponse> source(SourceEventsRequest request) {
173178
AbstractBufferedStream<SourceEventsResponse, Empty> result =
174-
new AbstractBufferedStream<SourceEventsResponse, Empty>(clientIdentification.getClientId(),
175-
BUFFER_SIZE,
176-
REFILL_BATCH) {
179+
new AbstractBufferedStream<>(clientIdentification.getClientId(),
180+
BUFFER_SIZE,
181+
REFILL_BATCH) {
177182
@Override
178183
protected SourceEventsResponse terminalMessage() {
179184
return SourceEventsResponse.newBuilder()
@@ -189,6 +194,27 @@ protected Empty buildFlowControlMessage(FlowControl flowControl) {
189194
return result;
190195
}
191196

197+
@Override
198+
public ResultStream<SnapshottedSourceEventsResponse> source(SnapshottedSourceRequest request) {
199+
AbstractBufferedStream<SnapshottedSourceEventsResponse, Empty> result =
200+
new AbstractBufferedStream<>(
201+
clientIdentification.getClientId(), BUFFER_SIZE, REFILL_BATCH
202+
) {
203+
@Override
204+
protected SnapshottedSourceEventsResponse terminalMessage() {
205+
return SnapshottedSourceEventsResponse.newBuilder()
206+
.build();
207+
}
208+
209+
@Override
210+
protected Empty buildFlowControlMessage(FlowControl flowControl) {
211+
return null;
212+
}
213+
};
214+
snapshottedEventStore.source(request, result);
215+
return result;
216+
}
217+
192218
@Override
193219
public CompletableFuture<GetTagsResponse> tagsFor(long sequence) {
194220
FutureStreamObserver<GetTagsResponse> future = new FutureStreamObserver<>(null);

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

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,21 +8,29 @@
88
import io.axoniq.axonserver.connector.ResultStream;
99
import io.axoniq.axonserver.connector.ResultStreamPublisher;
1010
import io.axoniq.axonserver.connector.event.DcbEventChannel;
11+
import io.axoniq.axonserver.connector.impl.HeaderAttachingInterceptor;
12+
import io.axoniq.axonserver.connector.impl.Headers;
1113
import io.axoniq.axonserver.connector.impl.ServerAddress;
14+
import io.axoniq.axonserver.grpc.event.dcb.AddSnapshotRequest;
1215
import io.axoniq.axonserver.grpc.event.dcb.AppendEventsResponse;
1316
import io.axoniq.axonserver.grpc.event.dcb.ConsistencyCondition;
1417
import io.axoniq.axonserver.grpc.event.dcb.Criterion;
18+
import io.axoniq.axonserver.grpc.event.dcb.DcbSnapshotStoreGrpc;
1519
import io.axoniq.axonserver.grpc.event.dcb.Event;
1620
import io.axoniq.axonserver.grpc.event.dcb.GetSequenceAtResponse;
1721
import io.axoniq.axonserver.grpc.event.dcb.GetTagsResponse;
1822
import io.axoniq.axonserver.grpc.event.dcb.GetTailResponse;
23+
import io.axoniq.axonserver.grpc.event.dcb.Snapshot;
24+
import io.axoniq.axonserver.grpc.event.dcb.SnapshottedSourceRequest;
1925
import io.axoniq.axonserver.grpc.event.dcb.SourceEventsRequest;
2026
import io.axoniq.axonserver.grpc.event.dcb.SourceEventsResponse;
2127
import io.axoniq.axonserver.grpc.event.dcb.StreamEventsRequest;
2228
import io.axoniq.axonserver.grpc.event.dcb.StreamEventsResponse;
2329
import io.axoniq.axonserver.grpc.event.dcb.Tag;
2430
import io.axoniq.axonserver.grpc.event.dcb.TaggedEvent;
2531
import io.axoniq.axonserver.grpc.event.dcb.TagsAndNamesCriterion;
32+
import io.grpc.ManagedChannel;
33+
import io.grpc.ManagedChannelBuilder;
2634
import org.junit.jupiter.api.*;
2735
import org.slf4j.Logger;
2836
import org.slf4j.LoggerFactory;
@@ -401,6 +409,72 @@ void sourceSingleTagAndEventName() {
401409
.verifyComplete();
402410
}
403411

412+
@Test
413+
void sourceWithSnapshotWhenNoneStored() {
414+
DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
415+
Tag tag = aTag();
416+
String eventName = aString();
417+
418+
long head = retrieveHead();
419+
TaggedEvent taggedEvent = taggedEvent(anEvent(aString(), eventName), tag);
420+
appendEvent(taggedEvent);
421+
422+
TagsAndNamesCriterion tagsAndName = TagsAndNamesCriterion.newBuilder().addTag(tag).addName(eventName).build();
423+
Criterion criterion = Criterion.newBuilder().setTagsAndNames(tagsAndName).build();
424+
SnapshottedSourceRequest request = SnapshottedSourceRequest.newBuilder()
425+
.setSnapshotKey(ByteString.copyFromUtf8(aString()))
426+
.addCriterion(criterion)
427+
.build();
428+
429+
StepVerifier.create(new ResultStreamPublisher<>(() -> dcbEventChannel.source(request)))
430+
.expectNextMatches(
431+
r -> !r.hasSnapshot()
432+
&& r.getEvent().getEvent().equals(taggedEvent.getEvent())
433+
&& r.getEvent().getSequence() == head
434+
)
435+
.expectNextMatches(r -> r.getConsistencyMarker() == head + 1L)
436+
.verifyComplete();
437+
}
438+
439+
@Test
440+
void sourceWithSnapshotWhenPresent() {
441+
DcbEventChannel dcbEventChannel = connection.dcbEventChannel();
442+
Tag tag = aTag();
443+
String eventName = aString();
444+
ByteString snapshotKey = ByteString.copyFromUtf8(aString());
445+
446+
long head = retrieveHead();
447+
TaggedEvent beforeSnapshot = taggedEvent(anEvent(aString(), eventName), tag);
448+
long snapshotSequence = appendEvent(beforeSnapshot).getSequenceOfTheFirstEvent();
449+
450+
Snapshot snapshot = Snapshot.newBuilder()
451+
.setName("snapshot-name")
452+
.setVersion("0.0.1")
453+
.setPayload(ByteString.copyFromUtf8("snapshot-payload"))
454+
.setTimestamp(Instant.now().toEpochMilli())
455+
.build();
456+
addSnapshot(snapshotKey, snapshotSequence, snapshot);
457+
458+
TaggedEvent afterSnapshot = taggedEvent(anEvent(aString(), eventName), tag);
459+
appendEvent(afterSnapshot);
460+
461+
TagsAndNamesCriterion tagsAndNames = TagsAndNamesCriterion.newBuilder().addTag(tag).addName(eventName).build();
462+
Criterion criterion = Criterion.newBuilder().setTagsAndNames(tagsAndNames).build();
463+
SnapshottedSourceRequest request = SnapshottedSourceRequest.newBuilder()
464+
.setSnapshotKey(snapshotKey)
465+
.addCriterion(criterion)
466+
.build();
467+
468+
StepVerifier.create(new ResultStreamPublisher<>(() -> dcbEventChannel.source(request)))
469+
.expectNextMatches(r -> r.hasSnapshot() && r.getSnapshot().equals(snapshot))
470+
.expectNextMatches(
471+
r -> r.getEvent().getEvent().equals(afterSnapshot.getEvent())
472+
&& r.getEvent().getSequence() > snapshotSequence
473+
)
474+
.expectNextMatches(r -> r.getConsistencyMarker() == head + 2L)
475+
.verifyComplete();
476+
}
477+
404478
@Test
405479
void noConditionAppend() {
406480
long head = retrieveHead();
@@ -1034,6 +1108,24 @@ private long retrieveHead() {
10341108
.getSequence();
10351109
}
10361110

1111+
private void addSnapshot(ByteString key, long sequence, Snapshot snapshot) {
1112+
ManagedChannel channel =
1113+
ManagedChannelBuilder.forAddress(axonServerAddress.getHostName(), axonServerAddress.getGrpcPort())
1114+
.usePlaintext()
1115+
.intercept(new HeaderAttachingInterceptor<>(Headers.CONTEXT, "default"))
1116+
.build();
1117+
try {
1118+
AddSnapshotRequest snapshotRequest = AddSnapshotRequest.newBuilder()
1119+
.setKey(key)
1120+
.setSequence(sequence)
1121+
.setSnapshot(snapshot)
1122+
.build();
1123+
DcbSnapshotStoreGrpc.newBlockingStub(channel).add(snapshotRequest);
1124+
} finally {
1125+
channel.shutdownNow();
1126+
}
1127+
}
1128+
10371129
private AppendEventsResponse appendEvent(TaggedEvent taggedEvent) {
10381130
return appendEventAsync(taggedEvent).join();
10391131
}

0 commit comments

Comments
 (0)