SpringCloud/spring-cloud-bus
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Allows setting output binding properties.

Browse Source 
This fixes the StreamBridge.send() binding name to BusConstants.OUTPUT and sets the BusConstants.OUTPUT destination to spring.cloud.bus.destination.

Fixes gh-244
This commit is contained in:
spencergibb
2021-05-26 16:32:31 -04:00
parent f98518e15a
commit fe97a24e73
5 changed files with 15 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
spring-cloud-bus-tests/src/test/java/org/springframework/cloud/bus/BusAmqpIntegrationTests.java
View File

@@ -33,6 +33,8 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.context.ApplicationListener;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.test.context.DynamicPropertyRegistry;
@@ -43,6 +45,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;
@SpringBootTest(webEnvironment = RANDOM_PORT, properties = { "management.endpoints.web.exposure.include=*",
"spring.cloud.stream.bindings.springCloudBusOutput.producer.errorChannelEnabled=true",
"logging.level.org.springframework.cloud.bus=TRACE", "spring.cloud.bus.id=app:1",
"spring.autoconfigure.exclude=org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration" })
@Testcontainers
@@ -53,6 +56,9 @@ public class BusAmqpIntegrationTests {
private static ConfigurableApplicationContext context;
@Autowired
private BindingServiceProperties bindingServiceProperties;
@DynamicPropertySource
static void properties(DynamicPropertyRegistry registry) {
registry.add("spring.rabbitmq.host", rabbitMQContainer::getHost);
@@ -87,6 +93,8 @@ public class BusAmqpIntegrationTests {
TestConfig remoteTestConfig = context.getBean(TestConfig.class);
assertThat(remoteTestConfig.latch.await(5, TimeUnit.SECONDS)).isTrue();
assertThat(testConfig.latch.await(5, TimeUnit.SECONDS)).isTrue();
ProducerProperties producerProperties = bindingServiceProperties.getProducerProperties(BusConstants.OUTPUT);
assertThat(producerProperties.isErrorChannelEnabled()).isTrue();
}
@SpringBootConfiguration

2
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusConstants.java
View File

@@ -33,7 +33,7 @@ public abstract class BusConstants {
public static final String OUTPUT = "springCloudBusOutput";
/**
* Name of the output channel for Spring Cloud Bus.
* Name of the binding destination for Spring Cloud Bus.
*/
public static final String DESTINATION = "springCloudBus";

1
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java
View File

@@ -70,6 +70,7 @@ public class BusEnvironmentPostProcessor implements EnvironmentPostProcessor {
BusConstants.INPUT);
String destination = environment.getProperty(PREFIX + ".destination", BusConstants.DESTINATION);
defaults.put("spring.cloud.stream.bindings." + BusConstants.INPUT + ".destination", destination);
defaults.put("spring.cloud.stream.bindings." + BusConstants.OUTPUT + ".destination", destination);
if (!environment.containsProperty(PREFIX + ".id")) {
defaults.put(PREFIX + ".id", IdUtils.getUnresolvedServiceId());
}

2
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/StreamBusBridge.java
View File

@@ -33,7 +33,7 @@ public class StreamBusBridge implements BusBridge {
public void send(RemoteApplicationEvent event) {
// TODO: configurable mimetype?
this.streamBridge.send(properties.getDestination(), MessageBuilder.withPayload(event).build());
this.streamBridge.send(BusConstants.OUTPUT, MessageBuilder.withPayload(event).build());
}
}

4
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusEnvironmentPostProcessorTests.java
View File

@@ -27,6 +27,7 @@ import static org.mockito.Mockito.mock;
import static org.springframework.cloud.bus.BusConstants.BUS_CONSUMER;
import static org.springframework.cloud.bus.BusConstants.DESTINATION;
import static org.springframework.cloud.bus.BusConstants.INPUT;
import static org.springframework.cloud.bus.BusConstants.OUTPUT;
import static org.springframework.cloud.bus.BusEnvironmentPostProcessor.DEFAULTS_PROPERTY_SOURCE_NAME;
import static org.springframework.cloud.bus.BusEnvironmentPostProcessor.OVERRIDES_PROPERTY_SOURCE_NAME;
@@ -39,6 +40,7 @@ public class BusEnvironmentPostProcessorTests {
assertThat(env.getProperty(FunctionProperties.PREFIX + ".definition")).isEqualTo(BUS_CONSUMER);
assertThat(env.getProperty("spring.cloud.stream.function.bindings." + BUS_CONSUMER + "-in-0")).isEqualTo(INPUT);
assertThat(env.getProperty("spring.cloud.stream.bindings." + INPUT + ".destination")).isEqualTo(DESTINATION);
assertThat(env.getProperty("spring.cloud.stream.bindings." + OUTPUT + ".destination")).isEqualTo(DESTINATION);
assertThat(env.getProperty(BusProperties.PREFIX + ".id")).isNotBlank();
assertThat(env.getPropertySources().contains(OVERRIDES_PROPERTY_SOURCE_NAME));
assertThat(env.getPropertySources().contains(DEFAULTS_PROPERTY_SOURCE_NAME));
@@ -56,6 +58,8 @@ public class BusEnvironmentPostProcessorTests {
assertThat(env.getProperty("spring.cloud.stream.function.bindings." + BUS_CONSUMER + "-in-0")).isEqualTo(INPUT);
assertThat(env.getProperty("spring.cloud.stream.bindings." + INPUT + ".destination"))
.isEqualTo("mydestination");
assertThat(env.getProperty("spring.cloud.stream.bindings." + OUTPUT + ".destination"))
.isEqualTo("mydestination");
assertThat(env.getProperty(idKey)).isEqualTo("app:1");
assertThat(env.getPropertySources().contains(OVERRIDES_PROPERTY_SOURCE_NAME));
assertThat(env.getPropertySources().contains(DEFAULTS_PROPERTY_SOURCE_NAME));