GH-1894 interim

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java

@@ -168,11 +168,8 @@ public class BinderFactoryAutoConfiguration {
 			ConfigurableApplicationContext configurableApplicationContext) {
 		Map<String, BinderType> binderTypes = new HashMap<>();
 		ClassLoader classLoader = configurableApplicationContext.getClassLoader();
-		// the above can never be null since it will default to
-		// ClassUtils.getDefaultClassLoader(..)
 		try {
 			Enumeration<URL> 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);
 					}
 				}

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<SubscribableChannel> channels = new ArrayList<>();
+	private final List<AbstractSubscribableChannel> 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;
+	}
 }

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);
+	}
+
 }

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<BlockingQueue<Message<?>>> messageQueues = new ArrayList<>();
+	private final Map<String, BlockingQueue<Message<byte[]>>> messageQueues = new LinkedHashMap<>();
 
+	public Message<byte[]> 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<byte[]> receive(long timeout, int channelIndex) {
+	public Message<byte[]> receive(long timeout, int bindingIndex) {
 		try {
-			return (Message<byte[]>) this.messageQueues.get(channelIndex).poll(timeout, TimeUnit.MILLISECONDS);
+			BlockingQueue<Message<byte[]>> 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<Message<?>> 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<Message<byte[]>> messageQueue = new LinkedTransferQueue<>();
+		this.messageQueues.put(bidningName, messageQueue);
+		this.getChannelByName(bidningName).subscribe(message -> this.messageQueues.get(bidningName).offer((Message<byte[]>) message));
 	}
 
 }