我正在尝试配置一组 spring 集成组件来使用来自 TCP 套接字的数据。基本协议是，打开连接后，系统会提示我输入用户名，然后输入密码，然后如果身份验证成功，数据就会在可用时流式传输给我。每 30 秒就会向我发送一条 ping 消息，以便我可以在没有数据传输的安静时段验证连接是否处于活动状态。
根据 spring-integration 文档，我设置了一个 TCP 网关。http://docs.spring.io/spring-integration/reference/html/ip.html#tcp-gateways http://docs.spring.io/spring-integration/reference/html/ip.html#tcp-gateways
<bean id="authInterceptorFactory"
class="org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain">
<property name="interceptors">
<array>
<bean class="com.socketfetching.AuthConnectionInterceptorFactory">
<constructor-arg value="Login Username:"/>
<constructor-arg value="${socket.username}"/>
<constructor-arg value="Password:"/>
<constructor-arg value="${socket.password}"/>
</bean>
</array>
</property>
</bean>
<bean id="lfSeserializer" class="org.springframework.integration.ip.tcp.serializer.ByteArrayLfSerializer"/>
<ip:tcp-connection-factory id="connectionFactory"
type="client"
host="${socket.url}"
port="${socket.port}"
single-use="false"
so-keep-alive="true"
interceptor-factory-chain="authInterceptorFactory"
deserializer="lfSeserializer"
serializer="lfSeserializer"
/>
<int:channel id="socketInitChannel"/>
<ip:tcp-inbound-gateway id="inGateway"
request-channel="clientBytes2StringChannel"
reply-channel="socketInitChannel"
connection-factory="connectionFactory"
reply-timeout="10000"
retry-interval="5000"
auto-startup="true"
client-mode="true"/>
InterceptorFactory 处理打开连接时发生的握手，并将预期的提示和我想要的响应作为参数。这种握手工作完美，我的应用程序正在从服务器接收其定期 ping。
client-mode=true 导致网关在启动时立即打开连接并等待用户名提示。
我的问题是连接丢失后的恢复。如果我终止网络连接，显然 ping 会停止发送，我希望我的网关能够检测到这一点并开始尝试定期重新连接。当我的网络连接恢复时，网关应该成功重新连接。
我认为重试间隔可能会处理这个问题，但它似乎没有任何效果。文档建议我使用 TaskScheduler 来实现此目的...但我不太确定如何将它与来自服务器的 ping 消息集成。
有什么建议吗？
编辑：
我找到了一个可行的解决方案，尽管我不确定它是否理想。我的网关上的重试间隔意味着每 5 秒将测试一次连接是否有效，并在需要时重新创建。它通过在我的 AuthConnectionInterceptor 上调用 isOpen() 来完成此操作。因此，我能够重写此方法来检查当前时间与通过拦截器的最后一条消息之间的增量。如果时间间隔太长，我会手动终止连接并触发重新连接。
这些课程的完整源代码如下......
拦截器工厂：
包 com.socketfetching；
import org.apache.log4j.Logger;
import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptor;
import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactory;
/**
* AuthConnectionInterceptorFactory
* Created by: Seth Kelly
* Date: 10/3/13
*/
public class AuthConnectionInterceptorFactory implements TcpConnectionInterceptorFactory {
private static Logger logger = Logger.getLogger(AuthConnectionInterceptorFactory.class);
private String usernamePrompt;
private String username;
private String passwordPrompt;
private String password;
public AuthConnectionInterceptorFactory(String usernamePrompt, String username, String passwordPrompt, String password) {
this.usernamePrompt = usernamePrompt;
this.username = username;
this.passwordPrompt = passwordPrompt;
this.password = password;
}
@Override
public TcpConnectionInterceptor getInterceptor() {
return new AuthConnectionInterceptor(usernamePrompt, username, passwordPrompt, password);
}
}
拦截器：
package com.socketfetching;
import org.apache.log4j.Logger;
import org.joda.time.DateTime;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.ip.tcp.connection.AbstractTcpConnectionInterceptor;
import org.springframework.integration.support.MessageBuilder;
/**
* AuthConnectionInterceptor
* Created by: Seth Kelly
* Date: 10/3/13
*
* Handles username/password authentication when opening a new TCP connection.
*/
public class AuthConnectionInterceptor extends AbstractTcpConnectionInterceptor {
private static Logger logger = Logger.getLogger(AuthConnectionInterceptor.class);
private String usernamePrompt;
private String username;
private String passwordPrompt;
private String password;
private Boolean usernameSent = false;
private Boolean passwordSent = false;
private static final String PING_PREFIX = "Ping";
private DateTime lastMsgReceived;
private Integer inactivityTimeout = 35000;
public AuthConnectionInterceptor(String usernamePrompt, String username, String passwordPrompt, String password) {
this.usernamePrompt = usernamePrompt;
this.username = username;
this.passwordPrompt = passwordPrompt;
this.password = password;
}
@Override
public boolean onMessage(Message<?> message) {
lastMsgReceived = new DateTime();
Boolean forwardMessage = true;
if(!this.isServer()) {
String payload = new String((byte[])message.getPayload());
if(!usernameSent) {
if(payload.equals(usernamePrompt)) {
try {
logger.debug("Sending username=" + username + "to authenticate socket.");
super.send(MessageBuilder.withPayload(username).build());
usernameSent = true;
forwardMessage = false;
} catch (Exception e) {
throw new MessagingException("Negotiation error", e);
}
}
else {
throw new MessagingException("Negotiation error. expected message=" + usernamePrompt +
" actual message=" + payload);
}
}
else if(!passwordSent) {
if(payload.equals(passwordPrompt)) {
try {
logger.debug("Sending password to authenticate socket.");
super.send(MessageBuilder.withPayload(password).build());
passwordSent = true;
forwardMessage = false;
} catch (Exception e) {
throw new MessagingException("Negotiation error", e);
}
}
else {
throw new MessagingException("Negotiation error. expected message=" + passwordPrompt +
" actual message=" + payload);
}
}
else if(payload.startsWith(PING_PREFIX)) {
//Just record that we received the ping.
forwardMessage = false;
}
}
if(forwardMessage)
return super.onMessage(message);
else
return true;
}
@Override
public boolean isOpen() {
DateTime now = new DateTime();
if((lastMsgReceived == null) ||
((now.getMillis() - lastMsgReceived.getMillis()) < inactivityTimeout)) {
return super.isOpen();
}
else
{
if(super.isOpen()) {
super.close();
}
return false;
}
}
public Integer getInactivityTimeout() {
return inactivityTimeout;
}
public void setInactivityTimeout(Integer inactivityTimeout) {
this.inactivityTimeout = inactivityTimeout;
}
}
