From 6b1d584c6c317be26a7ce951c1a008e14c98fd1c Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 16 Aug 2018 13:26:49 +0200 Subject: [PATCH] GH-1440: Add support for Supplier as Source Resolves spring-cloud/spring-cloud-stream#1440 Added initial support and configuration to create Source apps by simply providing a Supplier bean while also benefiting from composition provided by SCF Refactored/renamed some of the new classes/methods addressed PR comments * Polishing code style and fix typos --- .../function/FunctionConfiguration.java | 8 +- .../stream/function/FunctionSupport.java | 86 ------- .../IntegrationFlowFunctionSupport.java | 142 +++++++++++ .../stream/function/FunctionSupportTests.java | 134 ---------- .../SourceToFunctionsSupportTests.java | 232 ++++++++++++++++++ 5 files changed, 379 insertions(+), 223 deletions(-) delete mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionSupport.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java delete mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionSupportTests.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index 797aaa9aa..ff9ee83a1 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -34,10 +34,12 @@ import org.springframework.context.annotation.Configuration; public class FunctionConfiguration { @Bean - public FunctionSupport functionSupport(FunctionCatalog functionCatalog, FunctionInspector functionInspector, - CompositeMessageConverterFactory messageConverterFactory) { + public IntegrationFlowFunctionSupport functionSupport(FunctionCatalog functionCatalog, + FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory, + FunctionProperties functionProperties) { - return new FunctionSupport(functionCatalog, functionInspector, messageConverterFactory); + return new IntegrationFlowFunctionSupport(functionCatalog, functionInspector, messageConverterFactory, + functionProperties); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionSupport.java deleted file mode 100644 index e5d261322..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionSupport.java +++ /dev/null @@ -1,86 +0,0 @@ -/* - * Copyright 2018 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.function; - -import java.util.function.Consumer; - -import org.reactivestreams.Publisher; - -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -import org.springframework.cloud.function.context.FunctionCatalog; -import org.springframework.cloud.function.context.catalog.FunctionInspector; -import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; -import org.springframework.integration.dsl.IntegrationFlowBuilder; -import org.springframework.messaging.Message; -import org.springframework.util.Assert; - - - -/** - * - * @author Oleg Zhurakousky - * - * @since 2.1 - */ -public class FunctionSupport { - - private final FunctionCatalog functionCatalog; - - private final FunctionInspector functionInspector; - - private final CompositeMessageConverterFactory messageConverterFactory; - - public FunctionSupport(FunctionCatalog functionCatalog, FunctionInspector functionInspector, - CompositeMessageConverterFactory messageConverterFactory) { - - Assert.notNull(functionCatalog, "'functionCatalog' must not be null"); - Assert.notNull(functionInspector, "'functionInspector' must not be null"); - Assert.notNull(messageConverterFactory, "'messageConverterFactory' must not be null"); - this.functionCatalog = functionCatalog; - this.functionInspector = functionInspector; - this.messageConverterFactory = messageConverterFactory; - } - - public void applyFunctionToIntegrationFlow(String functionName, IntegrationFlowBuilder flowBuilder, - Consumer> outputProcessor) { - - FunctionInvoker functionInvoker = - new FunctionInvoker<>(functionName, this.functionCatalog, this.functionInspector, - this.messageConverterFactory); - - subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), outputProcessor); - } - - private Mono subscribeToOutput(Consumer> outputProcessor, - Publisher> outputPublisher) { - - Flux> output = outputProcessor == null - ? Flux.from(outputPublisher) - : Flux.from(outputPublisher).doOnNext(outputProcessor); - return output.then(); - } - - private void subscribeToInput(FunctionInvoker functionInvoker, Publisher> publisher, - Consumer> outputProcessor) { - - Flux> inputPublisher = Flux.from(publisher); - subscribeToOutput(outputProcessor, functionInvoker.apply(inputPublisher)).subscribe(); - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java new file mode 100644 index 000000000..fe1d1898a --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java @@ -0,0 +1,142 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.function; + +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.reactivestreams.Publisher; + +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import org.springframework.cloud.function.context.FunctionCatalog; +import org.springframework.cloud.function.context.catalog.FunctionInspector; +import org.springframework.cloud.function.core.FluxSupplier; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.integration.dsl.IntegrationFlowBuilder; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * + * @author Oleg Zhurakousky + * + * @since 2.1 + */ +public class IntegrationFlowFunctionSupport { + + private final FunctionCatalog functionCatalog; + + private final FunctionInspector functionInspector; + + private final CompositeMessageConverterFactory messageConverterFactory; + + private final FunctionProperties functionProperties; + + /** + * @param functionCatalog + * @param functionInspector + * @param messageConverterFactory + * @param functionProperties + */ + public IntegrationFlowFunctionSupport(FunctionCatalog functionCatalog, FunctionInspector functionInspector, + CompositeMessageConverterFactory messageConverterFactory, FunctionProperties functionProperties) { + + Assert.notNull(functionCatalog, "'functionCatalog' must not be null"); + Assert.notNull(functionInspector, "'functionInspector' must not be null"); + Assert.notNull(messageConverterFactory, "'messageConverterFactory' must not be null"); + Assert.notNull(functionProperties, "'functionProperties' must not be null"); + this.functionCatalog = functionCatalog; + this.functionInspector = functionInspector; + this.messageConverterFactory = messageConverterFactory; + this.functionProperties = functionProperties; + } + + /** + * Create an instance of the {@link IntegrationFlowBuilder} from a {@link Supplier} bean available in the context. + * The name of the bean must be provided via `spring.cloud.stream.function.name` property. + * @return instance of {@link IntegrationFlowBuilder} + * @throws IllegalStateException if the named Supplier can not be located. + */ + public IntegrationFlowBuilder integrationFlowFromNamedSupplier() { + if (StringUtils.hasText(this.functionProperties.getName())) { + Supplier supplier = functionCatalog.lookup(Supplier.class, this.functionProperties.getName()); + if (supplier instanceof FluxSupplier) { + supplier = ((FluxSupplier)supplier).getTarget(); + } + return integrationFlowFromProvidedSupplier(supplier) + .split(); + } + + throw new IllegalStateException( + "A Supplier is not specified in the 'spring.cloud.stream.function.name' property."); + } + + /** + * Create an instance of the {@link IntegrationFlowBuilder} from a provided {@link Supplier}. + * @return instance of {@link IntegrationFlowBuilder} + */ + public IntegrationFlowBuilder integrationFlowFromProvidedSupplier(Supplier supplier) { + return IntegrationFlows.from(supplier); + } + + /** + * Add a {@link Function} bean to the end of an integration flow. + * The name of the bean must be provided via `spring.cloud.stream.function.name` property. + *

+ * NOTE: If this method returns true, the integration flow is now represented + * as a Reactive Streams {@link Publisher} bean. + *

+ * @param flowBuilder instance of the {@link IntegrationFlowBuilder} representing + * the current state of the integration flow + * @param outputChannel channel where the output of a function will be sent + * @return true if {@link Function} was located and added and false if it wasn't. + */ + public boolean andThenFunction(IntegrationFlowBuilder flowBuilder, MessageChannel outputChannel) { + if (StringUtils.hasText(this.functionProperties.getName())) { + FunctionInvoker functionInvoker = + new FunctionInvoker<>(this.functionProperties.getName(), this.functionCatalog, + this.functionInspector, this.messageConverterFactory); + + subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), outputChannel::send); + return true; + } + return false; + } + + private Mono subscribeToOutput(Consumer> outputProcessor, + Publisher> outputPublisher) { + + Flux> output = outputProcessor == null + ? Flux.from(outputPublisher) + : Flux.from(outputPublisher).doOnNext(outputProcessor); + return output.then(); + } + + private void subscribeToInput(FunctionInvoker functionInvoker, Publisher> publisher, + Consumer> outputProcessor) { + + Flux> inputPublisher = Flux.from(publisher); + subscribeToOutput(outputProcessor, functionInvoker.apply(inputPublisher)).subscribe(); + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionSupportTests.java deleted file mode 100644 index 64d2591a6..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionSupportTests.java +++ /dev/null @@ -1,134 +0,0 @@ -/* - * Copyright 2018 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.stream.function; - -import java.nio.charset.StandardCharsets; - -import java.util.function.Function; -import java.util.function.Supplier; - -import org.junit.Test; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.WebApplicationType; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.binder.test.OutputDestination; -import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.context.ApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Import; -import org.springframework.integration.dsl.IntegrationFlow; -import org.springframework.integration.dsl.IntegrationFlowBuilder; -import org.springframework.integration.dsl.IntegrationFlows; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.util.MimeTypeUtils; -import org.springframework.util.StringUtils; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * - * @author Oleg Zhurakousky - * - * @since 2.1 - * - */ -public class FunctionSupportTests { - - @Test - public void testFunctionIsAppliedToMessageSource() { - ApplicationContext context = - new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(MyFunctionsConfiguration.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.name=toUpperCase", "--spring.jmx.enabled=false"); - - OutputDestination target = context.getBean(OutputDestination.class); - assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO FUNCTION".getBytes(StandardCharsets.UTF_8)); - } - - @Test - public void testComposedFunctionIsAppliedToMessageSource() { - ApplicationContext context = - new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(MyFunctionsConfiguration.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf", - "--spring.jmx.enabled=false"); - - OutputDestination target = context.getBean(OutputDestination.class); - assertThat(target.receive(1000).getPayload()) - .isEqualTo("HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8)); - } - - @SpringBootApplication - @Import(SourceConfiguration.class) - public static class MyFunctionsConfiguration { - - @Bean - public Function toUpperCase() { - return String::toUpperCase; - } - - @Bean - public Function concatWithSelf() { - return x -> x + ":" + x; - } - - } - - /** - * This configuration essentially emulates our existing app-starters for Sources - * and essentially demonstrates how a function(s) could be applied to an existing - * source via {@link FunctionSupport} class. - */ - @EnableBinding(Source.class) - public static class SourceConfiguration { - - @Autowired - private Source source; - - @Autowired - private FunctionProperties functionProperties; - - @Bean - public IntegrationFlow messageSourceFlow(FunctionSupport functionSupport) { - Supplier> messageSource = () -> MessageBuilder.withPayload("hello function") - .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build(); - - IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(messageSource); - - if (StringUtils.hasText(functionProperties.getName())) { - functionSupport - .applyFunctionToIntegrationFlow(functionProperties.getName(), flowBuilder, - message -> source.output().send(message)); - } - else { - flowBuilder = flowBuilder.channel(source.output()); - } - - return flowBuilder.get(); - } - - } - -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java new file mode 100644 index 000000000..854d0c24a --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java @@ -0,0 +1,232 @@ +/* + * Copyright 2018 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.function; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Function; +import java.util.function.Supplier; + +import org.junit.Test; + +import reactor.core.publisher.Flux; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlowBuilder; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.util.Assert; +import org.springframework.util.MimeTypeUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * + * @author Oleg Zhurakousky + * + * @since 2.1 + * + */ +public class SourceToFunctionsSupportTests { + + @Test + public void testFunctionIsAppliedToExistingMessageSource() { + try (ConfigurableApplicationContext context = + new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.name=toUpperCase", "--spring.jmx.enabled=false")) { + + OutputDestination target = context.getBean(OutputDestination.class); + assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO FUNCTION".getBytes(StandardCharsets.UTF_8)); + } + } + + @Test + public void testComposedFunctionIsAppliedToExistingMessageSource() { + try (ConfigurableApplicationContext context = + new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf", + "--spring.jmx.enabled=false")) { + OutputDestination target = context.getBean(OutputDestination.class); + assertThat(target.receive(1000).getPayload()) + .isEqualTo("HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8)); + } + } + + @Test + public void testMessageSourceIsCreatedFromProvidedSupplier() { + try (ConfigurableApplicationContext context = + new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.name=number", + "--spring.jmx.enabled=false")) { + + OutputDestination target = context.getBean(OutputDestination.class); + assertThat(target.receive(10000).getPayload()) + .isEqualTo("1".getBytes(StandardCharsets.UTF_8)); + assertThat(target.receive(10000).getPayload()) + .isEqualTo("2".getBytes(StandardCharsets.UTF_8)); + assertThat(target.receive(10000).getPayload()) + .isEqualTo("3".getBytes(StandardCharsets.UTF_8)); + //etc + } + } + + @Test + public void testMessageSourceIsCreatedFromProvidedSupplierComposedWithSingleFunction() { + try (ConfigurableApplicationContext context = + new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.name=number|concatWithSelf", + "--spring.jmx.enabled=false")) { + + OutputDestination target = context.getBean(OutputDestination.class); + assertThat(target.receive(10000).getPayload()) + .isEqualTo("11".getBytes(StandardCharsets.UTF_8)); + assertThat(target.receive(10000).getPayload()) + .isEqualTo("22".getBytes(StandardCharsets.UTF_8)); + assertThat(target.receive(10000).getPayload()) + .isEqualTo("33".getBytes(StandardCharsets.UTF_8)); + //etc + } + } + + @Test + public void testMessageSourceIsCreatedFromProvidedSupplierComposedWithMultipleFunctions() { + try (ConfigurableApplicationContext context = + new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.name=number|concatWithSelf|multiplyByTwo", + "--spring.jmx.enabled=false")) { + + OutputDestination target = context.getBean(OutputDestination.class); + assertThat(target.receive(10000).getPayload()) + .isEqualTo("22".getBytes(StandardCharsets.UTF_8)); + assertThat(target.receive(10000).getPayload()) + .isEqualTo("44".getBytes(StandardCharsets.UTF_8)); + assertThat(target.receive(10000).getPayload()) + .isEqualTo("66".getBytes(StandardCharsets.UTF_8)); + //etc + } + } + + @EnableAutoConfiguration + @Import(ProvidedMessageSourceConfiguration.class) + public static class SupplierConfiguration { + + AtomicInteger counter = new AtomicInteger(); + + @Bean + public Supplier number() { + return () -> String.valueOf(counter.incrementAndGet()); + } + + @Bean + public Function concatWithSelf() { + return x -> x + x; + } + + @Bean + public Function, Flux> multiplyByTwo() { + return x -> x.map(i -> String.valueOf(Integer.valueOf(i) * 2)); + } + + } + + @EnableAutoConfiguration + @Import(ExistingMessageSourceConfiguration.class) + public static class FunctionsConfiguration { + + @Bean + public Function toUpperCase() { + return String::toUpperCase; + } + + @Bean + public Function concatWithSelf() { + return x -> x + ":" + x; + } + + } + + /** + * This configuration essentially emulates our existing app-starters for Sources + * and essentially demonstrates how a function(s) could be applied to an existing + * source via {@link IntegrationFlowFunctionSupport} class. + */ + @EnableBinding(Source.class) + public static class ExistingMessageSourceConfiguration { + + @Autowired + private Source source; + + @Bean + public IntegrationFlow messageSourceFlow(IntegrationFlowFunctionSupport functionSupport) { + Supplier> messageSource = () -> MessageBuilder.withPayload("hello function") + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build(); + + IntegrationFlowBuilder flowBuilder = functionSupport.integrationFlowFromProvidedSupplier(messageSource); + + if (!functionSupport.andThenFunction(flowBuilder, this.source.output())) { + flowBuilder = flowBuilder.channel(this.source.output()); + } + + return flowBuilder.get(); + } + + } + + @EnableBinding(Source.class) + public static class ProvidedMessageSourceConfiguration { + + @Autowired + private Source source; + + @Autowired + private FunctionProperties functionProperties; + + @Bean + public IntegrationFlow messageSourceFlow(IntegrationFlowFunctionSupport functionSupport) { + Assert.hasText(this.functionProperties.getName(), "Supplier name must be provided"); + + return functionSupport + .integrationFlowFromNamedSupplier() + .channel(this.source.output()) + .get(); + } + + } + +}