feat: introducing factory for reducer (#22)

Signed-off-by: Yashash H L <[email protected]>

Author: yhl25

.DS_Store

@@ -1,4 +1,4 @@
-package io.numaproj.numaflow.examples.function.evenodd;
+package io.numaproj.numaflow.examples.function.map.evenodd;
 
 import io.numaproj.numaflow.function.Datum;
 import io.numaproj.numaflow.function.FunctionServer;

examples/src/main/java/io/numaproj/numaflow/examples/function/evenodd/EvenOddFunction.java renamed to examples/src/main/java/io/numaproj/numaflow/examples/function/map/evenodd/EvenOddFunction.java

@@ -1,4 +1,4 @@
-package io.numaproj.numaflow.examples.function.eventtimefilter;
+package io.numaproj.numaflow.examples.function.map.eventtimefilter;
 
 import io.numaproj.numaflow.function.Datum;
 import io.numaproj.numaflow.function.FunctionServer;

examples/src/main/java/io/numaproj/numaflow/examples/function/eventtimefilter/EventTimeFilterFunction.java renamed to examples/src/main/java/io/numaproj/numaflow/examples/function/map/eventtimefilter/EventTimeFilterFunction.java

@@ -1,4 +1,4 @@
-package io.numaproj.numaflow.examples.function.flatmap;
+package io.numaproj.numaflow.examples.function.map.flatmap;
 
 import io.numaproj.numaflow.function.Datum;
 import io.numaproj.numaflow.function.FunctionServer;

examples/src/main/java/io/numaproj/numaflow/examples/function/flatmap/FlatMapFunction.java renamed to examples/src/main/java/io/numaproj/numaflow/examples/function/map/flatmap/FlatMapFunction.java

@@ -1,4 +1,4 @@
-package io.numaproj.numaflow.examples.function.forward;
+package io.numaproj.numaflow.examples.function.map.forward;
 
 import io.numaproj.numaflow.function.Datum;
 import io.numaproj.numaflow.function.FunctionServer;

examples/src/main/java/io/numaproj/numaflow/examples/function/forward/ForwardFunction.java renamed to examples/src/main/java/io/numaproj/numaflow/examples/function/map/forward/ForwardFunction.java

@@ -0,0 +1,11 @@
+package io.numaproj.numaflow.examples.function.reduce.count;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+@Getter
+@AllArgsConstructor
+public class Config {
+    private int oddIncrementBy;
+    private int evenIncrementBy;
+}

examples/src/main/java/io/numaproj/numaflow/examples/function/reduce/count/Config.java

@@ -0,0 +1,72 @@
+package io.numaproj.numaflow.examples.function.reduce.count;
+
+import io.numaproj.numaflow.function.Datum;
+import io.numaproj.numaflow.function.FunctionServer;
+import io.numaproj.numaflow.function.Message;
+import io.numaproj.numaflow.function.metadata.Metadata;
+import io.numaproj.numaflow.function.reduce.Reducer;
+import io.numaproj.numaflow.function.reduce.ReducerFactory;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.Objects;
+
+@Slf4j
+@AllArgsConstructor
+public class EvenOddCounterFactory extends ReducerFactory<EvenOddCounterFactory.EvenOddCounter> {
+    private Config config;
+
+    @Override
+    public EvenOddCounter createReducer() {
+        return new EvenOddCounter(config);
+    }
+
+    @Slf4j
+    public static class EvenOddCounter extends Reducer {
+        private final Config config;
+        private int evenCount;
+        private int oddCount;
+
+        public EvenOddCounter(Config config) {
+            this.config = config;
+        }
+
+        @Override
+        public void addMessage(String key, Datum datum, Metadata md) {
+            try {
+                int val = Integer.parseInt(new String(datum.getValue()));
+                // increment based on the value specified in the config
+                if (val % 2 == 0) {
+                    evenCount += config.getEvenIncrementBy();
+                } else {
+                    oddCount += config.getOddIncrementBy();
+                }
+            } catch (NumberFormatException e) {
+                log.info("error while parsing integer - {}", e.getMessage());
+            }
+        }
+
+        @Override
+        public Message[] getOutput(String key, Metadata md) {
+            log.info(
+                    "even and odd count - {} {}, window - {} {}",
+                    evenCount,
+                    oddCount,
+                    md.getIntervalWindow().getStartTime().toString(),
+                    md.getIntervalWindow().getEndTime().toString());
+
+            if (Objects.equals(key, "even")) {
+                return new Message[]{Message.to(key, String.valueOf(evenCount).getBytes())};
+            } else {
+                return new Message[]{Message.to(key, String.valueOf(oddCount).getBytes())};
+            }
+        }
+    }
+
+    public static void main(String[] args) throws IOException {
+        log.info("counter udf was invoked");
+        Config config = new Config(1, 2);
+        new FunctionServer().registerReducerFactory(new EvenOddCounterFactory(config)).start();
+    }
+}

examples/src/main/java/io/numaproj/numaflow/examples/function/reduce/count/EvenOddCounterFactory.java

@@ -0,0 +1,21 @@
+package io.numaproj.numaflow.examples.function.reduce.sum;
+
+import io.numaproj.numaflow.function.FunctionServer;
+import io.numaproj.numaflow.function.reduce.ReducerFactory;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+
+@Slf4j
+public class SumFactory extends ReducerFactory<SumFunction> {
+
+    public static void main(String[] args) throws IOException {
+        log.info("sum udf was invoked");
+        new FunctionServer().registerReducerFactory(new SumFactory()).start();
+    }
+
+    @Override
+    public SumFunction createReducer() {
+        return new SumFunction();
+    }
+}

examples/src/main/java/io/numaproj/numaflow/examples/function/reduce/sum/SumFactory.java

@@ -1,30 +1,19 @@
-package io.numaproj.numaflow.examples.function.sum;
+package io.numaproj.numaflow.examples.function.reduce.sum;
 
 import io.numaproj.numaflow.function.Datum;
-import io.numaproj.numaflow.function.FunctionServer;
 import io.numaproj.numaflow.function.Message;
 import io.numaproj.numaflow.function.metadata.Metadata;
 import io.numaproj.numaflow.function.reduce.Reducer;
 import lombok.extern.slf4j.Slf4j;
 
-import java.io.IOException;
 
 @Slf4j
 public class SumFunction extends Reducer {
 
     private int sum = 0;
 
-    public SumFunction(String key, Metadata metadata) {
-        super(key, metadata);
-    }
-
-    public static void main(String[] args) throws IOException {
-        log.info("counter udf was invoked");
-        new FunctionServer().registerReducer(SumFunction.class).start();
-    }
-
     @Override
-    public void addMessage(Datum datum) {
+    public void addMessage(String key, Datum datum, Metadata md) {
         try {
             sum += Integer.parseInt(new String(datum.getValue()));
         } catch (NumberFormatException e) {
@@ -33,7 +22,7 @@ public void addMessage(Datum datum) {
     }
 
     @Override
-    public Message[] getOutput() {
+    public Message[] getOutput(String key, Metadata md) {
         return new Message[]{Message.toAll(String.valueOf(sum).getBytes())};
     }
 }

examples/src/main/java/io/numaproj/numaflow/examples/function/sum/SumFunction.java renamed to examples/src/main/java/io/numaproj/numaflow/examples/function/reduce/sum/SumFunction.java

@@ -16,6 +16,7 @@
 import io.numaproj.numaflow.function.map.MapHandler;
 import io.numaproj.numaflow.function.mapt.MapTHandler;
 import io.numaproj.numaflow.function.reduce.Reducer;
+import io.numaproj.numaflow.function.reduce.ReducerFactory;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
@@ -69,8 +70,8 @@ public FunctionServer registerMapperT(MapTHandler mapTHandler) {
         return this;
     }
 
-    public FunctionServer registerReducer(Class<? extends Reducer> groupByClass) {
-        this.functionService.setReduceHandler(groupByClass);
+    public FunctionServer registerReducerFactory(ReducerFactory<? extends Reducer> reducerFactory) {
+        this.functionService.setReduceHandler(reducerFactory);
         return this;
     }