diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java index 878a7eab3..8c5038f90 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java @@ -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 defaultCandidateConfigurations = new HashSet<>(); - for (Map.Entry binderConfigurationEntry : this.binderConfigurations - .entrySet()) { + for (Map.Entry 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 candidatesForBindableType = new ArrayList<>(); - for (String defaultCandidateConfiguration : defaultCandidateConfigurations) { - Binder 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 candidatesForBindableType = new ArrayList<>(); + for (String defaultCandidateConfiguration : defaultCandidateConfigurations) { + Binder 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 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 Binder 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 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> configurationClasses = new ArrayList>( @@ -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 binder = binderProducingContext.getBean(Binder.class); - if (this.listeners != null) { + if (!CollectionUtils.isEmpty(this.listeners)) { for (Listener binderFactoryListener : listeners) { binderFactoryListener.afterBinderContextInitialized(configurationName, binderProducingContext); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index bb78b95e8..02caeff12 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java @@ -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 void rescheduleConsumerBinding(final T input, final String inputName, final Binder binder, final ConsumerProperties consumerProperties, final String target, final LateBinding 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 void rescheduleProducerBinding(final T output, final String bindingTarget, final Binder binder, final ProducerProperties producerProperties, final LateBinding 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 Binder getBinder(String channelName, Class 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 Binder getBinder(String channelName, Class 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 implements Binding { private Binding delegate;