org.springframework.boot.context.properties.ConfigurationPropertiesBindingPostProcessor#org.springframework.messaging.converter.MessageConverter源码实例Demo

@Test
public void rqueueMessageSenderWithMessageConverters() throws IllegalAccessException {
  SimpleRqueueListenerContainerFactory factory = new SimpleRqueueListenerContainerFactory();
  MessageConverter messageConverter = new GenericMessageConverter();
  RqueueListenerConfig messageConfig = new RqueueListenerConfig();
  factory.setMessageConverters(Collections.singletonList(messageConverter));
  FieldUtils.writeField(messageConfig, "simpleRqueueListenerContainerFactory", factory, true);
  factory.setRedisConnectionFactory(mock(RedisConnectionFactory.class));
  assertNotNull(messageConfig.rqueueMessageSender(rqueueMessageTemplate));
  RqueueMessageSender messageSender = messageConfig.rqueueMessageSender(rqueueMessageTemplate);
  boolean messageConverterIsConfigured = false;
  for (MessageConverter converter : messageSender.getMessageConverters()) {
    messageConverterIsConfigured =
        messageConverterIsConfigured || converter.hashCode() == messageConverter.hashCode();
  }
  assertTrue(messageConverterIsConfigured);
}
 

/**
 * Create an instance of SimpAnnotationMethodMessageHandler with the given
 * message channels and broker messaging template.
 * @param clientInboundChannel the channel for receiving messages from clients (e.g. WebSocket clients)
 * @param clientOutboundChannel the channel for messages to clients (e.g. WebSocket clients)
 * @param brokerTemplate a messaging template to send application messages to the broker
 */
public SimpAnnotationMethodMessageHandler(SubscribableChannel clientInboundChannel,
		MessageChannel clientOutboundChannel, SimpMessageSendingOperations brokerTemplate) {

	Assert.notNull(clientInboundChannel, "clientInboundChannel must not be null");
	Assert.notNull(clientOutboundChannel, "clientOutboundChannel must not be null");
	Assert.notNull(brokerTemplate, "brokerTemplate must not be null");

	this.clientInboundChannel = clientInboundChannel;
	this.clientMessagingTemplate = new SimpMessagingTemplate(clientOutboundChannel);
	this.brokerTemplate = brokerTemplate;

	Collection<MessageConverter> converters = new ArrayList<MessageConverter>();
	converters.add(new StringMessageConverter());
	converters.add(new ByteArrayMessageConverter());
	this.messageConverter = new CompositeMessageConverter(converters);
}
 

public RocketMQMessageConverter() {
    List<MessageConverter> messageConverters = new ArrayList<>();
    ByteArrayMessageConverter byteArrayMessageConverter = new ByteArrayMessageConverter();
    byteArrayMessageConverter.setContentTypeResolver(null);
    messageConverters.add(byteArrayMessageConverter);
    messageConverters.add(new StringMessageConverter());
    if (JACKSON_PRESENT) {
        messageConverters.add(new MappingJackson2MessageConverter());
    }
    if (FASTJSON_PRESENT) {
        try {
            messageConverters.add(
                (MessageConverter)ClassUtils.forName(
                    "com.alibaba.fastjson.support.spring.messaging.MappingFastJsonMessageConverter",
                    ClassUtils.getDefaultClassLoader()).newInstance());
        } catch (ClassNotFoundException | IllegalAccessException | InstantiationException ignored) {
            //ignore this exception
        }
    }
    messageConverter = new CompositeMessageConverter(messageConverters);
}
 

public void testConfigureOutputChannelWithBadContentType() {
	BindingServiceProperties props = new BindingServiceProperties();
	BindingProperties bindingProps = new BindingProperties();
	bindingProps.setContentType("application/json");
	props.setBindings(Collections.singletonMap("foo", bindingProps));
	CompositeMessageConverterFactory converterFactory = new CompositeMessageConverterFactory(
			Collections.<MessageConverter>emptyList(), null);
	MessageConverterConfigurer configurer = new MessageConverterConfigurer(props,
			converterFactory.getMessageConverterForAllRegistered());
	QueueChannel out = new QueueChannel();
	configurer.configureOutputChannel(out, "foo");
	out.send(new GenericMessage<Foo>(new Foo(), Collections
			.<String, Object>singletonMap(MessageHeaders.CONTENT_TYPE, "bad/ct")));
	Message<?> received = out.receive(0);
	assertThat(received).isNotNull();
	assertThat(received.getPayload()).isInstanceOf(Foo.class);
}
 

private void sendHistoryToNewSubscriber(AbstractSubProtocolEvent ev) {
    Message<byte[]> msg = ev.getMessage();
    StompHeaderAccessor ha = StompHeaderAccessor.wrap(msg);
    String pattern = ha.getDestination();
    if(!pattern.startsWith(PREFIX)) {
        // we must send only to appropriate paths
        return;
    }
    MessageConverter messageConverter = this.simpMessagingTemplate.getMessageConverter();

    for(BusData data: buses.values()) {
        String dest = getDestination(data.getId());
        if(!this.pathMatcher.match(pattern, dest)) {
            continue;
        }
        for(Object obj: data.getEvents()) {
            StompHeaderAccessor mha = Stomp.createHeaders(ha.getSessionId(), ha.getSubscriptionId());
            mha.setDestination(dest);
            Message<?> message = messageConverter.toMessage(obj, mha.getMessageHeaders());
            clientChannel.send(message);
        }
    }
}
 

@Test
public void configureMessageConvertersCustom() {
	final MessageConverter testConverter = mock(MessageConverter.class);
	AbstractMessageBrokerConfiguration config = new BaseTestMessageBrokerConfig() {
		@Override
		protected boolean configureMessageConverters(List<MessageConverter> messageConverters) {
			messageConverters.add(testConverter);
			return false;
		}
	};

	CompositeMessageConverter compositeConverter = config.brokerMessageConverter();
	assertThat(compositeConverter.getConverters().size(), Matchers.is(1));
	Iterator<MessageConverter> iterator = compositeConverter.getConverters().iterator();
	assertThat(iterator.next(), Matchers.is(testConverter));
}
 

protected void initMessageConverter(MessageConverter messageConverter) {

		StringMessageConverter stringMessageConverter = new StringMessageConverter();
		stringMessageConverter.setSerializedPayloadClass(String.class);

		List<MessageConverter> messageConverters = new ArrayList<>();
		messageConverters.add(stringMessageConverter);

		if (messageConverter != null) {
			messageConverters.add(messageConverter);
		}
		else {
			MappingJackson2MessageConverter mappingJackson2MessageConverter = new MappingJackson2MessageConverter();
			mappingJackson2MessageConverter.setSerializedPayloadClass(String.class);
			messageConverters.add(mappingJackson2MessageConverter);
		}

		setMessageConverter(new CompositeMessageConverter(messageConverters));
	}
 

/**
 * Create an instance of SimpAnnotationMethodMessageHandler with the given
 * message channels and broker messaging template.
 * @param clientInboundChannel the channel for receiving messages from clients (e.g. WebSocket clients)
 * @param clientOutboundChannel the channel for messages to clients (e.g. WebSocket clients)
 * @param brokerTemplate a messaging template to send application messages to the broker
 */
public SimpAnnotationMethodMessageHandler(SubscribableChannel clientInboundChannel,
		MessageChannel clientOutboundChannel, SimpMessageSendingOperations brokerTemplate) {

	Assert.notNull(clientInboundChannel, "clientInboundChannel must not be null");
	Assert.notNull(clientOutboundChannel, "clientOutboundChannel must not be null");
	Assert.notNull(brokerTemplate, "brokerTemplate must not be null");

	this.clientInboundChannel = clientInboundChannel;
	this.clientMessagingTemplate = new SimpMessagingTemplate(clientOutboundChannel);
	this.brokerTemplate = brokerTemplate;

	Collection<MessageConverter> converters = new ArrayList<>();
	converters.add(new StringMessageConverter());
	converters.add(new ByteArrayMessageConverter());
	this.messageConverter = new CompositeMessageConverter(converters);
}
 

@Test
public void configureMessageConvertersCustomAndDefault() {
	final MessageConverter testConverter = mock(MessageConverter.class);

	AbstractMessageBrokerConfiguration config = new BaseTestMessageBrokerConfig() {
		@Override
		protected boolean configureMessageConverters(List<MessageConverter> messageConverters) {
			messageConverters.add(testConverter);
			return true;
		}
	};
	CompositeMessageConverter compositeConverter = config.brokerMessageConverter();

	assertThat(compositeConverter.getConverters().size(), Matchers.is(4));
	Iterator<MessageConverter> iterator = compositeConverter.getConverters().iterator();
	assertThat(iterator.next(), Matchers.is(testConverter));
	assertThat(iterator.next(), Matchers.instanceOf(StringMessageConverter.class));
	assertThat(iterator.next(), Matchers.instanceOf(ByteArrayMessageConverter.class));
	assertThat(iterator.next(), Matchers.instanceOf(MappingJackson2MessageConverter.class));
}
 

@Test
public void configureMessageConvertersCustom() {
	final MessageConverter testConverter = mock(MessageConverter.class);
	AbstractMessageBrokerConfiguration config = new BaseTestMessageBrokerConfig() {
		@Override
		protected boolean configureMessageConverters(List<MessageConverter> messageConverters) {
			messageConverters.add(testConverter);
			return false;
		}
	};

	CompositeMessageConverter compositeConverter = config.brokerMessageConverter();
	assertThat(compositeConverter.getConverters().size(), Matchers.is(1));
	Iterator<MessageConverter> iterator = compositeConverter.getConverters().iterator();
	assertThat(iterator.next(), Matchers.is(testConverter));
}
 

/**
 * Configure a {@link MessageConverter} to use to convert the payload of a message from
 * its serialized form with a specific MIME type to an Object matching the target method
 * parameter. The converter is also used when sending a message to the message broker.
 * @see CompositeMessageConverter
 */
public void setMessageConverter(MessageConverter converter) {
	this.messageConverter = converter;
	if (converter != null) {
		((AbstractMessageSendingTemplate<?>) this.clientMessagingTemplate).setMessageConverter(converter);
	}
}
 

public static Object convertMessageToObject(
    Message<String> message, List<MessageConverter> messageConverters) {
  notEmpty(messageConverters, "messageConverters cannot be empty");
  for (MessageConverter messageConverter : messageConverters) {
    try {
      return messageConverter.fromMessage(message, null);
    } catch (Exception e) {
    }
  }
  return null;
}
 

/**
 * Send message to queue of current session
 * @param subscriptionId
 * @param dest
 * @param msg
 */
public void sendToSubscription(String subscriptionId, String dest, Object msg) {
    Assert.notNull(subscriptionId, "subscriptionId is null");
    StompHeaderAccessor sha = createHeaders(sessionId, subscriptionId);
    MessageConverter messageConverter = this.template.getMessageConverter();
    sha.setDestination("/queue/" + dest);
    Message<?> message = messageConverter.toMessage(msg, sha.getMessageHeaders());
    clientChannel.send(message);
}
 

@Override
public boolean configureMessageConverters(List<MessageConverter> messageConverters) {
    messageConverters.add(new StringMessageConverter());
    messageConverters.add(new ByteArrayMessageConverter());

    GsonMessageConverter gsonMessageConverter = new GsonMessageConverter();
    gsonMessageConverter.setGson(Converters.registerDateTime(new GsonBuilder()).create());
    messageConverters.add(gsonMessageConverter);

    return false;
}
 

public WebSocketStompSession(WebSocketSession webSocketSession, MessageConverter messageConverter) {
	Assert.notNull(webSocketSession);
	Assert.notNull(messageConverter);
	this.id = webSocketSession.getId();
	this.webSocketSession = webSocketSession;
	this.messageConverter = messageConverter;
}
 

@Bean
@StreamMessageConverter
public MessageConverter bookingOrderMessageConverter() throws IOException {
  MessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(
      MimeType.valueOf("application/*+avro"));
  ((AvroSchemaMessageConverter) avroSchemaMessageConverter)
      .setSchemaLocation(new ClassPathResource("avro/billingOrder.avsc"));
  return avroSchemaMessageConverter;
}
 

@Bean
@StreamMessageConverter
public MessageConverter bookingOrderMessageConverter() throws IOException {
  LOG.info("avro message converter bean initialized.");
  MessageConverter avroSchemaMessageConverter = new AvroSchemaMessageConverter(
      MimeType.valueOf("application/*+avro"));
  ((AvroSchemaMessageConverter) avroSchemaMessageConverter)
      .setSchemaLocation(new ClassPathResource("avro/bookingOrder.avsc"));
  return avroSchemaMessageConverter;
}
 

@Override
public boolean configureMessageConverters(List<MessageConverter> converters) {
    MappingJackson2MessageConverter jacksonConverter = new MappingJackson2MessageConverter();
    jacksonConverter.setObjectMapper(objectMapper);
    converters.add(jacksonConverter);

    return true;
}
 

void addRemoteRegistryDto(Message<?> message, MessageConverter converter, long expirationPeriod) {
	UserRegistrySnapshot registry = (UserRegistrySnapshot) converter.fromMessage(message, UserRegistrySnapshot.class);
	if (registry != null && !registry.getId().equals(this.id)) {
		registry.init(expirationPeriod, this.sessionLookup);
		this.remoteRegistries.put(registry.getId(), registry);
	}
}
 

@Bean
public CompositeMessageConverter brokerMessageConverter() {
	List<MessageConverter> converters = new ArrayList<>();
	boolean registerDefaults = configureMessageConverters(converters);
	if (registerDefaults) {
		converters.add(new StringMessageConverter());
		converters.add(new ByteArrayMessageConverter());
		if (jackson2Present) {
			converters.add(createJacksonConverter());
		}
	}
	return new CompositeMessageConverter(converters);
}
 

private void convertAndSetMessage(Object o, Class<?> valueClass,
		MessageConverter messageConverter, Message<?> msg) {
	Object result = valueClass.isAssignableFrom(msg.getPayload().getClass())
			? msg.getPayload() : messageConverter.fromMessage(msg, valueClass);

	Assert.notNull(result, "Failed to convert message " + msg);

	keyValueThreadLocal.set(new KeyValue<>(o, result));
}
 

@BeforeEach
public void before() {
	List<MessageConverter> messageConverters = new ArrayList<>();
	JsonMapper jsonMapper = new GsonMapper(new Gson());
	messageConverters.add(NegotiatingMessageConverterWrapper.wrap(new JsonMessageConverter(jsonMapper)));
	messageConverters.add(NegotiatingMessageConverterWrapper.wrap(new ByteArrayMessageConverter()));
	messageConverters.add(NegotiatingMessageConverterWrapper.wrap(new StringMessageConverter()));
	this.messageConverter = new CompositeMessageConverter(messageConverters);

	this.conversionService = new DefaultConversionService();
}
 

/**
 * Convert from the given message to the given target class.
 * @param message the message to convert
 * @param targetClass the target class to convert the payload to
 * @return the converted payload of the reply message (never {@code null})
 */
@SuppressWarnings("unchecked")
@Nullable
protected <T> T doConvert(Message<?> message, Class<T> targetClass) {
	MessageConverter messageConverter = getMessageConverter();
	T value = (T) messageConverter.fromMessage(message, targetClass);
	if (value == null) {
		throw new MessageConversionException(message, "Unable to convert payload [" + message.getPayload() +
				"] to type [" + targetClass + "] using converter [" + messageConverter + "]");
	}
	return value;
}
 

@Test
@SuppressWarnings("unchecked")
@Ignore
public void testCompositeNonNativeSerdeUsingAvroContentType() {
	Random random = new Random();
	Sensor sensor = new Sensor();
	sensor.setId(UUID.randomUUID().toString() + "-v1");
	sensor.setAcceleration(random.nextFloat() * 10);
	sensor.setVelocity(random.nextFloat() * 100);
	sensor.setTemperature(random.nextFloat() * 50);

	List<MessageConverter> messageConverters = new ArrayList<>();
	messageConverters.add(new AvroSchemaMessageConverter(new AvroSchemaServiceManagerImpl()));
	CompositeMessageConverterFactory compositeMessageConverterFactory = new CompositeMessageConverterFactory(
			messageConverters, new ObjectMapper());
	MessageConverterDelegateSerde messageConverterDelegateSerde = new MessageConverterDelegateSerde(
			compositeMessageConverterFactory.getMessageConverterForAllRegistered());

	Map<String, Object> configs = new HashMap<>();
	configs.put("valueClass", Sensor.class);
	configs.put("contentType", "application/avro");
	messageConverterDelegateSerde.configure(configs, false);
	final byte[] serialized = messageConverterDelegateSerde.serializer().serialize(null,
			sensor);

	final Object deserialized = messageConverterDelegateSerde.deserializer()
			.deserialize(null, serialized);

	assertThat(deserialized).isEqualTo(sensor);
}
 

@Before
public void setup() throws Exception {
	this.method = MessageMethodArgumentResolverTests.class.getDeclaredMethod("handle",
			Message.class, Message.class, Message.class, Message.class, ErrorMessage.class, Message.class);

	this.converter = mock(MessageConverter.class);
	this.resolver = new MessageMethodArgumentResolver(this.converter);
}
 

@Test
public void customMessageConverterFailure() throws Exception {
	DefaultMessageHandlerMethodFactory instance = createInstance();
	MessageConverter messageConverter = new ByteArrayMessageConverter();
	instance.setMessageConverter(messageConverter);
	instance.afterPropertiesSet();

	InvocableHandlerMethod invocableHandlerMethod =
			createInvocableHandlerMethod(instance, "simpleString", String.class);

	thrown.expect(MessageConversionException.class);
	invocableHandlerMethod.invoke(MessageBuilder.withPayload(123).build());
}
 

@Test
@Ignore
public void testConfigureOutputChannelCannotConvert() {
	BindingServiceProperties props = new BindingServiceProperties();
	BindingProperties bindingProps = new BindingProperties();
	bindingProps.setContentType("foo/bar");
	props.setBindings(Collections.singletonMap("foo", bindingProps));
	MessageConverter converter = new AbstractMessageConverter(
			new MimeType("foo", "bar")) {

		@Override
		protected boolean supports(Class<?> clazz) {
			return true;
		}

		@Override
		protected Object convertToInternal(Object payload, MessageHeaders headers,
				Object conversionHint) {
			return null;
		}

	};
	CompositeMessageConverterFactory converterFactory = new CompositeMessageConverterFactory(
			Collections.<MessageConverter>singletonList(converter), null);
	MessageConverterConfigurer configurer = new MessageConverterConfigurer(props,
			converterFactory.getMessageConverterForAllRegistered());
	QueueChannel out = new QueueChannel();
	configurer.configureOutputChannel(out, "foo");
	try {
		out.send(new GenericMessage<Foo>(new Foo(),
				Collections.<String, Object>singletonMap(MessageHeaders.CONTENT_TYPE,
						"bad/ct")));
		fail("Expected MessageConversionException: " + out.receive(0));
	}
	catch (MessageConversionException e) {
		assertThat(e.getMessage())
				.endsWith("to the configured output type: 'foo/bar'");
	}
}
 

void addRemoteRegistryDto(Message<?> message, MessageConverter converter, long expirationPeriod) {
	UserRegistrySnapshot registry = (UserRegistrySnapshot) converter.fromMessage(message, UserRegistrySnapshot.class);
	if (registry != null && !registry.getId().equals(this.id)) {
		registry.init(expirationPeriod, this.sessionLookup);
		this.remoteRegistries.put(registry.getId(), registry);
	}
}
 

private CompositeMessageConverter createPayloadArgumentCompositeConverter() {
	List<MessageConverter> payloadArgumentConverters = new ArrayList<>(
			this.messageConverters);

	ObjectMessageConverter objectMessageConverter = new ObjectMessageConverter();
	objectMessageConverter.setStrictContentTypeMatch(true);
	payloadArgumentConverters.add(objectMessageConverter);

	payloadArgumentConverters.add(new SimpleMessageConverter());

	return new CompositeMessageConverter(payloadArgumentConverters);
}
 

@Test
public void configureMessageConvertersDefault() {
	AbstractMessageBrokerConfiguration config = new BaseTestMessageBrokerConfig();
	CompositeMessageConverter compositeConverter = config.brokerMessageConverter();

	List<MessageConverter> converters = compositeConverter.getConverters();
	assertThat(converters.size(), Matchers.is(3));
	assertThat(converters.get(0), Matchers.instanceOf(StringMessageConverter.class));
	assertThat(converters.get(1), Matchers.instanceOf(ByteArrayMessageConverter.class));
	assertThat(converters.get(2), Matchers.instanceOf(MappingJackson2MessageConverter.class));

	ContentTypeResolver resolver = ((MappingJackson2MessageConverter) converters.get(2)).getContentTypeResolver();
	assertEquals(MimeTypeUtils.APPLICATION_JSON, ((DefaultContentTypeResolver) resolver).getDefaultMimeType());
}