SpringCloud/spring-cloud-consul
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

get consul bus working again

Browse Source
This commit is contained in:
Spencer Gibb
2015-03-13 12:54:38 -06:00
parent 2ccd280734
commit a084b4b728
4 changed files with 151 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulBusAutoConfiguration.java
View File

@@ -33,6 +33,11 @@ public class ConsulBusAutoConfiguration {
@Autowired
ObjectMapper objectMapper;
@Bean
public EventService eventService() {
return new EventService();
}
@Bean
public ConsulOutboundEndpoint consulOutboundEndpoint() {
return new ConsulOutboundEndpoint();

28
spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/ConsulInboundChannelAdapter.java
View File

@@ -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<Event> events = eventService.watch();
List<Event> events = eventService.watch();
for (Event event : events) {
Map<String, Object> headers = new HashMap<>();
//Map<String, Object> 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

129
spring-cloud-consul-bus/src/main/java/org/springframework/cloud/consul/bus/EventService.java Normal file
View File

@@ -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<BigInteger> 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<Event> response = consul.eventFire(name, payload, new EventParams(), QueryParams.DEFAULT);
return response.getValue();
}
public Response<List<Event>> getEventsResponse() {
return consul.eventList(QueryParams.DEFAULT);
}
public List<Event> 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<Event> getEvents(BigInteger lastIndex) {
return filterEvents(readEvents(getEventsResponse()), lastIndex);
}
public List<Event> watch() {
return watch(lastIndex.get());
}
public List<Event> watch(BigInteger lastIndex) {
//TODO: parameterized or configurable watch time
long index = -1;
if (lastIndex != null) {
lastIndex.longValue();
}
Response<List<Event>> watch = consul.eventList(new QueryParams(2, index));
return filterEvents(readEvents(watch), lastIndex);
}
protected List<Event> readEvents(Response<List<Event>> response) {
setLastIndex(response);
return response.getValue();
}
/**
* from https://github.com/hashicorp/consul/blob/master/watch/funcs.go#L169-L194
*/
protected List<Event> filterEvents(List<Event> toFilter, BigInteger 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)) {
events = events.subList(i + 1, events.size());
break;
}
}
}
return events;
}
}

8
spring-cloud-consul-bus/src/test/java/org/springframework/cloud/consul/bus/ConsulBusIT.java
View File

@@ -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<String, Object> map = new HashMap<>();
map.put(JsonHeaders.TYPE_ID, RemoteApplicationEvent.class);*/
Message<?> message = transformer.transform(new GenericMessage<String>(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