diff --git a/pom.xml b/pom.xml index 723d0271..54270d8f 100644 --- a/pom.xml +++ b/pom.xml @@ -31,10 +31,10 @@ spring-cloud-consul-core spring-cloud-consul-config spring-cloud-consul-discovery - + spring-cloud-consul-binder spring-cloud-consul-sample spring-cloud-starter-consul - + spring-cloud-starter-consul-bus spring-cloud-starter-consul-config spring-cloud-starter-consul-discovery spring-cloud-starter-consul-all diff --git a/spring-cloud-consul-bus/pom.xml b/spring-cloud-consul-binder/pom.xml similarity index 62% rename from spring-cloud-consul-bus/pom.xml rename to spring-cloud-consul-binder/pom.xml index 99485bb8..48d1438e 100644 --- a/spring-cloud-consul-bus/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"> 4.0.0 - spring-cloud-consul-bus + spring-cloud-consul-binder jar - Spring Cloud Consul Bus - Spring Cloud Consul Bus + Spring Cloud Consul Binder + Spring Cloud Consul Binder org.springframework.cloud spring-cloud-consul - 1.0.1.BUILD-SNAPSHOT + 1.0.2.BUILD-SNAPSHOT .. @@ -29,29 +29,46 @@ org.springframework.cloud - spring-cloud-bus - true + spring-cloud-stream com.ecwid.consul consul-api true - - org.springframework.integration - spring-integration-java-dsl - org.projectlombok lombok provided + + org.springframework.cloud + spring-cloud-deployer-local + test + + + org.springframework.cloud + spring-cloud-stream-test-support-internal + test + org.springframework.boot spring-boot-starter-test test + + com.github.tomakehurst + wiremock + 1.58 + test + + + org.awaitility + awaitility + 2.0.0 + test + diff --git 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 new file mode 100644 index 00000000..b18321b4 --- /dev/null +++ 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 { + + private static final String BEAN_NAME_TEMPLATE = "outbound.%s"; + + private final EventService eventService; + + public ConsulBinder(EventService eventService) { + this.eventService = eventService; + } + + @Override + protected Binding 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 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); + } +} diff --git a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulInboundChannelAdapter.java b/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/ConsulInboundMessageProducer.java similarity index 65% rename from spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulInboundChannelAdapter.java rename to spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/ConsulInboundMessageProducer.java index a67ca8c5..b55f27f2 100644 --- a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulInboundChannelAdapter.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 events = eventService.watch(); for (Event event : events) { // Map headers = new HashMap<>(); @@ -71,8 +93,4 @@ public class ConsulInboundChannelAdapter extends MessageProducerSupport { .build()); } } - - @Override - protected void doStop() { - } } diff --git a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulOutboundEndpoint.java b/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/ConsulSendingHandler.java similarity index 61% rename from spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulOutboundEndpoint.java rename to spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/ConsulSendingHandler.java index a77ad9b1..f18fe60c 100644 --- a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulOutboundEndpoint.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 = consul.eventFire("springCloudBus", (String) payload, + Response event = consul.eventFire(this.eventName, (String) payload, new EventParams(), QueryParams.DEFAULT); // TODO: return event? - return null; + // return null; } } diff --git a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/EventService.java b/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/EventService.java similarity index 59% rename from spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/EventService.java rename to spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/EventService.java index 24e3cad4..b2820bc9 100644 --- a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/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 lastIndex = new AtomicReference<>(); + private AtomicReference 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 getEvents(BigInteger lastIndex) { + public List getEvents(Long lastIndex) { return filterEvents(readEvents(getEventsResponse()), lastIndex); } @@ -104,13 +91,13 @@ public class EventService { return watch(lastIndex.get()); } - public List watch(BigInteger lastIndex) { + public List watch(Long lastIndex) { // TODO: parameterized or configurable watch time long index = -1; if (lastIndex != null) { - index = lastIndex.longValue(); + index = lastIndex; } - Response> watch = consul.eventList(new QueryParams(properties.eventTimeout, index)); + Response> 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 filterEvents(List toFilter, BigInteger lastIndex) { + protected List filterEvents(List toFilter, Long lastIndex) { List 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; } diff --git 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 new file mode 100644 index 00000000..82e8236e --- /dev/null +++ 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 +} diff --git a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulBusProperties.java b/spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/config/ConsulBinderProperties.java similarity index 73% rename from spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulBusProperties.java rename to spring-cloud-consul-binder/src/main/java/org/springframework/cloud/consul/binder/config/ConsulBinderProperties.java index 8b1148e9..b7dc79f7 100644 --- a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulBusProperties.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; } diff --git 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 new file mode 100644 index 00000000..db28478f --- /dev/null +++ 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 diff --git a/spring-cloud-consul-binder/src/main/resources/META-INF/spring.binders b/spring-cloud-consul-binder/src/main/resources/META-INF/spring.binders new file mode 100644 index 00000000..c3f435cd --- /dev/null +++ b/spring-cloud-consul-binder/src/main/resources/META-INF/spring.binders @@ -0,0 +1,2 @@ +consul:\ +org.springframework.cloud.consul.binder.config.ConsulBinderConfiguration diff --git 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 new file mode 100644 index 00000000..280d4ba9 --- /dev/null +++ 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 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; + } + } +} diff --git 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 new file mode 100644 index 00000000..5158f549 --- /dev/null +++ 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 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 launchConsumers(String[] groups) throws InterruptedException { + Set consumers = new HashSet<>(); + + Map 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 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 appProperties = new HashMap<>(); + appProperties.put("server.port", String.valueOf(producerPort)); + List 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 properties, List 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 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; + } + } + +} diff --git 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 new file mode 100644 index 00000000..11e3a48b --- /dev/null +++ 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; + } + +} diff --git 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 new file mode 100644 index 00000000..88e3d049 --- /dev/null +++ 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 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; + } + } + +} diff --git a/spring-cloud-consul-binder/src/test/resources/application.yml b/spring-cloud-consul-binder/src/test/resources/application.yml new file mode 100644 index 00000000..65f59855 --- /dev/null +++ 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 diff --git 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 deleted file mode 100644 index 66272fe3..00000000 --- a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulBusAutoConfiguration.java +++ /dev/null @@ -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; - } - -} diff --git 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 deleted file mode 100644 index 6d76625f..00000000 --- a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/SimpleRemoteEvent.java +++ /dev/null @@ -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; - } -} diff --git a/spring-cloud-consul-bus/src/main/resources/META-INF/spring.factories b/spring-cloud-consul-bus/src/main/resources/META-INF/spring.factories deleted file mode 100644 index ad77841e..00000000 --- a/spring-cloud-consul-bus/src/main/resources/META-INF/spring.factories +++ /dev/null @@ -1,3 +0,0 @@ -# Auto Configuration -org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ -org.springframework.cloud.consul.bus.ConsulBusAutoConfiguration diff --git 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 deleted file mode 100644 index aaa4bb8c..00000000 --- a/spring-cloud-consul-bus/src/test/java/org/springframework/cloud/consul/bus/ConsulBusIT.java +++ /dev/null @@ -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 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 message; - - @Override - public void onApplicationEvent(RemoteApplicationEvent event) { - this.message = event; - } - } -} diff --git a/spring-cloud-consul-dependencies/pom.xml b/spring-cloud-consul-dependencies/pom.xml index 161076e1..f6e130be 100644 --- a/spring-cloud-consul-dependencies/pom.xml +++ b/spring-cloud-consul-dependencies/pom.xml @@ -17,6 +17,8 @@ 1.1.1.BUILD-SNAPSHOT 1.1.2.BUILD-SNAPSHOT 1.1.4.BUILD-SNAPSHOT + 1.0.0.BUILD-SNAPSHOT + 1.0.1.BUILD-SNAPSHOT 1.1.10 2.3.1 4.5 @@ -30,12 +32,11 @@ spring-cloud-consul-core ${project.version} - - + org.springframework.cloud spring-cloud-consul-config @@ -51,12 +52,11 @@ spring-cloud-starter-consul ${project.version} - - + org.springframework.cloud spring-cloud-starter-consul-config @@ -105,6 +105,24 @@ joda-time ${joda-time.version} + + org.springframework.cloud + spring-cloud-deployer-local + ${spring-cloud-deployer.version} + + + org.springframework.cloud + spring-cloud-stream-binder-test + ${spring-cloud-stream.version} + test + + + org.springframework.cloud + spring-cloud-stream-dependencies + ${spring-cloud-stream.version} + pom + import + org.springframework.cloud spring-cloud-bus-dependencies diff --git a/spring-cloud-starter-consul-all/pom.xml b/spring-cloud-starter-consul-all/pom.xml index 6314e93a..5fa3321e 100644 --- a/spring-cloud-starter-consul-all/pom.xml +++ b/spring-cloud-starter-consul-all/pom.xml @@ -20,10 +20,10 @@ ${basedir}/../.. - + org.springframework.cloud spring-cloud-starter-consul-config diff --git a/spring-cloud-starter-consul-bus/pom.xml b/spring-cloud-starter-consul-bus/pom.xml index 4dd5ff52..d5f23cf6 100644 --- a/spring-cloud-starter-consul-bus/pom.xml +++ b/spring-cloud-starter-consul-bus/pom.xml @@ -5,7 +5,7 @@ org.springframework.cloud spring-cloud-consul - 1.0.1.BUILD-SNAPSHOT + 1.0.2.BUILD-SNAPSHOT .. spring-cloud-starter-consul-bus @@ -26,7 +26,7 @@ org.springframework.cloud - spring-cloud-consul-bus + spring-cloud-consul-binder org.springframework.cloud diff --git a/src/test/resources/consul_acl/consul_anonymous_acl.json b/src/test/resources/consul_acl/consul_anonymous_acl.json index 39f16ac2..f2485952 100644 --- 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\"