INT-1507, INT-1545: Add groovy processor for payloads

- Remove unneeded config file
- Upgrade groovy
- fix potential permgen leak in groovy payload processor
- Test concurrent execution

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractScriptExecutingMessageProcessor.java

@@ -1,17 +1,14 @@
 /*
  * Copyright 2002-2010 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.
+ * 
+ * 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.integration.handler;
@@ -19,7 +16,6 @@ package org.springframework.integration.handler;
 import org.springframework.integration.Message;
 import org.springframework.integration.MessageHandlingException;
 import org.springframework.scripting.ScriptSource;
-import org.springframework.util.Assert;
 
 /**
  * Base {@link MessageProcessor} for scripting implementations to extend.
@@ -29,33 +25,29 @@ import org.springframework.util.Assert;
  */
 public abstract class AbstractScriptExecutingMessageProcessor<T> implements MessageProcessor<T> {
 
-	private final ScriptSource scriptSource;
-
-
-	/**
-	 * Create a processor for the given {@link ScriptSource}.
-	 */
-	public AbstractScriptExecutingMessageProcessor(ScriptSource scriptSource) {
-		Assert.notNull(scriptSource, "scriptSource must not be null");
-		this.scriptSource = scriptSource;
-	}
-
-
 	/**
 	 * Executes the script and returns the result.
 	 */
 	public final T processMessage(Message<?> message) {
 		try {
-			return this.executeScript(this.scriptSource, message);
-		}
-		catch (Exception e) {
+			return this.executeScript(getScriptSource(message), message);
+		} catch (Exception e) {
 			throw new MessageHandlingException(message, "failed to execute script", e);
 		}
 	}
 
 	/**
-	 * Subclasses must implement this method. In doing so, the execution context for the
-	 * script should be populated with the Message's 'payload' and 'headers' as variables. 
+	 * Subclasses must implement this method to create a script source, optionally using the message to locate or
+	 * create the script.
+	 * 
+	 * @param message the message being processed
+	 * @return a ScriptSource to use to create a script
+	 */
+	protected abstract ScriptSource getScriptSource(Message<?> message);
+
+	/**
+	 * Subclasses must implement this method. In doing so, the execution context for the script should be populated with
+	 * the Message's 'payload' and 'headers' as variables.
 	 */
 	protected abstract T executeScript(ScriptSource scriptSource, Message<?> message) throws Exception;
 

spring-integration-groovy/.springBeans

@@ -1,13 +1,13 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beansProjectDescription>
-	<version>1</version>
-	<pluginVersion><![CDATA[2.3.3.201005160334-M1]]></pluginVersion>
-	<configSuffixes>
-		<configSuffix><![CDATA[xml]]></configSuffix>
-	</configSuffixes>
-	<enableImports><![CDATA[false]]></enableImports>
-	<configs>
-	</configs>
-	<configSets>
-	</configSets>
-</beansProjectDescription>
+<?xml version="1.0" encoding="UTF-8"?>
+<beansProjectDescription>
+	<version>1</version>
+	<pluginVersion><![CDATA[2.5.0.201008251000-M3]]></pluginVersion>
+	<configSuffixes>
+		<configSuffix><![CDATA[xml]]></configSuffix>
+	</configSuffixes>
+	<enableImports><![CDATA[false]]></enableImports>
+	<configs>
+	</configs>
+	<configSets>
+	</configSets>
+</beansProjectDescription>

spring-integration-groovy/pom.xml

@@ -15,7 +15,7 @@
 		<dependency>
 			<groupId>org.codehaus.groovy</groupId>
 			<artifactId>groovy-all</artifactId>
-			<version>1.7.3</version>
+			<version>1.7.5</version>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.integration</groupId>

spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessor.java

@@ -16,15 +16,11 @@
 
 package org.springframework.integration.groovy;
 
-import groovy.lang.Binding;
 import groovy.lang.GString;
-import groovy.lang.GroovyObject;
-import groovy.lang.Script;
 
 import org.springframework.integration.Message;
 import org.springframework.integration.handler.AbstractScriptExecutingMessageProcessor;
 import org.springframework.scripting.ScriptSource;
-import org.springframework.scripting.groovy.GroovyObjectCustomizer;
 import org.springframework.scripting.groovy.GroovyScriptFactory;
 import org.springframework.util.Assert;
 
@@ -39,9 +35,15 @@ public class GroovyScriptExecutingMessageProcessor extends AbstractScriptExecuti
 
 	private final MessageContextBindingCustomizer customizer = new MessageContextBindingCustomizer();
 
+	private final ScriptSource scriptSource;
 
+
+	/**
+	 * Create a processor for the given {@link ScriptSource}.
+	 */
 	public GroovyScriptExecutingMessageProcessor(ScriptSource scriptSource) {
-		super(scriptSource);
+		Assert.notNull(scriptSource, "scriptSource must not be null");
+		this.scriptSource = scriptSource;
 		this.scriptFactory = new GroovyScriptFactory(this.getClass().getSimpleName(), this.customizer);
 	}
 
@@ -55,23 +57,9 @@ public class GroovyScriptExecutingMessageProcessor extends AbstractScriptExecuti
 		}
 	}
 
-
-	private static class MessageContextBindingCustomizer implements GroovyObjectCustomizer {
-
-		private volatile Message<?> message;
-
-		public void setMessage(Message<?> message) {
-			this.message = message;
-		}
-
-		public void customize(GroovyObject goo) {
-			Assert.state(goo instanceof Script, "Expected a Script");
-			if (this.message != null) {
-				Binding binding = ((Script) goo).getBinding();
-				binding.setVariable("payload", this.message.getPayload());
-				binding.setVariable("headers", this.message.getHeaders());
-			}
-		}
+	@Override
+	protected ScriptSource getScriptSource(Message<?> message) {
+		return scriptSource;
 	}
 
 }

spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyScriptPayloadMessageProcessor.java

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2002-2010 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.integration.groovy;
+
+import groovy.lang.GString;
+
+import java.util.Map;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.handler.AbstractScriptExecutingMessageProcessor;
+import org.springframework.scripting.ScriptSource;
+import org.springframework.scripting.groovy.GroovyScriptFactory;
+import org.springframework.scripting.support.StaticScriptSource;
+import org.springframework.util.Assert;
+
+/**
+ * @author Dave Syer
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public class GroovyScriptPayloadMessageProcessor extends AbstractScriptExecutingMessageProcessor<Object> {
+
+	private final Map<String, ?> map;
+
+	public GroovyScriptPayloadMessageProcessor() {
+		this(null);
+	}
+
+	public GroovyScriptPayloadMessageProcessor(Map<String, ?> map) {
+		this.map = map;
+	}
+
+	@Override
+	protected ScriptSource getScriptSource(Message<?> message) {
+		Object payload = message.getPayload();
+		Assert.isInstanceOf(String.class, payload, "Payload must be String containing Groovy script.");
+		String className = generateScriptName(message);
+		return new StaticScriptSource((String) payload, className);
+	}
+
+	@Override
+	protected Object executeScript(ScriptSource scriptSource, Message<?> message) throws Exception {
+		// Keeping everything local prevents PermGen (class instances) leaks...
+		MessageContextBindingCustomizer bindingCustomizer = new MessageContextBindingCustomizer(this.map);
+		bindingCustomizer.setMessage(message);
+		GroovyScriptFactory scriptFactory = new GroovyScriptFactory(this.getClass().getSimpleName(), bindingCustomizer);
+		Object result = scriptFactory.getScriptedObject(scriptSource, null);
+		return (result instanceof GString) ? result.toString() : result;
+	}
+	
+	protected String generateScriptName(Message<?> message) {
+		// Don't use the same script (class) name for all invocations by default
+		return getClass().getSimpleName() + message.getHeaders().getId().toString().replaceAll("-", "");
+	}
+
+}

spring-integration-groovy/src/main/java/org/springframework/integration/groovy/MessageContextBindingCustomizer.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2002-2010 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.integration.groovy;
+
+import groovy.lang.Binding;
+import groovy.lang.GroovyObject;
+import groovy.lang.Script;
+
+import java.util.Map;
+
+import org.springframework.integration.Message;
+import org.springframework.scripting.groovy.GroovyObjectCustomizer;
+import org.springframework.util.Assert;
+
+/**
+ * A groovy object customizer used internally by the groovy message processors. Not public because the customizer is not
+ * the best API for groovy context binding, but it's what we have in Spring right now.
+ * 
+ * @since 2.0
+ * @author Dave Syer
+ * 
+ */
+class MessageContextBindingCustomizer implements GroovyObjectCustomizer {
+
+	private volatile Message<?> message;
+	private final GroovyObjectCustomizer customizer;
+
+	public MessageContextBindingCustomizer() {
+		this((GroovyObjectCustomizer) null);
+	}
+
+	public MessageContextBindingCustomizer(Map<String, ?> context) {
+		this(new MapContextBindingCustomizer(context));
+	}
+
+	public MessageContextBindingCustomizer(GroovyObjectCustomizer customizer) {
+		this.customizer = customizer;
+	}
+
+	public void setMessage(Message<?> message) {
+		this.message = message;
+	}
+
+	public void customize(GroovyObject goo) {
+		Assert.state(goo instanceof Script, "Expected a Script");
+		if (customizer != null) {
+			customizer.customize(goo);
+		}
+		if (this.message != null) {
+			Binding binding = ((Script) goo).getBinding();
+			binding.setVariable("payload", this.message.getPayload());
+			binding.setVariable("headers", this.message.getHeaders());
+		}
+	}
+
+	private static class MapContextBindingCustomizer implements GroovyObjectCustomizer {
+
+		private final Map<String, ?> map;
+
+		public MapContextBindingCustomizer(Map<String, ?> map) {
+			this.map = map;
+		}
+
+		public void customize(GroovyObject goo) {
+			Assert.state(goo instanceof Script, "Expected a Script");
+			if (this.map != null) {
+				Binding binding = ((Script) goo).getBinding();
+				for (String key : map.keySet()) {
+					binding.setVariable(key, map.get(key));
+				}
+			}
+
+		}
+
+	}
+}

spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyScriptExecutingMessageProcessorTests.java

@@ -22,7 +22,9 @@ import static org.junit.Assert.assertFalse;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.concurrent.atomic.AtomicInteger;
 
+import org.junit.Rule;
 import org.junit.Test;
 import org.springframework.core.io.AbstractResource;
 import org.springframework.integration.Message;
@@ -31,6 +33,7 @@ import org.springframework.integration.handler.MessageProcessor;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.scripting.ScriptSource;
 import org.springframework.scripting.support.ResourceScriptSource;
+import org.springframework.test.annotation.Repeat;
 
 /**
  * @author Mark Fisher
@@ -40,15 +43,22 @@ import org.springframework.scripting.support.ResourceScriptSource;
  */
 public class GroovyScriptExecutingMessageProcessorTests {
 
+	@Rule
+	public RepeatProcessor repeater = new RepeatProcessor(4);
+
+	private AtomicInteger countHolder = new AtomicInteger();
+
 	@Test
+	@Repeat(20)
 	public void testSimpleExecution() throws Exception {
+		int count = countHolder.getAndIncrement();
 		String script = "return \"payload is $payload, header is $headers.testHeader\"";
-		Message<?> message = MessageBuilder.withPayload("foo").setHeader("testHeader", "bar").build();
+		Message<?> message = MessageBuilder.withPayload("foo").setHeader("testHeader", "bar"+count).build();
 		TestResource resource = new TestResource(script, "simpleTest");
 		ScriptSource scriptSource = new ResourceScriptSource(resource);
 		MessageProcessor<Object> processor = new GroovyScriptExecutingMessageProcessor(scriptSource);
 		Object result = processor.processMessage(message);
-		assertEquals("payload is foo, header is bar", result.toString());
+		assertEquals("payload is foo, header is bar"+count, result.toString());
 	}
 
 	@Test

spring-integration-groovy/src/test/java/org/springframework/integration/groovy/GroovyScriptPayloadMessageProcessorTests.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2002-2010 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.integration.groovy;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.springframework.integration.Message;
+import org.springframework.integration.handler.MessageProcessor;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.test.annotation.Repeat;
+
+/**
+ * @author Dave Syer
+ * @since 2.0
+ */
+public class GroovyScriptPayloadMessageProcessorTests {
+
+	@Rule
+	public RepeatProcessor repeater = new RepeatProcessor(4);
+
+	private AtomicInteger countHolder = new AtomicInteger();
+
+	private GroovyScriptPayloadMessageProcessor processor = new GroovyScriptPayloadMessageProcessor();
+
+	@Test
+	@Repeat(20)
+	public void testSimpleExecution() throws Exception {
+		int count = countHolder.getAndIncrement();
+		Message<?> message = MessageBuilder.withPayload("headers.foo" + count).setHeader("foo" + count, "bar").build();
+		Object result = processor.processMessage(message);
+		assertEquals("bar", result.toString());
+	}
+
+	@Test
+	public void testDoubleExecutionWithNewScript() throws Exception {
+		Message<?> message = MessageBuilder.withPayload("headers.foo").setHeader("foo", "bar").build();
+		Object result = processor.processMessage(message);
+		assertEquals("bar", result.toString());
+		message = MessageBuilder.withPayload("headers.bar").setHeader("bar", "spam").build();
+		result = processor.processMessage(message);
+		assertEquals("spam", result.toString());
+	}
+
+	@Test
+	public void testSimpleExecutionWithContext() throws Exception {
+		Message<?> message = MessageBuilder.withPayload("\"spam is $spam foo is $headers.foo\"")
+				.setHeader("foo", "bar").build();
+		MessageProcessor<Object> processor = new GroovyScriptPayloadMessageProcessor(Collections.singletonMap("spam",
+				"bucket"));
+		Object result = processor.processMessage(message);
+		assertEquals("spam is bucket foo is bar", result.toString());
+	}
+
+}

spring-integration-groovy/src/test/java/org/springframework/integration/groovy/RepeatProcessor.java

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2002-2010 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.integration.groovy;
+
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.rules.MethodRule;
+import org.junit.runners.model.FrameworkMethod;
+import org.junit.runners.model.Statement;
+import org.springframework.core.annotation.AnnotationUtils;
+import org.springframework.test.annotation.Repeat;
+
+/**
+ * A method rule that looks at Spring repeat annotations on methods and executes the test multiple times (without
+ * re-initializing the test case).
+ * 
+ * @author Dave Syer
+ * @since 2.0
+ * 
+ */
+public class RepeatProcessor implements MethodRule {
+
+	private final int concurrency;
+
+	public RepeatProcessor(int concurrency) {
+		this.concurrency = concurrency < 0 ? 0 : concurrency;
+	}
+
+	public Statement apply(final Statement base, FrameworkMethod method, Object target) {
+		Repeat repeat = AnnotationUtils.findAnnotation(method.getMethod(), Repeat.class);
+		if (repeat == null) {
+			return base;
+		}
+		final int repeats = repeat.value();
+		if (concurrency <= 0) {
+			return new Statement() {
+				@Override
+				public void evaluate() throws Throwable {
+					for (int i = 0; i < repeats; i++) {
+						try {
+							base.evaluate();
+						} catch (Throwable t) {
+							throw new IllegalStateException("Failed on iteration: " + i, t);
+						}
+					}
+				}
+			};
+		}
+		return new Statement() {
+			@Override
+			public void evaluate() throws Throwable {
+				List<Future<Boolean>> results = new ArrayList<Future<Boolean>>();
+				ExecutorService executor = Executors.newFixedThreadPool(concurrency);
+				try {
+					for (int i = 0; i < repeats; i++) {
+						final int count = i;
+						results.add(executor.submit(new Callable<Boolean>() {
+							public Boolean call() {
+								try {
+									base.evaluate();
+								} catch (Throwable t) {
+									throw new IllegalStateException("Failed on iteration: " + count, t);
+								}
+								return true;
+							}
+						}));
+					}
+					for (Future<Boolean> future : results) {
+						assertTrue("Null result from completer", future.get(10, TimeUnit.SECONDS));
+					}
+				} finally {
+					executor.shutdownNow();
+				}
+			}
+		};
+	}
+}