Spring Boot Integration - Send greet on connection initialization

Question

I am writing a simple TCP Server application with Spring Boot Integration Framework that has to send a greeting to the client on connection establishment. The workflow should be as follows:

Client connects to server
Server sends a greeting message to the client "Hello client"
Client sends a message to the server "Any string"
Server responds with "OK"
...

Currently steps 1, 3 and 4 are working, but I'm failing on step 2. My code so far:

@SpringBootApplication
@EnableIntegration
public class ExampleApp {

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

    // Create listener on port 1234
    @Bean
    public AbstractServerConnectionFactory serverConnectionFactory() {
        TcpNetServerConnectionFactory tcpNetServerConnectionFactory = new TcpNetServerConnectionFactory(1234);
        return tcpNetServerConnectionFactory;
    }

    // Inbound channel
    @Bean
    public MessageChannel requestChannel() {
        return new DirectChannel();
    }

    // Outbound channel
    @Bean
    public MessageChannel replyChannel() {
        return new DirectChannel();
    }

    // Inbound gateway
    @Bean
    public TcpInboundGateway tcpInboundGateway(AbstractServerConnectionFactory serverConnectionFactory, MessageChannel requestChannel, MessageChannel replyChannel) {
        TcpInboundGateway tcpInboundGateway = new TcpInboundGateway();
        tcpInboundGateway.setConnectionFactory(serverConnectionFactory);
        tcpInboundGateway.setRequestChannel(requestChannel);
        tcpInboundGateway.setReplyChannel(replyChannel);
        return tcpInboundGateway;
    }

    // Send reply for incoming message -> working
    @ServiceActivator(inputChannel = "requestChannel", outputChannel = "replyChannel")
    public Message<String> processMessage(Message<String> message) {
        Message reply = MessageBuilder
                .withPayload("OK")
                .setHeader(IpHeaders.CONNECTION_ID, message.getHeaders().get(IpHeaders.CONNECTION_ID, String.class))
                .build();
        return reply;
    }

    // Send greeting -> not working
    @Bean
    public ApplicationListener<TcpConnectionEvent> listener(MessageChannel replyChannel) {
        return tcpConnectionEvent -> {
            if (tcpConnectionEvent instanceof TcpConnectionOpenEvent) {
                Message<String> message = MessageBuilder
                        .withPayload("Hello client")
                        .setHeader(IpHeaders.CONNECTION_ID, tcpConnectionEvent.getConnectionId())
                        .build();
                replyChannel.send(message);
            }
        };
    }
}

If I connect to the server with

nc -C localhost 1234

connection is established, but I'm getting the following error in the log:

Failed to publish TcpConnectionOpenEvent [source=TcpNetConnection:localhost:37656:1234:187cfbc2-7e5d-4f4e-97de-1a3b55a4e264], [factory=serverConnectionFactory, connectionId=localhost:37656:1234:187cfbc2-7e5d-4f4e-97de-1a3b55a4e264] OPENED:Dispatcher has no subscribers for channel 'application.replyChannel'

If I send a String to the server he replies with "OK" as intended.

What am I missing to get this greeting message working?

Solution

Thanks to the comment of Gary Russel I found a solution. The Inbound gateway must be "split" into a Inbound/Outbound channel adapter pair. Here is the fully working example:

@SpringBootApplication
@EnableIntegration
public class ExampleApp {

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

    // Create listener on port 1234
    @Bean
    public AbstractServerConnectionFactory serverConnectionFactory() {
        TcpNetServerConnectionFactory tcpNetServerConnectionFactory = new TcpNetServerConnectionFactory(1234);
        return tcpNetServerConnectionFactory;
    }

    // Inbound channel
    @Bean
    public MessageChannel requestChannel() {
        return new DirectChannel();
    }

    // Outbound channel
    @Bean
    public MessageChannel replyChannel() {
        return new DirectChannel();
    }

    // Inbound channel adapter
    @Bean
    public TcpReceivingChannelAdapter receivingChannelAdapter(AbstractServerConnectionFactory serverConnectionFactory, MessageChannel requestChannel) {
        TcpReceivingChannelAdapter tcpReceivingChannelAdapter = new TcpReceivingChannelAdapter();
        tcpReceivingChannelAdapter.setConnectionFactory(serverConnectionFactory);
        tcpReceivingChannelAdapter.setOutputChannel(requestChannel);
        return tcpReceivingChannelAdapter;
    }

    // Outbound channel adapter
    @Bean
    @ServiceActivator(inputChannel = "replyChannel")
    public TcpSendingMessageHandler tcpSendingMessageHandler(AbstractServerConnectionFactory serverConnectionFactory) {
        TcpSendingMessageHandler tcpSendingMessageHandler = new TcpSendingMessageHandler();
        tcpSendingMessageHandler.setConnectionFactory(serverConnectionFactory);
        return tcpSendingMessageHandler;
    }

    // Send reply for incoming message -> working
    @ServiceActivator(inputChannel = "requestChannel", outputChannel = "replyChannel")
    public Message<String> processMessage(Message<String> message) {
        Message<String> reply = MessageBuilder
                .withPayload("OK")
                .setHeader(IpHeaders.CONNECTION_ID, message.getHeaders().get(IpHeaders.CONNECTION_ID, String.class))
                .build();
        return reply;
    }

    // Send greeting -> now working
    @Bean
    public ApplicationListener<TcpConnectionEvent> listener(MessageChannel replyChannel) {
        return tcpConnectionEvent -> {
            if (tcpConnectionEvent instanceof TcpConnectionOpenEvent) {
                Message<String> message = MessageBuilder
                        .withPayload("Hello client")
                        .setHeader(IpHeaders.CONNECTION_ID, tcpConnectionEvent.getConnectionId())
                        .build();
                replyChannel.send(message);
            }
        };
    }
}

Now the client gets the greeting "Hello client" on connection establishment and an "OK" as reply on each send message.

above error suggests that replyChannel has no subscriber, So you can bridge the replyChannel with a null channel. — Lakshman

Gary Russell Gary Russell · Accepted Answer · 2019-12-07T15:26:42

replyChannel.send(message);

You can't do that; the reply channel is wired up when the first request comes in.

In any case, you can't use a gateway like that, the reply channel is for a reply to a request not for sending some arbitrary message.

You have to use a pair of inbound/outbound channel adapters, instead of a gateway, to enable arbitrary two-way communication.

Spring Boot Integration - Send greet on connection initialization

Solution

1 Answers