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

Ensure messages flow as JSON

Browse Source 
This is for readability and also for compatibility with
Angel apps.
This commit is contained in:
Dave Syer
2016-03-18 10:35:08 +00:00
parent e5d5688618
commit 5a14eda63a
8 changed files with 232 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
docs/src/main/asciidoc/spring-cloud-bus.adoc
View File

@@ -97,4 +97,11 @@ to your app (and enable tracing). Or you could tap into the
NOTE: Any Bus application can trace acks, but sometimes it will be
useful to do this in a central service that can do more complex
queries on the data. Or forward it to a specialized tracing service.
queries on the data. Or forward it to a specialized tracing service.
== Broadcasting Your Own Events
The Bus can carry any event of type `RemoteApplicationEvent`, but the
default transport is JSON and the deserializer needs to know which
types are going to be used ahead of time. To register a new type you
can use `@JsonTypeName` on your custom class.

81
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java Normal file
View File

@@ -0,0 +1,81 @@
/*
* 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.bus;
import java.util.HashMap;
import java.util.Map;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.env.EnvironmentPostProcessor;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.MapPropertySource;
import org.springframework.core.env.MutablePropertySources;
import org.springframework.core.env.PropertySource;
/**
* {@link EnvironmentPostProcessor} that sets the default properties for the Bus.
*
* @author Dave Syer
*
* @since 1.0.0
*/
public class BusEnvironmentPostProcessor implements EnvironmentPostProcessor {
private static final String PROPERTY_SOURCE_NAME = "defaultProperties";
@Override
public void postProcessEnvironment(ConfigurableEnvironment environment,
SpringApplication application) {
Map<String, Object> map = new HashMap<String, Object>();
// Technically this is only needed on the consumer, but it's fine to be explicit
// on producers as well. It puts all consumers in the same "group", meaning they
// compete with each other and only one gets each message.
map.put("spring.cloud.stream.bindings." + SpringCloudBusClient.OUTPUT
+ ".content-type",
environment.getProperty("spring.cloud.bus.content-type",
"application/json"));
map.put("spring.cloud.stream.bindings." + SpringCloudBusClient.INPUT
+ ".content-type",
"application/x-java-object;type="
+ RemoteApplicationEvent.class.getName());
addOrReplace(environment.getPropertySources(), map);
}
private void addOrReplace(MutablePropertySources propertySources,
Map<String, Object> map) {
MapPropertySource target = null;
if (propertySources.contains(PROPERTY_SOURCE_NAME)) {
PropertySource<?> source = propertySources.get(PROPERTY_SOURCE_NAME);
if (source instanceof MapPropertySource) {
target = (MapPropertySource) source;
for (String key : map.keySet()) {
if (!target.containsProperty(key)) {
target.getSource().put(key, map.get(key));
}
}
}
}
if (target == null) {
target = new MapPropertySource(PROPERTY_SOURCE_NAME, map);
}
if (!propertySources.contains(PROPERTY_SOURCE_NAME)) {
propertySources.addLast(target);
}
}
}

6
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/AckRemoteApplicationEvent.java
View File

@@ -35,14 +35,14 @@ public class AckRemoteApplicationEvent extends RemoteApplicationEvent {
private final String ackId;
private final String ackDestinationService;
private final Class<? extends RemoteApplicationEvent> type;
private final Class<? extends RemoteApplicationEvent> event;
@SuppressWarnings("unused")
private AckRemoteApplicationEvent() {
super();
this.ackDestinationService = null;
this.ackId = null;
this.type = null;
this.event = null;
}
public AckRemoteApplicationEvent(Object source, String originService,
@@ -51,6 +51,6 @@ public class AckRemoteApplicationEvent extends RemoteApplicationEvent {
super(source, originService, destinationService);
this.ackDestinationService = ackDestinationService;
this.ackId = ackId;
this.type = type;
this.event = type;
}
}

2
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/TraceListener.java
View File

@@ -49,7 +49,7 @@ public class TraceListener {
protected Map<String, Object> getReceivedTrace(AckRemoteApplicationEvent event) {
Map<String, Object> map = new LinkedHashMap<String, Object>();
map.put("signal", "spring.cloud.bus.ack");
map.put("type", event.getType().getSimpleName());
map.put("event", event.getEvent().getSimpleName());
map.put("id", event.getAckId());
map.put("origin", event.getOriginService());
map.put("destination", event.getAckDestinationService());

121
spring-cloud-bus/src/main/java/org/springframework/cloud/bus/jackson/BusJacksonAutoConfiguration.java
View File

@@ -1,19 +1,36 @@
package org.springframework.cloud.bus.jackson;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.cloud.bus.BusAutoConfiguration;
import org.springframework.cloud.bus.ConditionalOnBusEnabled;
import org.springframework.cloud.bus.endpoint.RefreshBusEndpoint;
import org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent;
import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import org.springframework.cloud.stream.converter.AbstractFromMessageConverter;
import org.springframework.cloud.stream.converter.MessageConverterUtils;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.Resource;
import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
import org.springframework.core.io.support.ResourcePatternResolver;
import org.springframework.core.type.classreading.CachingMetadataReaderFactory;
import org.springframework.core.type.classreading.MetadataReader;
import org.springframework.core.type.classreading.MetadataReaderFactory;
import org.springframework.messaging.Message;
import org.springframework.util.ClassUtils;
import org.springframework.util.MimeTypeUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
/**
* @author Spencer Gibb
* @author Dave Syer
*/
@Configuration
@ConditionalOnBusEnabled
@@ -22,8 +39,102 @@ import com.fasterxml.jackson.databind.ObjectMapper;
public class BusJacksonAutoConfiguration {
@Bean
public SubtypeModule basicBusSubtypeModule() {
return new SubtypeModule(RefreshRemoteApplicationEvent.class,
EnvironmentChangeRemoteApplicationEvent.class);
public BusJacksonMessageConverter busJsonConverter() {
return new BusJacksonMessageConverter();
}
}
class BusJacksonMessageConverter extends AbstractFromMessageConverter {
private static final String DEFAULT_PACKAGE = ClassUtils
.getPackageName(RemoteApplicationEvent.class);
private static final String CLASS_RESOURCE_PATTERN = "/**/*.class";
private final ObjectMapper mapper = new ObjectMapper();
private ResourcePatternResolver resourcePatternResolver = new PathMatchingResourcePatternResolver();
private String[] packagesToScan = new String[] { DEFAULT_PACKAGE };
public void setPackagesToScan(String[] packagesToScan) {
List<String> packages = new ArrayList<>(Arrays.asList(packagesToScan));
if (!packages.contains(DEFAULT_PACKAGE)) {
packages.add(DEFAULT_PACKAGE);
}
this.packagesToScan = packages.toArray(new String[0]);
}
public BusJacksonMessageConverter() {
super(MimeTypeUtils.APPLICATION_JSON, MessageConverterUtils.X_JAVA_OBJECT);
mapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
mapper.registerModule(new SubtypeModule(findSubTypes()));
}
private Class<?>[] findSubTypes() {
List<Class<?>> types = new ArrayList<>();
if (this.packagesToScan != null) {
for (String pkg : this.packagesToScan) {
try {
String pattern = ResourcePatternResolver.CLASSPATH_ALL_URL_PREFIX
+ ClassUtils.convertClassNameToResourcePath(pkg)
+ CLASS_RESOURCE_PATTERN;
Resource[] resources = this.resourcePatternResolver
.getResources(pattern);
MetadataReaderFactory readerFactory = new CachingMetadataReaderFactory(
this.resourcePatternResolver);
for (Resource resource : resources) {
if (resource.isReadable()) {
MetadataReader reader = readerFactory
.getMetadataReader(resource);
String className = reader.getClassMetadata().getClassName();
try {
Class<?> type = ClassUtils.forName(className, null);
types.add(type);
}
catch (Exception e) {
}
}
}
}
catch (IOException ex) {
throw new IllegalStateException(
"Failed to scan classpath for remote event classes", ex);
}
}
}
return types.toArray(new Class<?>[0]);
}
@Override
protected Class<?>[] supportedPayloadTypes() {
return new Class<?>[] { String.class, byte[].class };
}
@Override
protected Class<?>[] supportedTargetTypes() {
return new Class[] { RemoteApplicationEvent.class }; // any type
}
@Override
public Object convertFromInternal(Message<?> message, Class<?> targetClass,
Object conversionHint) {
Object result = null;
try {
Object payload = message.getPayload();
if (payload instanceof byte[]) {
result = mapper.readValue((byte[]) payload, targetClass);
}
else if (payload instanceof String) {
result = mapper.readValue((String) payload, targetClass);
}
}
catch (Exception e) {
logger.error(e.getMessage(), e);
return null;
}
return result;
}
}

4
spring-cloud-bus/src/main/resources/META-INF/spring.factories
View File

@@ -1,3 +1,7 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.bus.BusAutoConfiguration,\
org.springframework.cloud.bus.jackson.BusJacksonAutoConfiguration
# Environment Post Processor
org.springframework.boot.env.EnvironmentPostProcessor=\
org.springframework.cloud.bus.BusEnvironmentPostProcessor

24
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationTests.java
View File

@@ -3,6 +3,7 @@ package org.springframework.cloud.bus;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@@ -92,9 +93,9 @@ public class BusAutoConfigurationTests {
OutboundMessageHandlerConfiguration outbound = this.context
.getBean(OutboundMessageHandlerConfiguration.class);
outbound.latch.await(2000L, TimeUnit.MILLISECONDS);
AckRemoteApplicationEvent message = (AckRemoteApplicationEvent) outbound.message
.getPayload();
assertEquals(refresh.getId(), message.getAckId());
String message = (String) outbound.message.getPayload();
assertTrue("Wrong ackId: " + message,
message.contains("\"ackId\":\"" + refresh.getId()));
}
@Test
@@ -189,7 +190,8 @@ public class BusAutoConfigurationTests {
}
@Configuration
@Import({ MessageConsumer.class, BusAutoConfiguration.class, TestSupportBinderAutoConfiguration.class,
@Import({ MessageConsumer.class, BusAutoConfiguration.class,
TestSupportBinderAutoConfiguration.class,
PropertyPlaceholderAutoConfiguration.class })
protected static class OutboundMessageHandlerConfiguration {
@@ -218,18 +220,20 @@ public class BusAutoConfigurationTests {
}
}
@Configuration
@MessageEndpoint
protected static class MessageConsumer {
@ServiceActivator(inputChannel=SpringCloudBusClient.OUTPUT)
public void handle(Message<?> msg) {}
@ServiceActivator(inputChannel = SpringCloudBusClient.OUTPUT)
public void handle(Message<?> msg) {
}
}
@Configuration
@Import({ MessageConsumer.class, BusAutoConfiguration.class, TestSupportBinderAutoConfiguration.class,
@Import({ MessageConsumer.class, BusAutoConfiguration.class,
TestSupportBinderAutoConfiguration.class,
PropertyPlaceholderAutoConfiguration.class })
protected static class InboundMessageHandlerConfiguration {

9
spring-cloud-bus/src/test/java/org/springframework/cloud/bus/jackson/SerializationTests.java
View File

@@ -24,6 +24,7 @@ import java.util.Collections;
import org.junit.Test;
import org.springframework.cloud.bus.event.EnvironmentChangeRemoteApplicationEvent;
import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent;
import org.springframework.cloud.bus.event.RemoteApplicationEvent;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -38,8 +39,8 @@ public class SerializationTests {
@Test
public void vanillaDeserialize() throws Exception {
this.mapper.registerModule(
new BusJacksonAutoConfiguration().basicBusSubtypeModule());
this.mapper.registerModule(new SubtypeModule(RefreshRemoteApplicationEvent.class,
EnvironmentChangeRemoteApplicationEvent.class));
EnvironmentChangeRemoteApplicationEvent source = new EnvironmentChangeRemoteApplicationEvent(
this, "foo", "bar", Collections.<String, String>emptyMap());
String value = this.mapper.writeValueAsString(source);
@@ -52,8 +53,8 @@ public class SerializationTests {
@Test
public void deserializeOldValueWithNoId() throws Exception {
this.mapper.registerModule(
new BusJacksonAutoConfiguration().basicBusSubtypeModule());
this.mapper.registerModule(new SubtypeModule(RefreshRemoteApplicationEvent.class,
EnvironmentChangeRemoteApplicationEvent.class));
EnvironmentChangeRemoteApplicationEvent source = new EnvironmentChangeRemoteApplicationEvent(
this, "foo", "bar", Collections.<String, String>emptyMap());
String value = this.mapper.writeValueAsString(source);