@Override
			public void bindProducer(String name, final MessageChannel moduleOutputChannel, Properties properties) {

				final MessageHandler handler = new MessageHandler() {

					@Override
					public void handleMessage(Message<?> message) throws MessagingException {
						messageReceived.set(message);
					}

				};

				if (consumerBound.get()) {
					Executors.newSingleThreadExecutor().execute(new Runnable() {

						@Override
						public void run() {
							try {
								Thread.sleep(1000); //delay the bind to exhibit the XD-2429 behavior
							}
							catch (InterruptedException e) {
								Thread.currentThread().interrupt();
							}
							((SubscribableChannel) moduleOutputChannel).subscribe(handler);
						}
					});
				}
				else {
					((SubscribableChannel) moduleOutputChannel).subscribe(handler);
				}
			}
@author Mark Fisher
@author Jennifer Hickey
@author Gary Russell
@author David Turanski
spring-xd-dirt/src/test/java/org/springframework/xd/dirt/plugins/stream/StreamPluginTests.java