diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java index 2fb3913e1..dd68be71a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java @@ -168,11 +168,8 @@ public class BinderFactoryAutoConfiguration { ConfigurableApplicationContext configurableApplicationContext) { Map binderTypes = new HashMap<>(); ClassLoader classLoader = configurableApplicationContext.getClassLoader(); - // the above can never be null since it will default to - // ClassUtils.getDefaultClassLoader(..) try { Enumeration resources = classLoader.getResources("META-INF/spring.binders"); - // see if test binder is available on the classpath and if so add it to the binderTypes try { BinderType bt = new BinderType("integration", new Class[] { @@ -180,7 +177,6 @@ public class BinderFactoryAutoConfiguration { binderTypes.put("integration", bt); } catch (Exception e) { -// e.printStackTrace(); // ignore. means test binder is not available } @@ -194,8 +190,7 @@ public class BinderFactoryAutoConfiguration { while (resources.hasMoreElements()) { URL url = resources.nextElement(); UrlResource resource = new UrlResource(url); - for (BinderType binderType : parseBinderConfigurations(classLoader, - resource)) { + for (BinderType binderType : parseBinderConfigurations(classLoader, resource)) { binderTypes.put(binderType.getDefaultName(), binderType); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java index 0ba9f0c60..86e5b787a 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/AbstractDestination.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2020 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. @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.test; import java.util.ArrayList; import java.util.List; +import org.springframework.integration.channel.AbstractSubscribableChannel; import org.springframework.messaging.SubscribableChannel; /** @@ -27,19 +28,28 @@ import org.springframework.messaging.SubscribableChannel; */ abstract class AbstractDestination { - private final List channels = new ArrayList<>(); + private final List channels = new ArrayList<>(); SubscribableChannel getChannel(int index) { return this.channels.get(index); } void setChannel(SubscribableChannel channel) { - this.channels.add(channel); - this.afterChannelIsSet(this.channels.size() - 1); + this.channels.add((AbstractSubscribableChannel) channel); + this.afterChannelIsSet(this.channels.size() - 1, ((AbstractSubscribableChannel) channel).getBeanName()); } - void afterChannelIsSet(int channelIndex) { + void afterChannelIsSet(int channelIndex, String name) { // noop } + SubscribableChannel getChannelByName(String name) { + //name = name + ".destination"; + for (AbstractSubscribableChannel subscribableChannel : channels) { + if (subscribableChannel.getBeanName().equals(name)) { + return subscribableChannel; + } + } + return null; + } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java index f8ecf1fd6..1be33bd2d 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2020 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. @@ -41,4 +41,8 @@ public class InputDestination extends AbstractDestination { this.getChannel(inputIndex).send(message); } + public void send(Message message, String bindingName) { + this.getChannelByName(bindingName).send(message); + } + } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java index 753982858..efccab0a3 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/OutputDestination.java @@ -17,7 +17,9 @@ package org.springframework.cloud.stream.binder.test; import java.util.ArrayList; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedTransferQueue; import java.util.concurrent.TimeUnit; @@ -34,17 +36,26 @@ import org.springframework.messaging.Message; */ public class OutputDestination extends AbstractDestination { - private final List>> messageQueues = new ArrayList<>(); + private final Map>> messageQueues = new LinkedHashMap<>(); + public Message receive(long timeout, String bindingName) { + try { + return this.messageQueues.get(bindingName).poll(timeout, TimeUnit.MILLISECONDS); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return null; + } /** * Allows to access {@link Message}s received by this {@link OutputDestination}. * @param timeout how long to wait before giving up * @return received message */ - @SuppressWarnings("unchecked") - public Message receive(long timeout, int channelIndex) { + public Message receive(long timeout, int bindingIndex) { try { - return (Message) this.messageQueues.get(channelIndex).poll(timeout, TimeUnit.MILLISECONDS); + BlockingQueue> destinationQueue = (new ArrayList<>(this.messageQueues.values())).get(bindingIndex); + return destinationQueue.poll(timeout, TimeUnit.MILLISECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -64,11 +75,12 @@ public class OutputDestination extends AbstractDestination { return this.receive(timeout, 0); } + @SuppressWarnings("unchecked") @Override - void afterChannelIsSet(int channelIndex) { - BlockingQueue> messageQueue = new LinkedTransferQueue<>(); - this.messageQueues.add(messageQueue); - this.getChannel(channelIndex).subscribe(message -> this.messageQueues.get(channelIndex).offer(message)); + void afterChannelIsSet(int channelIndex, String bidningName) { + BlockingQueue> messageQueue = new LinkedTransferQueue<>(); + this.messageQueues.put(bidningName, messageQueue); + this.getChannelByName(bidningName).subscribe(message -> this.messageQueues.get(bidningName).offer((Message) message)); } }