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

polishing

Browse Source
This commit is contained in:
Oleg Zhurakousky
2017-11-26 22:11:30 -05:00
parent 2bac583d81
commit acc4ecfccb
2 changed files with 51 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java
View File

@@ -37,6 +37,7 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.StandardEnvironment;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
@@ -96,14 +97,10 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
String configurationName;
// Fall back to a default if no argument is provided
if (StringUtils.isEmpty(name)) {
if (this.binderConfigurations.size() == 0) {
throw new IllegalStateException(
"A default binder has been requested, but there there is no binder available");
}
else if (!StringUtils.hasText(this.defaultBinder)) {
Assert.notEmpty(this.binderConfigurations, "A default binder has been requested, but there is no binder available");
if (!StringUtils.hasText(this.defaultBinder)) {
Set<String> defaultCandidateConfigurations = new HashSet<>();
for (Map.Entry<String, BinderConfiguration> binderConfigurationEntry : this.binderConfigurations
.entrySet()) {
for (Map.Entry<String, BinderConfiguration> binderConfigurationEntry : this.binderConfigurations.entrySet()) {
if (binderConfigurationEntry.getValue().isDefaultCandidate()) {
defaultCandidateConfigurations.add(binderConfigurationEntry.getKey());
}
@@ -113,36 +110,23 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
this.defaultBinderForBindingTargetType.put(bindingTargetType.getName(), configurationName);
}
else {
if (defaultCandidateConfigurations.size() > 1) {
List<String> candidatesForBindableType = new ArrayList<>();
for (String defaultCandidateConfiguration : defaultCandidateConfigurations) {
Binder<Object, ?, ?> binderInstance = getBinderInstance(defaultCandidateConfiguration);
Class<?> binderType = GenericsUtils.getParameterType(binderInstance.getClass(),
Binder.class, 0);
if (binderType.isAssignableFrom(bindingTargetType)) {
candidatesForBindableType.add(defaultCandidateConfiguration);
}
}
if (candidatesForBindableType.size() == 1) {
configurationName = candidatesForBindableType.iterator().next();
this.defaultBinderForBindingTargetType.put(bindingTargetType.getName(), configurationName);
}
else if (candidatesForBindableType.size() > 1) {
throw new IllegalStateException(
"A default binder has been requested, but there is more than one binder available for '"
+ bindingTargetType.getName() + "' : "
+ StringUtils.collectionToCommaDelimitedString(candidatesForBindableType)
+ ", and no default binder has been set.");
}
else {
throw new IllegalStateException("A default binder has been requested, but none of the "
+ "registered binders can bind a '" + bindingTargetType + "': "
+ StringUtils.collectionToCommaDelimitedString(defaultCandidateConfigurations));
List<String> candidatesForBindableType = new ArrayList<>();
for (String defaultCandidateConfiguration : defaultCandidateConfigurations) {
Binder<Object, ?, ?> binderInstance = getBinderInstance(defaultCandidateConfiguration);
Class<?> binderType = GenericsUtils.getParameterType(binderInstance.getClass(), Binder.class, 0);
if (binderType.isAssignableFrom(bindingTargetType)) {
candidatesForBindableType.add(defaultCandidateConfiguration);
}
}
if (candidatesForBindableType.size() == 1) {
configurationName = candidatesForBindableType.iterator().next();
this.defaultBinderForBindingTargetType.put(bindingTargetType.getName(), configurationName);
}
else {
throw new IllegalArgumentException(
"A default binder has been requested, but there is no default available");
throw new IllegalStateException("A default binder has been requested, but there is more than "
+ "one binder available for '" + bindingTargetType.getName() + "' : "
+ StringUtils.collectionToCommaDelimitedString(candidatesForBindableType)
+ ", and no default binder has been set.");
}
}
}
@@ -154,11 +138,8 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
configurationName = name;
}
Binder<T, ?, ?> binderInstance = getBinderInstance(configurationName);
if (!(GenericsUtils.getParameterType(binderInstance.getClass(), Binder.class, 0)
.isAssignableFrom(bindingTargetType))) {
throw new IllegalStateException(
"The binder '" + configurationName + "' cannot bind a " + bindingTargetType.getName());
}
Assert.state(GenericsUtils.getParameterType(binderInstance.getClass(), Binder.class, 0).isAssignableFrom(bindingTargetType),
"The binder '" + configurationName + "' cannot bind a " + bindingTargetType.getName());
return binderInstance;
}
@@ -166,9 +147,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
private <T> Binder<T, ?, ?> getBinderInstance(String configurationName) {
if (!this.binderInstanceCache.containsKey(configurationName)) {
BinderConfiguration binderConfiguration = this.binderConfigurations.get(configurationName);
if (binderConfiguration == null) {
throw new IllegalStateException("Unknown binder configuration: " + configurationName);
}
Assert.state(binderConfiguration != null, "Unknown binder configuration: " + configurationName);
BinderType binderType = this.binderTypeRegistry.get(binderConfiguration.getBinderType());
Assert.notNull(binderType, "Binder type " + binderConfiguration.getBinderType() + " is not defined");
Map<Object, Object> binderProperties = binderConfiguration.getProperties();
@@ -181,13 +160,7 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
// Initialize the domain with a unique name based on the bootstrapping context
// setting
ConfigurableEnvironment environment = this.context != null ? this.context.getEnvironment() : null;
String defaultDomain = environment != null ? environment.getProperty("spring.jmx.default-domain") : null;
if (defaultDomain == null) {
defaultDomain = "";
}
else {
defaultDomain += ".";
}
String defaultDomain = environment != null ? environment.getProperty("spring.jmx.default-domain.") : "";
args.add("--spring.jmx.default-domain=" + defaultDomain + "binder." + configurationName);
args.add("--spring.main.applicationContextClass=" + AnnotationConfigApplicationContext.class.getName());
List<Class<?>> configurationClasses = new ArrayList<Class<?>>(
@@ -208,17 +181,15 @@ public class DefaultBinderFactory implements BinderFactory, DisposableBean, Appl
if (useApplicationContextAsParent) {
springApplicationBuilder.parent(this.context);
}
if (useApplicationContextAsParent || (environment != null && binderConfiguration.isInheritEnvironment())) {
if (environment != null) {
StandardEnvironment binderEnvironment = new StandardEnvironment();
binderEnvironment.merge(environment);
springApplicationBuilder.environment(binderEnvironment);
}
if (environment != null && (useApplicationContextAsParent || binderConfiguration.isInheritEnvironment())) {
StandardEnvironment binderEnvironment = new StandardEnvironment();
binderEnvironment.merge(environment);
springApplicationBuilder.environment(binderEnvironment);
}
ConfigurableApplicationContext binderProducingContext = springApplicationBuilder
.run(args.toArray(new String[args.size()]));
Binder<T, ?, ?> binder = binderProducingContext.getBean(Binder.class);
if (this.listeners != null) {
if (!CollectionUtils.isEmpty(this.listeners)) {
for (Listener binderFactoryListener : listeners) {
binderFactoryListener.afterBinderContextInitialized(configurationName, binderProducingContext);
}

44
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2016 the original author or authors.
* Copyright 2015-2017 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.
@@ -138,22 +138,18 @@ public class BindingService {
public <T> void rescheduleConsumerBinding(final T input, final String inputName,
final Binder<T, ConsumerProperties, ?> binder, final ConsumerProperties consumerProperties,
final String target, final LateBinding<T> late, RuntimeException exception) {
if (exception instanceof IllegalStateException || exception instanceof IllegalArgumentException) {
throw exception;
}
assertNotIllegalException(exception);
this.log.error("Failed to create consumer binding; retrying in " +
this.bindingServiceProperties.getBindingRetryInterval() + " seconds", exception);
this.taskScheduler.schedule(() -> {
this.scheduleTask(() -> {
try {
late.setDelegate(binder.bindConsumer(target,
this.bindingServiceProperties.getGroup(inputName), input,
consumerProperties));
this.bindingServiceProperties.getGroup(inputName), input, consumerProperties));
}
catch (RuntimeException e) {
rescheduleConsumerBinding(input, inputName, binder, consumerProperties, target, late, e);
}
}, new Date(System.currentTimeMillis() +
this.bindingServiceProperties.getBindingRetryInterval() * 1_000));
});
}
@SuppressWarnings({ "unchecked", "rawtypes" })
@@ -209,20 +205,17 @@ public class BindingService {
public <T> void rescheduleProducerBinding(final T output, final String bindingTarget,
final Binder<T, ?, ProducerProperties> binder, final ProducerProperties producerProperties,
final LateBinding<T> late, final RuntimeException exception) {
if (exception instanceof IllegalStateException || exception instanceof IllegalArgumentException) {
throw exception;
}
assertNotIllegalException(exception);
this.log.error("Failed to create producer binding; retrying in " +
this.bindingServiceProperties.getBindingRetryInterval() + " seconds", exception);
this.taskScheduler.schedule(() -> {
this.scheduleTask(() -> {
try {
late.setDelegate(binder.bindProducer(bindingTarget, output, producerProperties));
}
catch (RuntimeException e) {
rescheduleProducerBinding(output, bindingTarget, binder, producerProperties, late, e);
}
}, new Date(System.currentTimeMillis() +
this.bindingServiceProperties.getBindingRetryInterval() * 1_000));
});
}
public void unbindConsumers(String inputName) {
@@ -247,11 +240,6 @@ public class BindingService {
}
}
protected <T> Binder<T, ?, ?> getBinder(String channelName, Class<T> bindableType) {
String binderConfigurationName = this.bindingServiceProperties.getBinder(channelName);
return binderFactory.getBinder(binderConfigurationName, bindableType);
}
/**
* Provided for backwards compatibility. Will be removed in a future version.
*
@@ -266,6 +254,11 @@ public class BindingService {
return this.bindingServiceProperties;
}
protected <T> Binder<T, ?, ?> getBinder(String channelName, Class<T> bindableType) {
String binderConfigurationName = this.bindingServiceProperties.getBinder(channelName);
return binderFactory.getBinder(binderConfigurationName, bindableType);
}
private void validate(Object properties) {
DataBinder dataBinder = new DataBinder(properties);
dataBinder.setValidator(validator);
@@ -275,6 +268,17 @@ public class BindingService {
}
}
private void scheduleTask(Runnable task) {
this.taskScheduler.schedule(task, new Date(System.currentTimeMillis() +
this.bindingServiceProperties.getBindingRetryInterval() * 1_000));
}
private void assertNotIllegalException(RuntimeException exception) throws RuntimeException {
if (exception instanceof IllegalStateException || exception instanceof IllegalArgumentException) {
throw exception;
}
}
private static class LateBinding<T> implements Binding<T> {
private Binding<T> delegate;