Add a script comparing normal and batch modes.

Utils/Dataflow/data4es/batch_stage/inp

@@ -0,0 +1,10 @@
+{}
+{"!@#$": true}
+{"df": 1}
+{"df": "abc"}
+{"df": "abc1"}
+{"df": "absdfg"}
+{"df": "Some text here."}
+{"df": "abc", "fd": true}
+{"d": "abc"}
+{"df": "abc"}

Utils/Dataflow/data4es/batch_stage/run.sh

@@ -0,0 +1,15 @@
+source ../../shell_lib/eop_filter
+
+cmd="./stage.py -m s"
+cmd_batch2="./stage.py -b 2 -m s"
+cmd_batch100="./stage.py -b 100 -m s"
+
+# Various tests that should produce the same results.
+
+# Stage chains.
+# these differ by size without eop_filter at the end
+cat inp | $cmd | eop_filter | $cmd | eop_filter > outp1
+cat inp | $cmd_batch2 | eop_filter | $cmd_batch2 | eop_filter > outp2
+cat inp | $cmd_batch100 | eop_filter | $cmd_batch100 | eop_filter > outp100
+cat inp | $cmd | eop_filter | $cmd_batch2 | eop_filter > outp12
+cat inp | $cmd_batch2 | eop_filter | $cmd | eop_filter > outp21

Utils/Dataflow/data4es/batch_stage/stage.py

@@ -63,12 +63,22 @@ def main(args):
     stage.set_output_message_type(messageType.JSON)
     stage.set_default_arguments(bnc='')
 
+    # Accept batch size from command line.
+    # This is cheating because batch size is supposed to be set by
+    # stage developer, not received from command line (so,
+    # from supervisor). However, this is done in this illustrative
+    # stage to simplify a process of comparing the results of
+    # normal mode and batch mode with different batch sizes.
+    stage.add_argument('-b', action='store', type=int, help='Batch size.',
+                       default=1, dest='bsize')
+
     stage.process = process
 
     exit_code = 0
     exc_info = None
     try:
         stage.configure(args)
+        stage.set_batch_size(stage.ARGS.bsize)
         stage.run()
     except (DataflowException, RuntimeError), err:
         if str(err):