allow to combine two matchers with a logical operator OR

fhussonnois · May 5, 2018 · 4c88b43 · 4c88b43
1 parent 800feb3
commit 4c88b43
Show file tree

Hide file tree

Showing 4 changed files with 42 additions and 10 deletions.
diff --git a/README.md b/README.md
@@ -143,7 +143,7 @@ QueryBuilder<String, Integer> builder = new QueryBuilder<>();
 <pre lang="java">
 builder.where(new SimpleMatcher<String, String>() {
     @Override
-    public boolean matches(Event<String, Integer> event) {
+    public boolean matches(Event<String, String> event) {
            return event.value().equals("A");
     }
 });
@@ -192,7 +192,17 @@ builder.fold("amount", new Aggregator<String, Integer, Integer>() {
         <td class="desc">Combines two simple patterns in a stage with a logical operator AND.</td>
         <td>
 <pre lang="java">
-builder.where( (event) -> event.value > 0).and((event) -> event.value < 10)    
+builder.where( (event) -> event.value() > 0).and((event) -> event.value() < 10)    
+</pre>
+        </td>
+    </tr>
+    </tr>
+    <tr>
+        <td><pre lang="java">or(Matcher)</pre></td>
+        <td class="desc">Combines two simple patterns in a stage with a logical operator OR.</td>
+        <td>
+<pre lang="java">
+builder.where( (event) -> event.value().equals("A")).or((event) -> event.value().equals("B")))    
 </pre>
         </td>
     </tr>

diff --git a/core/src/main/java/com/github/fhuss/kafka/streams/cep/pattern/Pattern.java b/core/src/main/java/com/github/fhuss/kafka/streams/cep/pattern/Pattern.java
@@ -145,13 +145,20 @@ void setWindow(long time, TimeUnit unit) {
         this.windowUnit = unit;
     }
 
-    void addPredicate(final Matcher<K, V> predicate) {
+    void andPredicate(final Matcher<K, V> predicate) {
         if (this.predicate == null)
             this.predicate = predicate;
         else
             this.predicate = Matcher.and(this.predicate, predicate);
     }
 
+    void orPredicate(final Matcher<K, V> predicate) {
+        if (this.predicate == null)
+            this.predicate = predicate;
+        else
+            this.predicate = Matcher.or(this.predicate, predicate);
+    }
+
     void setCardinality(final Cardinality cardinality) {
         this.cardinality = cardinality;
     }

diff --git a/core/src/main/java/com/github/fhuss/kafka/streams/cep/pattern/PredicateBuilder.java b/core/src/main/java/com/github/fhuss/kafka/streams/cep/pattern/PredicateBuilder.java
@@ -21,7 +21,7 @@
 
 public class PredicateBuilder<K, V> {
 
-    private Pattern<K, V> pattern;
+    private final Pattern<K, V> pattern;
 
     /**
      * Creates a  new {@link PredicateBuilder} instance.
@@ -32,17 +32,32 @@ public class PredicateBuilder<K, V> {
     }
 
     public PredicateBuilder<K, V> and(final SimpleMatcher<K, V> predicate) {
-        this.pattern.addPredicate(predicate);
+        this.pattern.andPredicate(predicate);
         return this;
     }
 
     public PredicateBuilder<K, V> and(final StatefulMatcher<K, V> predicate) {
-        this.pattern.addPredicate(predicate);
+        this.pattern.andPredicate(predicate);
         return this;
     }
 
     public PredicateBuilder<K, V> and(final SequenceMatcher<K, V> predicate) {
-        this.pattern.addPredicate(predicate);
+        this.pattern.andPredicate(predicate);
+        return this;
+    }
+
+    public PredicateBuilder<K, V> or(final SimpleMatcher<K, V> predicate) {
+        this.pattern.orPredicate(predicate);
+        return this;
+    }
+
+    public PredicateBuilder<K, V> or(final StatefulMatcher<K, V> predicate) {
+        this.pattern.orPredicate(predicate);
+        return this;
+    }
+
+    public PredicateBuilder<K, V> or(final SequenceMatcher<K, V> predicate) {
+        this.pattern.orPredicate(predicate);
         return this;
     }
 

diff --git a/core/src/main/java/com/github/fhuss/kafka/streams/cep/pattern/SelectBuilder.java b/core/src/main/java/com/github/fhuss/kafka/streams/cep/pattern/SelectBuilder.java
@@ -58,17 +58,17 @@ public SelectBuilder<K, V> strictContiguity() {
     }
 
     public PredicateBuilder<K, V> where(final SimpleMatcher<K, V> predicate) {
-        this.pattern.addPredicate(predicate);
+        this.pattern.andPredicate(predicate);
         return new PredicateBuilder<>(this.pattern);
     }
 
     public PredicateBuilder<K, V> where(final StatefulMatcher<K, V> predicate) {
-        this.pattern.addPredicate(predicate);
+        this.pattern.andPredicate(predicate);
         return new PredicateBuilder<>(this.pattern);
     }
 
     public PredicateBuilder<K, V> where(final SequenceMatcher<K, V> predicate) {
-        this.pattern.addPredicate(predicate);
+        this.pattern.andPredicate(predicate);
         return new PredicateBuilder<>(this.pattern);
     }