Merge pull request #4 from grainier/join-desugar

Change output processor to only return CURRENT events
ballerina-platform · Sep 19, 2018 · 2796453 · 2796453
2 parents 0a531f4 + de8c7ed
commit 2796453
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 7 deletions.
diff --git a/stdlib/streams/src/main/ballerina/streams/output-processor.bal b/stdlib/streams/src/main/ballerina/streams/output-processor.bal
@@ -24,14 +24,16 @@ public type OutputProcess object {
     public function process(StreamEvent[] streamEvents) {
         int index = 0;
         foreach event in streamEvents {
-            map outputData;
-            foreach k, v in event.data {
-                string[] s = k.split("\\.");
-                if (OUTPUT.equalsIgnoreCase(s[0])) {
-                    outputData[s[1]] = v;
+            if (event.eventType == "CURRENT") {
+                map outputData;
+                foreach k, v in event.data {
+                    string[] s = k.split("\\.");
+                    if (OUTPUT.equalsIgnoreCase(s[0])) {
+                        outputData[s[1]] = v;
+                    }
                 }
+                outputFunc(outputData);
             }
-            outputFunc(outputData);
         }
     }
 };

diff --git a/...-unit-test/src/test/java/org/ballerinalang/test/streaming/BallerinaStreamsV2JoinTest.java b/...-unit-test/src/test/java/org/ballerinalang/test/streaming/BallerinaStreamsV2JoinTest.java
@@ -44,7 +44,6 @@ public void setup() {
 
     @Test(description = "Test stream join query.")
     public void testStreamJoinQuery() {
-        System.setProperty("enable.siddhiRuntime", "true");
         BValue[] stocksWithPrices = BRunUtil.invoke(result, "startJoinQuery");
         Assert.assertNotNull(stocksWithPrices);
         Assert.assertEquals(stocksWithPrices.length, 2, "Expected events are not received");