Spring Cloud Consul Stream Binder

Uses consul events for a non-grouping stream binder. Also enables spring-cloud-starter-consul-bus. fixes gh-86
2016-06-29 13:30:28 -06:00
parent cc54bcbd62
commit a184d62d10
23 changed files with 1033 additions and 370 deletions
--- a/pom.xml
+++ b/pom.xml
@@ -31,10 +31,10 @@
 		<module>spring-cloud-consul-core</module>
 		<module>spring-cloud-consul-config</module>
 		<module>spring-cloud-consul-discovery</module>
-		<!--<module>spring-cloud-consul-bus</module>-->
+		<module>spring-cloud-consul-binder</module>
 		<module>spring-cloud-consul-sample</module>
 		<module>spring-cloud-starter-consul</module>
-		<!--<module>spring-cloud-starter-consul-bus</module>-->
+		<module>spring-cloud-starter-consul-bus</module>
 		<module>spring-cloud-starter-consul-config</module>
 		<module>spring-cloud-starter-consul-discovery</module>
 		<module>spring-cloud-starter-consul-all</module>
--- a/spring-cloud-consul-binder/pom.xml
+++ b/spring-cloud-consul-binder/pom.xml
@@ -4,15 +4,15 @@
 		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 	<modelVersion>4.0.0</modelVersion>

-	<artifactId>spring-cloud-consul-bus</artifactId>
+	<artifactId>spring-cloud-consul-binder</artifactId>
 	<packaging>jar</packaging>
-	<name>Spring Cloud Consul Bus</name>
-	<description>Spring Cloud Consul Bus</description>
+	<name>Spring Cloud Consul Binder</name>
+	<description>Spring Cloud Consul Binder</description>

 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-consul</artifactId>
-		<version>1.0.1.BUILD-SNAPSHOT</version>
+		<version>1.0.2.BUILD-SNAPSHOT</version>
 		<relativePath>..</relativePath>
 	</parent>

@@ -29,29 +29,46 @@
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-bus</artifactId>
-			<optional>true</optional>
+			<artifactId>spring-cloud-stream</artifactId>
 		</dependency>
 		<dependency>
 			<groupId>com.ecwid.consul</groupId>
 			<artifactId>consul-api</artifactId>
 			<optional>true</optional>
 		</dependency>
-		<dependency>
-			<groupId>org.springframework.integration</groupId>
-			<artifactId>spring-integration-java-dsl</artifactId>
-		</dependency>
 		<dependency>
 			<groupId>org.projectlombok</groupId>
 			<artifactId>lombok</artifactId>
 			<!-- Only needed at compile time -->
 			<scope>provided</scope>
 		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-deployer-local</artifactId>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud</groupId>
+			<artifactId>spring-cloud-stream-test-support-internal</artifactId>
+			<scope>test</scope>
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.boot</groupId>
 			<artifactId>spring-boot-starter-test</artifactId>
 			<scope>test</scope>
 		</dependency>
+		<dependency>
+			<groupId>com.github.tomakehurst</groupId>
+			<artifactId>wiremock</artifactId>
+			<version>1.58</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.awaitility</groupId>
+			<artifactId>awaitility</artifactId>
+			<version>2.0.0</version>
+			<scope>test</scope>
+		</dependency>

 	</dependencies>

--- a/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/ConsulBinder.java
+++ b/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/ConsulBinder.java
@@ -0,0 +1,68 @@
+/*
+ * Copyright 2013-2016 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.consul.binder;
+
+import org.springframework.cloud.stream.binder.AbstractBinder;
+import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.cloud.stream.binder.DefaultBinding;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.integration.endpoint.EventDrivenConsumer;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.util.Assert;
+
+/**
+ * @author Spencer Gibb
+ */
+public class ConsulBinder extends AbstractBinder<MessageChannel, ConsumerProperties, ProducerProperties> {
+
+	private static final String BEAN_NAME_TEMPLATE = "outbound.%s";
+
+	private final EventService eventService;
+
+	public ConsulBinder(EventService eventService) {
+		this.eventService = eventService;
+	}
+
+	@Override
+	protected Binding<MessageChannel> doBindConsumer(String name, String group, MessageChannel inputChannel, ConsumerProperties properties) {
+		ConsulInboundMessageProducer messageProducer = new ConsulInboundMessageProducer(this.eventService);
+		messageProducer.setOutputChannel(inputChannel);
+		messageProducer.setBeanFactory(this.getBeanFactory());
+		messageProducer.afterPropertiesSet();
+		messageProducer.start();
+
+		return new DefaultBinding<>(name, group, inputChannel, messageProducer);
+	}
+
+	@Override
+	protected Binding<MessageChannel> doBindProducer(String name, MessageChannel channel, ProducerProperties properties) {
+		Assert.isInstanceOf(SubscribableChannel.class, channel);
+
+		logger.debug("Binding Consul client to eventName " + name);
+		ConsulSendingHandler sendingHandler = new ConsulSendingHandler(this.eventService.getConsulClient(), name);
+
+		EventDrivenConsumer consumer = new EventDrivenConsumer((SubscribableChannel) channel, sendingHandler);
+		consumer.setBeanFactory(getBeanFactory());
+		consumer.setBeanName(String.format(BEAN_NAME_TEMPLATE, name));
+		consumer.afterPropertiesSet();
+		consumer.start();
+
+		return new DefaultBinding<>(name, null, channel, consumer);
+	}
+}
--- a/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/ConsulInboundMessageProducer.java
+++ b/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/ConsulInboundMessageProducer.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2015 the original author or authors.
+ * Copyright 2013-2016 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.
@@ -14,16 +14,17 @@
 * limitations under the License.
 */

-package org.springframework.cloud.consul.bus;
+package org.springframework.cloud.consul.binder;

 import static org.springframework.util.Base64Utils.decodeFromString;

-import java.io.IOException;
 import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;

-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.endpoint.MessageProducerSupport;
-import org.springframework.scheduling.annotation.Scheduled;

 import com.ecwid.consul.v1.event.model.Event;

@@ -32,11 +33,23 @@ import com.ecwid.consul.v1.event.model.Event;
 * Integration Messages, and sends the results to a Message Channel.
 * @author Spencer Gibb
 */
-public class ConsulInboundChannelAdapter extends MessageProducerSupport {
-	@Autowired
-	private EventService eventService;
+public class ConsulInboundMessageProducer extends MessageProducerSupport {

-	public ConsulInboundChannelAdapter() {
+	private EventService eventService;
+	private final ScheduledExecutorService scheduler;
+	private final Runnable eventsRunnable;
+	private ScheduledFuture<?> eventsHandle;
+
+	public ConsulInboundMessageProducer(EventService eventService) {
+		this.eventService = eventService;
+		this.scheduler = Executors.newScheduledThreadPool(1);
+		this.eventsRunnable = new Runnable() {
+
+			@Override
+			public void run() {
+				getEvents();
+			}
+		};
 	}

 	// link eventService to sendMessage
@@ -57,10 +70,19 @@ public class ConsulInboundChannelAdapter extends MessageProducerSupport {

 	@Override
 	protected void doStart() {
+		//TODO: make configurable
+		eventsHandle = this.scheduler.scheduleWithFixedDelay(eventsRunnable, 500, 500, TimeUnit.MILLISECONDS);
 	}

-	@Scheduled(fixedDelayString = "${spring.cloud.consul.bus.eventDelay:30000}")
-	public void getEvents() throws IOException {
+	@Override
+	protected void doStop() {
+		if (this.eventsHandle != null) {
+			this.eventsHandle.cancel(true);
+		}
+	}
+
+	// @Scheduled(fixedDelayString = "${spring.cloud.consul.binder.eventDelay:30000}")
+	public void getEvents() {
 		List<Event> events = eventService.watch();
 		for (Event event : events) {
 			// Map<String, Object> headers = new HashMap<>();
@@ -71,8 +93,4 @@ public class ConsulInboundChannelAdapter extends MessageProducerSupport {
 					.build());
 		}
 	}
-
-	@Override
-	protected void doStop() {
-	}
 }
--- a/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/ConsulSendingHandler.java
+++ b/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/ConsulSendingHandler.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2015 the original author or authors.
+ * Copyright 2013-2016 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.
@@ -14,9 +14,9 @@
 * limitations under the License.
 */

-package org.springframework.cloud.consul.bus;
+package org.springframework.cloud.consul.binder;

-import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.messaging.Message;

@@ -30,19 +30,28 @@ import com.ecwid.consul.v1.event.model.EventParams;
 * Adapter that converts and sends Messages as Consul events
 * @author Spencer Gibb
 */
-public class ConsulOutboundEndpoint extends AbstractReplyProducingMessageHandler {
+public class ConsulSendingHandler extends AbstractMessageHandler {

-	@Autowired
-	protected ConsulClient consul;
+	private final ConsulClient consul;
+	private final String eventName;
+
+	public ConsulSendingHandler(ConsulClient consul, String eventName) {
+		this.consul = consul;
+		this.eventName = eventName;
+	}

 	@Override
-	protected Object handleRequestMessage(Message<?> requestMessage) {
-		Object payload = requestMessage.getPayload();
+	protected void handleMessageInternal(Message<?> message) throws Exception {
+		if (logger.isTraceEnabled()) {
+			logger.trace("Publishing message" + message);
+		}
+
+		Object payload = message.getPayload();
 		// TODO: support headers
 		// TODO: support consul event filters: NodeFilter, ServiceFilter, TagFilter
-		Response<Event> event = consul.eventFire("springCloudBus", (String) payload,
+		Response<Event> event = consul.eventFire(this.eventName, (String) payload,
 				new EventParams(), QueryParams.DEFAULT);
 		// TODO: return event?
-		return null;
+		// return null;
 	}
 }
--- a/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/EventService.java
+++ b/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/EventService.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2015 the original author or authors.
+ * Copyright 2013-2016 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.
@@ -14,15 +14,14 @@
 * limitations under the License.
 */

-package org.springframework.cloud.consul.bus;
+package org.springframework.cloud.consul.binder;

-import java.math.BigInteger;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicReference;

 import javax.annotation.PostConstruct;

-import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.consul.binder.config.ConsulBinderProperties;

 import com.ecwid.consul.v1.ConsulClient;
 import com.ecwid.consul.v1.QueryParams;
@@ -36,16 +35,23 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 */
 public class EventService {

-	@Autowired
-	protected ConsulBusProperties properties;
+	protected ConsulBinderProperties properties;

-	@Autowired
 	protected ConsulClient consul;

-	@Autowired(required = false)
 	protected ObjectMapper objectMapper = new ObjectMapper();

-	private AtomicReference<BigInteger> lastIndex = new AtomicReference<>();
+	private AtomicReference<Long> lastIndex = new AtomicReference<>();
+
+	public EventService(ConsulBinderProperties properties, ConsulClient consul, ObjectMapper objectMapper) {
+		this.properties = properties;
+		this.consul = consul;
+		this.objectMapper = objectMapper;
+	}
+
+	public ConsulClient getConsulClient() {
+		return consul;
+	}

 	@PostConstruct
 	public void init() {
@@ -55,11 +61,11 @@ public class EventService {
 	private void setLastIndex(Response<?> response) {
 		Long consulIndex = response.getConsulIndex();
 		if (consulIndex != null) {
-			lastIndex.set(BigInteger.valueOf(consulIndex));
+			lastIndex.set(response.getConsulIndex());
 		}
 	}

-	public BigInteger getLastIndex() {
+	public Long getLastIndex() {
 		return lastIndex.get();
 	}

@@ -77,26 +83,7 @@ public class EventService {
 		return getEventsResponse().getValue();
 	}

-	/**
-	 * from https://github.com/hashicorp/consul/blob/master/api/event.go#L90-L104 //
-	 * IDToIndex is a bit of a hack. This simulates the index generation to // convert an
-	 * event ID into a WaitIndex. func (e *Event) IDToIndex(uuid string) uint64 { lower :=
-	 * uuid[0:8] + uuid[9:13] + uuid[14:18] upper := uuid[19:23] + uuid[24:36] lowVal, err
-	 * := strconv.ParseUint(lower, 16, 64) if err != nil { panic("Failed to convert " +
-	 * lower) } highVal, err := strconv.ParseUint(upper, 16, 64) if err != nil {
-	 * panic("Failed to convert " + upper) } return lowVal ^ highVal //^ bitwise XOR
-	 * integers }
-	 */
-	public BigInteger toIndex(String eventId) {
-		String lower = eventId.substring(0, 8) + eventId.substring(9, 13)
-				+ eventId.substring(14, 18);
-		String upper = eventId.substring(19, 23) + eventId.substring(24, 36);
-		BigInteger lowVal = new BigInteger(lower, 16);
-		BigInteger highVal = new BigInteger(upper, 16);
-		return lowVal.xor(highVal);
-	}
-
-	public List<Event> getEvents(BigInteger lastIndex) {
+	public List<Event> getEvents(Long lastIndex) {
 		return filterEvents(readEvents(getEventsResponse()), lastIndex);
 	}

@@ -104,13 +91,13 @@ public class EventService {
 		return watch(lastIndex.get());
 	}

-	public List<Event> watch(BigInteger lastIndex) {
+	public List<Event> watch(Long lastIndex) {
 		// TODO: parameterized or configurable watch time
 		long index = -1;
 		if (lastIndex != null) {
-			index = lastIndex.longValue();
+			index = lastIndex;
 		}
-		Response<List<Event>> watch = consul.eventList(new QueryParams(properties.eventTimeout, index));
+		Response<List<Event>> watch = consul.eventList(new QueryParams(properties.getEventTimeout(), index));
 		return filterEvents(readEvents(watch), lastIndex);
 	}

@@ -122,13 +109,13 @@ public class EventService {
 	/**
 	 * from https://github.com/hashicorp/consul/blob/master/watch/funcs.go#L169-L194
 	 */
-	protected List<Event> filterEvents(List<Event> toFilter, BigInteger lastIndex) {
+	protected List<Event> filterEvents(List<Event> toFilter, Long lastIndex) {
 		List<Event> events = toFilter;
 		if (lastIndex != null) {
 			for (int i = 0; i < events.size(); i++) {
 				Event event = events.get(i);
-				BigInteger eventIndex = toIndex(event.getId());
-				if (eventIndex.equals(lastIndex)) {
+				Long eventIndex = event.getWaitIndex();
+				if (lastIndex.equals(eventIndex)) {
 					events = events.subList(i + 1, events.size());
 					break;
 				}
--- a/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/config/ConsulBinderConfiguration.java
+++ b/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/config/ConsulBinderConfiguration.java
@@ -0,0 +1,65 @@
+/*
+ * Copyright 2013-2016 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.consul.binder.config;
+
+/**
+ */
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.consul.binder.ConsulBinder;
+import org.springframework.cloud.consul.binder.EventService;
+import org.springframework.cloud.stream.binder.Binder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+
+/**
+ * Configures the Consul binder.
+ *
+ * @author Spencer Gibb
+ */
+@Configuration
+@ConditionalOnMissingBean(Binder.class)
+@Import({ PropertyPlaceholderAutoConfiguration.class })
+@EnableConfigurationProperties({ConsulBinderProperties.class})
+public class ConsulBinderConfiguration {
+
+	@Autowired
+	private ConsulBinderProperties consulBinderProperties;
+
+	@Autowired(required = false)
+	protected ObjectMapper objectMapper = new ObjectMapper();
+
+	@Bean
+	@ConditionalOnMissingBean
+	public EventService eventService(ConsulClient consulClient) {
+		return new EventService(consulBinderProperties, consulClient, objectMapper);
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	public ConsulBinder consulClientBinder(EventService eventService) {
+		return new ConsulBinder(eventService);
+	}
+
+	//TODO: create consul client if needed
+}
--- a/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/config/ConsulBinderProperties.java
+++ b/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/config/ConsulBinderProperties.java
@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2015 the original author or authors.
+ * Copyright 2013-2016 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.
@@ -14,7 +14,7 @@
 * limitations under the License.
 */

-package org.springframework.cloud.consul.bus;
+package org.springframework.cloud.consul.binder.config;

 import lombok.Data;
 import org.springframework.boot.context.properties.ConfigurationProperties;
@@ -22,10 +22,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 /**
 * @author Spencer Gibb
 */
-@ConfigurationProperties("spring.cloud.consul.bus")
+@ConfigurationProperties("spring.cloud.stream.consul.binder")
@Data
-public class ConsulBusProperties {
-	boolean enabled = true;
-	int eventDelay = 10;
-	int eventTimeout = 2;
+public class ConsulBinderProperties {
+	private int eventTimeout = 5;
 }
--- a/spring-cloud-consul-binder/src/main/resources/META-INF/spring-cloud-stream/consul-binder.properties
+++ b/spring-cloud-consul-binder/src/main/resources/META-INF/spring-cloud-stream/consul-binder.properties
@@ -0,0 +1,18 @@
+#
+# Copyright 2013-2016 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.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+spring.cloud.stream.binder.consul.default.host=localhost
+spring.cloud.stream.binder.consul.default.port=8500
--- a/spring-cloud-consul-binder/src/main/resources/META-INF/spring.binders
+++ b/spring-cloud-consul-binder/src/main/resources/META-INF/spring.binders
@@ -0,0 +1,2 @@
+consul:\
+org.springframework.cloud.consul.binder.config.ConsulBinderConfiguration
--- a/spring-cloud-consul-binder/src/test/java/org/springframework/cloud/consul/binder/ConsulBinderApplicationTests.java
+++ b/spring-cloud-consul-binder/src/test/java/org/springframework/cloud/consul/binder/ConsulBinderApplicationTests.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright 2013-2016 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.consul.binder;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.aResponse;
+import static com.github.tomakehurst.wiremock.client.WireMock.get;
+import static com.github.tomakehurst.wiremock.client.WireMock.put;
+import static com.github.tomakehurst.wiremock.client.WireMock.putRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlPathMatching;
+import static com.github.tomakehurst.wiremock.client.WireMock.verify;
+import static org.awaitility.Awaitility.await;
+import static org.junit.Assert.assertNotNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.annotation.DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.test.SpringApplicationConfiguration;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.annotation.Output;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+import org.springframework.test.context.web.WebAppConfiguration;
+
+import com.ecwid.consul.v1.ConsulClient;
+import com.github.tomakehurst.wiremock.junit.WireMockRule;
+/**
+ * @author Spencer Gibb
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@SpringApplicationConfiguration(classes = ConsulBinderApplicationTests.Application.class)
+@WebAppConfiguration
+@DirtiesContext(classMode = AFTER_EACH_TEST_METHOD)
+public class ConsulBinderApplicationTests {
+	@Autowired
+	private Events events;
+
+	@Rule
+	public final WireMockRule wireMock = new WireMockRule(18500);
+
+	@Before
+	public void setUp() throws Exception {
+
+		wireMock.stubFor(put(urlPathMatching("/v1/event/fire/purchases"))
+				.willReturn(aResponse().withStatus(200)));
+
+		wireMock.stubFor(get(urlPathMatching("/v1/event/list"))
+				.willReturn(aResponse().withBody("[]")
+						.withStatus(200)
+						.withHeader("X-Consul-Index", "1")));
+	}
+
+	@Test
+	public void shouldInitializeConsulSource() {
+
+		assertNotNull(events);
+	}
+
+	@Test
+	public void shouldPublishTextConsulMessage() {
+
+		// given
+		final Message<String> message = MessageBuilder.withPayload("Hello Consul!")
+				.build();
+
+		// when
+		events.purchases().send(message);
+
+		// then
+		await().atMost(1, TimeUnit.SECONDS);
+		verify(1, putRequestedFor(urlPathMatching("/v1/event/fire/purchases")));
+	}
+
+	interface Events {
+
+		@Output
+		MessageChannel purchases();
+	}
+
+	@Configuration
+	@EnableAutoConfiguration
+	@EnableBinding(Events.class)
+	public static class Application {
+		@Bean
+		public ConsulClient consulClient() {
+			return new ConsulClient("localhost", 18500);
+		}
+
+		@Bean
+		public EventService eventService(ConsulClient consulClient) {
+			EventService eventService = mock(EventService.class);
+			when(eventService.getConsulClient()).thenReturn(consulClient);
+			return eventService;
+		}
+	}
+}
--- a/spring-cloud-consul-binder/src/test/java/org/springframework/cloud/consul/binder/ConsulBinderTests.java
+++ b/spring-cloud-consul-binder/src/test/java/org/springframework/cloud/consul/binder/ConsulBinderTests.java
@@ -0,0 +1,393 @@
+/*
+ * Copyright 2015 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.consul.binder;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.cloud.consul.binder.test.consumer.TestConsumer;
+import org.springframework.cloud.consul.binder.test.producer.TestProducer;
+import org.springframework.cloud.deployer.spi.app.AppDeployer;
+import org.springframework.cloud.deployer.spi.core.AppDefinition;
+import org.springframework.cloud.deployer.spi.core.AppDeploymentRequest;
+import org.springframework.cloud.deployer.spi.local.LocalAppDeployer;
+import org.springframework.cloud.deployer.spi.local.LocalDeployerProperties;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.UrlResource;
+import org.springframework.util.SocketUtils;
+import org.springframework.web.client.ResourceAccessException;
+import org.springframework.web.client.RestTemplate;
+
+/**
+ * Tests for {@link org.springframework.cloud.consul.binder.ConsulBinder}.
+ *
+ * @author Spencer Gibb
+ */
+public class ConsulBinderTests {
+	private static final Logger logger = LoggerFactory.getLogger(ConsulBinderTests.class);
+
+	/**
+	 * Timeout value in milliseconds for operations to complete.
+	 */
+	private static final long TIMEOUT = 30000;
+
+	/**
+	 * Payload of test message.
+	 */
+	public static final String MESSAGE_PAYLOAD = "hello world";
+
+	/**
+	 * Name of binding used for producer and consumer bindings.
+	 */
+	public static final String BINDING_NAME = "test";
+
+	/**
+	 * Deployer to launch producer and consumer test applications.
+	 */
+	private final AppDeployer deployer;
+
+	/**
+	 * Rest template for communicating with producer/consumer test applications.
+	 */
+	private final RestTemplate restTemplate = new RestTemplate();
+
+
+	public ConsulBinderTests() {
+		LocalDeployerProperties properties = new LocalDeployerProperties();
+		properties.setDeleteFilesOnExit(false);
+		this.deployer = new ClasspathDeployer(properties);
+	}
+
+	/**
+	 * Test basic message sending functionality.
+	 *
+	 * @throws Exception
+	 */
+	@Test
+	public void testMessageSendReceive() throws Exception {
+		testMessageSendReceive(null, false);
+	}
+
+	/**
+	 * Test usage of partition selector.
+	 *
+	 * @throws Exception
+	 */
+	/*@Test
+	public void testPartitionedMessageSendReceive() throws Exception {
+		testMessageSendReceive(null, true);
+	}*/
+
+	/**
+	 * Test consumer group functionality.
+	 *
+	 * @throws Exception
+	 */
+	/*@Test
+	public void testMessageSendReceiveConsumerGroups() throws Exception {
+		testMessageSendReceive(new String[]{"a", "b"}, false);
+	}*/
+
+	/**
+	 * Test message sending functionality.
+	 *
+	 * @param groups consumer groups; may be {@code null}
+	 * @param partitioned if true, execute test with a partition selector
+	 * @throws Exception
+	 */
+	private void testMessageSendReceive(String[] groups, boolean partitioned) throws Exception {
+		Set<AppId> consumers = null;
+		AppId producer = null;
+
+		try {
+			consumers = launchConsumers(groups);
+			producer = launchProducer(partitioned);
+
+			for (AppId consumer : consumers) {
+				assertEquals(MESSAGE_PAYLOAD, waitForMessage(consumer.port));
+			}
+
+			if (partitioned) {
+				assertTrue(partitionSelectorUsed(producer.port));
+			}
+		}
+		finally {
+			if (producer != null) {
+				shutdownApplication(producer.id);
+			}
+			if (consumers != null) {
+				for (AppId consumer : consumers) {
+					shutdownApplication(consumer.id);
+				}
+			}
+		}
+	}
+
+	/**
+	 * Launch one or more consumers based on the number of consumer groups.
+	 * Blocks execution until the consumers are bound.
+	 *
+	 * @param groups consumer groups; may be {@code null}
+	 * @return a set of {@link AppId}s for the consumers
+	 * @throws InterruptedException
+	 */
+	private Set<AppId> launchConsumers(String[] groups) throws InterruptedException {
+		Set<AppId> consumers = new HashSet<>();
+
+		Map<String, String> appProperties = new HashMap<>();
+		int consumerCount = groups == null ? 1 : groups.length;
+		for (int i = 0; i < consumerCount; i++) {
+			int consumerPort = SocketUtils.findAvailableTcpPort();
+			appProperties.put("server.port", String.valueOf(consumerPort));
+			List<String> args = new ArrayList<>();
+			args.add(String.format("--server.port=%d", consumerPort));
+			args.add("--debug");
+			if (groups != null) {
+				args.add(String.format("--group=%s", groups[i]));
+			}
+			consumers.add(new AppId(launchApplication(TestConsumer.class, appProperties, args), consumerPort));
+		}
+		for (AppId app : consumers) {
+			waitForConsumer(app.port);
+		}
+
+		return consumers;
+	}
+
+	/**
+	 * Launch a producer that publishes a test message.
+	 *
+	 * @param partitioned if true, configure producer to use a partition selector
+	 * @return {@link AppId} for producer
+	 */
+	private AppId launchProducer(boolean partitioned) {
+		int producerPort = SocketUtils.findAvailableTcpPort();
+		Map<String, String> appProperties = new HashMap<>();
+		appProperties.put("server.port", String.valueOf(producerPort));
+		List<String> args = new ArrayList<>();
+		args.add(String.format("--server.port=%d", producerPort));
+		args.add(String.format("--partitioned=%b", partitioned));
+		args.add("--debug");
+
+		return new AppId(launchApplication(TestProducer.class, appProperties, args), producerPort);
+	}
+
+	/**
+	 * Block the executing thread until the consumer is bound.
+	 *
+	 * @param port server port of the consumer application
+	 * @throws InterruptedException if the thread is interrupted
+	 * @throws AssertionError if the consumer is not bound after
+	 * {@value #TIMEOUT} milliseconds
+	 */
+	private void waitForConsumer(int port) throws InterruptedException {
+		long start = System.currentTimeMillis();
+		while (System.currentTimeMillis() < start + TIMEOUT) {
+			if (isConsumerBound(port)) {
+				return;
+			}
+			else {
+				Thread.sleep(1000);
+			}
+		}
+		assertTrue("Consumer not bound", isConsumerBound(port));
+	}
+
+	/**
+	 * Return {@code true} if the consumer at the provided port is bound.
+	 *
+	 * @param port http port for consumer
+	 * @return true if consumer is bound
+	 */
+	private boolean isConsumerBound(int port) {
+		try {
+			return restTemplate.getForObject(
+					String.format("http://localhost:%d/is-bound", port), Boolean.class);
+		}
+		catch (ResourceAccessException e) {
+			logger.trace("isConsumerBound", e);
+			return false;
+		}
+	}
+
+	/**
+	 * Return the most recent payload message a consumer received.
+	 *
+	 * @param port http port for consumer
+	 * @return the most recent payload message a consumer received;
+	 *         may be {@code null}
+	 */
+	private String getConsumerMessagePayload(int port) {
+		try {
+			return restTemplate.getForObject(
+					String.format("http://localhost:%d/message-payload", port), String.class);
+		}
+		catch (ResourceAccessException e) {
+			logger.debug("getConsumerMessagePayload", e);
+			return null;
+		}
+	}
+
+	/**
+	 * Return {@code true} if the producer made use of a custom partition selector.
+	 *
+	 * @param port http port for producer
+	 * @return true if the producer used a custom partition selector
+	 */
+	private boolean partitionSelectorUsed(int port) throws InterruptedException {
+		try {
+			return restTemplate.getForObject(
+					String.format("http://localhost:%d/partition-strategy-invoked", port),
+					Boolean.class);
+		}
+		catch (ResourceAccessException e) {
+			logger.debug("partitionSelectorUsed", e);
+			return false;
+		}
+	}
+
+	/**
+	 * Block the executing thread until a message is received by the
+	 * consumer application, or until {@value #TIMEOUT} milliseconds elapses.
+	 *
+	 * @param port server port of the consumer application
+	 * @return the message payload that was received
+	 * @throws InterruptedException if the thread is interrupted
+	 */
+	private String waitForMessage(int port) throws InterruptedException {
+		long start = System.currentTimeMillis();
+		String message = null;
+		while (System.currentTimeMillis() < start + TIMEOUT) {
+			message = getConsumerMessagePayload(port);
+			if (message == null) {
+				Thread.sleep(1000);
+			}
+			else {
+				break;
+			}
+		}
+		return message;
+	}
+
+	/**
+	 * Launch an application in a separate JVM.
+	 *
+	 * @param clz the main class to launch
+	 * @param properties the properties to pass to the application
+	 * @param args the command line arguments for the application
+	 * @return a string identifier for the application
+	 */
+	private String launchApplication(Class<?> clz, Map<String, String> properties, List<String> args) {
+		Resource resource = new UrlResource(clz.getProtectionDomain().getCodeSource().getLocation());
+
+		properties.put(AppDeployer.GROUP_PROPERTY_KEY, "test-group");
+		properties.put("main", clz.getName());
+		properties.put("classpath", System.getProperty("java.class.path"));
+
+		String appName = String.format("%s-%s", clz.getSimpleName(), properties.get("server.port"));
+		AppDefinition definition = new AppDefinition(appName, properties);
+
+		AppDeploymentRequest request = new AppDeploymentRequest(definition, resource, properties, args);
+		return this.deployer.deploy(request);
+	}
+
+	/**
+	 * Shut down the application with the provided id.
+	 *
+	 * @param id id of application to shut down
+	 */
+	private void shutdownApplication(String id) {
+		this.deployer.undeploy(id);
+	}
+
+	private static class ClasspathDeployer extends LocalAppDeployer {
+
+		/**
+		 * Instantiates a new local app deployer.
+		 *
+		 * @param properties the properties
+		 */
+		ClasspathDeployer(LocalDeployerProperties properties) {
+			super(properties);
+		}
+
+		/**
+		 * Builds the jar execution command.
+		 *
+		 * @param jarPath the jar path
+		 * @param request the request
+		 * @return the string[]
+		 */
+		protected String[] buildJarExecutionCommand(String jarPath, AppDeploymentRequest request) {
+
+			ArrayList<String> commands = new ArrayList<>();
+			commands.add(super.getLocalDeployerProperties().getJavaCmd());
+			commands.add("-cp");
+			commands.add(request.getDefinition().getProperties().get("classpath"));
+			commands.add(request.getDefinition().getProperties().get("main"));
+			commands.addAll(request.getCommandlineArguments());
+
+			return commands.toArray(new String[commands.size()]);
+		}
+	}
+
+
+	/**
+	 * String identification and http port for a launched application.
+	 */
+	private static class AppId {
+		final String id;
+		final int port;
+
+		AppId(String id, int port) {
+			this.id = id;
+			this.port = port;
+		}
+
+		@Override
+		public boolean equals(Object o) {
+			if (this == o) {
+				return true;
+			}
+			if (o == null || getClass() != o.getClass()) {
+				return false;
+			}
+
+			AppId appId = (AppId) o;
+			return port == appId.port && id.equals(appId.id);
+		}
+
+		@Override
+		public int hashCode() {
+			int result = id.hashCode();
+			result = 31 * result + port;
+			return result;
+		}
+	}
+
+}
--- a/spring-cloud-consul-binder/src/test/java/org/springframework/cloud/consul/binder/test/consumer/TestConsumer.java
+++ b/spring-cloud-consul-binder/src/test/java/org/springframework/cloud/consul/binder/test/consumer/TestConsumer.java
@@ -0,0 +1,107 @@
+/*
+ * Copyright 2013-2016 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.consul.binder.test.consumer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cloud.consul.binder.ConsulBinder;
+import org.springframework.cloud.consul.binder.ConsulBinderTests;
+import org.springframework.cloud.consul.binder.config.ConsulBinderConfiguration;
+import org.springframework.cloud.stream.binder.ConsumerProperties;
+import org.springframework.context.annotation.ComponentScan;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHandler;
+import org.springframework.messaging.MessagingException;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.ExecutorSubscribableChannel;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * Consumer application that binds a channel to a {@link ConsulBinder}
+ * and stores the received message payload.
+ */
+@RestController
+@Import(ConsulBinderConfiguration.class)
+@Configuration
+@EnableAutoConfiguration
+public class TestConsumer implements ApplicationRunner {
+	private static final Logger logger = LoggerFactory.getLogger(TestConsumer.class);
+
+	/**
+	 * Flag that indicates if the consumer has been bound.
+	 */
+	private volatile boolean isBound = false;
+
+	/**
+	 * Payload of last received message.
+	 */
+	private volatile String messagePayload;
+
+	@Autowired
+	private ConsulBinder binder;
+
+	/**
+	 * Main method.
+	 *
+	 * @param args if present, first arg is consumer group name
+	 * @throws Exception
+	 */
+	public static void main(String[] args) throws Exception {
+		SpringApplication.run(TestConsumer.class, args);
+	}
+
+	@Override
+	public void run(ApplicationArguments args) throws Exception {
+		logger.info("Consumer running with binder {}", binder);
+		SubscribableChannel consumerChannel = new ExecutorSubscribableChannel();
+		consumerChannel.subscribe(new MessageHandler() {
+			@Override
+			public void handleMessage(Message<?> message) throws MessagingException {
+				messagePayload = (String) message.getPayload();
+				logger.info("Received message: {}", messagePayload);
+			}
+		});
+		String group = null;
+
+		if (args.containsOption("group")) {
+			group = args.getOptionValues("group").get(0);
+		}
+
+		binder.bindConsumer(ConsulBinderTests.BINDING_NAME, group, consumerChannel,
+				new ConsumerProperties());
+		isBound = true;
+	}
+
+	@RequestMapping("/is-bound")
+	public boolean isBound() {
+		return isBound;
+	}
+
+	@RequestMapping("/message-payload")
+	public String getMessagePayload() {
+		return messagePayload;
+	}
+
+}
--- a/spring-cloud-consul-binder/src/test/java/org/springframework/cloud/consul/binder/test/producer/TestProducer.java
+++ b/spring-cloud-consul-binder/src/test/java/org/springframework/cloud/consul/binder/test/producer/TestProducer.java
@@ -0,0 +1,104 @@
+/*
+ * Copyright 2013-2016 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.consul.binder.test.producer;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.cloud.consul.binder.ConsulBinder;
+import org.springframework.cloud.consul.binder.ConsulBinderTests;
+import org.springframework.cloud.consul.binder.config.ConsulBinderConfiguration;
+import org.springframework.cloud.stream.binder.PartitionSelectorStrategy;
+import org.springframework.cloud.stream.binder.ProducerProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.SubscribableChannel;
+import org.springframework.messaging.support.ExecutorSubscribableChannel;
+import org.springframework.messaging.support.GenericMessage;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * Producer application that binds a channel to a {@link ConsulBinder}
+ * and sends a test message.
+ */
+@RestController
+@Import(ConsulBinderConfiguration.class)
+@Configuration
+@EnableAutoConfiguration
+public class TestProducer implements ApplicationRunner {
+
+	private static final Logger logger = LoggerFactory.getLogger(TestProducer.class);
+
+	@Autowired
+	private ConsulBinder binder;
+
+	public static void main(String[] args) {
+		SpringApplication.run(TestProducer.class, args);
+	}
+
+	@Override
+	public void run(ApplicationArguments args) throws Exception {
+		if (args.containsOption("partitioned")
+				&& Boolean.valueOf(args.getOptionValues("partitioned").get(0))) {
+			binder.setPartitionSelector(stubPartitionSelectorStrategy());
+		}
+		SubscribableChannel producerChannel = producerChannel();
+		ProducerProperties properties = new ProducerProperties();
+		properties.setPartitionKeyExpression(new SpelExpressionParser().parseExpression("payload"));
+		binder.bindProducer(ConsulBinderTests.BINDING_NAME, producerChannel, properties);
+
+		Message<String> message = new GenericMessage<>(ConsulBinderTests.MESSAGE_PAYLOAD);
+		logger.info("Writing message to binder {}", binder);
+		producerChannel.send(message);
+	}
+
+	@Bean
+	public SubscribableChannel producerChannel() {
+		return new ExecutorSubscribableChannel();
+	}
+
+	@Bean
+	public StubPartitionSelectorStrategy stubPartitionSelectorStrategy() {
+		return new StubPartitionSelectorStrategy();
+	}
+
+	@RequestMapping("/partition-strategy-invoked")
+	public boolean partitionStrategyInvoked() {
+		return stubPartitionSelectorStrategy().invoked;
+	}
+
+
+	public static class StubPartitionSelectorStrategy implements PartitionSelectorStrategy {
+		public volatile boolean invoked = false;
+
+		@Override
+		public int selectPartition(Object key, int partitionCount) {
+			logger.info("Selecting partition for key {}; partition count: {}", key, partitionCount);
+			invoked = true;
+			return 1;
+		}
+	}
+
+}
--- a/spring-cloud-consul-binder/src/test/resources/application.yml
+++ b/spring-cloud-consul-binder/src/test/resources/application.yml
@@ -0,0 +1,10 @@
+spring:
+  cloud:
+    stream:
+      binders:
+        purchases:
+          type: consul
+      consul:
+        binder:
+#          host: localhost
+#          port: 18500
--- a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulBusAutoConfiguration.java
+++ b/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulBusAutoConfiguration.java
@@ -1,96 +0,0 @@
-/*
- * Copyright 2013-2015 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.consul.bus;
-
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.boot.autoconfigure.AutoConfigureAfter;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.cloud.bus.BusAutoConfiguration;
-import org.springframework.cloud.bus.event.RemoteApplicationEvent;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.dsl.IntegrationFlow;
-import org.springframework.integration.dsl.IntegrationFlows;
-import org.springframework.integration.dsl.support.Transformers;
-import org.springframework.integration.support.json.Jackson2JsonObjectMapper;
-import org.springframework.messaging.MessageChannel;
-import org.springframework.scheduling.annotation.EnableScheduling;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-/**
- * @author Spencer Gibb
- */
-@Configuration
-@ConditionalOnConsulEnabled
-@ConditionalOnProperty(value = "spring.cloud.consul.bus.enabled", matchIfMissing = true)
-@AutoConfigureAfter(BusAutoConfiguration.class)
-@EnableScheduling
-@EnableConfigurationProperties
-public class ConsulBusAutoConfiguration {
-	@Autowired
-	@Qualifier("cloudBusInboundChannel")
-	MessageChannel cloudBusInboundChannel;
-
-	@Autowired
-	ObjectMapper objectMapper;
-
-	@Bean
-	public ConsulBusProperties consulBusProperties() {
-		return new ConsulBusProperties();
-	}
-
-	@Bean
-	public EventService eventService() {
-		return new EventService();
-	}
-
-	@Bean
-	public ConsulOutboundEndpoint consulOutboundEndpoint() {
-		return new ConsulOutboundEndpoint();
-	}
-
-	@Bean
-	public IntegrationFlow cloudBusConsulOutboundFlow(
-			@Qualifier("cloudBusOutboundChannel") MessageChannel cloudBusOutboundChannel) {
-		return IntegrationFlows.from(cloudBusOutboundChannel)
-		// TODO: put the json headers as part of the message, here?
-				.transform(Transformers.toJson()).handle(consulOutboundEndpoint()).get();
-	}
-
-	@Bean
-	public IntegrationFlow cloudBusConsulInboundFlow() {
-		return IntegrationFlows
-				.from(consulInboundChannelAdapter())
-				.transform(
-                        Transformers.fromJson(RemoteApplicationEvent.class,
-                                new Jackson2JsonObjectMapper(objectMapper)))
-				.channel(cloudBusInboundChannel) // now set in consulInboundChannelAdapter
-													// bean
-				.get();
-	}
-
-	@Bean
-	public ConsulInboundChannelAdapter consulInboundChannelAdapter() {
-		ConsulInboundChannelAdapter adapter = new ConsulInboundChannelAdapter();
-		adapter.setOutputChannel(cloudBusInboundChannel);
-		return adapter;
-	}
-
-}
--- a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/SimpleRemoteEvent.java
+++ b/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/SimpleRemoteEvent.java
@@ -1,51 +0,0 @@
-/*
- * Copyright 2013-2015 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.consul.bus;
-
-import lombok.Data;
-
-import org.springframework.cloud.bus.event.RemoteApplicationEvent;
-
-import com.fasterxml.jackson.annotation.JsonTypeName;
-
-/**
- * @author Spencer Gibb
- */
-@JsonTypeName("simple")
-@Data
-public class SimpleRemoteEvent extends RemoteApplicationEvent {
-
-	private String message;
-
-	private SimpleRemoteEvent() {
-	}
-
-	public SimpleRemoteEvent(Object source, String originService,
-			String destinationService, String message) {
-		super(source, originService, destinationService);
-		this.message = message;
-	}
-
-	public SimpleRemoteEvent(Object source, String originService, String message) {
-		super(source, originService);
-		this.message = message;
-	}
-
-	public boolean canEqual(Object other) {
-		return other instanceof RemoteApplicationEvent;
-	}
-}
--- a/spring-cloud-consul-bus/src/main/resources/META-INF/spring.factories
+++ b/spring-cloud-consul-bus/src/main/resources/META-INF/spring.factories
@@ -1,3 +0,0 @@
-# Auto Configuration
-org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-org.springframework.cloud.consul.bus.ConsulBusAutoConfiguration
--- a/spring-cloud-consul-bus/src/test/java/org/springframework/cloud/consul/bus/ConsulBusIT.java
+++ b/spring-cloud-consul-bus/src/test/java/org/springframework/cloud/consul/bus/ConsulBusIT.java
@@ -1,128 +0,0 @@
-/*
- * Copyright 2013-2015 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.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.cloud.consul.bus;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-import org.junit.FixMethodOrder;
-import org.junit.Test;
-import org.junit.runners.MethodSorters;
-import org.springframework.boot.builder.SpringApplicationBuilder;
-import org.springframework.cloud.bus.BusAutoConfiguration;
-import org.springframework.cloud.bus.event.RemoteApplicationEvent;
-import org.springframework.cloud.bus.jackson.SubtypeModule;
-import org.springframework.cloud.consul.ConsulAutoConfiguration;
-import org.springframework.context.ApplicationListener;
-import org.springframework.context.ConfigurableApplicationContext;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Import;
-import org.springframework.integration.config.EnableIntegration;
-import org.springframework.integration.dsl.support.Transformers;
-import org.springframework.integration.json.JsonToObjectTransformer;
-import org.springframework.integration.support.json.Jackson2JsonObjectMapper;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.support.GenericMessage;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-
-/**
- * @author Spencer Gibb
- */
-@FixMethodOrder(MethodSorters.NAME_ASCENDING)
-public class ConsulBusIT {
-
-	@Test
-	public void test001ConsulOutboundEndpoint_HandleRequestMessage() {
-		ConfigurableApplicationContext context = getOutboundContext();
-		context.publishEvent(new SimpleRemoteEvent(this, "testService", "testMessage"));
-	}
-
-	private ConfigurableApplicationContext getOutboundContext() {
-		System.setProperty("spring.cloud.config.enabled", "false");
-		ConfigurableApplicationContext context = new SpringApplicationBuilder()
-				.web(false).sources(OutboundConfig.class).run();
-		context.setId("testService");
-		return context;
-	}
-
-	/*
-	 * @Test public void test002ConsulInboundChannelAdapter() {
-	 * ConfigurableApplicationContext inbound = getInboundContext();
-	 * ConfigurableApplicationContext outbound = getOutboundContext();
-	 * outbound.publishEvent(new TestMessage(this, "testService", "inboundTestService",
-	 * "testMessage"));
-	 * 
-	 * InboundConfig inboundConfig = inbound.getBean(InboundConfig.class);
-	 * assertNotNull("message was null", inboundConfig.message); }
-	 * 
-	 * private ConfigurableApplicationContext getInboundContext() {
-	 * System.setProperty("spring.cloud.config.enabled", "false");
-	 * ConfigurableApplicationContext context = new SpringApplicationBuilder() .web(false)
-	 * .sources(InboundConfig.class) .run(); context.setId("inboundTestService"); return
-	 * context; }
-	 */
-
-	protected static final String JSON_PAYLOAD = "{\"type\":\"simple\",\"timestamp\":1416349427372,\"originService\":\"testService\",\"destinationService\":null,\"message\":\"testMessage\"}";
-
-	@Test
-	public void test003JsonToObject() {
-		ObjectMapper objectMapper = new ObjectMapper();
-		objectMapper.registerModule(new SubtypeModule(SimpleRemoteEvent.class));
-		JsonToObjectTransformer transformer = Transformers.fromJson(
-				RemoteApplicationEvent.class, new Jackson2JsonObjectMapper(objectMapper));
-		/*
-		 * HashMap<String, Object> map = new HashMap<>(); map.put(JsonHeaders.TYPE_ID,
-		 * RemoteApplicationEvent.class);
-		 */
-		Message<?> message = transformer.transform(new GenericMessage<>(JSON_PAYLOAD));
-		Object payload = message.getPayload();
-		assertTrue("payload is of wrong type", payload instanceof RemoteApplicationEvent);
-		assertTrue("payload is of wrong type", payload instanceof SimpleRemoteEvent);
-		SimpleRemoteEvent event = (SimpleRemoteEvent) payload;
-		assertEquals("payload is wrong", "testMessage", event.getMessage());
-	}
-
-	@Configuration
-	@Import({ ConsulAutoConfiguration.class, BusAutoConfiguration.class,
-			ConsulBusAutoConfiguration.class })
-	@EnableIntegration
-	public static class OutboundConfig {
-
-		@Bean
-		public ObjectMapper objectMapper() {
-			ObjectMapper objectMapper = new ObjectMapper();
-			objectMapper.registerModule(new SubtypeModule(SimpleRemoteEvent.class));
-			return objectMapper;
-		}
-	}
-
-	@Configuration
-	@Import({ ConsulAutoConfiguration.class, BusAutoConfiguration.class,
-			ConsulBusAutoConfiguration.class })
-	@EnableIntegration
-	public static class InboundConfig implements
-			ApplicationListener<RemoteApplicationEvent> {
-		RemoteApplicationEvent message;
-
-		@Override
-		public void onApplicationEvent(RemoteApplicationEvent event) {
-			this.message = event;
-		}
-	}
-}
--- a/spring-cloud-consul-dependencies/pom.xml
+++ b/spring-cloud-consul-dependencies/pom.xml
@@ -17,6 +17,8 @@
 		<spring-cloud-bus.version>1.1.1.BUILD-SNAPSHOT</spring-cloud-bus.version>
 		<spring-cloud-commons.version>1.1.2.BUILD-SNAPSHOT</spring-cloud-commons.version>
 		<spring-cloud-netflix.version>1.1.4.BUILD-SNAPSHOT</spring-cloud-netflix.version>
+		<spring-cloud-deployer.version>1.0.0.BUILD-SNAPSHOT</spring-cloud-deployer.version>
+		<spring-cloud-stream.version>1.0.1.BUILD-SNAPSHOT</spring-cloud-stream.version>
 		<consul-api.version>1.1.10</consul-api.version>
 		<gson.version>2.3.1</gson.version>
 		<httpclient.version>4.5</httpclient.version>
@@ -30,12 +32,11 @@
 				<artifactId>spring-cloud-consul-core</artifactId>
 				<version>${project.version}</version>
 			</dependency>
-			<!-- TODO: add this back when consul-bus is implemented -->
-			<!--<dependency>
+			<dependency>
 				<groupId>org.springframework.cloud</groupId>
-				<artifactId>spring-cloud-consul-bus</artifactId>
+				<artifactId>spring-cloud-consul-binder</artifactId>
 				<version>${project.version}</version>
-			</dependency>-->
+			</dependency>
 			<dependency>
 				<groupId>org.springframework.cloud</groupId>
 				<artifactId>spring-cloud-consul-config</artifactId>
@@ -51,12 +52,11 @@
 				<artifactId>spring-cloud-starter-consul</artifactId>
 				<version>${project.version}</version>
 			</dependency>
-			<!-- TODO: add this back when consul-bus is implemented -->
-			<!--<dependency>
+			<dependency>
 				<groupId>org.springframework.cloud</groupId>
 				<artifactId>spring-cloud-starter-consul-bus</artifactId>
 				<version>${project.version}</version>
-			</dependency>-->
+			</dependency>
 			<dependency>
 				<groupId>org.springframework.cloud</groupId>
 				<artifactId>spring-cloud-starter-consul-config</artifactId>
@@ -105,6 +105,24 @@
 				<artifactId>joda-time</artifactId>
 				<version>${joda-time.version}</version>
 			</dependency>
+			<dependency>
+                <groupId>org.springframework.cloud</groupId>
+                <artifactId>spring-cloud-deployer-local</artifactId>
+                <version>${spring-cloud-deployer.version}</version>
+            </dependency>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-stream-binder-test</artifactId>
+				<version>${spring-cloud-stream.version}</version>
+				<scope>test</scope>
+			</dependency>
+			<dependency>
+				<groupId>org.springframework.cloud</groupId>
+				<artifactId>spring-cloud-stream-dependencies</artifactId>
+				<version>${spring-cloud-stream.version}</version>
+				<type>pom</type>
+				<scope>import</scope>
+			</dependency>
 			<dependency>
 				<groupId>org.springframework.cloud</groupId>
 				<artifactId>spring-cloud-bus-dependencies</artifactId>
--- a/spring-cloud-starter-consul-all/pom.xml
+++ b/spring-cloud-starter-consul-all/pom.xml
@@ -20,10 +20,10 @@
 		<main.basedir>${basedir}/../..</main.basedir>
 	</properties>
 	<dependencies>
-		<!--<dependency>
+		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-starter-consul-bus</artifactId>
-		</dependency>-->
+		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
 			<artifactId>spring-cloud-starter-consul-config</artifactId>
--- a/spring-cloud-starter-consul-bus/pom.xml
+++ b/spring-cloud-starter-consul-bus/pom.xml
@@ -5,7 +5,7 @@
 	<parent>
 		<groupId>org.springframework.cloud</groupId>
 		<artifactId>spring-cloud-consul</artifactId>
-		<version>1.0.1.BUILD-SNAPSHOT</version>
+		<version>1.0.2.BUILD-SNAPSHOT</version>
 		<relativePath>..</relativePath>
 	</parent>
 	<artifactId>spring-cloud-starter-consul-bus</artifactId>
@@ -26,7 +26,7 @@
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
-			<artifactId>spring-cloud-consul-bus</artifactId>
+			<artifactId>spring-cloud-consul-binder</artifactId>
 		</dependency>
 		<dependency>
 			<groupId>org.springframework.cloud</groupId>
--- a/src/test/resources/consul_acl/consul_anonymous_acl.json
+++ b/src/test/resources/consul_acl/consul_anonymous_acl.json
@@ -8,6 +8,11 @@
        \"policy\": \"write\"
      }
    },
+    \"event\": {
+      \"\": {
+        \"policy\": \"write\"
+      }
+    },
    \"service\": {
      \"\": {
        \"policy\": \"write\"