使用 XMPP 处理传入消息的最佳方法

Question

是否有变通办法让 Spring 处理来自 XMPP 的传入消息？我尝试了许多不同的配置来让入站通道适配器响应传入的 XMPP 消息，但没有任何反应。我知道它们出现在 Spring Integration 层（我可以在日志中看到）但它们被忽略了。有没有办法让它们进入我的应用层？如果可以的话，我希望避免对 Spring Integration 本身进行更改。

这是我的集成配置：

<int-xmpp:inbound-channel-adapter id="gcmIn" 
                                      channel="gcmInChannel"
                                      xmpp-connection="gcmConnection"
                                      auto-startup="true"
    />
    <bean id="inboundBean" class="example.integration.GcmInputHandler"/>

    <int:service-activator input-channel="gcmInChannel" output-channel="nullChannel" ref="inboundBean" method="handle"/>

使用出站通道适配器可以正常工作。我可以 100% 轻松地通过 GCM 发送消息。但是 inbound 什么也没做，即使我知道消息正在进来。

谢谢

Answer 1

不是很干净，您需要覆盖 ChatMessageListeningEndpoint，它会删除所有空正文消息。 然后需要将其用作配置中的入站通道适配器。

另外需要在 Smack Provider Manager 上注册 GCM 包扩展，否则会丢失 JSON 消息。

正在处理一个示例项目——所以如果您需要更多帮助，请告诉我，我会尽快发布一个链接，它以某种可以理解的方式工作。

这里是一个示例 GCM 输入适配器

    public class GcmMessageListeningEndpoint extends ChatMessageListeningEndpoint {
    private static final Logger LOG = LoggerFactory.getLogger(GcmMessageListeningEndpoint.class);
    @Setter
    protected PacketListener packetListener = new GcmPacketListener();
    protected XmppHeaderMapper headerMapper = new DefaultXmppHeaderMapper();

    public GcmMessageListeningEndpoint(XMPPConnection connection) {
        super(connection);


        ProviderManager.addExtensionProvider(GcmPacketExtension.GCM_ELEMENT_NAME, GcmPacketExtension.GCM_NAMESPACE,
                new PacketExtensionProvider() {
                    @Override
                    public PacketExtension parseExtension(XmlPullParser parser) throws Exception {
                        String json = parser.nextText();
                        return new GcmPacketExtension(json);
                    }
                });
    }

    @Override
    public void setHeaderMapper(XmppHeaderMapper headerMapper) {
        super.setHeaderMapper(headerMapper);
        this.headerMapper = headerMapper;
        if (this.headerMapper == null) throw new IllegalArgumentException("Null XmppHeaderMapper isn't supported!");
    }

    public String getComponentType() {
        return "xmpp:inbound-channel-adapter-gcm";
    }

    @Override
    protected void doStart() {
        Assert.isTrue(this.initialized, this.getComponentName() + " [" + this.getComponentType() + "] must be initialized");
        this.xmppConnection.addPacketListener(this.packetListener, null);
    }

    @Override
    protected void doStop() {
        if (this.xmppConnection != null) {
            this.xmppConnection.removePacketListener(this.packetListener);
        }
    }

    class GcmPacketListener implements PacketListener {

        @Override
        public void processPacket(Packet packet) throws NotConnectedException {
            if (packet instanceof org.jivesoftware.smack.packet.Message) {
                org.jivesoftware.smack.packet.Message xmppMessage = (org.jivesoftware.smack.packet.Message) packet;
                Map<String, ?> mappedHeaders = headerMapper.toHeadersFromRequest(xmppMessage);
                sendMessage(MessageBuilder.withPayload(xmppMessage).copyHeaders(mappedHeaders).build());
            } else {
                LOG.warn("Unsuported Packet {}", packet);
            }
        }
    }
}

这里 inbound-channel-adapter 的新配置删除了 XML 中的配置：

@Bean 
public GcmMessageListeningEndpoint inboundAdpater(XMPPConnection connection, MessageChannel gcmInChannel) {
    GcmMessageListeningEndpoint endpoint = new GcmMessageListeningEndpoint(connection);
    endpoint.setOutputChannel(gcmInChannel);
    return endpoint;
}