package com.quantiply.rico.local;

import com.quantiply.rico.Configuration;
import com.quantiply.rico.Envelope;
import com.quantiply.rico.Processor;
import com.quantiply.rico.Configurator;
import com.quantiply.rico.serde.JSONStringSerde;
import com.quantiply.rico.serde.StringSerde;

import javax.script.ScriptException;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.List;

public class LocalRunner {
    private final String _configPath;
    private final List<Envelope<?>> _localCache = new ArrayList<>();
    private StringSerde<Object> _serde;
    private int BATCH_SIZE ;
    private Processor _task;
    private boolean _isWindowTriggered;

    public LocalRunner(String configPath) {
        _configPath = configPath;
    }

    public void init() throws Exception {
        Class clazz;
        // Read the config.
        Configurator cfg = new Configurator(_configPath);

        Configuration localCfg = cfg.get("local");

        System.out.println("Local Config :" + localCfg);

        _isWindowTriggered = false; //TODO - get this from config?

        // Initialize Serde
        String serdeClass = localCfg.getString("string-serde");
        clazz = Class.forName(serdeClass);
        _serde = (StringSerde<Object>) clazz.newInstance();
        _serde.init(localCfg);

        // Instantiate the processor
        String processorClass = localCfg.getString("processor.class");
        String processorName = localCfg.getString("processor.name");

        BATCH_SIZE = localCfg.getInt("batch.size");

        System.out.println(processorName + " config :" + cfg.get(processorName));

        LocalContext context = new LocalContext(cfg.get(processorName));
        clazz = Class.forName(processorClass);
        _task = (Processor) clazz.newInstance();
        _task.init(cfg.get(processorName), context);

        // TODO: Add a timer for window.
    }

    public void close() throws Exception {
        _task.shutdown();
    }

    public void run() throws Exception{
        // Read from STDIN
        BufferedReader br =
                new BufferedReader(new InputStreamReader(System.in));

        String input;

        while((input=br.readLine()) != null) {
            Envelope<Object> event = _serde.fromString(input);
            _localCache.add(event);

            if (_isWindowTriggered) {
                output(_task.window());
            }

            if (_localCache.size() >= BATCH_SIZE) {
                output(_task.process(_localCache));
                _localCache.clear();
            }
        }
    }

    private void output(List<Envelope<Object>> results) {
        if (results != null) {
            results.stream().forEach((result) -> {
                StringWriter stringWriter = new StringWriter();
                try {
                    _serde.writeTo(result, stringWriter);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
                System.out.println(stringWriter.toString());
            });
        }
    }

    public static void main(String[] args) {
        try {
            String configPath = args[0];
            LocalRunner runner = new LocalRunner(configPath);
            runner.init();
            runner.run();
        } catch (Exception e) {
            if(! (e instanceof ScriptException)){
                e.printStackTrace();
            } else {
                System.out.println(e.getMessage());
            }
        }

        // TODO: Handle Ctrl + C
    }
}

@@ -95,24 +95,28 @@
 ly.rico.
-Envelope
+Configurator
 ;%0Aimport
@@ -135,25 +135,24 @@
 ly.rico.
-Processor
+Envelope
 ;%0Aimport
@@ -167,34 +167,31 @@
 ntiply.rico.
-Configurat
+Process
 or;%0Aimport c
@@ -218,12 +218,8 @@
 rde.
-JSON
 Stri
@@ -238,44 +238,56 @@
 ort 
-com.quantiply.rico.serde.StringSerde
+org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory
 ;%0A%0Ai
@@ -495,16 +495,98 @@
 unner %7B%0A
+    private final static Logger LOG = LoggerFactory.getLogger(LocalRunner.class);%0A
     priv
@@ -1118,34 +1118,24 @@
         
-System.out.println
+LOG.info
 (%22Local 
@@ -1696,63 +1696,51 @@
     
-System.out.println(processorName + %22 config :%22 + cfg.ge
+LOG.info(%22Processor %5B%25s%5D Config : %25s%22.forma
 t(pr
@@ -1746,24 +1746,34 @@
 rocessorName
+, localCfg
 ));%0A%0A       
@@ -3420,32 +3420,40 @@
 );%0A        %7D
+%0A       
  catch (
 Exception e)
@@ -3444,54 +3444,8 @@
 ch (
-Exception e) %7B%0A            if(! (e instanceof 
 Scri
@@ -3459,16 +3459,14 @@
 tion
-))%7B%0A    
+ e) %7B%0A
     
@@ -3506,25 +3506,40 @@
     
+%7D%0A
     
-%7D else %7B%0A    
+    catch (Exception e) %7B%0A
     
@@ -3545,35 +3545,35 @@
          System.
-out
+err
 .println(e.getMe
@@ -3565,16 +3565,56 @@
 println(
+%22Unexpected exception in LocalRunner%22 + 
 e.getMes
@@ -3626,33 +3626,52 @@
 ));%0A            
-%7D
+e.printStackTrace();
 %0A        %7D%0A%0A    

clean up