Весенняя интеграция TCP / IP проблема с закрытым соединением

Question

Я использую пружинную интеграцию в качестве модуля шлюза полнодуплексной системы связи. этот поток является клиентским приложением <-> весенний интеграционный ip-модуль (siid) <-> серверное приложение Проблема в том, что когда клиентское приложение закрыто, ssid не может закрыть соединение со стороной серверного приложения? вот мой код

    // siid connet to client

@Bean
public TcpNetServerConnectionFactory server(){
    TcpNetServerConnectionFactory server=new TcpNetServerConnectionFactory(1234);
    server.setMapper(new TcpSerMapper()); // use 'mapper' attribute in XML
    MySerializer mySeri=new MySerializer();
    server.setDeserializer(mySeri);
    server.setSerializer(mySeri);
    return server;
    }

// inboundGateway, inChannel as reqeustChannel
@Bean
public TcpInboundGateway inGate(){
    TcpInboundGateway inGate=new TcpInboundGateway();
    inGate.setConnectionFactory(server());
    inGate.setRequestChannelName("inChannel");
    inGate.setReplyChannelName("outputChannel");
    return inGate;
    }

// serviceActivator to get inChannel's payload msg and send though a gateway.
@ServiceActivator(inputChannel = "inChannel")
public  byte[]doClientForward(Message<?> msg){
    byte[]msgPayload=(byte[])(msg.getPayload());
    byte[]sendResult=null;
    ToTCP toTcp=(ToTCP)contextBean.get("toTcpBean"); // ToTCP is a gateway
    sendResult=toTcp.sends((msgPayload),"localhost",7779);
    QueueChannel outputChannel=(QueueChannel)contextBean.get("outputChannel");
    return sendResult;
    }

public static class DynamicSerSeri extends AbstractPooledBufferByteArraySerializer {
protected byte[] doDeserialize(InputStream inputStream, byte[] buffer) throws IOException {
    byte[] bytes = this.copyBuffer(inputStream, buffer);
    return bytes;
}

public void serialize(byte[] object, OutputStream outputStream) throws IOException {
    outputStream.write(object);
}

public byte[] copyBuffer(InputStream inputStream, byte[] buffer) throws IOException {

    int n = 0;
    int bite = 0;
    try {
        while (true) {
            bite = inputStream.read(); // blocked here
            this.setMaxMessageSize(inputStream.available() + 1);
            buffer = new byte[inputStream.available() + 1];
            if (bite < 0 && n == 0) {
                throw new SoftEndOfStreamException("Stream closed between payloads");
            }
            checkClosure(bite);
            buffer[n++] = (byte) bite;
            if (bite == -1) {
                break;

            }
            if (n == this.maxMessageSize) {
                break;
            }
        }
        return buffer;
    } catch (SoftEndOfStreamException e) {
        throw e; // I was stuck here. when client closed, cf can't receive this exception and send close singnal to server side
    } catch (IOException e) {
        publishEvent(e, buffer, n);
        throw e;
    } catch (RuntimeException e) {
        publishEvent(e, buffer, n);
        throw e;
    }
}

}


@MessagingGateway()
public interface ToTCP {
@Gateway(requestChannel = "toTcp.input", replyChannel = "outputChannel")
public byte[] sends(byte[] data, @Header("host") String host, @Header("port") int port);
}

@Bean
public IntegrationFlow toTcp() {
    return f -> f.route(new ClientTcpRouter());
}

// I am not sure I understand IntegrationFlowContext,but it works
public static class ClientTcpRouter extends AbstractMessageRouter {
@Autowired
private IntegrationFlowContext flowContext;

@Override
protected synchronized Collection<MessageChannel> determineTargetChannels(Message<?> message) {
    // connection to server side.
    TcpNetClientConnectionFactory cf = new TcpNetClientConnectionFactory(host, port); //?? this connection factory does's closed when inGate's connection factory throw SoftEndOfStreamException
    TcpOutboundGateway handler = new TcpOutboundGateway();
    handler.setConnectionFactory(cf);
    cf.setDeserializer(new DynamicSerSeri());
    cf.setSerializer(new DynamicSerSeri());
    IntegrationFlow flow = f -> f.handle(handler);
    IntegrationFlowContext.IntegrationFlowRegistration flowRegistration =
            this.flowContext.registration(flow)
                    .addBean(cf)
                    .id(hostPort + ".flow")
                    .register();
    MessageChannel inputChannel = flowRegistration.getInputChannel();
    this.subFlows.put(hostPort, inputChannel);
    return inputChannel;
}
}

TcpInboundGateway получает соединение от клиента в inputChannel, и я использую serviceActivator, чтобы получить полезную нагрузку inputChannel и отправить на сторону сервера по TcpOutboundGateway, который имеет фабрику соединений со стороной сервера. когда клиент закрыл соединение с помощью spring -gration-ip-module, TcpInboundGateway может получить исключение в SoftEndOfStreamException, но я не знаю, как закрыть соединение TcpOutboundGateway со стороны сервера.

Answer 1

Используйте метод ApplicationListener bean или @EventListener для прослушивания событий TCP .

Когда вы впервые открываете исходящее соединение, вы получите TcpConnectionOpenEvent.По умолчанию он публикуется (и будет приниматься) в вызывающем потоке.Вы можете связать идентификатор исходящего соединения с входящим.

Прослушивание TcpConnectionCloseEvent из фабрики входящих соединений;затем вы можете закрыть исходящее соединение, используя connectionId.

outboundFactory.closeConnection(connectionId);

РЕДАКТИРОВАТЬ

Поскольку вы используете TcpNetServerConnectionFactory, вы можете использовать ThreadAffinityClientConnectionFactory который будет автоматически связывать исходящее соединение с входящим соединением.

Когда вы закрываете событие для входящего соединения, оно будет в том же потоке, поэтому вы можете просто вызвать releaseConnection() в этом потоке иисходящее соединение закроется.

Вот пример

@SpringBootApplication
public class So55207274Application {

    public static void main(String[] args) {
        SpringApplication.run(So55207274Application.class, args);
    }

    @Bean
    public IntegrationFlow flow() {
        return IntegrationFlows.from(Tcp.inboundGateway(server()))
                .log()
                .handle(Tcp.outboundGateway(threadBoundClient()))
                .get();
    }

    @Bean
    public TcpNetServerConnectionFactory server() {
        return new TcpNetServerConnectionFactory(1234);
    }

    @Bean
    public ThreadAffinityClientConnectionFactory threadBoundClient() {
        return new ThreadAffinityClientConnectionFactory(client());
    }

    public TcpNetClientConnectionFactory client() {
        TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", 1235);
        client.setSingleUse(true);
        return client;
    }

    @EventListener
    public void listen(TcpConnectionCloseEvent event) {
        if (event.getConnectionFactoryName().equals("server")) {
            try {
                threadBoundClient().releaseConnection();
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        }
        System.out.println(event);
    }

    // Test server

    @Bean
    public IntegrationFlow test() {
        return IntegrationFlows.from(Tcp.inboundGateway(Tcp.netServer(1235)))
                .transform(Transformers.objectToString())
                .<String, String>transform(p -> p.toUpperCase())
                .get();
    }

}