@Test public void testDoConvertMessage() throws Exception { DefaultRocketMQListenerContainer listenerContainer = new DefaultRocketMQListenerContainer(); Method doConvertMessage = DefaultRocketMQListenerContainer.class.getDeclaredMethod("doConvertMessage", MessageExt.class); doConvertMessage.setAccessible(true); listenerContainer.setRocketMQListener(new RocketMQListener<String>() { @Override public void onMessage(String message) { } }); Field messageType = DefaultRocketMQListenerContainer.class.getDeclaredField("messageType"); messageType.setAccessible(true); messageType.set(listenerContainer, String.class); MessageExt messageExt = new MessageExt(0, System.currentTimeMillis(), null, System.currentTimeMillis(), null, null); messageExt.setBody("hello".getBytes()); String result = (String) doConvertMessage.invoke(listenerContainer, messageExt); assertThat(result).isEqualTo("hello"); listenerContainer.setRocketMQListener(new RocketMQListener<MessageExt>() { @Override public void onMessage(MessageExt message) { } }); Field messageType2 = DefaultRocketMQListenerContainer.class.getDeclaredField("messageType"); messageType2.setAccessible(true); messageType2.set(listenerContainer, MessageExt.class); messageExt = new MessageExt(0, System.currentTimeMillis(), null, System.currentTimeMillis(), null, null); messageExt.setBody("hello".getBytes()); MessageExt result2 = (MessageExt) doConvertMessage.invoke(listenerContainer, messageExt); assertThat(result2).isEqualTo(messageExt); listenerContainer.setRocketMQListener(new RocketMQListener<User>() { @Override public void onMessage(User message) { } }); }
@SuppressWarnings("unchecked") private Object doConvertMessage(MessageExt messageExt) { if (Objects.equals(messageType, MessageExt.class)) { return messageExt; } else { String str = new String(messageExt.getBody(), Charset.forName(charset)); if (Objects.equals(messageType, String.class)) { return str; } else { try { if (messageType instanceof Class) { return this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) messageType); } else { return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) messageType).getRawType(), methodParameter); } } catch (Exception e) { log.info("convert failed. str:{}, msgType:{}", str, messageType); throw new RuntimeException("cannot convert message to " + messageType, e); } } } }
DefaultRocketMQListenerContainer implements InitializingBean,
    RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware { @SuppressWarnings("unchecked") private Object doConvertMessage(MessageExt messageExt) { if (Objects.equals(messageType, MessageExt.class)) { return messageExt; } else { String str = new String(messageExt.getBody(), Charset.forName(charset)); if (Objects.equals(messageType, String.class)) { return str; } else { try { if (messageType instanceof Class) { return this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) messageType); } else { return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) messageType).getRawType(), methodParameter); } } catch (Exception e) { log.info("convert failed. str:{}, msgType:{}", str, messageType); throw new RuntimeException("cannot convert message to " + messageType, e); } } } } }
DefaultRocketMQListenerContainer implements InitializingBean,
    RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware { @SuppressWarnings("unchecked") private Object doConvertMessage(MessageExt messageExt) { if (Objects.equals(messageType, MessageExt.class)) { return messageExt; } else { String str = new String(messageExt.getBody(), Charset.forName(charset)); if (Objects.equals(messageType, String.class)) { return str; } else { try { if (messageType instanceof Class) { return this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) messageType); } else { return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) messageType).getRawType(), methodParameter); } } catch (Exception e) { log.info("convert failed. str:{}, msgType:{}", str, messageType); throw new RuntimeException("cannot convert message to " + messageType, e); } } } }  }
DefaultRocketMQListenerContainer implements InitializingBean,
    RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware { @SuppressWarnings("unchecked") private Object doConvertMessage(MessageExt messageExt) { if (Objects.equals(messageType, MessageExt.class)) { return messageExt; } else { String str = new String(messageExt.getBody(), Charset.forName(charset)); if (Objects.equals(messageType, String.class)) { return str; } else { try { if (messageType instanceof Class) { return this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) messageType); } else { return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) messageType).getRawType(), methodParameter); } } catch (Exception e) { log.info("convert failed. str:{}, msgType:{}", str, messageType); throw new RuntimeException("cannot convert message to " + messageType, e); } } } }  long getSuspendCurrentQueueTimeMillis(); void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis); int getDelayLevelWhenNextConsume(); void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume); String getNameServer(); void setNameServer(String nameServer); AccessChannel getAccessChannel(); void setAccessChannel(AccessChannel accessChannel); String getConsumerGroup(); void setConsumerGroup(String consumerGroup); String getTopic(); void setTopic(String topic); int getConsumeThreadMax(); String getCharset(); void setCharset(String charset); MessageConverter getMessageConverter(); DefaultRocketMQListenerContainer setMessageConverter(MessageConverter messageConverter); RocketMQListener getRocketMQListener(); void setRocketMQListener(RocketMQListener rocketMQListener); RocketMQReplyListener getRocketMQReplyListener(); void setRocketMQReplyListener(RocketMQReplyListener rocketMQReplyListener); RocketMQMessageListener getRocketMQMessageListener(); void setRocketMQMessageListener(RocketMQMessageListener anno); ConsumeMode getConsumeMode(); SelectorType getSelectorType(); void setSelectorExpression(String selectorExpression); String getSelectorExpression(); MessageModel getMessageModel(); DefaultMQPushConsumer getConsumer(); void setConsumer(DefaultMQPushConsumer consumer); @Override void destroy(); @Override boolean isAutoStartup(); @Override void stop(Runnable callback); @Override void start(); @Override void stop(); @Override boolean isRunning(); @Override int getPhase(); @Override void afterPropertiesSet(); @Override void setApplicationContext(ApplicationContext applicationContext); @Override String toString(); void setName(String name); }
DefaultRocketMQListenerContainer implements InitializingBean,
    RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware { @SuppressWarnings("unchecked") private Object doConvertMessage(MessageExt messageExt) { if (Objects.equals(messageType, MessageExt.class)) { return messageExt; } else { String str = new String(messageExt.getBody(), Charset.forName(charset)); if (Objects.equals(messageType, String.class)) { return str; } else { try { if (messageType instanceof Class) { return this.getMessageConverter().fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) messageType); } else { return ((SmartMessageConverter) this.getMessageConverter()).fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) ((ParameterizedType) messageType).getRawType(), methodParameter); } } catch (Exception e) { log.info("convert failed. str:{}, msgType:{}", str, messageType); throw new RuntimeException("cannot convert message to " + messageType, e); } } } }  long getSuspendCurrentQueueTimeMillis(); void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis); int getDelayLevelWhenNextConsume(); void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume); String getNameServer(); void setNameServer(String nameServer); AccessChannel getAccessChannel(); void setAccessChannel(AccessChannel accessChannel); String getConsumerGroup(); void setConsumerGroup(String consumerGroup); String getTopic(); void setTopic(String topic); int getConsumeThreadMax(); String getCharset(); void setCharset(String charset); MessageConverter getMessageConverter(); DefaultRocketMQListenerContainer setMessageConverter(MessageConverter messageConverter); RocketMQListener getRocketMQListener(); void setRocketMQListener(RocketMQListener rocketMQListener); RocketMQReplyListener getRocketMQReplyListener(); void setRocketMQReplyListener(RocketMQReplyListener rocketMQReplyListener); RocketMQMessageListener getRocketMQMessageListener(); void setRocketMQMessageListener(RocketMQMessageListener anno); ConsumeMode getConsumeMode(); SelectorType getSelectorType(); void setSelectorExpression(String selectorExpression); String getSelectorExpression(); MessageModel getMessageModel(); DefaultMQPushConsumer getConsumer(); void setConsumer(DefaultMQPushConsumer consumer); @Override void destroy(); @Override boolean isAutoStartup(); @Override void stop(Runnable callback); @Override void start(); @Override void stop(); @Override boolean isRunning(); @Override int getPhase(); @Override void afterPropertiesSet(); @Override void setApplicationContext(ApplicationContext applicationContext); @Override String toString(); void setName(String name);  }