file-based registry for serialized functions

README.md

@@ -1,31 +1,19 @@
-Example Stream Application:
+Register a Function:
 
 ```
-java -jar spring-cloud-function-stream-1.0.0.BUILD-SNAPSHOT.jar
-     --server.port=8081
-     --spring.cloud.stream.bindings.input.destination=words
-     --spring.cloud.stream.bindings.output.destination=uppercaseWords
-     --function.name=uppercase
-     --function.code="f -> f.map(s -> s.toString().toUpperCase())"
+./register.sh -n uppercase -f "f->f.map(s->s.toString().toUpperCase())"
 ```
 
-Example REST Application:
+Run a Stream Processing Microservice using that Function:
 
 ```
-java -jar spring-cloud-function-web-1.0.0.BUILD-SNAPSHOT.jar
-     --web.path=/demo
-     --function.name=uppercase
-     --function.code="f -> f.map(s -> s.toString().toUpperCase())"
+./stream.sh -i words -o uppercaseWords -f uppercase
 ```
 
-Using Scripts:
+Run a REST Microservice using that Function:
 
 ```
-FUN="f->f.map(s->s.toString().toUpperCase())"
-
-./stream.fun -i words -o uppercaseWords -f $FUN
-
-./web.fun -p /words -f $FUN
+./web.sh -p /words -f uppercase
 ```
 
 (more docs soon)

scripts/register.sh

@@ -0,0 +1,16 @@
+#!/bin/bash
+
+while getopts ":n:f:" opt; do
+  case $opt in
+    n)
+      NAME=$OPTARG
+      ;;
+    f)
+      FUNC=$OPTARG
+      ;;
+    esac
+done
+
+java -jar ../spring-cloud-function-core/target/spring-cloud-function-core-1.0.0.BUILD-SNAPSHOT-registrar.jar\
+ $NAME\
+ $FUNC

scripts/stream.sh

@@ -17,6 +17,4 @@ done
 java -jar ../spring-cloud-function-stream/target/spring-cloud-function-stream-1.0.0.BUILD-SNAPSHOT.jar\
  --spring.cloud.stream.bindings.input.destination=$IN\
  --spring.cloud.stream.bindings.output.destination=$OUT\
- --function.name=func\
- --function.code=$FUNC
-
+ --function.name=$FUNC

scripts/web.sh

@@ -13,6 +13,5 @@ done
 
 java -jar ../spring-cloud-function-web/target/spring-cloud-function-web-1.0.0.BUILD-SNAPSHOT.jar\
  --web.path=$WEBPATH\
- --function.name=func\
- --function.code=$FUNC
+ --function.name=$FUNC
 

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/CompiledFunctionFactory.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2016 the original author or authors.
+ * 
+ * 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 org.springframework.cloud.function.compiler;
+
+import java.util.List;
+import java.util.function.Function;
+
+import org.springframework.cloud.function.compiler.java.CompilationResult;
+
+/**
+ * @author Mark Fisher
+ */
+public class CompiledFunctionFactory<T, R> implements FunctionFactory<T, R> {
+
+	private final Function<T, R> function;
+
+	private final byte[] generatedClassBytes;
+
+	public CompiledFunctionFactory(CompilationResult compilationResult) {
+		List<Class<?>> clazzes = compilationResult.getCompiledClasses();
+		Function<T, R> function = null;
+		for (Class<?> clazz: clazzes) {
+			if (clazz.getName().equals(FunctionCompiler.GENERATED_FUNCTION_FACTORY_CLASS_NAME)) {
+				try {
+					@SuppressWarnings("unchecked")
+					FunctionFactory<T, R> functionFactory = (FunctionFactory<T, R>) clazz.newInstance();
+					function = functionFactory.getFunction();
+				}
+				catch (Exception e) {
+					throw new IllegalArgumentException("Unexpected problem during retrieval of Function from compiled class", e);
+				}
+			}
+		}
+		if (function == null) {
+			throw new IllegalArgumentException("Failed to extract Function from compilation result.");
+		}
+		this.function = function;
+		this.generatedClassBytes = compilationResult.getClassBytes(FunctionCompiler.GENERATED_FUNCTION_FACTORY_CLASS_NAME);
+	}
+
+	public Function<T, R> getFunction() {
+		return function;
+	}
+
+	public byte[] getGeneratedClassBytes() {
+		return generatedClassBytes;
+	}
+}

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/FunctionCompiler.java

@@ -17,7 +17,6 @@
 package org.springframework.cloud.function.compiler;
 
 import java.util.List;
-import java.util.function.Function;
 import java.util.regex.Matcher;
 
 import org.slf4j.Logger;
@@ -34,6 +33,10 @@ import org.springframework.cloud.function.compiler.java.RuntimeJavaCompiler;
  */
 public class FunctionCompiler {
 
+	private final static String PACKAGE = "org.springframework.cloud.function.compiler";
+
+	public final static String GENERATED_FUNCTION_FACTORY_CLASS_NAME = PACKAGE + ".GeneratedFunctionFactory";
+
 	private static Logger logger = LoggerFactory.getLogger(FunctionCompiler.class);
 
 	// Newlines in the property are escaped
@@ -42,10 +45,6 @@ public class FunctionCompiler {
 	// Individual double-quote characters are represented by two double quotes in the DSL
 	private static final String DOUBLE_DOUBLE_QUOTE = Matcher.quoteReplacement("\"\"");
 
-	private final static String PACKAGE = "org.springframework.cloud.function.compiler";
-
-	private final static String MAIN_COMPILED_CLASS_NAME = PACKAGE + ".GeneratedFunctionFactory";
-
 	/**
 	 * The user supplied code snippet is inserted into the template and then the result is compiled
 	 */
@@ -63,7 +62,7 @@ public class FunctionCompiler {
 	private final RuntimeJavaCompiler compiler = new RuntimeJavaCompiler();
 
 	/**
-	 * Produce a Function instance by:<ul>
+	 * Produce a CompiledFunctionFactory instance by:<ul>
 	 * <li>Decoding the code String to process any newlines/double-double-quotes
 	 * <li>Insert the code into the source code template for a class
 	 * <li>Compiling the class using the JDK provided Java Compiler
@@ -72,35 +71,21 @@ public class FunctionCompiler {
 	 * <li>Returning that instance.
 	 * </ul>
 	 *
-	 * @return a Function instance
+	 * @return a CompiledFunctionFactory instance
 	 */
-	public <T, R> Function<T, R> compile(String code) {
+	public <T, R> CompiledFunctionFactory<T, R> compile(String code) {
 		logger.info("Initial code property value :'{}'", code);
  		code = decode(code);
  		if (code.startsWith("\"") && code.endsWith("\"")) {
  			code = code.substring(1,code.length()-1);
  		}
  		if (!code.startsWith("return ") && !code.endsWith(";")) {
- 			code = "return " + code + ";";
+ 			code = "return (Function<Flux<Object>,Flux<Object>> & java.io.Serializable) " + code + ";";
  		}
 		logger.info("Processed code property value :\n{}\n", code);
 		CompilationResult compilationResult = buildAndCompileSourceCode(code);
 		if (compilationResult.wasSuccessful()) {
-			List<Class<?>> clazzes = compilationResult.getCompiledClasses();
-			logger.info("Compilation resulted in #{} classes", clazzes.size());
-			for (Class<?> clazz: clazzes) {
-				if (clazz.getName().equals(MAIN_COMPILED_CLASS_NAME)) {
-					try {
-						FunctionFactory functionFactory = (FunctionFactory) clazz.newInstance();
-						return functionFactory.getFunction();
-					}
-					catch (Exception e) {
-						logger.error("Unexpected problem during retrieval of Function from compiled class", e);
-					}
-				}
-				System.out.println(clazz.getName());
-			}
-			logger.error("Failed to find the expected compiled class");
+			return new CompiledFunctionFactory<>(compilationResult);
 		}
 		List<CompilationMessage> compilationMessages = compilationResult.getCompilationMessages();
 		throw new CompilationFailedException(compilationMessages);
@@ -119,7 +104,7 @@ public class FunctionCompiler {
 	 */
 	private CompilationResult buildAndCompileSourceCode(String methodBody) {
 		String sourceCode = makeSourceClassDefinition(methodBody);
-		return compiler.compile(MAIN_COMPILED_CLASS_NAME, sourceCode);
+		return compiler.compile(GENERATED_FUNCTION_FACTORY_CLASS_NAME, sourceCode);
 	}
 
 	private static String decode(String input) {
@@ -133,7 +118,7 @@ public class FunctionCompiler {
 	 * @param methodBody the code to insert into the Reactive source class template
 	 * @return a complete Java Class definition
 	 */
-	public static String makeSourceClassDefinition(String methodBody) {
+	private static String makeSourceClassDefinition(String methodBody) {
 		return String.format(SOURCE_CODE_TEMPLATE, methodBody);
 	}
 

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/FunctionFactory.java

@@ -21,8 +21,8 @@ import java.util.function.Function;
 /**
  * @author Mark Fisher
  */
-public interface FunctionFactory {
+public interface FunctionFactory<T, R> {
 
-	<T, R> Function<T, R> getFunction();
+	Function<T, R> getFunction();
 
 }

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/java/CompilationResult.java

@@ -18,7 +18,9 @@ package org.springframework.cloud.function.compiler.java;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Holder for the results of compilation. If compilation was successful the set
@@ -28,6 +30,7 @@ import java.util.List;
  * warning messages collected.
  * 
  * @author Andy Clement
+ * @author Mark Fisher
  */
 public class CompilationResult {
 
@@ -37,10 +40,20 @@ public class CompilationResult {
 
 	List<Class<?>> compiledClasses = new ArrayList<>();
 
+	private Map<String, byte[]> classBytes = new HashMap<>();
+
 	public CompilationResult(boolean successfulCompilation) {
 		this.successfulCompilation = successfulCompilation;
 	}
 
+	public void addClassBytes(String name, byte[] bytes) {
+		this.classBytes.put(name, bytes);
+	}
+
+	public byte[] getClassBytes(String classname) {
+		return this.classBytes.get(classname);
+	}
+
 	public boolean wasSuccessful() {
 		return successfulCompilation;
 	}

spring-cloud-function-compiler/src/main/java/org/springframework/cloud/function/compiler/java/RuntimeJavaCompiler.java

@@ -94,6 +94,7 @@ public class RuntimeJavaCompiler {
 				for (CompiledClassDefinition ccd: ccds) {
 					Class<?> clazz = ccl.defineClass(ccd.getClassName(), ccd.getBytes());
 					classes.add(clazz);
+					compilationResult.addClassBytes(ccd.getClassName(), ccd.getBytes());
 				}
 			} catch (IOException ioe) {
 				logger.debug("Unexpected exception defining classes",ioe);

spring-cloud-function-core/pom.xml

@@ -33,4 +33,15 @@
 		</dependency>
 	</dependencies>
 
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.springframework.boot</groupId>
+				<artifactId>spring-boot-maven-plugin</artifactId>
+				<configuration>
+					<classifier>registrar</classifier>
+				</configuration>
+			</plugin>
+		</plugins>
+	</build>
 </project>

spring-cloud-function-core/src/main/java/org/springframework/cloud/function/FunctionRegistrar.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2016 the original author or authors.
+ * 
+ * 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 org.springframework.cloud.function;
+
+import org.springframework.cloud.function.registry.FileSystemFunctionRegistry;
+
+/**
+ * @author Mark Fisher
+ */
+public class FunctionRegistrar {
+
+	public static void main(String[] args) {
+		if (args.length != 2) {
+			System.err.println("USAGE: java FunctionRegistrar functionName functionBody");
+			System.exit(1);
+		}
+		FileSystemFunctionRegistry registry = new FileSystemFunctionRegistry();
+		registry.register(args[0], args[1]);
+	}
+}

spring-cloud-function-core/src/main/java/org/springframework/cloud/function/registry/FileSystemFunctionRegistry.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2016 the original author or authors.
+ * 
+ * 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 org.springframework.cloud.function.registry;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.function.Function;
+
+import org.springframework.cloud.function.compiler.CompiledFunctionFactory;
+import org.springframework.util.Assert;
+import org.springframework.util.FileCopyUtils;
+
+/**
+ * @author Mark Fisher
+ */
+public class FileSystemFunctionRegistry extends FunctionRegistrySupport {
+
+	private final File directory;
+
+	public FileSystemFunctionRegistry() {
+		this(new File("/tmp/function-registry"));
+	}
+
+	public FileSystemFunctionRegistry(File directory) {
+		Assert.notNull(directory, "Directory must not be null");
+		if (!directory.exists()) {
+			directory.mkdirs();
+		}
+		else {
+			Assert.isTrue(directory.isDirectory(),
+					String.format("%s is not a directory.", directory.getAbsolutePath()));
+		}
+		this.directory = directory;
+	}
+
+	@Override
+	public <T, R> Function<T, R> lookup(String name) {
+		try {
+			byte[] bytes = FileCopyUtils.copyToByteArray(new File(this.directory, fileName(name)));
+			return this.deserialize(bytes);
+		}
+		catch (IOException e) {
+			throw new IllegalArgumentException(String.format("failed to lookup function: %s", name), e); 
+		}
+	}
+
+	public void register(String name, String function) {
+		CompiledFunctionFactory<?, ?> factory = this.compile(function);
+		File file = new File(this.directory, fileName(name));
+		try {
+			FileCopyUtils.copy(factory.getGeneratedClassBytes(), file);
+		}
+		catch (IOException e) {
+			throw new IllegalArgumentException(String.format("failed to register function: %s", name), e);
+		}
+	}
+
+	private static String fileName(String functionName) {
+		return String.format("%s.%s", functionName, "fun");
+	}
+}

spring-cloud-function-core/src/main/java/org/springframework/cloud/function/registry/FunctionRegistry.java

@@ -23,14 +23,10 @@ import java.util.function.Function;
  */
 public interface FunctionRegistry {
 
-	<T, R> Function<T, R> lookup(String name);
-
-	void register(String name, Function<?, ?> function);
-
 	void register(String name, String function);
 
-	void compose(String composedFunctionName, Function<?, ?>... functions);
+	<T, R> Function<T, R> lookup(String name);
 
-	void compose(String composedFunctionName, String... functionNames);
+	<T, R> Function<T, R> compose(String... functionNames);
 
 }

spring-cloud-function-core/src/main/java/org/springframework/cloud/function/registry/FunctionRegistrySupport.java

@@ -18,8 +18,10 @@ package org.springframework.cloud.function.registry;
 
 import java.util.function.Function;
 
+import org.springframework.cloud.function.compiler.CompiledFunctionFactory;
 import org.springframework.cloud.function.compiler.FunctionCompiler;
-import org.springframework.util.Assert;
+import org.springframework.cloud.function.compiler.FunctionFactory;
+import org.springframework.cloud.function.compiler.java.SimpleClassLoader;
 
 /**
  * @author Mark Fisher
@@ -28,31 +30,31 @@ public abstract class FunctionRegistrySupport implements FunctionRegistry {
 
 	private final FunctionCompiler compiler = new FunctionCompiler();
 
+	private final SimpleClassLoader classLoader = new SimpleClassLoader(FunctionRegistrySupport.class.getClassLoader());
+
 	@Override
-	public void register(String name, String code) {
-		Function<?, ?> function = compiler.compile(code);
-		this.register(name, function);
-	}
-
 	@SuppressWarnings("unchecked")
-	public void compose(String name, Function<?, ?>... functions) {
-		Assert.isTrue(functions != null && functions.length > 1, "more than one Function is required");
-		@SuppressWarnings("rawtypes")
-		Function function = functions[0];
-		for (int i = 1; i < functions.length; i++) {
-			function = function.andThen(functions[i]);
-		}
-		this.register(name, function);
-	}
-
-	@SuppressWarnings("unchecked")
-	public void compose(String composedFunctionName, String... functionNames) {
-		Assert.isTrue(functionNames != null && functionNames.length > 1, "more than one Function is required");
+	public <T, R> Function<T, R> compose(String... functionNames) {
 		@SuppressWarnings("rawtypes")
 		Function function = this.lookup(functionNames[0]);
 		for (int i = 1; i < functionNames.length; i++) {
 			function = function.andThen(this.lookup(functionNames[i]));
 		}
-		this.register(composedFunctionName, function);
+		return function;
+	}
+
+	protected <T, R> CompiledFunctionFactory<T, R> compile(String code) {
+		return this.compiler.compile(code);
+	}
+
+	@SuppressWarnings("unchecked")
+	protected <T, R> Function<T, R> deserialize(byte[] bytes) {
+		Class<?> factoryClass = this.classLoader.defineClass(FunctionCompiler.GENERATED_FUNCTION_FACTORY_CLASS_NAME, bytes);
+		try {
+			return ((FunctionFactory<T, R>) factoryClass.newInstance()).getFunction();
+		}
+		catch (InstantiationException | IllegalAccessException e) {
+			throw new IllegalArgumentException("failed to deserialize function", e);
+		}
 	}
 }

spring-cloud-function-core/src/main/java/org/springframework/cloud/function/registry/InMemoryFunctionRegistry.java

@@ -27,12 +27,13 @@ public class InMemoryFunctionRegistry extends FunctionRegistrySupport {
 	private final ConcurrentHashMap<String, Function<?, ?>> map = new ConcurrentHashMap<>();
 
 	@Override
+	@SuppressWarnings("unchecked")
 	public Function<?, ?> lookup(String name) {
 		return this.map.get(name);
 	}
 
 	@Override
-	public void register(String name, Function<?, ?> function) {
-		this.map.put(name, function);
+	public void register(String name, String function) {
+		this.map.put(name, this.compile(function).getFunction());
 	}
 }

spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/FunctionConfigurationProperties.java

@@ -22,12 +22,10 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
  * @author Mark Fisher
  */
 @ConfigurationProperties(prefix = "function")
-public class StreamConfigurationProperties {
+public class FunctionConfigurationProperties {
 
 	private String name;
 
-	private String code;
-
 	public String getName() {
 		return name;
 	}
@@ -35,12 +33,4 @@ public class StreamConfigurationProperties {
 	public void setName(String name) {
 		this.name = name;
 	}
-
-	public String getCode() {
-		return code;
-	}
-
-	public void setCode(String code) {
-		this.code = code;
-	}
 }

spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamConfiguration.java

@@ -19,8 +19,8 @@ package org.springframework.cloud.function.stream;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.function.invoker.AbstractFunctionInvoker;
+import org.springframework.cloud.function.registry.FileSystemFunctionRegistry;
 import org.springframework.cloud.function.registry.FunctionRegistry;
-import org.springframework.cloud.function.registry.InMemoryFunctionRegistry;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.messaging.Processor;
 import org.springframework.context.annotation.Bean;
@@ -29,17 +29,15 @@ import org.springframework.context.annotation.Bean;
  * @author Mark Fisher
  */
 @EnableBinding(Processor.class)
-@EnableConfigurationProperties(StreamConfigurationProperties.class)
+@EnableConfigurationProperties(FunctionConfigurationProperties.class)
 public class StreamConfiguration {
 
 	@Autowired
-	private StreamConfigurationProperties properties;
+	private FunctionConfigurationProperties properties;
 
 	@Bean
 	public FunctionRegistry registry() {
-		FunctionRegistry registry = new InMemoryFunctionRegistry();
-		registry.register(properties.getName(), properties.getCode());
-		return registry;
+		return new FileSystemFunctionRegistry();
 	}
 
 	@Bean

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RestConfiguration.java

@@ -30,6 +30,7 @@ import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.function.registry.FileSystemFunctionRegistry;
 import org.springframework.cloud.function.registry.FunctionRegistry;
 import org.springframework.cloud.function.registry.InMemoryFunctionRegistry;
 import org.springframework.context.annotation.Bean;
@@ -59,9 +60,7 @@ public class RestConfiguration {
 
 	@Bean
 	public FunctionRegistry registry() {
-		FunctionRegistry registry = new InMemoryFunctionRegistry();
-		registry.register(functionProperties.getName(), functionProperties.getCode());
-		return registry;
+		return new FileSystemFunctionRegistry();
 	}
 
 	@Bean