/*
# Licensed Materials - Property of IBM
# Copyright IBM Corp. 2015  
 */
package com.ibm.streamsx.topology.internal.functional.ops;

import java.util.logging.Logger;

import com.ibm.streams.operator.AbstractOperator;
import com.ibm.streams.operator.OperatorContext;
import com.ibm.streams.operator.logging.TraceLevel;
import com.ibm.streams.operator.model.Parameter;
import com.ibm.streams.operator.model.SharedLoader;
import com.ibm.streamsx.topology.function.Initializable;
import com.ibm.streamsx.topology.internal.functional.FunctionalHelper;

/**
 * 
 * Common code for operators with inputs and outputs.
 * 
 */
@SharedLoader
public abstract class FunctionFunctor extends AbstractOperator implements Functional {

    public static final String FUNCTIONAL_LOGIC_PARAM = "functionalLogic";
    static final Logger trace = Logger.getLogger("com.ibm.streamsx.topology.operators");

    // parameters
    private String functionalLogic;
    private String[] jar;
    
    /**
     * Logic (function) used by this operator,
     * will be closed upon shutdown.
     */
    private Object logicInstance;

    public final String getFunctionalLogic() {
        return functionalLogic;
    }

    @Parameter
    public void setFunctionalLogic(String logic) {
        this.functionalLogic = logic;
    }

    public final String[] getJar() {
        return jar;
    }

    @Parameter(optional = true)
    public final void setJar(String[] jar) {
        this.jar = jar;
    }

    @Override
    public synchronized void initialize(OperatorContext context)
            throws Exception {
        super.initialize(context);
        FunctionalHelper.addLibraries(this, getJar());
    }
    
    @Override
    public void shutdown() throws Exception {
        closeLogic(logicInstance);
        super.shutdown();
    }
    
    public void setLogic(Object logicInstance) throws Exception {
        this.logicInstance = logicInstance;
        initializeLogic(getOperatorContext(), logicInstance);
    }
    
    static void initializeLogic(OperatorContext context, Object logicInstance) throws Exception {
        if (logicInstance instanceof Initializable) {
            ((Initializable) logicInstance).initialize(new FunctionOperatorContext(context));
        }
    }
    
    /**
     * If logicInstance implements AutoCloseable
     * then shut it down by calling it close() method.
     */
    static void closeLogic(Object logicInstance) {
        if (logicInstance instanceof AutoCloseable) {
            try {
                synchronized (logicInstance) {
                    ((AutoCloseable) logicInstance).close();
                }
            } catch (Exception e) {
                trace.log(TraceLevel.ERROR,
                        "Exception " +  e.getMessage() + " closing function instance:" + logicInstance, e);
            }
        }       
    }
}

@@ -541,16 +541,81 @@
 lHelper;
+%0Aimport com.ibm.streamsx.topology.internal.logic.WrapperFunction;
 %0A%0A/**%0A *
@@ -2150,32 +2150,55 @@
 ows Exception %7B%0A
+        for (;;) %7B%0A    
         if (logi
@@ -2227,32 +2227,36 @@
 nitializable) %7B%0A
+    
             ((In
@@ -2329,24 +2329,244 @@
 (context));%0A
+            %7D%0A            if (logicInstance instanceof WrapperFunction) %7B%0A                logicInstance = ((WrapperFunction) logicInstance).getWrappedFunction();%0A            %7D else %7B%0A                break;%0A            %7D%0A
         %7D%0A  
@@ -2737,32 +2737,55 @@
 ogicInstance) %7B%0A
+        for (;;) %7B%0A    
         if (logi
@@ -2834,16 +2834,20 @@
         
+    
 try %7B%0A  
@@ -2860,16 +2860,20 @@
         
+   
+ 
 synchron
@@ -2887,32 +2887,36 @@
 ogicInstance) %7B%0A
+    
                 
@@ -2968,34 +2968,42 @@
                 
+   
+ 
 %7D%0A
+    
             %7D ca
@@ -3038,16 +3038,20 @@
         
+    
 trace.lo
@@ -3073,115 +3073,342 @@
 ROR,
-%0A                        %22Exception %22 +  e.getMessage() + %22 closing function instance:%22 + logicInstance, e)
+ %22Exception %22 + e.getMessage()%0A                            + %22 closing function instance:%22 + logicInstance, e);%0A                %7D%0A            %7D%0A            if (logicInstance instanceof WrapperFunction) %7B%0A                logicInstance = ((WrapperFunction) logicInstance).getWrappedFunction();%0A            %7D else %7B%0A                break
 ;%0A  

Fix #152 Initializable/AutoCloseable now called for wrapped functions