From 5a14eda63a0c9f48c08b6e00c9765f72385d3efe Mon Sep 17 00:00:00 2001 From: Dave Syer Date: Fri, 18 Mar 2016 10:35:08 +0000 Subject: [PATCH] Ensure messages flow as JSON This is for readability and also for compatibility with Angel apps. --- docs/src/main/asciidoc/spring-cloud-bus.adoc | 9 +- .../bus/BusEnvironmentPostProcessor.java | 81 ++++++++++++ .../bus/event/AckRemoteApplicationEvent.java | 6 +- .../cloud/bus/event/TraceListener.java | 2 +- .../jackson/BusJacksonAutoConfiguration.java | 121 +++++++++++++++++- .../main/resources/META-INF/spring.factories | 4 + .../cloud/bus/BusAutoConfigurationTests.java | 24 ++-- .../cloud/bus/jackson/SerializationTests.java | 9 +- 8 files changed, 232 insertions(+), 24 deletions(-) create mode 100644 spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java diff --git a/docs/src/main/asciidoc/spring-cloud-bus.adoc b/docs/src/main/asciidoc/spring-cloud-bus.adoc index f43eaef..d35d2b3 100644 --- a/docs/src/main/asciidoc/spring-cloud-bus.adoc +++ b/docs/src/main/asciidoc/spring-cloud-bus.adoc @@ -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. \ No newline at end of file +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. \ No newline at end of file diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java new file mode 100644 index 0000000..64f87ca --- /dev/null +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/BusEnvironmentPostProcessor.java @@ -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 map = new HashMap(); + // 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 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); + } + } + +} diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/AckRemoteApplicationEvent.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/AckRemoteApplicationEvent.java index a13c2ac..d5aa25b 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/AckRemoteApplicationEvent.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/AckRemoteApplicationEvent.java @@ -35,14 +35,14 @@ public class AckRemoteApplicationEvent extends RemoteApplicationEvent { private final String ackId; private final String ackDestinationService; - private final Class type; + private final Class 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; } } diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/TraceListener.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/TraceListener.java index 09e2228..cec0317 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/TraceListener.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/event/TraceListener.java @@ -49,7 +49,7 @@ public class TraceListener { protected Map getReceivedTrace(AckRemoteApplicationEvent event) { Map map = new LinkedHashMap(); 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()); diff --git a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/jackson/BusJacksonAutoConfiguration.java b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/jackson/BusJacksonAutoConfiguration.java index 09ad1a9..3fc990b 100644 --- a/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/jackson/BusJacksonAutoConfiguration.java +++ b/spring-cloud-bus/src/main/java/org/springframework/cloud/bus/jackson/BusJacksonAutoConfiguration.java @@ -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 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> 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; } } diff --git a/spring-cloud-bus/src/main/resources/META-INF/spring.factories b/spring-cloud-bus/src/main/resources/META-INF/spring.factories index 086e276..8645269 100644 --- a/spring-cloud-bus/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-bus/src/main/resources/META-INF/spring.factories @@ -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 \ No newline at end of file diff --git a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationTests.java b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationTests.java index add7508..eea1a9a 100644 --- a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationTests.java +++ b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/BusAutoConfigurationTests.java @@ -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 { diff --git a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/jackson/SerializationTests.java b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/jackson/SerializationTests.java index be2766a..19e532a 100644 --- a/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/jackson/SerializationTests.java +++ b/spring-cloud-bus/src/test/java/org/springframework/cloud/bus/jackson/SerializationTests.java @@ -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.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.emptyMap()); String value = this.mapper.writeValueAsString(source);