add default broadcasting in spring-cloud-starter-bus-rocketmq
*/
package org.springframework.cloud.bus.rocketmq.env;
+import static org.springframework.cloud.bus.SpringCloudBusClient.INPUT;
+
+import java.util.HashMap;
+import java.util.Map;
+
import org.springframework.boot.SpringApplication;
import org.springframework.boot.env.EnvironmentPostProcessor;
import org.springframework.cloud.bus.BusEnvironmentPostProcessor;
@@ -25,20 +30,16 @@ import org.springframework.core.env.MapPropertySource;
import org.springframework.core.env.MutablePropertySources;
import org.springframework.core.env.PropertySource;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.springframework.cloud.bus.SpringCloudBusClient.INPUT;
-
/**
- * The lowest precedence {@link EnvironmentPostProcessor} configures default RocketMQ Bus Properties that will be
- * appended into {@link SpringApplication#defaultProperties}
+ * The lowest precedence {@link EnvironmentPostProcessor} configures default RocketMQ Bus
+ * Properties that will be appended into {@link SpringApplication#defaultProperties}
*
* @author <a href="mailto:mercyblitz@gmail.com">Mercy</a>
* @see BusEnvironmentPostProcessor
* @since 0.2.1
*/
-public class RocketMQBusEnvironmentPostProcessor implements EnvironmentPostProcessor, Ordered {
+public class RocketMQBusEnvironmentPostProcessor
+ implements EnvironmentPostProcessor, Ordered {
/**
* The name of {@link PropertySource} of {@link SpringApplication#defaultProperties}
@@ -46,7 +47,8 @@ public class RocketMQBusEnvironmentPostProcessor implements EnvironmentPostProce
private static final String PROPERTY_SOURCE_NAME = "defaultProperties";
@Override
- public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) {
+ public void postProcessEnvironment(ConfigurableEnvironment environment,
+ SpringApplication application) {
addDefaultPropertySource(environment);
@@ -64,7 +66,15 @@ public class RocketMQBusEnvironmentPostProcessor implements EnvironmentPostProce
private void configureDefaultProperties(Map<String, Object> source) {
// Required Properties
String groupBindingPropertyName = createBindingPropertyName(INPUT, "group");
+ String broadcastingPropertyName = createRocketMQPropertyName(INPUT,
+ "broadcasting");
source.put(groupBindingPropertyName, "rocketmq-bus-group");
+ source.put(broadcastingPropertyName, "true");
+ }
+
+ private String createRocketMQPropertyName(String channel, String propertyName) {
+ return "spring.cloud.stream.rocketmq.bindings." + INPUT + ".consumer."
+ + propertyName;
}
private String createBindingPropertyName(String channel, String propertyName) {
@@ -72,7 +82,8 @@ public class RocketMQBusEnvironmentPostProcessor implements EnvironmentPostProce
}
/**
- * Copy from {@link BusEnvironmentPostProcessor#addOrReplace(MutablePropertySources, Map)}
+ * Copy from
+ * {@link BusEnvironmentPostProcessor#addOrReplace(MutablePropertySources, Map)}
*
* @param propertySources {@link MutablePropertySources}
* @param map Default RocketMQ Bus Properties
