OccurrentSubscriptionFilter to a record

changelog.md

@@ -1,3 +1,6 @@
+### Changelog next version
+* Converted `org.occurrent.subscription.OccurrentSubscriptionFilter` from a Java class to a record. This means that the `public final` filter instance field is now a record property. So if you ever used `occurrentSubscriptionFilter.filter` to access the underluying filter, you now need to do `occurrentSubscriptionFilter.filter()` instead.  
+
 ### 0.19.7 (2024-11-01)
 * Implemented "in" conditions so you can now do e.g. `subscriptionModel.subscribe("id", OccurrentSubscriptionFilter.filter(Filter.streamVersion(Condition.in(12L, 14L))`. There's also a Kotlin extension function, `isIn`, which can be imported from `org.occurrent.condition.isIn`.
 * Upgraded kotlin from 2.0.20 to 2.0.21

common/mongodb/native/filter-bsonfilter-conversion/src/main/java/org/occurrent/mongodb/spring/filterbsonfilterconversion/internal/FilterToBsonFilterConverter.java

@@ -59,19 +59,12 @@ private static Bson innerConvert(String fieldNamePrefix, TimeRepresentation time
             Condition<?> conditionToUse = resolveSpecialCases(timeRepresentation, scf);
             String fieldName = fieldNameOf(fieldNamePrefix, scf.fieldName());
             criteria = convertConditionToBsonCriteria(fieldName, conditionToUse);
-        } else if (filter instanceof CompositionFilter) {
-            CompositionFilter cf = (CompositionFilter) filter;
+        } else if (filter instanceof CompositionFilter cf) {
             Bson[] composedBson = cf.filters().stream().map(f -> innerConvert(fieldNamePrefix, timeRepresentation, f)).toArray(Bson[]::new);
-            switch (cf.operator()) {
-                case AND:
-                    criteria = Filters.and(composedBson);
-                    break;
-                case OR:
-                    criteria = Filters.or(composedBson);
-                    break;
-                default:
-                    throw new IllegalStateException("Unexpected value: " + cf.operator());
-            }
+            criteria = switch (cf.operator()) {
+                case AND -> Filters.and(composedBson);
+                case OR -> Filters.or(composedBson);
+            };
         } else {
             throw new IllegalStateException("Unexpected filter: " + filter.getClass().getName());
         }

subscription/core/src/main/java/org/occurrent/subscription/OccurrentSubscriptionFilter.java

@@ -23,13 +23,10 @@
 /**
  * An implementation of {@code SubscriptionFilter} for Occurrent {@link Filter}'s.
  */
-public class OccurrentSubscriptionFilter implements SubscriptionFilter {
+public record OccurrentSubscriptionFilter(Filter filter) implements SubscriptionFilter {
 
-    public final Filter filter;
-
-    public OccurrentSubscriptionFilter(Filter filter) {
+    public OccurrentSubscriptionFilter {
         Objects.requireNonNull(filter, Filter.class.getSimpleName() + " cannot be null");
-        this.filter = filter;
     }
 
     public static OccurrentSubscriptionFilter filter(Filter filter) {

subscription/inmemory/src/main/java/org/occurrent/subscription/inmemory/InMemorySubscriptionModel.java

@@ -166,7 +166,7 @@ private static Filter getFilter(SubscriptionFilter filter) {
         if (filter == null) {
             f = Filter.all();
         } else if (filter instanceof OccurrentSubscriptionFilter) {
-            f = ((OccurrentSubscriptionFilter) filter).filter;
+            f = ((OccurrentSubscriptionFilter) filter).filter();
         } else {
             throw new IllegalArgumentException(InMemorySubscriptionModel.class.getSimpleName() + " only support filters of type " + OccurrentSubscriptionFilter.class.getName());
         }

subscription/mongodb/native/blocking/src/main/java/org/occurrent/subscription/mongodb/nativedriver/blocking/NativeMongoSubscriptionModel.java

@@ -202,7 +202,7 @@ private static List<Bson> createPipeline(TimeRepresentation timeRepresentation,
         if (filter == null) {
             pipeline = Collections.emptyList();
         } else if (filter instanceof OccurrentSubscriptionFilter) {
-            Filter occurrentFilter = ((OccurrentSubscriptionFilter) filter).filter;
+            Filter occurrentFilter = ((OccurrentSubscriptionFilter) filter).filter();
             Bson bson = FilterToBsonFilterConverter.convertFilterToBsonFilter(MongoFilterSpecification.FULL_DOCUMENT, timeRepresentation, occurrentFilter);
             pipeline = Collections.singletonList(match(bson));
         } else if (filter instanceof MongoFilterSpecification.MongoJsonFilterSpecification) {

subscription/mongodb/spring/common/src/main/java/org/occurrent/subscription/mongodb/spring/internal/ApplyFilterToChangeStreamOptionsBuilder.java

@@ -45,7 +45,7 @@ public static ChangeStreamOptions applyFilter(TimeRepresentation timeRepresentat
         if (filter == null) {
             changeStreamOptions = changeStreamOptionsBuilder.build();
         } else if (filter instanceof OccurrentSubscriptionFilter) {
-            Filter occurrentFilter = ((OccurrentSubscriptionFilter) filter).filter;
+            Filter occurrentFilter = ((OccurrentSubscriptionFilter) filter).filter();
             Criteria criteria = convertFilterToCriteria(FULL_DOCUMENT, timeRepresentation, occurrentFilter);
             changeStreamOptions = changeStreamOptionsBuilder.filter(newAggregation(match(criteria))).build();
         } else if (filter instanceof MongoJsonFilterSpecification) {

subscription/util/blocking/catchup-subscription/src/main/java/org/occurrent/subscription/blocking/durable/catchup/CatchupSubscriptionModel.java

@@ -310,7 +310,7 @@ private static Filter deriveFilterToUseDuringCatchupPhase(SubscriptionFilter fil
         if (filter == null) {
             catchupFilter = timeFilter;
         } else {
-            Filter userSuppliedFilter = ((OccurrentSubscriptionFilter) filter).filter;
+            Filter userSuppliedFilter = ((OccurrentSubscriptionFilter) filter).filter();
             catchupFilter = timeFilter.and(userSuppliedFilter);
         }
         return catchupFilter;