SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implement task-launch-request-function

Browse Source 
Removed task-launch-request from stream-applications-core

Fix Checkstyle errors
This commit is contained in:
David Turanski
2020-06-15 11:31:08 -04:00
committed by Soby Chacko
parent d962ba65ea
commit 1a6be1c031
24 changed files with 616 additions and 766 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
applications/stream-applications-core/common/stream-applications-task-launch-request-common/README.adoc
View File

@@ -1,52 +0,0 @@
=== `Data Flow Task Launch Request Support` Common Module
This artifact contains a Spring Boot auto-configuration providing components to produce a Task Launch Request payload.
In theory any message source can be used to launch a Task. The task launch request is compatible with the
task-launcher-dataflow sink.
==== Data Flow Task Launch Request
Data Flow Task Launch Request requests require a Data Flow server with an existing task definition.
The task launch request contains the name of the task to launch. This must the name of a defined task in Data Flow.
You may optionally provide command line arguments and deployment properties.
===== Task Name
The task name is a required field. This may be statically configured by setting `task.launch.request.task-name`,
extracted from the Message by setting `task.launch.request.task-name-expression`.
You may also override the default implementation of the `TaskNameMessageMapper` bean to enable more complex runtime task name mappings.
===== Task Command Line Arguments
The launched task often requires additional data which may be passed as command line arguments.
The `task.launch.request.args` property accepts a comma delimited string of key-value pairs, for example
`key1=val1,key2=val2`. In addition, a `task.launch.request.arg-expressions` allows you to use SpEL expressions to evaluate
message contents to provide command line arguments.
For example, `task.launch.request.arg-expressions=foo=payload.toUpperCase(),bar=payload.substring(0,2)`.
You may also provide override implementation of the `CommandLineArgumentsMessageMapper` bean to implement more complex logic.
===== Task Deployment Properties
Deployment properties are platform-specific configuration used by the `TaskLauncher` and are always statically configured by
setting `task.launch.request.deployment-properties` and apply to every task launch request.
=== Configuration
To enable any stream app to transform its output to a Task Launch Request, include a dependency on this module
[source,xml]
----
<dependency>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>app-starters-task-launch-request-common</artifactId>
</dependency>
----
Setting the application property `spring.cloud.stream.function.definition=taskLaunchRequest` is required to execute the transformation.
You may safely add the above dependency to applications which optionally produce a task launch request.
`TaskLaunchRequestIntegrationTests` provides some configuration examples.

41
applications/stream-applications-core/common/stream-applications-task-launch-request-common/pom.xml
View File

@@ -1,41 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>stream-applications-core</artifactId>
<groupId>org.springframework.cloud.stream.app</groupId>
<version>3.0.0-SNAPSHOT</version>
<relativePath>../..</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>stream-applications-task-launch-request-common</artifactId>
<name>stream-applications-task-launch-request-common</name>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
<type>test-jar</type>
<scope>test</scope>
<classifier>test-binder</classifier>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

163
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/java/org/springframework/cloud/stream/app/tasklaunchrequest/DataFlowTaskLaunchRequestAutoConfiguration.java
View File

@@ -1,163 +0,0 @@
/*
* Copyright 2018-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.stream.app.tasklaunchrequest;
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 javax.annotation.PostConstruct;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.app.tasklaunchrequest.support.CommandLineArgumentsMessageMapper;
import org.springframework.cloud.stream.app.tasklaunchrequest.support.TaskLaunchRequestSupplier;
import org.springframework.cloud.stream.app.tasklaunchrequest.support.TaskNameMessageMapper;
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.messaging.Message;
import org.springframework.util.StringUtils;
/**
* @author David Turanski
**/
@Configuration
@EnableConfigurationProperties(DataflowTaskLaunchRequestProperties.class)
public class DataFlowTaskLaunchRequestAutoConfiguration {
/**
* Task launch request function name.
*/
public final static String TASK_LAUNCH_REQUEST_FUNCTION_NAME = "taskLaunchRequest";
@Autowired
private BeanFactory beanFactory;
private EvaluationContext evaluationContext;
/**
* A {@link java.util.function.Function} to transform a {@link Message} payload to a {@link DataFlowTaskLaunchRequest}.
*
* @param taskLaunchRequestMessageProcessor a {@link TaskLaunchRequestMessageProcessor}.
* @return a {code DataFlowTaskLaunchRequest} Message.
*/
@Bean(name = TASK_LAUNCH_REQUEST_FUNCTION_NAME)
@ConditionalOnMissingBean(TaskLaunchRequestFunction.class)
public TaskLaunchRequestFunction taskLaunchRequest(TaskLaunchRequestMessageProcessor taskLaunchRequestMessageProcessor) {
return message -> taskLaunchRequestMessageProcessor.postProcessMessage(message);
}
@Bean
@ConditionalOnMissingBean(TaskNameMessageMapper.class)
public TaskNameMessageMapper taskNameMessageMapper(DataflowTaskLaunchRequestProperties taskLaunchRequestProperties) {
if (StringUtils.hasText(taskLaunchRequestProperties.getTaskNameExpression())) {
SpelExpressionParser expressionParser = new SpelExpressionParser();
Expression taskNameExpression = expressionParser.parseExpression(taskLaunchRequestProperties.getTaskNameExpression());
return new ExpressionEvaluatingTaskNameMessageMapper(taskNameExpression, this.evaluationContext);
}
return message -> taskLaunchRequestProperties.getTaskName();
}
@Bean
@ConditionalOnMissingBean(CommandLineArgumentsMessageMapper.class)
public CommandLineArgumentsMessageMapper commandLineArgumentsMessageMapper(
DataflowTaskLaunchRequestProperties dataflowTaskLaunchRequestProperties) {
return new ExpressionEvaluatingCommandLineArgsMapper(dataflowTaskLaunchRequestProperties.getArgExpressions(),
this.evaluationContext);
}
@Bean
public TaskLaunchRequestSupplier taskLaunchRequestInitializer(
DataflowTaskLaunchRequestProperties taskLaunchRequestProperties) {
return new DataflowTaskLaunchRequestPropertiesInitializer(taskLaunchRequestProperties);
}
@Bean
public TaskLaunchRequestMessageProcessor taskLaunchRequestMessageProcessor(
TaskLaunchRequestSupplier taskLaunchRequestInitializer,
TaskNameMessageMapper taskNameMessageMapper,
CommandLineArgumentsMessageMapper commandLineArgumentsMessageMapper) {
return new TaskLaunchRequestMessageProcessor(taskLaunchRequestInitializer,
taskNameMessageMapper,
commandLineArgumentsMessageMapper);
}
@PostConstruct
public void createEvaluationContext() {
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
}
static class DataflowTaskLaunchRequestPropertiesInitializer extends TaskLaunchRequestSupplier {
DataflowTaskLaunchRequestPropertiesInitializer(
DataflowTaskLaunchRequestProperties taskLaunchRequestProperties) {
this.commandLineArgumentSupplier(
() -> new ArrayList<>(taskLaunchRequestProperties.getArgs())
);
this.deploymentPropertiesSupplier(
() -> KeyValueListParser.parseCommaDelimitedKeyValuePairs(
taskLaunchRequestProperties.getDeploymentProperties())
);
this.taskNameSupplier(() -> taskLaunchRequestProperties.getTaskName());
}
}
static class ExpressionEvaluatingCommandLineArgsMapper implements CommandLineArgumentsMessageMapper {
private final Map<String, Expression> 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<String> processMessage(Message<?> message) {
return evaluateArgExpressions(message);
}
private Collection<String> evaluateArgExpressions(Message<?> message) {
List<String> results = new LinkedList<>();
this.argExpressionsMap.forEach((k, expression) ->
results.add(String.format("%s=%s", k, expression.getValue(this.evaluationContext, message))));
return results;
}
}
}

38
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/java/org/springframework/cloud/stream/app/tasklaunchrequest/ExpressionEvaluatingTaskNameMessageMapper.java
View File

@@ -1,38 +0,0 @@
/*
* Copyright 2019-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.stream.app.tasklaunchrequest;
import org.springframework.cloud.stream.app.tasklaunchrequest.support.TaskNameMessageMapper;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.messaging.Message;
public class ExpressionEvaluatingTaskNameMessageMapper implements TaskNameMessageMapper {
private final Expression expression;
private final EvaluationContext evaluationContext;
public ExpressionEvaluatingTaskNameMessageMapper(Expression expression, EvaluationContext evaluationContext) {
this.evaluationContext = evaluationContext;
this.expression = expression;
}
@Override
public String processMessage(Message<?> message) {
return expression.getValue(evaluationContext, message).toString();
}
}

2
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/resources/META-INF/spring-configuration-metadata-whitelist.properties
View File

@@ -1,2 +0,0 @@
configuration-properties.classes=\
org.springframework.cloud.stream.app.tasklaunchrequest.DataflowTaskLaunchRequestProperties

2
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/resources/META-INF/spring.factories
View File

@@ -1,2 +0,0 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.stream.app.tasklaunchrequest.DataFlowTaskLaunchRequestAutoConfiguration

88
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/test/java/org/springframework/cloud/stream/app/tasklaunchrequest/KeyValueListParserTests.java
View File

@@ -1,88 +0,0 @@
/*
* 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.stream.app.tasklaunchrequest;
import java.util.Map;
import org.junit.Test;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Chris Schaefer
* @author David Turanski
*/
public class KeyValueListParserTests {
@Test
public void testParseSimpleDeploymentProperty() {
Map<String, String> deploymentProperties = KeyValueListParser.parseCommaDelimitedKeyValuePairs(
"app.sftp.param=value");
assertThat(deploymentProperties.size()).isEqualTo(1);
assertThat(deploymentProperties.containsKey("app.sftp.param")).isTrue();
assertThat(deploymentProperties.get("app.sftp.param")).isEqualTo("value");
}
@Test
public void testParseSimpleDeploymentPropertyMultipleValues() {
Map<String, String> deploymentProperties = KeyValueListParser.parseCommaDelimitedKeyValuePairs(
"app.sftp.param=value1,value2,value3");
assertThat(deploymentProperties.size()).isEqualTo(1);
assertThat(deploymentProperties.containsKey("app.sftp.param")).isTrue();
assertThat(deploymentProperties.get("app.sftp.param")).isEqualTo("value1,value2,value3");
}
@Test
public void testParseSpelExpressionMultipleValues() {
Map<String, String> argExpressions = KeyValueListParser.parseCommaDelimitedKeyValuePairs(
"arg1=payload.substr(0,2),arg2=headers['foo'],arg3=headers['bar']==false");
assertThat(argExpressions.size()).isEqualTo(3);
assertThat(argExpressions.containsKey("arg1")).isTrue();
assertThat(argExpressions.get("arg1")).isEqualTo("payload.substr(0,2)");
assertThat(argExpressions.containsKey("arg2")).isTrue();
assertThat(argExpressions.get("arg2")).isEqualTo("headers['foo']");
}
@Test
public void testParseMultipleDeploymentPropertiesSingleValue() {
Map<String, String> deploymentProperties = KeyValueListParser.parseCommaDelimitedKeyValuePairs(
"app.sftp.param=value1,app.sftp.other.param=value2");
assertThat(deploymentProperties.size()).isEqualTo(2);
assertThat(deploymentProperties.containsKey("app.sftp.param")).isTrue();
assertThat(deploymentProperties.get("app.sftp.param")).isEqualTo("value1");
assertThat(deploymentProperties.containsKey("app.sftp.other.param")).isTrue();
assertThat(deploymentProperties.get("app.sftp.other.param")).isEqualTo("value2");
}
@Test
public void testParseMultipleDeploymentPropertiesMultipleValues() {
DataflowTaskLaunchRequestProperties taskLaunchRequestProperties = new DataflowTaskLaunchRequestProperties();
Map<String, String> deploymentProperties = KeyValueListParser.parseCommaDelimitedKeyValuePairs(
"app.sftp.param=value1,value2,app.sftp.other.param=other1,other2");
assertThat(deploymentProperties.size()).isEqualTo(2);
assertThat(deploymentProperties.containsKey("app.sftp.param")).isTrue();
assertThat(deploymentProperties.get("app.sftp.param")).isEqualTo("value1,value2");
assertThat(deploymentProperties.get("app.sftp.other.param")).isEqualTo("other1,other2");
}
}

313
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/test/java/org/springframework/cloud/stream/app/tasklaunchrequest/TaskLaunchRequestIntegrationTests.java
View File

@@ -1,313 +0,0 @@
/*
* Copyright 2018-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.stream.app.tasklaunchrequest;
import java.io.IOException;
import java.util.Collections;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Before;
import org.junit.Test;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.app.tasklaunchrequest.support.CommandLineArgumentsMessageMapper;
import org.springframework.cloud.stream.app.tasklaunchrequest.support.TaskNameMessageMapper;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollectorAutoConfiguration;
import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.entry;
/**
* @author David Turanski
**/
public class TaskLaunchRequestIntegrationTests {
private ApplicationContextRunner applicationContextRunner;
@Before
public void setUp() {
applicationContextRunner =
new ApplicationContextRunner().withUserConfiguration(TestChannelBinderConfiguration.class, TestApp.class);
}
@Test
public void noTaskLaunchRequestPropertiesAreRequired() {
applicationContextRunner.withPropertyValues("spring.jmx.enabled=false")
.run(context -> {
MessageChannel input = context.getBean("input", MessageChannel.class);
OutputDestination target = context.getBean(OutputDestination.class);
Message<byte[]> message =
MessageBuilder.withPayload("hello".getBytes()).build();
input.send(message);
Message<byte[]> response = target.receive(1000);
assertThat(response.getPayload()).isEqualTo(message.getPayload());
});
}
@Test
public void simpleDataflowTaskLaunchRequest() {
applicationContextRunner.withPropertyValues(
"spring.jmx.enabled=false",
"spring.cloud.stream.function.definition=taskLaunchRequest",
"task.launch.request.task-name=foo")
.run(context -> {
DataFlowTaskLaunchRequest dataFlowTaskLaunchRequest = verifyAndreceiveDataFlowTaskLaunchRequest(context);
assertThat(dataFlowTaskLaunchRequest.getTaskName()).isEqualTo("foo");
assertThat(dataFlowTaskLaunchRequest.getCommandlineArguments()).hasSize(0);
assertThat(dataFlowTaskLaunchRequest.getDeploymentProperties()).hasSize(0);
});
}
@Test
public void dataflowTaskLaunchRequestWithArgsAndDeploymentProperties() {
applicationContextRunner.withPropertyValues(
"spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequest",
"task.launch.request.task-name=foo", "task.launch.request.args=foo=bar,baz=boo",
"task.launch.request.deploymentProperties=count=3")
.run(context -> {
DataFlowTaskLaunchRequest dataFlowTaskLaunchRequest = verifyAndreceiveDataFlowTaskLaunchRequest(context);
assertThat(dataFlowTaskLaunchRequest.getTaskName()).isEqualTo("foo");
assertThat(dataFlowTaskLaunchRequest.getCommandlineArguments()).containsExactlyInAnyOrder("foo=bar",
"baz=boo");
assertThat(dataFlowTaskLaunchRequest.getDeploymentProperties()).containsOnly(entry("count", "3"));
});
}
@Test
public void dataflowTaskLaunchRequestWithCommandLineArgsMessageMapper() {
applicationContextRunner.withPropertyValues(
"spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequest",
"task.launch.request.task-name=foo", "enhanceTLRArgs=true")
.run(context -> {
DataFlowTaskLaunchRequest dataFlowTaskLaunchRequest = verifyAndreceiveDataFlowTaskLaunchRequest(context);
assertThat(dataFlowTaskLaunchRequest.getTaskName()).isEqualTo("foo");
assertThat(dataFlowTaskLaunchRequest.getCommandlineArguments()).hasSize(1);
assertThat(dataFlowTaskLaunchRequest.getCommandlineArguments()).containsExactly("runtimeArg");
});
}
@Test
public void taskLaunchRequestWithArgExpressions() {
applicationContextRunner.withPropertyValues(
"spring.jmx.enabled=false",
"spring.cloud.stream.function.definition=taskLaunchRequest",
"task.launch.request.task-name=foo",
"task.launch.request.arg-expressions=foo=payload.toUpperCase(),bar=payload.substring(0,2)")
.run(context -> {
MessageChannel input = context.getBean("input", MessageChannel.class);
OutputDestination target = context.getBean(OutputDestination.class);
Message<String> message = MessageBuilder.withPayload("hello").build();
input.send(message);
ObjectMapper objectMapper = context.getBean(ObjectMapper.class);
Message<byte[]> response = target.receive(1000);
assertThat(response).isNotNull();
DataFlowTaskLaunchRequest request = objectMapper.readValue(response.getPayload(),
DataFlowTaskLaunchRequest.class);
assertThat(request.getCommandlineArguments()).containsExactlyInAnyOrder("foo=HELLO", "bar=he");
});
}
@Test
public void taskLaunchRequestWithIntPayload() {
applicationContextRunner.withPropertyValues(
"spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequest",
"task.launch.request.task-name=foo",
"task.launch.request.arg-expressions=i=payload")
.run(context -> {
ObjectMapper objectMapper = context.getBean(ObjectMapper.class);
MessageChannel input = context.getBean("input", MessageChannel.class);
OutputDestination target = context.getBean(OutputDestination.class);
Message<Integer> message =
MessageBuilder.withPayload(123).build();
input.send(message);
Message<byte[]> response = target.receive(1000);
assertThat(response).isNotNull();
DataFlowTaskLaunchRequest request = objectMapper.readValue(response.getPayload(),
DataFlowTaskLaunchRequest.class);
assertThat(request.getCommandlineArguments()).containsExactly("i=123");
});
}
@Test
public void taskNameExpression() {
applicationContextRunner.withPropertyValues(
"spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequest",
"task.launch.request.task-name-expression=payload+'_task'")
.run(context -> {
MessageChannel input = context.getBean("input", MessageChannel.class);
OutputDestination target = context.getBean(OutputDestination.class);
ObjectMapper objectMapper = context.getBean(ObjectMapper.class);
Message<String> message = MessageBuilder.withPayload("foo").build();
input.send(message);
Message<byte[]> response = target.receive(1000);
assertThat(response).isNotNull();
DataFlowTaskLaunchRequest request = objectMapper.readValue(response.getPayload(),
DataFlowTaskLaunchRequest.class);
assertThat(request.getTaskName()).isEqualTo("foo_task");
});
}
@Test
public void customTaskNameExtractor() {
applicationContextRunner.withPropertyValues(
"spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequest",
"customTaskNameExtractor=true")
.run(context -> {
MessageChannel input = context.getBean("input", MessageChannel.class);
OutputDestination target = context.getBean(OutputDestination.class);
ObjectMapper objectMapper = context.getBean(ObjectMapper.class);
Message<String> message = MessageBuilder.withPayload("foo").build();
input.send(message);
Message<byte[]> response = target.receive(1000);
assertThat(response).isNotNull();
DataFlowTaskLaunchRequest request = objectMapper.readValue(response.getPayload(),
DataFlowTaskLaunchRequest.class);
assertThat(request.getTaskName()).isEqualTo("fooTask");
});
//TODO: Workaround for https://github.com/spring-cloud/spring-cloud-stream/issues/1876
applicationContextRunner.withPropertyValues(
"spring.jmx.enabled=false", "spring.cloud.stream.function.definition=taskLaunchRequest",
"customTaskNameExtractor=true")
.run(context -> {
MessageChannel input = context.getBean("input", MessageChannel.class);
OutputDestination target = context.getBean(OutputDestination.class);
ObjectMapper objectMapper = context.getBean(ObjectMapper.class);
Message<String> message = MessageBuilder.withPayload("bar").build();
input.send(message);
Message<byte[]> response = target.receive(1000);
assertThat(response).isNotNull();
DataFlowTaskLaunchRequest request = objectMapper.readValue(response.getPayload(),
DataFlowTaskLaunchRequest.class);
assertThat(request.getTaskName()).isEqualTo("defaultTask");
});
}
private DataFlowTaskLaunchRequest verifyAndreceiveDataFlowTaskLaunchRequest(ApplicationContext context)
throws IOException {
MessageChannel input = context.getBean("input", MessageChannel.class);
OutputDestination target = context.getBean(OutputDestination.class);
MessageBuilder<byte[]> builder = MessageBuilder.withPayload(new byte[]{});
ObjectMapper objectMapper = context.getBean(ObjectMapper.class);
input.send(builder.build());
Message<byte[]> message = target.receive(1000);
assertThat(message).isNotNull();
return objectMapper.readValue(message.getPayload(),
DataFlowTaskLaunchRequest.class);
}
@EnableAutoConfiguration(exclude = {TestSupportBinderAutoConfiguration.class,
MessageCollectorAutoConfiguration.class})
@EnableBinding(Processor.class)
static class TestApp {
@Bean
ObjectMapper objectMapper() {
return new ObjectMapper();
}
@Bean
@ConditionalOnProperty("customTaskNameExtractor")
TaskNameMessageMapper taskNameExtractor() {
return message -> ((String) (message.getPayload())).equalsIgnoreCase("foo") ?
"fooTask" :
"defaultTask";
}
@Bean
@ConditionalOnProperty("enhanceTLRArgs")
CommandLineArgumentsMessageMapper commandLineArgumentsProvider() {
return message -> Collections.singletonList("runtimeArg");
}
@Bean
public IntegrationFlow flow() {
return IntegrationFlows.from(Processor.INPUT)
.channel(Processor.OUTPUT).get();
}
}
}

1
applications/stream-applications-core/pom.xml
View File

@@ -56,7 +56,6 @@
<module>common/stream-applications-test-support</module>
<module>common/stream-applications-postprocessor-common</module>
<module>common/stream-applications-micrometer-common</module>
<module>common/stream-applications-task-launch-request-common</module>
<module>common/stream-applications-security-common</module>
</modules>

22
functions/function/task-launch-request-function/README.adoc Normal file
View File

@@ -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.

37
functions/function/task-launch-request-function/pom.xml Normal file
View File

@@ -0,0 +1,37 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>1.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent</relativePath>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>task-launch-request-function</artifactId>
<name>task-launch-request-function</name>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-json</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
</project>

4
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/java/org/springframework/cloud/stream/app/tasklaunchrequest/support/CommandLineArgumentsMessageMapper.java → functions/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/CommandLineArgumentsMessageMapper.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* 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.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.tasklaunchrequest.support;
package org.springframework.cloud.fn.task.launch.request;
import java.util.Collection;

4
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/java/org/springframework/cloud/stream/app/tasklaunchrequest/KeyValueListParser.java → functions/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/KeyValueListParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* 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.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.tasklaunchrequest;
package org.springframework.cloud.fn.task.launch.request;
import java.util.ArrayList;
import java.util.HashMap;

24
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/java/org/springframework/cloud/stream/app/tasklaunchrequest/DataFlowTaskLaunchRequest.java → functions/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequest.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* 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.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.tasklaunchrequest;
package org.springframework.cloud.fn.task.launch.request;
import java.util.ArrayList;
import java.util.Collection;
@@ -24,7 +24,7 @@ import java.util.Map;
import com.fasterxml.jackson.annotation.JsonProperty;
public class DataFlowTaskLaunchRequest {
public class TaskLaunchRequest {
@JsonProperty("args")
private List<String> commandlineArguments = new ArrayList<>();
@@ -34,31 +34,31 @@ public class DataFlowTaskLaunchRequest {
@JsonProperty("name")
private String taskName;
public List<String> getCommandlineArguments() {
return this.commandlineArguments;
}
public void setCommandlineArguments(List<String> commandlineArguments) {
this.commandlineArguments = new ArrayList<>(commandlineArguments);
}
public Map<String, String> getDeploymentProperties() {
return this.deploymentProperties;
public List<String> getCommandlineArguments() {
return this.commandlineArguments;
}
public void setDeploymentProperties(Map<String, String> deploymentProperties) {
this.deploymentProperties = deploymentProperties;
}
public String getTaskName() {
return this.taskName;
public Map<String, String> getDeploymentProperties() {
return this.deploymentProperties;
}
public void setTaskName(String taskName) {
this.taskName = taskName;
}
public DataFlowTaskLaunchRequest addCommmandLineArguments(Collection<String> args) {
public String getTaskName() {
return this.taskName;
}
public TaskLaunchRequest addCommmandLineArguments(Collection<String> args) {
this.commandlineArguments.addAll(args);
return this;
}

6
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/java/org/springframework/cloud/stream/app/tasklaunchrequest/TaskLaunchRequestFunction.java → functions/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunction.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* 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.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.tasklaunchrequest;
package org.springframework.cloud.fn.task.launch.request;
import java.util.function.Function;
@@ -26,6 +26,6 @@ import org.springframework.messaging.Message;
* @author David Turanski
**/
@FunctionalInterface
public interface TaskLaunchRequestFunction extends Function<Message<?>, Message<DataFlowTaskLaunchRequest>> {
public interface TaskLaunchRequestFunction extends Function<Message<?>, Message<TaskLaunchRequest>> {
}

184
functions/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionConfiguration.java Normal file
View File

@@ -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<String, Expression> 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<String> processMessage(Message<?> message) {
return evaluateArgExpressions(message);
}
private Collection<String> evaluateArgExpressions(Message<?> message) {
List<String> results = new LinkedList<>();
this.argExpressionsMap.forEach((k, expression) -> results
.add(String.format("%s=%s", k, expression.getValue(this.evaluationContext, message))));
return results;
}
}
}

8
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/java/org/springframework/cloud/stream/app/tasklaunchrequest/DataflowTaskLaunchRequestProperties.java → functions/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionProperties.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* 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.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.tasklaunchrequest;
package org.springframework.cloud.fn.task.launch.request;
import java.util.ArrayList;
import java.util.List;
@@ -27,14 +27,14 @@ import org.springframework.util.StringUtils;
import org.springframework.validation.annotation.Validated;
/**
* Base Properties to create a {@link DataFlowTaskLaunchRequest}.
* Base Properties to create a {@link TaskLaunchRequest}.
*
* @author Chris Schaefer
* @author David Turanski
*/
@Validated
@ConfigurationProperties("task.launch.request")
public class DataflowTaskLaunchRequestProperties {
public class TaskLaunchRequestFunctionProperties {
/**
* Comma separated list of optional args in key=value format.

32
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/java/org/springframework/cloud/stream/app/tasklaunchrequest/TaskLaunchRequestMessageProcessor.java → functions/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestMessageProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* 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.
@@ -14,11 +14,8 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.tasklaunchrequest;
package org.springframework.cloud.fn.task.launch.request;
import org.springframework.cloud.stream.app.tasklaunchrequest.support.CommandLineArgumentsMessageMapper;
import org.springframework.cloud.stream.app.tasklaunchrequest.support.TaskLaunchRequestSupplier;
import org.springframework.cloud.stream.app.tasklaunchrequest.support.TaskNameMessageMapper;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.core.MessagePostProcessor;
@@ -27,15 +24,17 @@ import org.springframework.util.Assert;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.StringUtils;
public class TaskLaunchRequestMessageProcessor implements MessagePostProcessor {
class TaskLaunchRequestMessageProcessor implements MessagePostProcessor {
private final TaskNameMessageMapper taskNameMessageMapper;
private final CommandLineArgumentsMessageMapper commandLineArgumentsMessageMapper;
private final TaskLaunchRequestSupplier taskLaunchRequestInitializer;
public TaskLaunchRequestMessageProcessor(TaskLaunchRequestSupplier taskLaunchRequestInitializer,
TaskNameMessageMapper taskNameMessageMapper,
CommandLineArgumentsMessageMapper commandLIneArgumentsMessageMapper) {
TaskLaunchRequestMessageProcessor(TaskLaunchRequestSupplier taskLaunchRequestInitializer,
TaskNameMessageMapper taskNameMessageMapper,
CommandLineArgumentsMessageMapper commandLIneArgumentsMessageMapper) {
this.taskLaunchRequestInitializer = taskLaunchRequestInitializer;
@@ -46,24 +45,23 @@ public class TaskLaunchRequestMessageProcessor implements MessagePostProcessor {
}
@Override
public Message<DataFlowTaskLaunchRequest> postProcessMessage(Message<?> message) {
DataFlowTaskLaunchRequest taskLaunchRequest = taskLaunchRequestInitializer.get();
public Message<TaskLaunchRequest> 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 " + DataFlowTaskLaunchRequest.class.getName());
Assert.hasText(taskLaunchRequest.getTaskName(),
() -> "'taskName' is required in " + TaskLaunchRequest.class.getName());
}
taskLaunchRequest.addCommmandLineArguments(commandLineArgumentsMessageMapper.processMessage(message));
MessageBuilder<DataFlowTaskLaunchRequest> builder
= MessageBuilder.withPayload(taskLaunchRequest).copyHeaders(message.getHeaders());
MessageBuilder<TaskLaunchRequest> builder = MessageBuilder.withPayload(taskLaunchRequest)
.copyHeaders(message.getHeaders());
return adjustHeaders(builder).build();
}
private MessageBuilder<DataFlowTaskLaunchRequest> adjustHeaders(MessageBuilder<DataFlowTaskLaunchRequest> builder) {
private MessageBuilder<TaskLaunchRequest> adjustHeaders(MessageBuilder<TaskLaunchRequest> builder) {
builder.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
return builder;
}

27
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/java/org/springframework/cloud/stream/app/tasklaunchrequest/support/TaskLaunchRequestSupplier.java → functions/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestSupplier.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* 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.
@@ -14,21 +14,21 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.tasklaunchrequest.support;
package org.springframework.cloud.fn.task.launch.request;
import java.util.List;
import java.util.Map;
import java.util.function.Supplier;
import org.springframework.cloud.stream.app.tasklaunchrequest.DataFlowTaskLaunchRequest;
import org.springframework.util.Assert;
public class TaskLaunchRequestSupplier implements Supplier<DataFlowTaskLaunchRequest> {
class TaskLaunchRequestSupplier implements Supplier<TaskLaunchRequest> {
private Supplier<String> taskNameSupplier;
private Supplier<List<String>> commandLineArgumentsSupplier;
private Supplier<Map<String, String>> deploymentPropertiesSupplier;
private Supplier<List<String>> commandLineArgumentsSupplier;
private Supplier<Map<String, String>> deploymentPropertiesSupplier;
public TaskLaunchRequestSupplier taskNameSupplier(Supplier<String> taskNameSupplier) {
this.taskNameSupplier = taskNameSupplier;
@@ -40,27 +40,28 @@ public class TaskLaunchRequestSupplier implements Supplier<DataFlowTaskLaunchReq
return this;
}
public TaskLaunchRequestSupplier deploymentPropertiesSupplier(Supplier<Map<String, String>> deploymentPropertiesSupplier) {
public TaskLaunchRequestSupplier deploymentPropertiesSupplier(
Supplier<Map<String, String>> deploymentPropertiesSupplier) {
this.deploymentPropertiesSupplier = deploymentPropertiesSupplier;
return this;
}
@Override
public DataFlowTaskLaunchRequest get() {
public TaskLaunchRequest get() {
Assert.notNull(this.taskNameSupplier, "'taskNameSupplier' is required.");
DataFlowTaskLaunchRequest dataFlowTaskLaunchRequest = new DataFlowTaskLaunchRequest();
dataFlowTaskLaunchRequest.setTaskName(this.taskNameSupplier.get());
TaskLaunchRequest taskLaunchRequest = new TaskLaunchRequest();
taskLaunchRequest.setTaskName(this.taskNameSupplier.get());
if (this.commandLineArgumentsSupplier != null) {
dataFlowTaskLaunchRequest.setCommandlineArguments(this.commandLineArgumentsSupplier.get());
taskLaunchRequest.setCommandlineArguments(this.commandLineArgumentsSupplier.get());
}
if (this.deploymentPropertiesSupplier != null) {
dataFlowTaskLaunchRequest.setDeploymentProperties(this.deploymentPropertiesSupplier.get());
taskLaunchRequest.setDeploymentProperties(this.deploymentPropertiesSupplier.get());
}
return dataFlowTaskLaunchRequest;
return taskLaunchRequest;
}
}

4
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/main/java/org/springframework/cloud/stream/app/tasklaunchrequest/support/TaskNameMessageMapper.java → functions/function/task-launch-request-function/src/main/java/org/springframework/cloud/fn/task/launch/request/TaskNameMessageMapper.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* 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.
@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.tasklaunchrequest.support;
package org.springframework.cloud.fn.task.launch.request;
import org.springframework.integration.handler.MessageProcessor;

97
functions/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/KeyValueListParserTests.java Normal file
View File

@@ -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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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"));
}
}

211
functions/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionApplicationTests.java Normal file
View File

@@ -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<String> message = MessageBuilder.withPayload("hello").build();
TaskLaunchRequestFunction taskLaunchRequestFunction = context.getBean(TaskLaunchRequestFunction.class);
Message<TaskLaunchRequest> 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<Integer> message = MessageBuilder.withPayload(123).build();
Message<TaskLaunchRequest> 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<TaskLaunchRequest> 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<String> message = MessageBuilder.withPayload("foo").build();
Message<TaskLaunchRequest> 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<TaskLaunchRequest> 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");
}
}
}

21
applications/stream-applications-core/common/stream-applications-task-launch-request-common/src/test/java/org/springframework/cloud/stream/app/tasklaunchrequest/TaskLaunchRequestPropertiesTests.java → functions/function/task-launch-request-function/src/test/java/org/springframework/cloud/fn/task/launch/request/TaskLaunchRequestFunctionPropertiesTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* 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.
@@ -14,11 +14,11 @@
* limitations under the License.
*/
package org.springframework.cloud.stream.app.tasklaunchrequest;
package org.springframework.cloud.fn.task.launch.request;
import java.util.List;
import org.junit.Test;
import org.junit.jupiter.api.Test;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.test.util.TestPropertyValues;
@@ -32,18 +32,18 @@ import static org.assertj.core.api.Assertions.assertThat;
/**
* @author David Turanski
**/
public class TaskLaunchRequestPropertiesTests {
public class TaskLaunchRequestFunctionPropertiesTests {
@Test
public void deploymentPropertiesCanBeCustomized() {
DataflowTaskLaunchRequestProperties properties = getBatchProperties(
TaskLaunchRequestFunctionProperties properties = getBatchProperties(
"task.launch.request.deploymentProperties:prop1=val1,prop2=val2");
assertThat(properties.getDeploymentProperties()).isEqualTo("prop1=val1,prop2=val2");
}
@Test
public void parametersCanBeCustomized() {
DataflowTaskLaunchRequestProperties properties = getBatchProperties(
TaskLaunchRequestFunctionProperties properties = getBatchProperties(
"task.launch.request.args:jp1=jpv1,jp2=jpv2");
List<String> args = properties.getArgs();
@@ -53,7 +53,7 @@ public class TaskLaunchRequestPropertiesTests {
assertThat(args.get(1)).isEqualTo("jp2=jpv2");
}
private DataflowTaskLaunchRequestProperties getBatchProperties(String... var) {
private TaskLaunchRequestFunctionProperties getBatchProperties(String... var) {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
if (var != null) {
@@ -63,14 +63,13 @@ public class TaskLaunchRequestPropertiesTests {
context.register(Conf.class);
context.refresh();
return context.getBean(DataflowTaskLaunchRequestProperties.class);
return context.getBean(TaskLaunchRequestFunctionProperties.class);
}
@Configuration
@EnableIntegration
@EnableConfigurationProperties(DataflowTaskLaunchRequestProperties.class)
@Import(DataFlowTaskLaunchRequestAutoConfiguration.class)
@EnableConfigurationProperties(TaskLaunchRequestFunctionProperties.class)
@Import(TaskLaunchRequestFunctionConfiguration.class)
static class Conf {
}

1
functions/pom.xml
View File

@@ -75,6 +75,7 @@
<module>function/payload-converter-function</module>
<module>function/splitter-function</module>
<module>function/tasklauncher-function</module>
<module>function/task-launch-request-function</module>
<module>supplier/file-supplier</module>
<module>supplier/ftp-supplier</module>