study/Spring-Cloud-Alibaba
1
0
Fork 0
mirror of https://gitee.com/mirrors/Spring-Cloud-Alibaba.git synced 2021-06-26 13:25:11 +08:00
Code Issues Releases Wiki Activity

Updates to spring cloud greenwich

Browse Source
This commit is contained in:
fangjian0423
2019-02-19 17:01:09 +08:00
parent 94a3bc2cc2
commit 1dc80d7a6f
6 changed files with 41 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/RocketMQMessageChannelBinder.java
View File

@@ -22,6 +22,7 @@ import org.apache.rocketmq.client.producer.TransactionCheckListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
@@ -53,19 +54,18 @@ public class RocketMQMessageChannelBinder extends
private static final Logger logger = LoggerFactory
.getLogger(RocketMQMessageChannelBinder.class);
private final RocketMQExtendedBindingProperties extendedBindingProperties;
private final RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties;
private final InstrumentationManager instrumentationManager;
private final ConsumersManager consumersManager;
private RocketMQExtendedBindingProperties extendedBindingProperties = new RocketMQExtendedBindingProperties();
public RocketMQMessageChannelBinder(ConsumersManager consumersManager,
RocketMQExtendedBindingProperties extendedBindingProperties,
RocketMQTopicProvisioner provisioningProvider,
RocketMQBinderConfigurationProperties rocketBinderConfigurationProperties,
InstrumentationManager instrumentationManager) {
super(null, provisioningProvider);
this.consumersManager = consumersManager;
this.extendedBindingProperties = extendedBindingProperties;
this.rocketBinderConfigurationProperties = rocketBinderConfigurationProperties;
this.instrumentationManager = instrumentationManager;
}
@@ -143,6 +143,16 @@ public class RocketMQMessageChannelBinder extends
return extendedBindingProperties.getExtendedProducerProperties(channelName);
}
@Override
public String getDefaultsPrefix() {
return extendedBindingProperties.getDefaultsPrefix();
}
@Override
public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
return extendedBindingProperties.getExtendedPropertiesEntryClass();
}
private <T> T getClassConfiguration(String destName, String className,
Class<T> interfaceClass) {
if (StringUtils.isEmpty(className)) {
@@ -184,4 +194,9 @@ public class RocketMQMessageChannelBinder extends
return null;
}
public void setExtendedBindingProperties(
RocketMQExtendedBindingProperties extendedBindingProperties) {
this.extendedBindingProperties = extendedBindingProperties;
}
}

3
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/config/RocketMQBinderAutoConfiguration.java
View File

@@ -64,8 +64,9 @@ public class RocketMQBinderAutoConfiguration {
RocketMQTopicProvisioner provisioningProvider,
ConsumersManager consumersManager) {
RocketMQMessageChannelBinder binder = new RocketMQMessageChannelBinder(
consumersManager, extendedBindingProperties, provisioningProvider,
consumersManager, provisioningProvider,
rocketBinderConfigurationProperties, instrumentationManager);
binder.setExtendedBindingProperties(extendedBindingProperties);
return binder;
}

6
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQBindingProperties.java
View File

@@ -16,16 +16,19 @@
package org.springframework.cloud.stream.binder.rocketmq.properties;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
/**
* @author Timur Valiev
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
public class RocketMQBindingProperties {
public class RocketMQBindingProperties implements BinderSpecificPropertiesProvider {
private RocketMQConsumerProperties consumer = new RocketMQConsumerProperties();
private RocketMQProducerProperties producer = new RocketMQProducerProperties();
@Override
public RocketMQConsumerProperties getConsumer() {
return consumer;
}
@@ -34,6 +37,7 @@ public class RocketMQBindingProperties {
this.consumer = consumer;
}
@Override
public RocketMQProducerProperties getProducer() {
return producer;
}

64
spring-cloud-stream-binder-rocketmq/src/main/java/org/springframework/cloud/stream/binder/rocketmq/properties/RocketMQExtendedBindingProperties.java
View File

@@ -16,71 +16,27 @@
package org.springframework.cloud.stream.binder.rocketmq.properties;
import java.util.HashMap;
import java.util.Map;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.ExtendedBindingProperties;
import org.springframework.cloud.stream.binder.AbstractExtendedBindingProperties;
import org.springframework.cloud.stream.binder.BinderSpecificPropertiesProvider;
/**
* @author Timur Valiev
* @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
*/
@ConfigurationProperties("spring.cloud.stream.rocketmq")
public class RocketMQExtendedBindingProperties implements
ExtendedBindingProperties<RocketMQConsumerProperties, RocketMQProducerProperties> {
public class RocketMQExtendedBindingProperties extends
AbstractExtendedBindingProperties<RocketMQConsumerProperties, RocketMQProducerProperties, RocketMQBindingProperties> {
private Map<String, RocketMQBindingProperties> bindings = new HashMap<>();
private static final String DEFAULTS_PREFIX = "spring.cloud.stream.rocketmq.default";
public Map<String, RocketMQBindingProperties> getBindings() {
return this.bindings;
}
public void setBindings(Map<String, RocketMQBindingProperties> bindings) {
this.bindings = bindings;
@Override
public String getDefaultsPrefix() {
return DEFAULTS_PREFIX;
}
@Override
public synchronized RocketMQConsumerProperties getExtendedConsumerProperties(
String channelName) {
if (bindings.containsKey(channelName)) {
if (bindings.get(channelName).getConsumer() != null) {
return bindings.get(channelName).getConsumer();
}
else {
RocketMQConsumerProperties properties = new RocketMQConsumerProperties();
this.bindings.get(channelName).setConsumer(properties);
return properties;
}
}
else {
RocketMQConsumerProperties properties = new RocketMQConsumerProperties();
RocketMQBindingProperties rbp = new RocketMQBindingProperties();
rbp.setConsumer(properties);
bindings.put(channelName, rbp);
return properties;
}
}
@Override
public synchronized RocketMQProducerProperties getExtendedProducerProperties(
String channelName) {
if (bindings.containsKey(channelName)) {
if (bindings.get(channelName).getProducer() != null) {
return bindings.get(channelName).getProducer();
}
else {
RocketMQProducerProperties properties = new RocketMQProducerProperties();
this.bindings.get(channelName).setProducer(properties);
return properties;
}
}
else {
RocketMQProducerProperties properties = new RocketMQProducerProperties();
RocketMQBindingProperties rbp = new RocketMQBindingProperties();
rbp.setProducer(properties);
bindings.put(channelName, rbp);
return properties;
}
public Class<? extends BinderSpecificPropertiesProvider> getExtendedPropertiesEntryClass() {
return RocketMQBindingProperties.class;
}
}