Skip to content

Commit fc9f6d3

Browse files
claudevanrogu
claude
authored and
vanrogu
committed
Remove consistent readmodel support
Consistent (synchronous) readmodel updates added complexity without sufficient benefit. All long-lived readmodels now use eventually consistent processing via the existing async ProjectorProcessor infrastructure. Removed: - ConsistentEventProcessor interface - CONSISTENT enum value from Consistency - consistent() method from LiveModelSpecification and LongLivedReadModelSpecification interfaces - Synchronous event subscription in BoundedContextImpl - updateSharedConsistentModels method from ReadModelModule - consistentReadModels collection and constructor parameter - All consistent model test fixtures from EventDispatchingToReadModelsTest https://claude.ai/code/session_01BS1GKM5XqW4kQyauxndPoZ
1 parent 1a66386 commit fc9f6d3

File tree

7 files changed

+41
-158
lines changed

7 files changed

+41
-158
lines changed

‎sliceworkz-eventmodeling-api/src/main/java/org/sliceworkz/eventmodeling/readmodels/LiveModelSpecification.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,6 @@ public interface LiveModelSpecification<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, O
2525

2626
BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> live();
2727

28-
BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> consistent();
29-
3028
BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> eventuallyConsistent();
3129

3230
/**

‎sliceworkz-eventmodeling-api/src/main/java/org/sliceworkz/eventmodeling/readmodels/LongLivedReadModelSpecification.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,6 @@ public interface LongLivedReadModelSpecification<DOMAIN_EVENT_TYPE, INBOUND_EVEN
2323

2424
BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> live();
2525

26-
BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> consistent();
27-
2826
BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> eventuallyConsistent();
2927

3028
LongLivedReadModelSpecification<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> ephemeral();

‎sliceworkz-eventmodeling-impl/src/main/java/org/sliceworkz/eventmodeling/module/boundedcontext/BoundedContextBuilderImpl.java‎

Lines changed: 4 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -309,10 +309,9 @@ public <T extends BoundedContext<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND
309309
LOGGER.warn("no features rootPackage");
310310
}
311311

312-
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> consistentReadModels = longLivedReadModelSpecs.stream().filter(s->s.consistency()==Consistency.CONSISTENT).map(LongLivedReadModelSpecificationImpl::readModel).collect(Collectors.toCollection(ArrayList::new));
313-
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentSharedReadModels = longLivedReadModelSpecs.stream().filter(s->s.consistency()==Consistency.EVENTUALLY_CONSISTENT&&s.isShared()).map(LongLivedReadModelSpecificationImpl::readModel).collect(Collectors.toCollection(ArrayList::new));
314-
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentLocalReadModels = longLivedReadModelSpecs.stream().filter(s->s.consistency()==Consistency.EVENTUALLY_CONSISTENT&&s.isLocal()&&!s.isEphemeral()).map(LongLivedReadModelSpecificationImpl::readModel).collect(Collectors.toCollection(ArrayList::new));
315-
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentEphemeralReadModels = longLivedReadModelSpecs.stream().filter(s->s.consistency()==Consistency.EVENTUALLY_CONSISTENT&&s.isLocal()&&s.isEphemeral()).map(LongLivedReadModelSpecificationImpl::readModel).collect(Collectors.toCollection(ArrayList::new));
312+
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentSharedReadModels = longLivedReadModelSpecs.stream().filter(s->s.isShared()).map(LongLivedReadModelSpecificationImpl::readModel).collect(Collectors.toCollection(ArrayList::new));
313+
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentLocalReadModels = longLivedReadModelSpecs.stream().filter(s->s.isLocal()&&!s.isEphemeral()).map(LongLivedReadModelSpecificationImpl::readModel).collect(Collectors.toCollection(ArrayList::new));
314+
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentEphemeralReadModels = longLivedReadModelSpecs.stream().filter(s->s.isLocal()&&s.isEphemeral()).map(LongLivedReadModelSpecificationImpl::readModel).collect(Collectors.toCollection(ArrayList::new));
316315

317316
Collection<Translator<INBOUND_EVENT_TYPE,DOMAIN_EVENT_TYPE>> translators = translatorSpecs.stream().map(i->(Translator<INBOUND_EVENT_TYPE,DOMAIN_EVENT_TYPE>)i).collect(Collectors.toCollection(ArrayList::new));
318317
InboundModule<DOMAIN_EVENT_TYPE,INBOUND_EVENT_TYPE,OUTBOUND_EVENT_TYPE> im = new InboundModule<>(name, inboundEventStream, translators, instance, meterRegistry);
@@ -322,7 +321,7 @@ public <T extends BoundedContext<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND
322321

323322
AutomationModule<DOMAIN_EVENT_TYPE,INBOUND_EVENT_TYPE,OUTBOUND_EVENT_TYPE> am = new AutomationModule<>(name, domainEventStream, automations, instance, meterRegistry);
324323

325-
ReadModelModule<DOMAIN_EVENT_TYPE> rmm = new ReadModelModule<DOMAIN_EVENT_TYPE>(name, domainEventStream, readAllInStoreEventStream, liveModelSpecs, consistentReadModels, eventuallyConsistentSharedReadModels, eventuallyConsistentLocalReadModels, eventuallyConsistentEphemeralReadModels, instance, meterRegistry);
324+
ReadModelModule<DOMAIN_EVENT_TYPE> rmm = new ReadModelModule<DOMAIN_EVENT_TYPE>(name, domainEventStream, readAllInStoreEventStream, liveModelSpecs, eventuallyConsistentSharedReadModels, eventuallyConsistentLocalReadModels, eventuallyConsistentEphemeralReadModels, instance, meterRegistry);
326325
DCBModule<DOMAIN_EVENT_TYPE, OUTBOUND_EVENT_TYPE> dcb = new DCBModule<DOMAIN_EVENT_TYPE, OUTBOUND_EVENT_TYPE>(name, instance, rmm, domainEventStream, outboundEventStream, meterRegistry);
327326

328327
AggregateModule<DOMAIN_EVENT_TYPE> aggregateModule = new AggregateModule<DOMAIN_EVENT_TYPE>(name, instance, aggregateSpecifications, domainEventStream, meterRegistry);
@@ -355,7 +354,6 @@ public class LiveModelSpecificationImpl implements LiveModelSpecification<DOMAIN
355354

356355
private BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> builder;
357356
private Class<? extends ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> readModelClass;
358-
private Consistency consistency = Consistency.LIVE;
359357
private LiveModelSnapshotSpecificationImpl<?,DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> snapshotSpecification;
360358

361359
public LiveModelSpecificationImpl ( BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> builder, Class<? extends ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> readModelClass ) {
@@ -365,15 +363,9 @@ public LiveModelSpecificationImpl ( BoundedContextBuilder<DOMAIN_EVENT_TYPE, INB
365363

366364
@Override
367365
public BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> live ( ) {
368-
this.consistency = Consistency.LIVE;
369366
return builder;
370367
}
371368

372-
@Override
373-
public BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> consistent ( ) {
374-
throw new IllegalArgumentException("LIVE read model - cannot be updated CONSISTENTly");
375-
}
376-
377369
@Override
378370
public BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> eventuallyConsistent ( ) {
379371
throw new IllegalArgumentException("LIVE read model - cannot be updated EVENTUALLY CONSISTENT");
@@ -389,10 +381,6 @@ public <SNAPSHOT_TYPE> LiveModelSnapshotSpecification<DOMAIN_EVENT_TYPE, INBOUND
389381
return snapshotSpecification;
390382
}
391383

392-
public Consistency consistency ( ) {
393-
return consistency;
394-
}
395-
396384
public Class<? extends ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> readModelClass ( ) {
397385
return readModelClass;
398386
}
@@ -422,7 +410,6 @@ public class LongLivedReadModelSpecificationImpl implements LongLivedReadModelSp
422410
private boolean ephemeral = false;
423411
private BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> builder;
424412
private ReadModelWithMetaData<DOMAIN_EVENT_TYPE> readModel;
425-
private Consistency consistency = Consistency.EVENTUALLY_CONSISTENT;
426413

427414
public LongLivedReadModelSpecificationImpl ( BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> builder, ReadModelWithMetaData<DOMAIN_EVENT_TYPE> readModel ) {
428415
this.builder = builder;
@@ -434,18 +421,8 @@ public BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVE
434421
throw new IllegalArgumentException("state-based readmodel - cannot be live rendered");
435422
}
436423

437-
@Override
438-
public BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> consistent ( ) {
439-
if ( ! isShared() ) {
440-
throw new IllegalArgumentException("CONSISTENT updates are only supported for SHARED ReadMoodels. LIVE, LOCAL and EPHEMERAL ReadModels can not by updated CONSISTENTly.");
441-
}
442-
this.consistency = Consistency.CONSISTENT;
443-
return builder;
444-
}
445-
446424
@Override
447425
public BoundedContextBuilder<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE> eventuallyConsistent ( ) {
448-
this.consistency = Consistency.EVENTUALLY_CONSISTENT;
449426
return builder;
450427
}
451428

@@ -482,10 +459,6 @@ public boolean isLocal ( ) {
482459
return !isShared();
483460
}
484461

485-
public Consistency consistency ( ) {
486-
return consistency;
487-
}
488-
489462
public ReadModelWithMetaData<DOMAIN_EVENT_TYPE> readModel ( ) {
490463
return readModel;
491464
}
@@ -502,10 +475,4 @@ private static void logEventTypes ( String type, Class<?> rootEventClass ) {
502475
}
503476
}
504477

505-
public enum Consistency {
506-
LIVE,
507-
CONSISTENT,
508-
EVENTUALLY_CONSISTENT
509-
}
510-
511478
}

‎sliceworkz-eventmodeling-impl/src/main/java/org/sliceworkz/eventmodeling/module/boundedcontext/BoundedContextImpl.java‎

Lines changed: 1 addition & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,6 @@
3636
import org.sliceworkz.eventmodeling.module.automation.AutomationModule;
3737
import org.sliceworkz.eventmodeling.module.boundedcontext.KernelEvent.BoundedContextStarted;
3838
import org.sliceworkz.eventmodeling.module.dcb.DCBModule;
39-
import org.sliceworkz.eventmodeling.module.eventdispatching.ConsistentEventProcessor;
4039
import org.sliceworkz.eventmodeling.module.inbound.InboundModule;
4140
import org.sliceworkz.eventmodeling.module.outbound.OutboundModule;
4241
import org.sliceworkz.eventmodeling.module.readmodels.ReadModelModule;
@@ -53,7 +52,7 @@
5352
import io.micrometer.core.instrument.Counter;
5453
import io.micrometer.core.instrument.MeterRegistry;
5554

56-
public class BoundedContextImpl<DOMAIN_EVENT_TYPE,INBOUND_EVENT_TYPE,OUTBOUND_EVENT_TYPE> implements AllCapabilities<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE>, UnboundedReadModelCapability<DOMAIN_EVENT_TYPE>, ConsistentEventProcessor<DOMAIN_EVENT_TYPE> {
55+
public class BoundedContextImpl<DOMAIN_EVENT_TYPE,INBOUND_EVENT_TYPE,OUTBOUND_EVENT_TYPE> implements AllCapabilities<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE>, UnboundedReadModelCapability<DOMAIN_EVENT_TYPE> {
5756

5857
private static final Logger LOGGER = LoggerFactory.getLogger(BoundedContextImpl.class);
5958

@@ -99,7 +98,6 @@ public BoundedContextImpl (
9998
this.undeployedFeatureSlices = undeployedFeatureSlices;
10099

101100
this.domainEventStream = domainEventStream;
102-
this.domainEventStream.subscribe(this::syncDomainEventHandler); // subscribe the kernel to the stream's event appends, to deliver to consistent consumers
103101

104102
this.readmodelModule = readmodelModule;
105103
this.inboundModule = inboundModule;
@@ -297,16 +295,6 @@ public <T extends Aggregate<DOMAIN_EVENT_TYPE>> T aggregate(Class<T> aggregateCl
297295

298296

299297

300-
/*
301-
* EVENT DISPATCHING
302-
*/
303-
304-
@Override
305-
public void syncDomainEventHandler(List<? extends Event<DOMAIN_EVENT_TYPE>> events) {
306-
// update consistent readmodels
307-
readmodelModule.updateSharedConsistentModels(events.stream());
308-
}
309-
310298
@SuppressWarnings("unchecked")
311299
@Override
312300
public <T extends FeatureSliceConfiguration<DOMAIN_EVENT_TYPE, INBOUND_EVENT_TYPE, OUTBOUND_EVENT_TYPE>> List<T> getDeployedFeatureSlices() {

‎sliceworkz-eventmodeling-impl/src/main/java/org/sliceworkz/eventmodeling/module/eventdispatching/ConsistentEventProcessor.java‎

Lines changed: 0 additions & 29 deletions
This file was deleted.

‎sliceworkz-eventmodeling-impl/src/main/java/org/sliceworkz/eventmodeling/module/readmodels/ReadModelModule.java‎

Lines changed: 0 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,6 @@ public class ReadModelModule<DOMAIN_EVENT_TYPE> implements LifecycleCapability {
6161
private EventSource<DOMAIN_EVENT_TYPE> domainEventStream;
6262
private EventSource<Object> allInStorageEventStream;
6363
private Map<Class<? extends ReadModelWithMetaData<DOMAIN_EVENT_TYPE>>, LiveModelInfo<DOMAIN_EVENT_TYPE>> liveModels = new HashMap<>();
64-
private Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> consistentReadModels = new ArrayList<>();
6564
private Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentSharedReadModels = new ArrayList<>();
6665
private Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentLocalReadModels = new ArrayList<>();
6766
private String boundedContext;
@@ -90,7 +89,6 @@ public <LMSI extends LiveModelSpecificationAccessor<DOMAIN_EVENT_TYPE>> ReadMode
9089
EventStream<DOMAIN_EVENT_TYPE> domainEventStream,
9190
EventStream<Object> allInStorageEventStream,
9291
List<LMSI> liveModelSpecs,
93-
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> consistentReadModels,
9492
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentSharedReadModels,
9593
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentLocalReadModels,
9694
Collection<ReadModelWithMetaData<DOMAIN_EVENT_TYPE>> eventuallyConsistentEphemeralReadModels,
@@ -127,10 +125,6 @@ public <LMSI extends LiveModelSpecificationAccessor<DOMAIN_EVENT_TYPE>> ReadMode
127125
counterSnapshotWrite));
128126
}
129127

130-
for ( ReadModelWithMetaData<DOMAIN_EVENT_TYPE> consistentReadModel : consistentReadModels ) {
131-
this.consistentReadModels.add(consistentReadModel);
132-
}
133-
134128
for ( ReadModelWithMetaData<DOMAIN_EVENT_TYPE> eventuallyConsistentSharedReadModel : eventuallyConsistentSharedReadModels ) {
135129
this.eventuallyConsistentSharedReadModels.add(eventuallyConsistentSharedReadModel);
136130
}
@@ -268,16 +262,6 @@ private Constructor<?> selectConstructor ( Class<?> readModelClass, Object[] con
268262
throw new IllegalArgumentException("no public constructor found on " + readModelClass + " for parameters " + constructorParams);
269263
}
270264

271-
/**
272-
* Updates consistent readmodels shared over all BoundedContext instances (eg: database, CDN, ...)
273-
* This method will only be called on a single BoundedContext instance, the one where the event is initially delivered (due to consistency)
274-
*/
275-
public void updateSharedConsistentModels ( Stream<? extends Event<DOMAIN_EVENT_TYPE>> events ) {
276-
events.forEach(e->{
277-
consistentReadModels.stream().filter(rm->rm.eventQuery().matches(e)).forEach(rm->rm.when(e));
278-
});
279-
}
280-
281265
// TODO is this the way? this can be done by the processors at start also ...
282266
/**
283267
* Initialized all ephemeral/inmemory models before start

0 commit comments

Comments
 (0)