chore: adding examples, release version (#11)

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

Author: yhl25

.github/workflows/maven-publish.yml

@@ -5,7 +5,7 @@ name: Maven Package
 
 on:
   release:
-    types: [created]
+    types: [ created ]
 
 jobs:
   build:
@@ -16,19 +16,19 @@ jobs:
       packages: write
 
     steps:
-    - uses: actions/checkout@v3
-    - name: Set up JDK 11
-      uses: actions/setup-java@v3
-      with:
-        java-version: '11'
-        distribution: 'temurin'
-        server-id: github # Value of the distributionManagement/repository/id field of the pom.xml
-        settings-path: ${{ github.workspace }} # location for the settings.xml file
+      - uses: actions/checkout@v3
+      - name: Set up JDK 11
+        uses: actions/setup-java@v3
+        with:
+          java-version: '11'
+          distribution: 'temurin'
+          server-id: github # Value of the distributionManagement/repository/id field of the pom.xml
+          settings-path: ${{ github.workspace }} # location for the settings.xml file
 
-    - name: Build with Maven
-      run: mvn -B package --file pom.xml
+      - name: Build with Maven
+        run: mvn -B package --file pom.xml
 
-    - name: Publish to GitHub Packages Apache Maven
-      run: mvn deploy -s $GITHUB_WORKSPACE/settings.xml
-      env:
-        GITHUB_TOKEN: ${{ github.token }}
+      - name: Publish to GitHub Packages Apache Maven
+        run: mvn deploy -s $GITHUB_WORKSPACE/settings.xml
+        env:
+          GITHUB_TOKEN: ${{ github.token }}

examples/src/main/java/io/numaproj/numaflow/examples/function/evenodd/Forward.java

@@ -0,0 +1,26 @@
+package io.numaproj.numaflow.examples.function.evenodd;
+
+import io.numaproj.numaflow.function.FunctionServer;
+import io.numaproj.numaflow.function.Message;
+import io.numaproj.numaflow.function.map.MapFunc;
+import io.numaproj.numaflow.function.v1.Udfunction;
+
+import java.io.IOException;
+import java.util.logging.Logger;
+
+public class Forward {
+    private static final Logger logger = Logger.getLogger(Forward.class.getName());
+
+    private static Message[] process(String key, Udfunction.Datum data) {
+        int value = Integer.parseInt(new String(data.getValue().toByteArray()));
+        if (value % 2 == 0) {
+            return new Message[]{Message.to("even", data.getValue().toByteArray())};
+        }
+        return new Message[]{Message.to("odd", data.getValue().toByteArray())};
+    }
+
+    public static void main(String[] args) throws IOException {
+        logger.info("Forward invoked");
+        new FunctionServer().registerMapper(new MapFunc(Forward::process)).start();
+    }
+}

examples/src/main/java/io/numaproj/numaflow/examples/function/flatmap/FlatMap.java

@@ -0,0 +1,29 @@
+package io.numaproj.numaflow.examples.function.flatmap;
+
+import io.numaproj.numaflow.function.FunctionServer;
+import io.numaproj.numaflow.function.Message;
+import io.numaproj.numaflow.function.map.MapFunc;
+import io.numaproj.numaflow.function.v1.Udfunction;
+
+import java.io.IOException;
+import java.util.logging.Logger;
+
+public class FlatMap {
+    private static final Logger logger = Logger.getLogger(FlatMap.class.getName());
+
+    private static Message[] process(String key, Udfunction.Datum data) {
+        String msg = new String(data.getValue().toByteArray());
+        String[] strs = msg.split(",");
+        Message[] results = new Message[strs.length];
+
+        for (int i = 0; i < strs.length; i++) {
+            results[i] = Message.toAll(strs[i].getBytes());
+        }
+        return results;
+    }
+
+    public static void main(String[] args) throws IOException {
+        logger.info("Flatmap invoked");
+        new FunctionServer().registerMapper(new MapFunc(FlatMap::process)).start();
+    }
+}

examples/src/main/java/io/numaproj/numaflow/examples/sink/simple/SimpleSink.java

@@ -2,7 +2,6 @@
 
 import io.numaproj.numaflow.sink.Response;
 import io.numaproj.numaflow.sink.SinkDatumStream;
-import io.numaproj.numaflow.sink.SinkDatumStreamImpl;
 import io.numaproj.numaflow.sink.SinkFunc;
 import io.numaproj.numaflow.sink.SinkServer;
 import io.numaproj.numaflow.sink.v1.Udsink;
@@ -20,7 +19,7 @@ private static List<Response> process(SinkDatumStream datumStream) {
 
         while (true) {
             Udsink.Datum datum = datumStream.ReadMessage();
-            // DONE indicates the end of the input
+            // EOF indicates the end of the input
             if (datum == SinkDatumStream.EOF) {
                 break;
             }

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>io.numaproj.numaflow</groupId>
     <artifactId>numaflow-java</artifactId>
-    <version>0.3.0-SNAPSHOT</version>
+    <version>0.3.1</version>
     <packaging>jar</packaging>
 
     <name>numaflow-java</name>

settings.xml

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<settings xmlns="http://maven.apache.org/SETTINGS/1.0.0"
-          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+<settings xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+          xmlns="http://maven.apache.org/SETTINGS/1.0.0"
           xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.0.0 http://maven.apache.org/xsd/settings-1.0.0.xsd">
     <activeProfiles>
         <activeProfile>github</activeProfile>
@@ -13,8 +13,12 @@
                 <repository>
                     <id>central</id>
                     <url>https://repo1.maven.org/maven2</url>
-                    <releases><enabled>true</enabled></releases>
-                    <snapshots><enabled>false</enabled></snapshots>
+                    <releases>
+                        <enabled>true</enabled>
+                    </releases>
+                    <snapshots>
+                        <enabled>false</enabled>
+                    </snapshots>
                 </repository>
                 <repository>
                     <id>github</id>
@@ -32,4 +36,4 @@
             <password>${env.JAVA_SDK_MAVEN_PASSWORD}</password>
         </server>
     </servers>
-</settings>
+</settings>

src/main/java/io/numaproj/numaflow/function/reduce/ReduceDatumStream.java

@@ -7,9 +7,10 @@
  * reduce handlers to read input messages.
  */
 public interface ReduceDatumStream {
+    // EOF indicates the end of input
+    Udfunction.Datum EOF = Udfunction.Datum.newBuilder().setKey("EOF").build();
+
     /* ReadMessage can be used to read message from the stream
      * returns null if there are no more messages to consume.*/
     Udfunction.Datum ReadMessage();
-    // EOF indicates the end of input
-    Udfunction.Datum EOF = Udfunction.Datum.newBuilder().setKey("EOF").build();
 }

src/main/java/io/numaproj/numaflow/function/reduce/ReduceDatumStreamImpl.java

@@ -23,7 +23,8 @@ public Udfunction.Datum ReadMessage() {
         try {
             readMessage = blockingQueue.take();
         } catch (InterruptedException e) {
-            logger.severe("Error occurred while reading message from datum stream" + e.getMessage());
+            logger.severe(
+                    "Error occurred while reading message from datum stream" + e.getMessage());
             Thread.interrupted();
         }
         return readMessage;

src/main/java/io/numaproj/numaflow/sink/SinkDatumStream.java

@@ -1,16 +1,16 @@
 package io.numaproj.numaflow.sink;
 
-import io.numaproj.numaflow.function.v1.Udfunction;
 import io.numaproj.numaflow.sink.v1.Udsink;
 
 /**
  * SinkDatumStream is an interface which will be passed to
  * sink handlers to read input messages.
  */
 public interface SinkDatumStream {
+    // EOF indicates the end of input
+    Udsink.Datum EOF = Udsink.Datum.newBuilder().setKey("EOF").build();
+
     /* ReadMessage can be used to read message from the stream
      * returns null if there are no more messages to consume.*/
     Udsink.Datum ReadMessage();
-    // EOF indicates the end of input
-    Udsink.Datum EOF = Udsink.Datum.newBuilder().setKey("EOF").build();
 }

src/main/java/io/numaproj/numaflow/sink/SinkDatumStreamImpl.java

@@ -23,7 +23,8 @@ public Udsink.Datum ReadMessage() {
         try {
             readMessage = blockingQueue.take();
         } catch (InterruptedException e) {
-            logger.severe("Error occurred while reading message from datum stream" + e.getMessage());
+            logger.severe(
+                    "Error occurred while reading message from datum stream" + e.getMessage());
             Thread.interrupted();
         }
         return readMessage;