diff --git a/function/task-launch-request-function/README.adoc b/function/task-launch-request-function/README.adoc new file mode 100644 index 00000000..c3e8b65e --- /dev/null +++ b/function/task-launch-request-function/README.adoc @@ -0,0 +1,22 @@ +# Task Launch Request Function + +This module provides a function that can be reused and composed in other applications to transform the output to a link:src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequest.java[TaskLaunchRequest] +that can be used as input to the Tasklauncher function to launch a task. + +## Beans for injection + +You can import the `TaskLaunchRequestFunctionConfiguration` in a Spring Boot application and then inject the following bean. + +`taskLaunchRequestFunction` as a link:src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunction.java[TaskLaunchRequestFunction]. + +You can use `taskLaunchRequestFunction` as a qualifier when injecting. + +Once injected, you can use the `apply` method of the `Function` to invoke it and get the result. + +## Configuration Options + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionProperties.java[TaskLaunchRequestFunctionProperties.java] + +## Examples + +See this link:src/test/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionApplicationTests.java[test suite] for examples of how this function is used. diff --git a/function/task-launch-request-function/pom.xml b/function/task-launch-request-function/pom.xml new file mode 100644 index 00000000..2ce59d34 --- /dev/null +++ b/function/task-launch-request-function/pom.xml @@ -0,0 +1,37 @@ + + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + 4.0.0 + + task-launch-request-function + task-launch-request-function + + + + org.springframework.integration + spring-integration-core + + + org.springframework.boot + spring-boot-starter-json + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-starter-test + test + + + + diff --git a/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/CommandLineArgumentsMessageMapper.java b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/CommandLineArgumentsMessageMapper.java new file mode 100644 index 00000000..8cac1b0f --- /dev/null +++ b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/CommandLineArgumentsMessageMapper.java @@ -0,0 +1,24 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import java.util.Collection; + +import org.springframework.integration.handler.MessageProcessor; + +public interface CommandLineArgumentsMessageMapper extends MessageProcessor> { +} diff --git a/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/KeyValueListParser.java b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/KeyValueListParser.java new file mode 100644 index 00000000..48f01f0e --- /dev/null +++ b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/KeyValueListParser.java @@ -0,0 +1,66 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.Map; + +import org.springframework.util.StringUtils; + +/** + * Parses a comma delimited list of key value pairs in which the values can contain commas as well. + * + * @author Chris Schaeffer + * @author David Turanski + **/ +abstract class KeyValueListParser { + + static Map parseCommaDelimitedKeyValuePairs(String value) { + Map keyValuePairs = new HashMap<>(); + + if (StringUtils.isEmpty(value)) { + return keyValuePairs; + } + + ArrayList pairs = new ArrayList<>(); + + String[] candidates = StringUtils.commaDelimitedListToStringArray(value); + + for (int i = 0; i < candidates.length; i++) { + if (i > 0 && !candidates[i].contains("=")) { + pairs.add(pairs.get(pairs.size() - 1) + "," + candidates[i]); + } + else { + pairs.add(candidates[i]); + } + } + + for (String pair : pairs) { + addKeyValuePair(pair, keyValuePairs); + } + + return keyValuePairs; + } + + private static void addKeyValuePair(String pair, Map properties) { + int firstEquals = pair.indexOf('='); + if (firstEquals != -1) { + properties.put(pair.substring(0, firstEquals).trim(), pair.substring(firstEquals + 1).trim()); + } + } +} diff --git a/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequest.java b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequest.java new file mode 100644 index 00000000..3644499b --- /dev/null +++ b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequest.java @@ -0,0 +1,65 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import com.fasterxml.jackson.annotation.JsonProperty; + +public class TaskLaunchRequest { + @JsonProperty("args") + private List commandlineArguments = new ArrayList<>(); + + @JsonProperty("deploymentProps") + private Map deploymentProperties = new HashMap<>(); + + @JsonProperty("name") + private String taskName; + + public void setCommandlineArguments(List commandlineArguments) { + this.commandlineArguments = new ArrayList<>(commandlineArguments); + } + + public List getCommandlineArguments() { + return this.commandlineArguments; + } + + public void setDeploymentProperties(Map deploymentProperties) { + this.deploymentProperties = deploymentProperties; + } + + public Map getDeploymentProperties() { + return this.deploymentProperties; + } + + public void setTaskName(String taskName) { + this.taskName = taskName; + } + + public String getTaskName() { + return this.taskName; + } + + public TaskLaunchRequest addCommmandLineArguments(Collection args) { + this.commandlineArguments.addAll(args); + return this; + } +} diff --git a/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunction.java b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunction.java new file mode 100644 index 00000000..8497b395 --- /dev/null +++ b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunction.java @@ -0,0 +1,31 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import java.util.function.Function; + +import org.springframework.messaging.Message; + +/** + * A marker interface useful for unambiguous dependency injection of this Function. + * + * @author David Turanski + **/ +@FunctionalInterface +public interface TaskLaunchRequestFunction extends Function, Message> { + +} diff --git a/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionConfiguration.java b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionConfiguration.java new file mode 100644 index 00000000..a341e484 --- /dev/null +++ b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionConfiguration.java @@ -0,0 +1,184 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.util.StringUtils; + +/** + * Configuration for a {@link TaskLaunchRequestFunction}, provided as a common function that can be composed with other Suppliers or + * Functions to transform any {@link Message} to a {@link TaskLaunchRequest} which may be used as input to the {@code TaskLauncherFunction} to launch a task. + * + * Command line arguments used by the task, as well as the task name itself may be statically configured or extracted from + * the message contents, using SpEL. See {@link TaskLaunchRequestFunctionProperties} for details. + * + * It is also possible to provide your own implementations of {@link CommandLineArgumentsMessageMapper} and {@link TaskNameMessageMapper}. + * + * @author David Turanski + **/ +@Configuration +@EnableConfigurationProperties(TaskLaunchRequestFunctionProperties.class) +public class TaskLaunchRequestFunctionConfiguration { + + /** + * The function name. + */ + public final static String TASK_LAUNCH_REQUEST_FUNCTION_NAME = "taskLaunchRequestFunction"; + + /** + * A {@link java.util.function.Function} to transform a {@link Message} payload to a + * {@link TaskLaunchRequest}. + * + * @param taskLaunchRequestMessageProcessor a {@link TaskLaunchRequestMessageProcessor}. + * + * @return a {@code TaskLaunchRequest} Message. + */ + @Bean(name = TASK_LAUNCH_REQUEST_FUNCTION_NAME) + public TaskLaunchRequestFunction taskLaunchRequest( + TaskLaunchRequestMessageProcessor taskLaunchRequestMessageProcessor) { + return message -> taskLaunchRequestMessageProcessor.postProcessMessage(message); + } + + @Bean + public TaskLaunchRequestSupplier taskLaunchRequestInitializer( + TaskLaunchRequestFunctionProperties taskLaunchRequestProperties) { + return new TaskLaunchRequestPropertiesInitializer(taskLaunchRequestProperties); + } + + @SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection") + @Bean + public TaskLaunchRequestMessageProcessor taskLaunchRequestMessageProcessor( + TaskLaunchRequestSupplier taskLaunchRequestInitializer, + TaskLaunchRequestFunctionProperties properties, + EvaluationContext evaluationContext, + @Nullable TaskNameMessageMapper taskNameMessageMapper, + @Nullable CommandLineArgumentsMessageMapper commandLineArgumentsMessageMapper) { + + if (taskNameMessageMapper == null) { + taskNameMessageMapper = taskNameMessageMapper(properties, evaluationContext); + } + + if (commandLineArgumentsMessageMapper == null) { + commandLineArgumentsMessageMapper = commandLineArgumentsMessageMapper(properties, evaluationContext); + } + + return new TaskLaunchRequestMessageProcessor(taskLaunchRequestInitializer, + taskNameMessageMapper, + commandLineArgumentsMessageMapper); + } + + @Bean + public EvaluationContext evaluationContext(BeanFactory beanFactory) { + return ExpressionUtils.createStandardEvaluationContext(beanFactory); + } + + private TaskNameMessageMapper taskNameMessageMapper(TaskLaunchRequestFunctionProperties taskLaunchRequestProperties, + EvaluationContext evaluationContext) { + if (StringUtils.hasText(taskLaunchRequestProperties.getTaskNameExpression())) { + SpelExpressionParser expressionParser = new SpelExpressionParser(); + Expression taskNameExpression = expressionParser + .parseExpression(taskLaunchRequestProperties.getTaskNameExpression()); + return new ExpressionEvaluatingTaskNameMessageMapper(taskNameExpression, evaluationContext); + } + + return message -> taskLaunchRequestProperties.getTaskName(); + } + + private CommandLineArgumentsMessageMapper commandLineArgumentsMessageMapper( + TaskLaunchRequestFunctionProperties taskLaunchRequestFunctionProperties, + EvaluationContext evaluationContext) { + return new ExpressionEvaluatingCommandLineArgsMapper(taskLaunchRequestFunctionProperties.getArgExpressions(), + evaluationContext); + } + + private static class TaskLaunchRequestPropertiesInitializer extends TaskLaunchRequestSupplier { + TaskLaunchRequestPropertiesInitializer( + TaskLaunchRequestFunctionProperties taskLaunchRequestProperties) { + + this.commandLineArgumentSupplier( + () -> new ArrayList<>(taskLaunchRequestProperties.getArgs())); + + this.deploymentPropertiesSupplier( + () -> KeyValueListParser.parseCommaDelimitedKeyValuePairs( + taskLaunchRequestProperties.getDeploymentProperties())); + + this.taskNameSupplier(() -> taskLaunchRequestProperties.getTaskName()); + } + } + + private static class ExpressionEvaluatingTaskNameMessageMapper implements TaskNameMessageMapper { + + private final Expression expression; + private final EvaluationContext evaluationContext; + + ExpressionEvaluatingTaskNameMessageMapper(Expression expression, EvaluationContext evaluationContext) { + this.evaluationContext = evaluationContext; + this.expression = expression; + } + + @Override + public String processMessage(Message message) { + return expression.getValue(evaluationContext, message).toString(); + } + } + + private static class ExpressionEvaluatingCommandLineArgsMapper implements CommandLineArgumentsMessageMapper { + private final Map argExpressionsMap; + + private final EvaluationContext evaluationContext; + + ExpressionEvaluatingCommandLineArgsMapper(String argExpressions, EvaluationContext evaluationContext) { + this.evaluationContext = evaluationContext; + this.argExpressionsMap = new HashMap<>(); + if (StringUtils.hasText(argExpressions)) { + SpelExpressionParser expressionParser = new SpelExpressionParser(); + + KeyValueListParser.parseCommaDelimitedKeyValuePairs(argExpressions).forEach( + (k, v) -> argExpressionsMap.put(k, expressionParser.parseExpression(v))); + } + } + + @Override + public Collection processMessage(Message message) { + return evaluateArgExpressions(message); + } + + private Collection evaluateArgExpressions(Message message) { + List results = new LinkedList<>(); + this.argExpressionsMap.forEach((k, expression) -> results + .add(String.format("%s=%s", k, expression.getValue(this.evaluationContext, message)))); + return results; + } + } + +} diff --git a/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionProperties.java b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionProperties.java new file mode 100644 index 00000000..bdde0504 --- /dev/null +++ b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionProperties.java @@ -0,0 +1,113 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import java.util.ArrayList; +import java.util.List; + +import javax.validation.constraints.AssertFalse; +import javax.validation.constraints.NotNull; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.util.StringUtils; +import org.springframework.validation.annotation.Validated; + +/** + * Base Properties to create a {@link TaskLaunchRequest}. + * + * @author Chris Schaefer + * @author David Turanski + */ +@Validated +@ConfigurationProperties("task.launch.request") +public class TaskLaunchRequestFunctionProperties { + + /** + * Comma separated list of optional args in key=value format. + */ + private List args = new ArrayList<>(); + + /** + * Comma separated list of option args as SpEL expressions in key=value format. + */ + private String argExpressions = ""; + + /** + * Comma delimited list of deployment properties to be applied to the + * TaskLaunchRequest. + */ + private String deploymentProperties = ""; + + /** + * The Data Flow task name. + */ + private String taskName; + + + /** + * A SpEL expression to extract the task name from each Message, using the Message as the evaluation context. + */ + private String taskNameExpression; + + @NotNull + public List getArgs() { + return this.args; + } + + public void setArgs(List args) { + this.args = new ArrayList<>(args); + } + + @NotNull + public String getDeploymentProperties() { + return this.deploymentProperties; + } + + public void setDeploymentProperties(String deploymentProperties) { + this.deploymentProperties = deploymentProperties; + } + + public String getTaskName() { + return taskName; + } + + public void setTaskName(String taskName) { + this.taskName = taskName; + } + + public String getTaskNameExpression() { + return taskNameExpression; + } + + public void setTaskNameExpression(String taskNameExpression) { + this.taskNameExpression = taskNameExpression; + } + + public String getArgExpressions() { + return argExpressions; + } + + public void setArgExpressions(String argExpressions) { + this.argExpressions = argExpressions; + } + + @AssertFalse(message = "Cannot specify both 'taskName' and 'taskNameExpression'.") + public boolean isTaskNameAndTaskNameExpressionSet() { + return StringUtils.hasText(this.taskName) && StringUtils.hasText(this.taskNameExpression); + } + +} diff --git a/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestMessageProcessor.java b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestMessageProcessor.java new file mode 100644 index 00000000..1bbea190 --- /dev/null +++ b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestMessageProcessor.java @@ -0,0 +1,68 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.core.MessagePostProcessor; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.Assert; +import org.springframework.util.MimeTypeUtils; +import org.springframework.util.StringUtils; + +class TaskLaunchRequestMessageProcessor implements MessagePostProcessor { + + private final TaskNameMessageMapper taskNameMessageMapper; + + private final CommandLineArgumentsMessageMapper commandLineArgumentsMessageMapper; + + private final TaskLaunchRequestSupplier taskLaunchRequestInitializer; + + TaskLaunchRequestMessageProcessor(TaskLaunchRequestSupplier taskLaunchRequestInitializer, + TaskNameMessageMapper taskNameMessageMapper, + CommandLineArgumentsMessageMapper commandLIneArgumentsMessageMapper) { + + this.taskLaunchRequestInitializer = taskLaunchRequestInitializer; + + this.taskNameMessageMapper = taskNameMessageMapper; + + this.commandLineArgumentsMessageMapper = commandLIneArgumentsMessageMapper; + + } + + @Override + public Message postProcessMessage(Message message) { + TaskLaunchRequest taskLaunchRequest = taskLaunchRequestInitializer.get(); + + if (!StringUtils.hasText(taskLaunchRequest.getTaskName())) { + taskLaunchRequest.setTaskName(taskNameMessageMapper.processMessage(message)); + Assert.hasText(taskLaunchRequest.getTaskName(), + () -> "'taskName' is required in " + TaskLaunchRequest.class.getName()); + } + + taskLaunchRequest.addCommmandLineArguments(commandLineArgumentsMessageMapper.processMessage(message)); + + MessageBuilder builder = MessageBuilder.withPayload(taskLaunchRequest) + .copyHeaders(message.getHeaders()); + return adjustHeaders(builder).build(); + } + + private MessageBuilder adjustHeaders(MessageBuilder builder) { + builder.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON); + return builder; + } +} diff --git a/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestSupplier.java b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestSupplier.java new file mode 100644 index 00000000..1d6bb373 --- /dev/null +++ b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestSupplier.java @@ -0,0 +1,67 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import java.util.List; +import java.util.Map; +import java.util.function.Supplier; + +import org.springframework.util.Assert; + +class TaskLaunchRequestSupplier implements Supplier { + + private Supplier taskNameSupplier; + + private Supplier> commandLineArgumentsSupplier; + + private Supplier> deploymentPropertiesSupplier; + + public TaskLaunchRequestSupplier taskNameSupplier(Supplier taskNameSupplier) { + this.taskNameSupplier = taskNameSupplier; + return this; + } + + public TaskLaunchRequestSupplier commandLineArgumentSupplier(Supplier> commandLineArgumentsSupplier) { + this.commandLineArgumentsSupplier = commandLineArgumentsSupplier; + return this; + } + + public TaskLaunchRequestSupplier deploymentPropertiesSupplier( + Supplier> deploymentPropertiesSupplier) { + this.deploymentPropertiesSupplier = deploymentPropertiesSupplier; + return this; + } + + @Override + public TaskLaunchRequest get() { + + Assert.notNull(this.taskNameSupplier, "'taskNameSupplier' is required."); + + TaskLaunchRequest taskLaunchRequest = new TaskLaunchRequest(); + taskLaunchRequest.setTaskName(this.taskNameSupplier.get()); + + if (this.commandLineArgumentsSupplier != null) { + taskLaunchRequest.setCommandlineArguments(this.commandLineArgumentsSupplier.get()); + } + + if (this.deploymentPropertiesSupplier != null) { + taskLaunchRequest.setDeploymentProperties(this.deploymentPropertiesSupplier.get()); + } + + return taskLaunchRequest; + } +} diff --git a/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskNameMessageMapper.java b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskNameMessageMapper.java new file mode 100644 index 00000000..09eeb938 --- /dev/null +++ b/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskNameMessageMapper.java @@ -0,0 +1,23 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import org.springframework.integration.handler.MessageProcessor; + +@FunctionalInterface +public interface TaskNameMessageMapper extends MessageProcessor { +} diff --git a/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/KeyValueListParserTests.java b/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/KeyValueListParserTests.java new file mode 100644 index 00000000..28d35361 --- /dev/null +++ b/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/KeyValueListParserTests.java @@ -0,0 +1,97 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import java.util.Map; + +import org.junit.jupiter.api.Test; + +import static junit.framework.TestCase.assertEquals; +import static junit.framework.TestCase.assertTrue; + +/** + * @author Chris Schaefer + * @author David Turanski + */ +public class KeyValueListParserTests { + + @Test + public void testParseSimpleDeploymentProperty() { + Map deploymentProperties = KeyValueListParser.parseCommaDelimitedKeyValuePairs( + "app.sftp.param=value"); + assertTrue("Invalid number of deployment properties: " + deploymentProperties.size(), + deploymentProperties.size() == 1); + assertTrue("Expected deployment key not found", deploymentProperties.containsKey("app.sftp.param")); + assertEquals("Invalid deployment value", "value", deploymentProperties.get("app.sftp.param")); + } + + @Test + public void testParseSimpleDeploymentPropertyMultipleValues() { + Map deploymentProperties = KeyValueListParser.parseCommaDelimitedKeyValuePairs( + "app.sftp.param=value1,value2,value3"); + + assertTrue("Invalid number of deployment properties: " + deploymentProperties.size(), + deploymentProperties.size() == 1); + assertTrue("Expected deployment key not found", deploymentProperties.containsKey("app.sftp.param")); + assertEquals("Invalid deployment value", "value1,value2,value3", deploymentProperties.get("app.sftp.param")); + } + + @Test + public void testParseSpelExpressionMultipleValues() { + Map argExpressions = KeyValueListParser.parseCommaDelimitedKeyValuePairs( + "arg1=payload.substr(0,2),arg2=headers['foo'],arg3=headers['bar']==false"); + + assertTrue("Invalid number of deployment properties: " + argExpressions.size(), + argExpressions.size() == 3); + assertTrue("Expected deployment key not found", argExpressions.containsKey("arg1")); + assertEquals("Invalid deployment value", "payload.substr(0,2)", argExpressions.get("arg1")); + + assertTrue("Expected deployment key not found", argExpressions.containsKey("arg2")); + assertEquals("Invalid deployment value", "headers['foo']", argExpressions.get("arg2")); + + assertTrue("Expected deployment key not found", argExpressions.containsKey("arg3")); + assertEquals("Invalid deployment value", "headers['bar']==false", argExpressions.get("arg3")); + } + + @Test + public void testParseMultipleDeploymentPropertiesSingleValue() { + Map deploymentProperties = KeyValueListParser.parseCommaDelimitedKeyValuePairs( + "app.sftp.param=value1,app.sftp.other.param=value2"); + + assertTrue("Invalid number of deployment properties: " + deploymentProperties.size(), + deploymentProperties.size() == 2); + assertTrue("Expected deployment key not found", deploymentProperties.containsKey("app.sftp.param")); + assertEquals("Invalid deployment value", "value1", deploymentProperties.get("app.sftp.param")); + assertTrue("Expected deployment key not found", deploymentProperties.containsKey("app.sftp.other.param")); + assertEquals("Invalid deployment value", "value2", deploymentProperties.get("app.sftp.other.param")); + } + + @Test + public void testParseMultipleDeploymentPropertiesMultipleValues() { + TaskLaunchRequestFunctionProperties taskLaunchRequestProperties = new TaskLaunchRequestFunctionProperties(); + + Map deploymentProperties = KeyValueListParser.parseCommaDelimitedKeyValuePairs( + "app.sftp.param=value1,value2,app.sftp.other.param=other1,other2"); + + assertTrue("Invalid number of deployment properties: " + deploymentProperties.size(), + deploymentProperties.size() == 2); + assertTrue("Expected deployment key not found", deploymentProperties.containsKey("app.sftp.param")); + assertEquals("Invalid deployment value", "value1,value2", deploymentProperties.get("app.sftp.param")); + assertTrue("Expected deployment key not found", deploymentProperties.containsKey("app.sftp.other.param")); + assertEquals("Invalid deployment value", "other1,other2", deploymentProperties.get("app.sftp.other.param")); + } +} diff --git a/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionApplicationTests.java b/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionApplicationTests.java new file mode 100644 index 00000000..ddd8bfca --- /dev/null +++ b/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionApplicationTests.java @@ -0,0 +1,211 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import java.io.IOException; +import java.util.Collections; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.test.annotation.DirtiesContext; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.entry; + +/** + * @author David Turanski + **/ +public class TaskLaunchRequestFunctionApplicationTests { + + private SpringApplicationBuilder springApplicationBuilder; + + @BeforeEach + public void setUp() { + springApplicationBuilder = new SpringApplicationBuilder(TestApplication.class) + .web(WebApplicationType.NONE); + } + + @Test + @DirtiesContext + public void simpleDataflowTaskLaunchRequest() throws IOException { + + ApplicationContext context = springApplicationBuilder.properties( + "spring.jmx.enabled=false", + "spring.cloud.stream.function.definition=taskLaunchRequestFunction", + "task.launch.request.task-name=foo") + .run(); + + TaskLaunchRequest taskLaunchRequest = verifyAndreceiveTaskLaunchRequest(context); + + assertThat(taskLaunchRequest.getTaskName()).isEqualTo("foo"); + assertThat(taskLaunchRequest.getCommandlineArguments()).hasSize(0); + assertThat(taskLaunchRequest.getDeploymentProperties()).hasSize(0); + } + + @Test + @DirtiesContext + public void dataflowTaskLaunchRequestWithArgsAndDeploymentProperties() throws IOException { + + ApplicationContext context = springApplicationBuilder.properties( + "spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequestFunction", + "task.launch.request.task-name=foo", "task.launch.request.args=foo=bar,baz=boo", + "task.launch.request.deploymentProperties=count=3") + .run(); + TaskLaunchRequest taskLaunchRequest = verifyAndreceiveTaskLaunchRequest(context); + + assertThat(taskLaunchRequest.getTaskName()).isEqualTo("foo"); + assertThat(taskLaunchRequest.getCommandlineArguments()).containsExactlyInAnyOrder("foo=bar", + "baz=boo"); + assertThat(taskLaunchRequest.getDeploymentProperties()).containsOnly(entry("count", "3")); + } + + @Test + @DirtiesContext + public void taskLaunchRequestWithCommandLineArgsMessageMapper() throws IOException { + + ApplicationContext context = springApplicationBuilder.properties( + "spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequestFunction", + "task.launch.request.task-name=foo", "enhanceTLRArgs=true") + .run(); + + TaskLaunchRequest taskLaunchRequest = verifyAndreceiveTaskLaunchRequest(context); + + assertThat(taskLaunchRequest.getTaskName()).isEqualTo("foo"); + assertThat(taskLaunchRequest.getCommandlineArguments()).hasSize(1); + assertThat(taskLaunchRequest.getCommandlineArguments()).containsExactly("runtimeArg"); + + } + + @Test + @DirtiesContext + public void taskLaunchRequestWithArgExpressions() throws IOException { + ApplicationContext context = springApplicationBuilder.properties( + "spring.jmx.enabled=false", + "spring.cloud.stream.function.definition=taskLaunchRequestFunction", + "task.launch.request.task-name=foo", + "task.launch.request.arg-expressions=foo=payload.toUpperCase(),bar=payload.substring(0,2)") + .run(); + + Message message = MessageBuilder.withPayload("hello").build(); + + TaskLaunchRequestFunction taskLaunchRequestFunction = context.getBean(TaskLaunchRequestFunction.class); + + Message response = taskLaunchRequestFunction.apply(message); + + assertThat(response).isNotNull(); + TaskLaunchRequest request = response.getPayload(); + assertThat(request.getCommandlineArguments()).containsExactlyInAnyOrder("foo=HELLO", "bar=he"); + } + + @Test + @DirtiesContext + public void taskLaunchRequestWithIntPayload() throws IOException { + ApplicationContext context = springApplicationBuilder.properties( + "spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequestFunction", + "task.launch.request.task-name=foo", + "task.launch.request.arg-expressions=i=payload") + .run(); + + TaskLaunchRequestFunction taskLaunchRequestFunction = context.getBean(TaskLaunchRequestFunction.class); + + Message message = MessageBuilder.withPayload(123).build(); + + Message response = taskLaunchRequestFunction.apply(message); + + assertThat(response).isNotNull(); + + TaskLaunchRequest request = response.getPayload(); + assertThat(request.getCommandlineArguments()).containsExactly("i=123"); + } + + @Test + @DirtiesContext + public void taskNameExpression() throws IOException { + ApplicationContext context = springApplicationBuilder.properties( + "spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequestFunction", + "task.launch.request.task-name-expression=payload+'_task'") + .run(); + + TaskLaunchRequestFunction taskLaunchRequestFunction = context.getBean(TaskLaunchRequestFunction.class); + + Message message = MessageBuilder.withPayload("foo").build(); + + Message response = taskLaunchRequestFunction.apply(message); + assertThat(response).isNotNull(); + + TaskLaunchRequest request = response.getPayload(); + assertThat(request.getTaskName()).isEqualTo("foo_task"); + } + + @Test + @DirtiesContext + public void customTaskNameExtractor() throws IOException { + ApplicationContext context = springApplicationBuilder.properties( + "spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequestFunction", + "customTaskNameExtractor=true") + .run(); + TaskLaunchRequestFunction taskLaunchRequestFunction = context.getBean(TaskLaunchRequestFunction.class); + + Message message = MessageBuilder.withPayload("foo").build(); + + Message response = taskLaunchRequestFunction.apply(message); + assertThat(response).isNotNull(); + + TaskLaunchRequest request = response.getPayload(); + assertThat(request.getTaskName()).isEqualTo("fooTask"); + + message = MessageBuilder.withPayload("bar").build(); + response = taskLaunchRequestFunction.apply(message); + request = response.getPayload(); + + assertThat(request.getTaskName()).isEqualTo("defaultTask"); + } + + private TaskLaunchRequest verifyAndreceiveTaskLaunchRequest(ApplicationContext context) + throws IOException { + TaskLaunchRequestFunction taskLaunchRequestFunction = context.getBean(TaskLaunchRequestFunction.class); + Message message = taskLaunchRequestFunction + .apply(MessageBuilder.withPayload(new byte[] {}).build()); + assertThat(message).isNotNull(); + return message.getPayload(); + } + + @SpringBootApplication + protected static class TestApplication { + + @Bean + @ConditionalOnProperty("customTaskNameExtractor") + TaskNameMessageMapper taskNameExtractor() { + return message -> ((String) (message.getPayload())).equalsIgnoreCase("foo") ? "fooTask" : "defaultTask"; + } + + @Bean + @ConditionalOnProperty("enhanceTLRArgs") + CommandLineArgumentsMessageMapper commandLineArgumentsProvider() { + return message -> Collections.singletonList("runtimeArg"); + } + } +} diff --git a/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionPropertiesTests.java b/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionPropertiesTests.java new file mode 100644 index 00000000..a452a0e9 --- /dev/null +++ b/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionPropertiesTests.java @@ -0,0 +1,76 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.fn.task.launch.request; + +import java.util.List; + +import org.junit.jupiter.api.Test; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.util.TestPropertyValues; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.config.EnableIntegration; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author David Turanski + **/ +public class TaskLaunchRequestFunctionPropertiesTests { + + @Test + public void deploymentPropertiesCanBeCustomized() { + TaskLaunchRequestFunctionProperties properties = getBatchProperties( + "task.launch.request.deploymentProperties:prop1=val1,prop2=val2"); + assertThat(properties.getDeploymentProperties()).isEqualTo("prop1=val1,prop2=val2"); + } + + @Test + public void parametersCanBeCustomized() { + TaskLaunchRequestFunctionProperties properties = getBatchProperties( + "task.launch.request.args:jp1=jpv1,jp2=jpv2"); + List args = properties.getArgs(); + + assertThat(args).isNotNull(); + assertThat(args).hasSize(2); + assertThat(args.get(0)).isEqualTo("jp1=jpv1"); + assertThat(args.get(1)).isEqualTo("jp2=jpv2"); + } + + private TaskLaunchRequestFunctionProperties getBatchProperties(String... var) { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + + if (var != null) { + TestPropertyValues.of(var).applyTo(context); + } + + context.register(Conf.class); + context.refresh(); + + return context.getBean(TaskLaunchRequestFunctionProperties.class); + } + + @Configuration + @EnableIntegration + @EnableConfigurationProperties(TaskLaunchRequestFunctionProperties.class) + @Import(TaskLaunchRequestFunctionConfiguration.class) + static class Conf { + + } +} diff --git a/pom.xml b/pom.xml index f9d96bba..8d5983e7 100644 --- a/pom.xml +++ b/pom.xml @@ -75,6 +75,7 @@ function/payload-converter-function function/splitter-function function/tasklauncher-function + function/task-launch-request-function supplier/file-supplier supplier/ftp-supplier