//
//   Copyright 2016  Cityzen Data
//
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
//   Unless required by applicable law or agreed to in writing, software
//   distributed under the License is distributed on an "AS IS" BASIS,
//   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//   See the License for the specific language governing permissions and
//   limitations under the License.
//

package io.warp10.script;

import io.warp10.WarpConfig;
import io.warp10.script.WarpScriptStack.Macro;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Semaphore;

public class WarpScriptExecutor {

  private static final String WARP10_CONFIG = "warp10.config";
  
  public static enum StackSemantics {
    // Single stack which will be shared among calling threads
    SYNCHRONIZED,
    // One stack per thread
    PERTHREAD,
    // A new stack for every execution
    NEW,
  };
  
  /**
   * Macro this executor will tirelessly apply
   */
  private final Macro macro;

  /**
   * Single stack to use when StackSemantics is SYNCHRONIZED, null otherwise
   */
  private final WarpScriptStack stack;
  
  /**
   * Semaphore for synchronizing threads
   */
  private final Semaphore sem;

  private final StackSemantics semantics;
  
  private final Map<String,Object> symbolTable;  
  
  private static final Properties properties;
  
  private ThreadLocal<WarpScriptStack> perThreadStack = new ThreadLocal<WarpScriptStack>() {
    @Override
    protected WarpScriptStack initialValue() {
      MemoryWarpScriptStack stack = new MemoryWarpScriptStack(null, null, null, properties);
      
      try {
        stack.exec(WarpScriptLib.BOOTSTRAP);
      } catch (WarpScriptException e) {
        throw new RuntimeException(e);
      }

      return stack;
    }
  };

  static {
    //
    // Initialize WarpConfig
    //
  
    try {
      WarpConfig.setProperties((String) System.getProperty(WARP10_CONFIG));
      properties = WarpConfig.getProperties();
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }
  
  public WarpScriptExecutor(StackSemantics semantics, String script) throws WarpScriptException {
    this(semantics, script, null);
  }
  
  public WarpScriptExecutor(StackSemantics semantics, String script, Map<String,Object> symbols) throws WarpScriptException {

    this.semantics = semantics;
    
    if (StackSemantics.SYNCHRONIZED.equals(semantics)) {
      this.stack = perThreadStack.get();
      this.sem = new Semaphore(1);
    } else {
      this.stack = null;
      this.sem = new Semaphore(Integer.MAX_VALUE);
    }
    
    //
    // Attempt to execute the script code on the stack
    // to define a macro
    //
      
    WarpScriptStack stack = new MemoryWarpScriptStack(null, null, null, new Properties());    
    stack.exec(WarpScriptLib.BOOTSTRAP);

    this.symbolTable = new HashMap<String,Object>();
    
    if (null != symbols) {
      this.symbolTable.putAll(symbols);
    }
        
    StringBuilder sb = new StringBuilder();
    sb.append(WarpScriptStack.MACRO_START);
    sb.append("\n");
    sb.append(script);
    sb.append("\n");
    sb.append(WarpScriptStack.MACRO_END);
    
    stack.execMulti(sb.toString());
      
    if (1 != stack.depth()) {
      throw new WarpScriptException("Stack depth was not 1 after the code execution.");
    }
    
    if (!(stack.peek() instanceof Macro)) {
      throw new WarpScriptException("No macro was found on top of the stack.");
    }
    
    Macro macro = (Macro) stack.pop();
    
    this.macro = macro;    
  }

  public List<Object> exec(List<Object> input) throws WarpScriptException {
    try {
      this.sem.acquire();
    } catch (InterruptedException ie) {
      throw new WarpScriptException("Got interrupted while attempting to acquire semaphore.");
    }
    
    WarpScriptStack stack = this.stack;

    try {
      if (null == stack) {
        if (StackSemantics.PERTHREAD.equals(this.semantics)) {
          stack = perThreadStack.get();
        } else if (StackSemantics.NEW.equals(this.semantics)) {
          stack = new MemoryWarpScriptStack(null, null, null, properties);
          stack.exec(WarpScriptLib.BOOTSTRAP);
        } else {
          throw new WarpScriptException("Invalid stack semantics.");
        }
      }
      
      //
      // Update the symbol table if 'symbolTable' is set
      //
      
      if (null != this.symbolTable) {
        stack.save();
        MemoryWarpScriptStack.StackContext context = (MemoryWarpScriptStack.StackContext) stack.peek();
        context.symbolTable.putAll(this.symbolTable);
        stack.restore();
      }
      
      //
      // Push the parameters onto the stack
      //
      
      for (Object o: input) {
        stack.push(o);
      }
      
      //
      // Execute the macro
      //
      
      stack.exec(this.macro);
      
      //
      // Pop the output off the stack
      //
      
      List<Object> output = new ArrayList<Object>();
      
      while(stack.depth() > 0) {
        output.add(stack.pop());
      }
      
      java.util.Collections.reverse(output);
      
      return output;
    } finally {
      // Clear the stack
      stack.clear();
      this.sem.release();
    }
  }
  
  /**
   * Store a symbol in the symbol table.
   */
  public WarpScriptExecutor store(String key, Object value) {
    this.symbolTable.put(key, value);
    return this;
  }
}

@@ -710,16 +710,60 @@
 k.Macro;
+%0Aimport org.apache.hadoop.util.Progressable;
 %0A%0Aimport
@@ -1240,19 +1240,17 @@
 W,%0A  %7D;%0A
-  
 %0A
+
   /**%0A  
@@ -2385,27 +2385,25 @@
 ;%0A    %7D%0A  %7D%0A
-  
 %0A
+
   public War
@@ -2527,19 +2527,17 @@
 l);%0A  %7D%0A
-  
 %0A
+
   public
@@ -2653,16 +2653,215 @@
 eption %7B
+%0A    this(semantics,script,symbols,null);%0A  %7D%0A%0A  public WarpScriptExecutor(StackSemantics semantics, String script, Map%3CString,Object%3E symbols, Progressable progressable) throws WarpScriptException %7B
 %0A%0A    th
@@ -2985,24 +2985,116 @@
 tack.get();%0A
+      this.stack.setAttribute(WarpScriptStack.ATTRIBUTE_HADOOP_PROGRESSABLE, progressable);%0A
       this.s
@@ -3407,20 +3407,101 @@
 ties());
+%0A
     
+stack.setAttribute(WarpScriptStack.ATTRIBUTE_HADOOP_PROGRESSABLE, progressable);
 %0A    sta
@@ -5850,24 +5850,53 @@
 %7D finally %7B%0A
+      if (null != stack) %7B%0A  
       // Cle
@@ -5910,24 +5910,26 @@
 stack%0A      
+ 
+ 
 stack.clear(
@@ -5927,24 +5927,32 @@
 ck.clear();%0A
+      %7D%0A
       this.s

add Progressable parameter in constructor