package com.ibm.streamsx.topology.internal.functional.ops;

import com.ibm.streams.operator.OperatorContext;
import com.ibm.streamsx.topology.function.FunctionContext;
import com.ibm.streamsx.topology.internal.functional.FunctionalHandler;
import com.ibm.streamsx.topology.internal.functional.StatelessFunctionalHandler;

class FunctionalOpUtils {
    
    
    static <T> FunctionalHandler<T> createFunctionHandler(OperatorContext context, FunctionContext functionContext, String functionalLogic) throws Exception {
        FunctionalHandler<T> handler = new StatelessFunctionalHandler<T>(functionContext, functionalLogic);
        return handler;
    }

}

@@ -94,32 +94,151 @@
 peratorContext;%0A
+import com.ibm.streams.operator.state.CheckpointContext;%0Aimport com.ibm.streams.operator.state.CheckpointContext.Kind;%0A
 import com.ibm.s
@@ -272,32 +272,32 @@
 unctionContext;%0A
-
 import com.ibm.s
@@ -637,16 +637,285 @@
         
+CheckpointContext cc = context.getOptionalContext(CheckpointContext.class);%0A        if (cc != null) %7B%0A            if (cc.getKind() == Kind.OPERATOR_DRIVEN)%0A                throw new IllegalStateException(); // TODO compile time checks.%0A            %0A            Stateful
 Function
@@ -942,28 +942,27 @@
  = new State
-less
+ful
 FunctionalHa
@@ -1005,16 +1005,71 @@
 Logic);%0A
+            context.registerStateHandler(handler);%0A    
         
@@ -1088,13 +1088,106 @@
 er;%0A    
+  
+  %7D%0A        return new StatelessFunctionalHandler%3CT%3E(functionContext, functionalLogic);%0A    
 %7D
-%0A
 %0A%7D%0A

Use a stateful funcion handler if checkpointing is enabled