@Test
	// XD 2429
	public void testBindOrder() {
		ModuleDefinition moduleDefinition = TestModuleDefinitions.dummy("testing", ModuleType.processor);
		Module module = mock(Module.class);
		when(module.getDescriptor()).thenReturn(new ModuleDescriptor.Builder()
				.setGroup("foo")
				.setIndex(1)
				.setModuleDefinition(moduleDefinition)
				.build());
		when(module.getType()).thenReturn(moduleDefinition.getType());
		when(module.getName()).thenReturn(moduleDefinition.getName());
		final AtomicReference<Message<?>> messageReceived = new AtomicReference<>();
		LocalMessageBus bus = new LocalMessageBus() {

			final AtomicBoolean consumerBound = new AtomicBoolean();

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

			@Override
			public void bindConsumer(String name, MessageChannel moduleInputChannel, Properties properties) {
				this.consumerBound.set(true);
			}

		};
		when(module.getComponent(MessageBus.class)).thenReturn(bus);
		DirectChannel input = new DirectChannel();
		input.setBeanName("test.input.channel");
		DirectChannel output = new DirectChannel();
		output.setBeanName("test.output.channel");
		BridgeHandler handler = new BridgeHandler();
		handler.setOutputChannel(output);
		EventDrivenConsumer bridge = new EventDrivenConsumer(input, handler);
		bridge.start();
		when(module.getComponent("input", MessageChannel.class)).thenReturn(input);
		when(module.getComponent("output", MessageChannel.class)).thenReturn(output);
		ModuleDeploymentProperties props = new ModuleDeploymentProperties();
		props.setTrackHistory(true);
		when(module.getDeploymentProperties()).thenReturn(props);
		when(module.getComponent(MessageBuilderFactory.class)).thenReturn(new DefaultMessageBuilderFactory());
		StreamPlugin plugin = new StreamPlugin(bus, zkConnection);
		plugin.preProcessModule(module);
		plugin.postProcessModule(module);
		input.send(new GenericMessage<String>("foo"));
		assertNotNull(messageReceived.get());
		@SuppressWarnings("unchecked")
		Collection<Map<String, Object>> xdHistory =
				(Collection<Map<String, Object>>) messageReceived.get().getHeaders().get(XdHeaders.XD_HISTORY);
		assertTrue(xdHistory != null);
		assertEquals("testing", xdHistory.iterator().next().get("module"));
	}
@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