Merge pull request #134 from dilini-muthumala/master

Introduce EOF back into other modes

Author: dnwick

component/src/main/java/io/siddhi/extension/io/file/FileSource.java

@@ -602,7 +602,6 @@ public StateFactory<FileSourceState> init(SourceEventListener sourceEventListene
         createInitialSourceConf();
         updateSourceConf();
         getPattern();
-        validateEOFConfigs();
         if (MetricsDataHolder.getInstance().getMetricService() != null &&
                 MetricsDataHolder.getInstance().getMetricManagementService().isEnabled()) {
             try {
@@ -1058,21 +1057,4 @@ public void restore(Map<String, Object> map) {
                         (List<String>) map.get(Constants.PROCESSED_FILE_LIST));
         }
     }
-
-    private void validateEOFConfigs() {
-        for (String property: requiredProperties) {
-            /**
-             * EOF property will only be supported under REGEX mode with tailing disabled, going forward.
-             * TEXT_FULL mode is supported in this version to keep backward compatibility
-             */
-            if (property.equalsIgnoreCase(Constants.EOF)) {
-                if (!((Constants.REGEX.equalsIgnoreCase(mode) && !isTailingEnabled)
-                        || Constants.TEXT_FULL.equalsIgnoreCase(mode))) { 
-                    throw new SiddhiAppCreationException("Transport property trp:eof can be provided only " +
-                            "on regex mode when tailing disabled or in Text Full mode. Given mode: " + mode
-                            + ". Tailing enabled: " + isTailingEnabled);
-                }
-            }
-        }
-    }
 }

component/src/main/java/io/siddhi/extension/io/file/processors/FileProcessor.java

@@ -323,21 +323,6 @@ public String getId() {
         return "file-message-processor";
     }
 
-    private String[] getRequiredPropertyValues(CarbonMessage carbonMessage) {
-        String[] values = new String[requiredProperties.length];
-        int i = 0;
-        for (String propertyKey : requiredProperties) {
-            Object value = carbonMessage.getProperty(propertyKey);
-            if (value != null) {
-                values[i++] = value.toString();
-            } else {
-                log.error("Failed to find required transport property '" + propertyKey + "'. Assigning null value");
-                values[i++] = null;
-            }
-        }
-        return values;
-    }
-
     private void increaseMetrics(int byteLength) {
         metrics.getTotalFileReadCount().inc();
         metrics.getTotalReadsMetrics().inc();
@@ -358,8 +343,23 @@ private void increaseTailingMetrics() {
         metrics.getTailEnabledFilesMap().replace(Utils.getShortFilePath(fileURI), System.currentTimeMillis());
     }
 
+    private String[] getRequiredPropertyValues(CarbonMessage carbonMessage) {
+        String[] values = new String[requiredProperties.length];
+        int i = 0;
+        for (String propertyKey : requiredProperties) {
+            Object value = carbonMessage.getProperty(propertyKey);
+            if (value != null) {
+                values[i++] = value.toString();
+            } else {
+                log.error("Failed to find required transport property '" + propertyKey + "'. Assigning null value");
+                values[i++] = null;
+            }
+        }
+        return values;
+    }
+
     /**
-     * In Regex mode, the user may request for property trp:eol which is not extracted from the carbonMessage.
+     * In Regex mode, the user may request for property trp:eof which is not extracted from the carbonMessage.
      * @param eof Whether EOF reached or not
      * @return required properties array
      */

component/src/test/java/io/siddhi/extension/io/file/FileSourceBinaryChunkedModeTestCase.java

@@ -92,15 +92,15 @@ public void siddhiIoFileTestForBinaryChunkedWithDirUri() throws InterruptedExcep
                 "move.after.process='file:/" + moveAfterProcessDir + "', " +
                 "tailing='false', " +
                 "@map(type='binaryPassThrough', " +
-                "@attributes(buffer='0', fileName = 'trp:file.name', " +
+                "@attributes(buffer='0', eof = 'trp:eof', fileName = 'trp:file.name', " +
                 "sequenceNumber = 'trp:sequence.number', length = 'trp:content.length')))\n" +
-                "define stream FooStream (buffer object, fileName string, sequenceNumber int, " +
+                "define stream FooStream (buffer object, eof bool, fileName string, sequenceNumber int, " +
                 "length int);\n" +
                 "@sink(type='log')\n" +
-                "define stream BarStream (fileName string, sequenceNumber int, length int); ";
+                "define stream BarStream (eof bool, fileName string, sequenceNumber int, length int); ";
         String query = "" +
                 "from FooStream " +
-                "select fileName, sequenceNumber, length " +
+                "select eof, fileName, sequenceNumber, length " +
                 "insert into BarStream; ";
         SiddhiManager siddhiManager = new SiddhiManager();
         SiddhiAppRuntime siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(streams + query);
@@ -127,7 +127,7 @@ public void receive(Event[] events) {
         siddhiAppRuntime.shutdown();
     }
 
-    @Test(expectedExceptions = SiddhiAppCreationException.class)
+    @Test
     public void siddhiIoFileTestForBinaryChunkedWithFileUri() throws InterruptedException {
         log.info("Siddhi IO File Test with binary.chunked mode and binaryPassThrough Mapper with File Uri");
         File file = new File(dirUri + "/binary/apache.bin");

component/src/test/java/io/siddhi/extension/io/file/FileSourceLineModeTestCase.java

@@ -1003,7 +1003,7 @@ public void siddhiIoFileReadLineInBinaryFileTest() {
         siddhiAppRuntime.start();
     }
 
-    @Test(expectedExceptions = SiddhiAppCreationException.class)
+    @Test
     public void siddhiIoFileTestForEOFAndFileName() throws InterruptedException {
         log.info("test SiddhiIoFile [mode=line] Test for EOF and File Name");
         String streams = "" +
@@ -1046,7 +1046,7 @@ public void receive(Event[] events) {
         siddhiAppRuntime.shutdown();
     }
 
-    @Test(expectedExceptions = SiddhiAppCreationException.class)
+    @Test
     public void siddhiIoFileTestForSkipHeader() throws InterruptedException {
         log.info("test SiddhiIoFile header.present parameter Test");
         String streams = "" +
@@ -1082,7 +1082,7 @@ public void receive(Event[] events) {
         siddhiAppRuntime.shutdown();
     }
 
-    @Test(expectedExceptions = SiddhiAppCreationException.class)
+    @Test
     public void siddhiIoFileTestWithoutSkipHeader() throws InterruptedException {
         log.info("test SiddhiIoFile without header.present parameter Test");
         String streams = "" +