diff --git a/spring-cloud-stream-binder-rabbit-deployer/pom.xml b/spring-cloud-stream-binder-rabbit-deployer/pom.xml index 21cbfd17a..097207b8a 100644 --- a/spring-cloud-stream-binder-rabbit-deployer/pom.xml +++ b/spring-cloud-stream-binder-rabbit-deployer/pom.xml @@ -10,7 +10,7 @@ org.springframework.cloud spring-cloud-stream-binder-rabbit-parent - 3.1.2-SNAPSHOT + 3.2.0-SNAPSHOT @@ -23,6 +23,11 @@ spring-cloud-function-deployer ${project.version} + + org.springframework.cloud + spring-cloud-function-rsocket + ${project.version} + org.springframework.boot spring-boot-configuration-processor diff --git a/spring-cloud-stream-binder-rabbit-deployer/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployer.java b/spring-cloud-stream-binder-rabbit-deployer/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployer.java index d3d5341ff..b05dcc96b 100644 --- a/spring-cloud-stream-binder-rabbit-deployer/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployer.java +++ b/spring-cloud-stream-binder-rabbit-deployer/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployer.java @@ -16,11 +16,19 @@ package org.springframework.cloud.stream.binder.rabbit.deployer; +import java.util.function.Consumer; import java.util.function.Function; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.cloud.function.context.FunctionProperties; +import org.springframework.cloud.stream.function.StreamBridge; import org.springframework.context.annotation.Bean; +import org.springframework.core.env.Environment; +import org.springframework.messaging.Message; +import org.springframework.messaging.rsocket.RSocketRequester; +import org.springframework.messaging.support.MessageBuilder; /** @@ -32,24 +40,42 @@ import org.springframework.context.annotation.Bean; @SpringBootApplication public class RabbitDeployer { + /* + * SpringApplication.run(RabbitDeployer.class, + "--spring.cloud.function.definition=reverseFunction", + "--spring.cloud.function.location=/bootjar-1.0.0.RELEASE-exec.jar", + "--spring.cloud.function.function-class=function.example.ReverseFunction" + ); + */ public static void main(String[] args) { - SpringApplication.run(RabbitDeployer.class, - //target/it/bootjar/target/bootjar-1.0.0.RELEASE-exec.jar - "--spring.cloud.function.definition=reverseFunction;echo", - "--spring.cloud.function.location=/Users/ozhurakousky/dev/repo/spring-cloud-function/spring-cloud-function-deployer/target/it/bootjar/target/bootjar-1.0.0.RELEASE-exec.jar", - "--spring.cloud.function.function-class=function.example.ReverseFunction"); + SpringApplication.run(RabbitDeployer.class); } - @Bean - public Function echo() { - // java -jar timesource-kafka.jar + @ConditionalOnProperty(name = FunctionProperties.PREFIX + ".rsocket.enabled", matchIfMissing = true) + public Function, Message> gateway(StreamBridge bridge) { + return message -> { + String destinationName = (String) message.getHeaders().get("target_destination"); + bridge.send(destinationName, message); + return MessageBuilder.withPayload("Successfully sent to reverseFunction-in-0".getBytes()).build(); + }; + } - // java -jar rabbit-bundle.jar -Dspring.cloud.function.location=.....jar - // java -jar rabbit-kafka-bundle.jar -Dspring.cloud.function.location=.....jar - - // java -jar rabbit-rsocket-bundle.jar = GATEWAY - return v -> v; + /* + * Just like any other stream bean. This one will subscribe to broker destination (using regular stream mechanisms) + * and using some configuration provided by the user will propagate message to remote (rsocket) subscriber + */ + @Bean + @ConditionalOnProperty(name = FunctionProperties.PREFIX + ".rsocket.enabled", matchIfMissing = true) + public Consumer> delegatingConsumer(RSocketRequester.Builder rsocketRequesterBuilder, Environment environment) { + String host = environment.getProperty("spring.cloud.function.rsocket.subscriber.host"); + String port = environment.getProperty("spring.cloud.function.rsocket.subscriber.port"); + return message -> { +// rsocketRequesterBuilder.tcp("host", Integer.valueOf(port)) +// .route("pojoToString") +// .data(message) +// .retrieveMono(String.class); + }; } // Step-1 - rabbit-bundle.jar(time) | rabbit-bundle.jar(log) - Step One - local @@ -59,6 +85,4 @@ public class RabbitDeployer { //http | rabbit-rsocket-bundle.jar(producer) | python | rabbit-rsocket-bundle.jar(consumer) | rabbit-bundle.jar(log) - //http => pyjon - } diff --git a/spring-cloud-stream-binder-rabbit-deployer/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployerEnvironmentPostProcessor.java b/spring-cloud-stream-binder-rabbit-deployer/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployerEnvironmentPostProcessor.java new file mode 100644 index 000000000..c5b2adbe1 --- /dev/null +++ b/spring-cloud-stream-binder-rabbit-deployer/src/main/java/org/springframework/cloud/stream/binder/rabbit/deployer/RabbitDeployerEnvironmentPostProcessor.java @@ -0,0 +1,41 @@ +/* + * Copyright 2021-2021 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.binder.rabbit.deployer; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.env.EnvironmentPostProcessor; +import org.springframework.core.env.ConfigurableEnvironment; + +/** + * + * @author Oleg Zhurakousky + * + * @since 3.2 + * + */ +class RabbitDeployerEnvironmentPostProcessor implements EnvironmentPostProcessor { + + @Override + public void postProcessEnvironment(ConfigurableEnvironment environment, + SpringApplication application) { + if (!environment.containsProperty("spring.cloud.function.rsocket.enabled")) { + environment.getSystemProperties().putIfAbsent("spring.cloud.function.rsocket.enabled", false); + } + + } + +} diff --git a/spring-cloud-stream-binder-rabbit-deployer/src/main/resources/META-INF/spring.binders b/spring-cloud-stream-binder-rabbit-deployer/src/main/resources/META-INF/spring.binders deleted file mode 100644 index 7a65c4682..000000000 --- a/spring-cloud-stream-binder-rabbit-deployer/src/main/resources/META-INF/spring.binders +++ /dev/null @@ -1,2 +0,0 @@ -rabbit:\ -org.springframework.cloud.stream.binder.rabbit.config.RabbitServiceAutoConfiguration diff --git a/spring-cloud-stream-binder-rabbit-deployer/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-binder-rabbit-deployer/src/main/resources/META-INF/spring.factories index c299f8ddf..175847543 100644 --- a/spring-cloud-stream-binder-rabbit-deployer/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-stream-binder-rabbit-deployer/src/main/resources/META-INF/spring.factories @@ -1 +1,2 @@ -org.springframework.boot.autoconfigure.EnableAutoConfiguration=org.springframework.cloud.stream.binder.rabbit.config.ExtendedBindingHandlerMappingsProviderConfiguration +org.springframework.boot.env.EnvironmentPostProcessor:\ +org.springframework.cloud.stream.binder.rabbit.deployer.RabbitDeployerEnvironmentPostProcessor \ No newline at end of file diff --git a/spring-cloud-stream-binder-rabbit-deployer/src/test/resources/log4j.properties b/spring-cloud-stream-binder-rabbit-deployer/src/test/resources/log4j.properties deleted file mode 100644 index 335956bc0..000000000 --- a/spring-cloud-stream-binder-rabbit-deployer/src/test/resources/log4j.properties +++ /dev/null @@ -1,8 +0,0 @@ -log4j.rootCategory=DEBUG, stdout - -# standard logging including calling site -log4j.appender.stdout=org.apache.log4j.ConsoleAppender -log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %40.40c:%4L - %m%n - -log4j.category.org.springframework.cloud.binder.rabbit=DEBUG