I am writing a spring boot application which acts as a sort of middle man between two external clients say serviceA and serviceB. Communication will be via socket connection i.e. tcp/ip allthrough. So serviceA always initiate the communication, my applicaiton receives the message and does some processing, logging to db et al, then the message is relayed to serviceB. ServiceB sends a response, which I am also processing, after which the response is sent back to serviceA.
When I telnet "localhost <port>", I am able to initiate the communication and the processing all the way to serviceC, i am able to get a response also.
However, the goal is not to use telnet but to receive a message from an external client (serviceA) via socket connection. When a client application tries to connect to my app, there is no activity at all on my application.
SO WHY IS THERE NO ACTIVITY?
Note that the message to be received at the tcpInboundFlow is a string. Is this a case of the serializer/deserializer being used by the TcpInboundGateway?
I am making use of spring integration tcp. By using Java dsl and a mix of annotations based configuration based on the documentation, i have acheived some level of functionality.
I use the tcpInboundGateway making use of serverConnectionFactory to recieve the message (string format)
I use a router to determine channel based on some business logic
I do further processing using transformers
Then I make use of a tcpOutboundGateway to push the message to the external client.
I make use of a serviceActivator to process the response from the external client.
Below are code snippets of some of the components used. All the channels are directChannel implementations (they are not shown)
@Bean
public IntegrationFlow tcpInboundFlow() {
return IntegrationFlow.from(tcpInboundGateway())
.handle("messageProcessingService", "processMessage")
.route(mtiRouter())
.get();
}
@Bean
@Router(inputChannel = "handlerOutputChannel")
public AbstractMessageRouter mtiRouter() {
return new AbstractMessageRouter() {
@Override
protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
ISOMsg isoMsg = (ISOMsg) message.getPayload();
try {
String mti = isoMsg.getMTI();
log.info("reached the router");
return switch (mti) {
case "0800" -> Collections.singleton(networkManagementChannel());
case "0420" -> Collections.singleton(reversalChannel());
case "0200" -> Collections.singleton(financialRequestChannel());
default -> throw new TmsIsoException("Unsupported MTI: " + mti);
};
} catch (ISOException e) {
log.info("failed to reach the router");
throw new TmsIsoException("Failed to get mti from iso msg");
}
}
};
}
@Bean
public IntegrationFlow financialRequestFlow() {
return IntegrationFlow.from(financialRequestChannel())
.transform("financialRequestProcessingService", "processFinancialRequestMessage")
.transform("transformationService", "keyMessage")
.handle(tcpOutboundGateway())
.transform("responseService", "processResponseMessageForPurchase")
.get();
} @Bean
public TcpNetServerConnectionFactory serverConnectionFactory() {
TcpNetServerConnectionFactory factory = new TcpNetServerConnectionFactory(30002);
factory.setSerializer(new ByteArrayCrLfSerializer());
factory.setDeserializer(new ByteArrayCrLfSerializer());
// factory.setSoTimeout(60000);
return factory;
}
@Bean
public TcpNetClientConnectionFactory clientConnectionFactory() {
TcpNetClientConnectionFactory factory =
new TcpNetClientConnectionFactory("196.46.20.30", 5334);
factory.setSerializer(customIsoMessageSerializer);
factory.setDeserializer(customIsoMessageDeserializer);
return factory;
}
@Bean
public TcpInboundGateway tcpInboundGateway() {
TcpInboundGateway gateway = new TcpInboundGateway();
gateway.setConnectionFactory(serverConnectionFactory());
gateway.setRequestChannel(requestChannel());
gateway.setReplyChannel(replyChannel());
// gateway.setErrorChannel(errorChannel());
gateway.setReplyTimeout(60000);
gateway.setBeanName("tcpIn");
return gateway;
}
@Bean
public TcpOutboundGateway tcpOutboundGateway() {
TcpOutboundGateway gateway = new TcpOutboundGateway();
gateway.setConnectionFactory(clientConnectionFactory());
gateway.setRemoteTimeout(60000);
gateway.setReplyChannelName("replyChannel");
return gateway;
}
The above case was due to the serializer/deserializer being used; since the client-side connection was established. For information about some standard deserializers and serializers in Spring TCP provided, view here
In my case, I elected to use a ByteStxEtxSerializer and ByteStxEtxDeserializer
So when the client writes to the socket, it passes in a byte that must be formatted to the requirement of the serializer(named above) used by my serverConnectionFactory.
To format the byte, the below was written on the client end
@Component public class MessageFormatter {
private static final byte STX = 0x02; // Start of Text
private static final byte ETX = 0x03; // End of Text
private static final byte ESC = 0x1B; // Escape character
/**
* Converts a raw string into a byte array formatted with STX and ETX bytes,
* escaping any existing STX, ETX, and ESC bytes in the message.
*
* @param rawMessage the raw string message to be formatted
* @return the formatted byte array
*/
public static byte[] formatMessageWithStxEtx(String rawMessage) {
// Convert the raw string to a byte array
byte[] rawBytes = rawMessage.getBytes(StandardCharsets.UTF_8);
// Use a ByteArrayOutputStream to handle dynamic byte array creation
ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
// Add the STX byte at the beginning
byteStream.write(STX);
// Escape any STX, ETX, and ESC bytes in the raw message
for (byte b : rawBytes) {
if (b == STX || b == ETX || b == ESC) {
byteStream.write(ESC); // Write the escape byte
}
byteStream.write(b); // Write the original byte
}
// Add the ETX byte at the end
byteStream.write(ETX);
return byteStream.toByteArray();
}
//This method below is how it is written on the client
public String makeSocketCall(String ipAddress, String port, String
messageToBeSentViaTcp) {
Socket socket = null;
try {
socket = new Socket(ipAddress, Integer.parseInt(port));
//formatter converts the string into a byte array that has an Stx and Etx
//demarcator at the start and end of the array respectively.
byte[] byteRequest = MessageFormatter.formatMessageWithStxEtx(messageToBeSentViaTcp);
DataOutputStream dOut = new DataOutputStream(socket.getOutputStream());
DataInputStream dIn = new DataInputStream(socket.getInputStream());
dOut.write(byteRequest);
String response = dIn.readLine();
dOut.close();
dIn.close();
socket.close();
return response;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
As for the server, nothing changed from the previous code. The most important take away is to settle on a serializer and the format of the message which is to be transmitted via tcp