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 index 8abb03b1..a9020664 100644 --- 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 @@ -33,6 +33,11 @@ public class ConsulBusAutoConfiguration { @Autowired ObjectMapper objectMapper; + @Bean + public EventService eventService() { + return new EventService(); + } + @Bean public ConsulOutboundEndpoint consulOutboundEndpoint() { return new ConsulOutboundEndpoint(); diff --git a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulInboundChannelAdapter.java b/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulInboundChannelAdapter.java index ee9f773f..803cec16 100644 --- a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulInboundChannelAdapter.java +++ b/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulInboundChannelAdapter.java @@ -1,14 +1,15 @@ package org.springframework.cloud.consul.bus; -import com.fasterxml.jackson.databind.ObjectMapper; +import static org.springframework.util.Base64Utils.decodeFromString; + +import java.io.IOException; +import java.util.List; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.scheduling.annotation.Scheduled; -import java.io.IOException; -import java.util.HashMap; -import java.util.List; -import java.util.Map; +import com.ecwid.consul.v1.event.model.Event; /** * Adapter that receives Messages from Consul Events, converts them into @@ -16,11 +17,8 @@ import java.util.Map; * @author Spencer Gibb */ public class ConsulInboundChannelAdapter extends MessageProducerSupport { - //@Autowired - //private EventService eventService; - @Autowired - private ObjectMapper objectMapper; + private EventService eventService; public ConsulInboundChannelAdapter() { } @@ -45,16 +43,16 @@ public class ConsulInboundChannelAdapter extends MessageProducerSupport { @Scheduled(fixedDelayString = "10") public void getEvents() throws IOException { - /*FIXME: List events = eventService.watch(); + List events = eventService.watch(); for (Event event : events) { - Map headers = new HashMap<>(); + //Map headers = new HashMap<>(); //headers.put(MessageHeaders.REPLY_CHANNEL, outputChannel.) + String decoded = new String(decodeFromString(event.getPayload())); sendMessage(getMessageBuilderFactory() - //TODO: deal with odd objectMapper thing - .withPayload(objectMapper.readValue(event.getDecoded(), String.class)) - //TODO: support headers + .withPayload(decoded) + //TODO: support headers .build()); - }*/ + } } @Override diff --git a/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/EventService.java b/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/EventService.java new file mode 100644 index 00000000..0d61e085 --- /dev/null +++ b/spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/EventService.java @@ -0,0 +1,129 @@ +package org.springframework.cloud.consul.bus; + +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 com.ecwid.consul.v1.ConsulClient; +import com.ecwid.consul.v1.QueryParams; +import com.ecwid.consul.v1.Response; +import com.ecwid.consul.v1.event.model.Event; +import com.ecwid.consul.v1.event.model.EventParams; +import com.fasterxml.jackson.databind.ObjectMapper; +/** + * @author Spencer Gibb + */ +public class EventService { + + @Autowired + protected ConsulClient consul; + + @Autowired(required = false) + protected ObjectMapper objectMapper = new ObjectMapper(); + + private AtomicReference lastIndex = new AtomicReference<>(); + + @PostConstruct + public void init() { + setLastIndex(getEventsResponse()); + } + + private void setLastIndex(Response response) { + Long consulIndex = response.getConsulIndex(); + if (consulIndex != null) { + lastIndex.set(BigInteger.valueOf(consulIndex)); + } + } + + public BigInteger getLastIndex() { + return lastIndex.get(); + } + + public Event fire(String name, String payload) { + Response response = consul.eventFire(name, payload, new EventParams(), QueryParams.DEFAULT); + return response.getValue(); + } + + public Response> getEventsResponse() { + return consul.eventList(QueryParams.DEFAULT); + } + + public List getEvents() { + return getEventsResponse().getValue(); + } + + /** + * from https://github.com/armon/consul-api/blob/master/event.go#L92-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); + BigInteger index = lowVal.xor(highVal); + return index; + } + + public List getEvents(BigInteger lastIndex) { + return filterEvents(readEvents(getEventsResponse()), lastIndex); + } + + public List watch() { + return watch(lastIndex.get()); + } + + public List watch(BigInteger lastIndex) { + //TODO: parameterized or configurable watch time + long index = -1; + if (lastIndex != null) { + lastIndex.longValue(); + } + Response> watch = consul.eventList(new QueryParams(2, index)); + return filterEvents(readEvents(watch), lastIndex); + } + + protected List readEvents(Response> response) { + setLastIndex(response); + return response.getValue(); + } + + /** + * from https://github.com/hashicorp/consul/blob/master/watch/funcs.go#L169-L194 + */ + protected List filterEvents(List toFilter, BigInteger 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)) { + events = events.subList(i + 1, events.size()); + break; + } + } + } + return events; + } + +} + 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 index 58d12139..74a87203 100644 --- 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,8 +1,6 @@ package org.springframework.cloud.consul.bus; -import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import com.fasterxml.jackson.databind.ObjectMapper; -import lombok.Data; import org.junit.FixMethodOrder; import org.junit.Test; import org.junit.runners.MethodSorters; @@ -66,7 +64,7 @@ public class ConsulBusIT { return context; }*/ - protected static final String JSON_PAYLOAD = "{\"type\":\"simple\",\"timestamp\":1416349427372,\"originService\":\"testService\",\"destinationService\":null,\"headers\":{},\"message\":\"testMessage\"}"; + protected static final String JSON_PAYLOAD = "{\"type\":\"simple\",\"timestamp\":1416349427372,\"originService\":\"testService\",\"destinationService\":null,\"message\":\"testMessage\"}"; @Test public void test003JsonToObject() { @@ -75,10 +73,12 @@ public class ConsulBusIT { 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/*, map*/)); + 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