Spring Integration XMPP 支持教程

引言

XMPP（可扩展消息与存在协议）是一种用于即时通讯和在线状态信息的开放协议，广泛应用于聊天应用（如Google Talk、Facebook Chat等）。Spring Integration提供了强大的XMPP支持，使开发者能够轻松集成XMPP功能到企业应用中。

🚀 添加依赖

首先在项目中添加Spring Integration XMPP依赖：

Gradle

// build.gradle.kts
dependencies {
    implementation("org.springframework.integration:spring-integration-xmpp:6.5.1")
}

Maven

<!-- pom.xml -->
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-xmpp</artifactId>
    <version>6.5.1</version>
</dependency>

🔌 XMPP 连接配置

配置XMPP连接是使用所有适配器的前提：

@Configuration
class XmppConfig {

    @Bean
    fun xmppConnection(): XmppConnectionFactoryBean {
        return XmppConnectionFactoryBean().apply {
            user = "your_username"
            password = "your_password"
            host = "xmpp.example.com"
            port = 5222
            resource = "SpringIntegrationApp"
            subscriptionMode = SubscriptionMode.accept_all
        }
    }
}

连接管理技巧

• 自动重连：当连接断开时，Spring会自动尝试重新连接
• 连接监听：内置ConnectionListener记录连接事件（DEBUG级别）
• 订阅模式：使用subscriptionMode控制其他用户订阅处理策略

📨 消息适配器

入站消息适配器（接收消息）

@Configuration
class InboundConfig {

    @Bean
    fun inboundAdapter(connection: XmppConnection): MessageProducerSupport {
        return object : MessageProducerSupport() {
            init {

                val listener = ChatMessageListeningEndpoint(connection).apply {
                    setOutputChannelName("xmppInboundChannel")
                    payloadExpression = SpelExpressionParser().parseExpression(
                        "#extension?.json ?: payload"
                    )
                }
                listener.start()
            }
        }
    }

    @Bean
    fun xmppInboundChannel(): MessageChannel {
        return DirectChannel()
    }
}

消息处理流程时序图

IMPORTANT

payloadExpression 允许从XMPP扩展中提取数据：

• #extension?.json：提取GCM/FCM扩展的JSON
• #extension?.bodies[0]：提取XHTML扩展内容

出站消息适配器（发送消息）

@Configuration
class OutboundConfig {

    @Bean
    fun outboundAdapter(connection: XmppConnection): MessageHandler {
        return ChatMessageSendingMessageHandler(connection).apply {
            setExtensionProvider(GcmExtensionProvider())
        }
    }

    @Bean
    @ServiceActivator(inputChannel = "outboundChannel")
    fun serviceActivator(): MessageHandler {
        return outboundAdapter(xmppConnection())
    }
}

// 使用示例
fun sendMessage() {
    val message = MessageBuilder.withPayload("""{"message":"Hello XMPP!"}""")
        .setHeader(XmppHeaders.CHAT_TO, "recipient@example.com")
        .build()

    outboundChannel.send(message)
}

注意

发送消息时必须设置XmppHeaders.CHAT_TO头部，指定消息接收者

👤 在线状态(Presence)适配器

入站状态适配器（接收状态更新）

@Bean
fun presenceInboundAdapter(connection: XmppConnection): MessageProducerSupport {
    return object : MessageProducerSupport() {
        init {
            val endpoint = PresenceListeningEndpoint(connection).apply {
                setOutputChannelName("presenceChannel")
            }
            endpoint.start()
        }
    }
}

出站状态适配器（广播状态更新）

@Bean
@ServiceActivator(inputChannel = "presenceOutChannel")
fun presenceOutboundAdapter(connection: XmppConnection): MessageHandler {
    return PresenceSendingMessageHandler(connection)
}

// 更新状态示例
fun updateStatus() {
    val presence = Presence(Presence.Type.available).apply {
        status = "Working remotely"
    }

    val message = MessageBuilder.withPayload(presence).build()
    presenceOutChannel.send(message)
}

⚙️ 高级配置

SASL认证配置

@Configuration
class AdvancedXmppConfig {

    @Bean
    fun xmppConnection(): XMPPConnection {

        SASLAuthentication.supportSASLMechanism("EXTERNAL", 0)

        val config = ConnectionConfiguration("localhost", 5223).apply {
            keystorePath = "path/to/truststore.jks"
            isSecurityEnabled = true
            socketFactory = SSLSocketFactory.getDefault()
        }

        return XMPPConnection(config)
    }
}

自定义头部映射

@Bean
fun headerMapper(): DefaultXmppHeaderMapper {
    return DefaultXmppHeaderMapper().apply {
        // 映射所有标准头部+自定义头部
        setRequestHeaderNames("STANDARD_REQUEST_HEADERS", "custom_*", "!excluded_header")
    }
}

头部映射技巧

使用!前缀排除特定头部，例如："STANDARD_REQUEST_HEADERS,!excluded_header"

🌐 XMPP扩展协议(XEP)支持

XMPP通过扩展协议提供额外功能：

// 注册自定义扩展提供程序
ProviderManager.addIQProvider("query", "custom:namespace", CustomIQProvider())
ProviderManager.addExtensionProvider("ext", "custom:namespace", CustomExtProvider())

// 使用GCM扩展
fun sendGcmMessage() {
    val gcmJson = """{"to":"device_token","data":{"message":"Hello GCM"}}"""

    val message = MessageBuilder.withPayload(gcmJson)
        .setHeader(XmppHeaders.CHAT_TO, "gcm@example.com")
        .build()

    outboundChannel.send(message)
}

标准XMPP消息

val textMessage = MessageBuilder.withPayload("Hello!")
    .setHeader(XmppHeaders.CHAT_TO, "user@example.com")
    .build()

GCM扩展消息

val gcmMessage = MessageBuilder.withPayload("""{"to":"device_id"}""")
    .setHeader(XmppHeaders.CHAT_TO, "gcm@example.com")
    .build()

💡 最佳实践与常见问题

推荐实践

1. 连接复用：所有适配器共享同一XMPP连接
2. 异常处理：添加错误通道处理消息发送失败
3. 性能优化：对高频消息使用异步通道
4. 安全配置：始终启用TLS加密

常见问题解决

Q: 消息发送失败，如何诊断？
A: 启用DEBUG日志查看详细通信过程：

logging.level.org.springframework.integration.xmpp=DEBUG
logging.level.org.jivesoftware.smack=DEBUG

Q: 如何处理Google Cloud Messaging(GCM)扩展？
A: 使用payload表达式提取JSON：

payloadExpression = SpelExpressionParser().parseExpression(
    "#extension?.json ?: payload"
)

Q: 如何过滤特定类型的消息？
A: 注入自定义StanzaFilter：

@Bean
fun stanzaFilter(): StanzaFilter {
    return StanzaTypeFilter(Message::class.java)
}

@Bean
fun inboundAdapter(connection: XmppConnection, filter: StanzaFilter) {
    return ChatMessageListeningEndpoint(connection).apply {
        stanzaFilter = filter
        // ...
    }
}

总结

Spring Integration的XMPP支持提供了强大而灵活的企业集成能力。通过本教程，您已学习到：

1. XMPP连接配置与管理 ✅
2. 消息收发适配器的使用技巧 📩
3. 在线状态监控与广播 👥
4. 高级配置与扩展协议支持 ⚙️
5. 常见问题解决方案 🔧

在实际应用中，结合Spring Integration的消息路由和转换能力，可以构建出功能丰富的实时通信系统。