Spring Integration TCP doesn't send messages

Question

I've been trying to create a consumer and a producer using the Spring Integration TCP. I've achived some success on the listening part (I got the message sended by a telnet normally), but when I try to send this same message to the terminal client, nothing happens.

These are my classes:

@EnableIntegration
@IntegrationComponentScan
@Configuration
public class TcpIntegration {

  @Value("${tcp.port}")
  private Integer port;

  @MessagingGateway(defaultRequestChannel="toTcp")
  public interface Gateway {
    String viaTcp(String in);
  }

  @Bean
  @ServiceActivator(inputChannel="toTcp")
  public TcpSendingMessageHandler tcpOutGate(AbstractClientConnectionFactory connectionFactory) {
    TcpSendingMessageHandler gate = new TcpSendingMessageHandler();
    gate.setConnectionFactory(connectionFactory);
    return gate;
  }

  @Bean
  public TcpReceivingChannelAdapter tcpInGate(AbstractServerConnectionFactory connectionFactory)  {
    TcpReceivingChannelAdapter inGate = new TcpReceivingChannelAdapter();
    inGate.setConnectionFactory(connectionFactory);
    inGate.setOutputChannel(fromTcp());
    return inGate;
  }

  @Bean
  public MessageChannel fromTcp() {
    return new DirectChannel();
  }

  @Bean
  public AbstractClientConnectionFactory clientCF() {
    return new TcpNetClientConnectionFactory("localhost", this.port);
  }

  @Bean
  public AbstractServerConnectionFactory serverCF() {
    return new TcpNetServerConnectionFactory(this.port);
  }
}

TcpListener

@MessageEndpoint
@AllArgsConstructor
public class TcpListener {

  private final Gateway gateway;

  @ServiceActivator(inputChannel = "fromTcp")
  public void convert(String payload) {
    System.out.println(payload);
    gateway.viaTcp(payload);
  }
}

Why doesn't it work?

Answer 1

You need to tell the adapter which socket to send the messsage on by setting the ip_connnetionId header.

  @ServiceActivator(inputChannel = "fromTcp")
  public void convert(String payload, @Header(IpHeaders.CONNECTION_ID String cid) {
    System.out.println(payload);
    gateway.viaTcp(payload, cid);
  }

  @MessagingGateway(defaultRequestChannel="toTcp")
  public interface Gateway {
    @Gateway(@GatewayHeader(IPHeaders.CONNECTION_ID, "#args[1]"))
    String viaTcp(String in, String cid);
  }

EDIT

My syntax was a little bit off; here's a working example...

@SpringBootApplication
public class So65597331Application {

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

    @MessagingGateway(defaultRequestChannel = "toTcp")
    public interface Gate {
        @Gateway(payloadExpression = "#args[0]",
                headers = @GatewayHeader(name = IpHeaders.CONNECTION_ID, expression = "#args[1]"))
        String viaTcp(String in, String cid);
    }

    @Bean
    @ServiceActivator(inputChannel = "toTcp")
    public TcpSendingMessageHandler tcpOutGate(AbstractServerConnectionFactory connectionFactory) {
        TcpSendingMessageHandler gate = new TcpSendingMessageHandler();
        gate.setConnectionFactory(connectionFactory);
        return gate;
    }

    @Autowired
    private Gate gateway;

    @ServiceActivator(inputChannel = "fromTcp")
    public void convert(String payload, @Header(IpHeaders.CONNECTION_ID) String cid) {
        System.out.println(payload);
        gateway.viaTcp(payload, cid);
    }

    @Bean
    public TcpReceivingChannelAdapter tcpInGate(AbstractServerConnectionFactory connectionFactory) {
        TcpReceivingChannelAdapter inGate = new TcpReceivingChannelAdapter();
        inGate.setConnectionFactory(connectionFactory);
        inGate.setOutputChannel(fromTcp());
        return inGate;
    }

    @Bean
    public MessageChannel fromTcp() {
        return new DirectChannel();
    }

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

}

$ telnet localhost 1234
Trying ::1...
Connected to localhost.
Escape character is '^]'.
foo
foo
Connection closed by foreign host.