具有弹簧集成的 UDP 服务器

Question

我想知道是否可以使用能够接受请求并返回响应的 spring-integration 框架创建一个 UDP 服务器。

对于 TCP，有允许请求/响应处理的 TCP 网关，但对于 UDP，我没有看到类似的东西。

设置 UDP 侦听器和接收数据包很容易，但是我看不到如何返回响应，因为我只能将它路由到预定义的输出通道。

我也看不到发送者的 IP 和端口，因为转换器没有接收 DatagramPacket 对象，而只接收数据。

这是我的配置：

<int:channel id="ChannelIn" />

<ip:udp-inbound-channel-adapter id="ChannelReceiver"
    channel="ChannelIn"
    port="5555"
    multicast="false"
    check-length="false" 
    pool-size="10"
    />

<int:transformer
    ref="datagramToPacketTransformer"
    input-channel="ChannelIn"
    output-channel="ChannelSA" 
    method="toPacket"/>

<int:channel id="ChannelSA" />

<int:service-activator id="ChannelActivator" 
    input-channel="ChannelSA"
    ref="PacketHandler"
    method="process"
/>

Answer 1

我找到了处理 UDP 的方法。下面是我的配置

<int-ip:udp-inbound-channel-adapter id="ChannelReceiver"
                                    channel="serverBytes2StringChannel"
                                    port="9000"
                                    multicast="false"
                                    check-length="false" 
                                    pool-size="10"
                                    lookup-host="false"
/>

<int:transformer id="serverBytes2String"
                 input-channel="serverBytes2StringChannel"
                 output-channel="udpSA" 
                 expression="new String(payload)"

/>

<int:service-activator input-channel="udpSA" 
                       ref="deviceService" 
                       method="udpMessage"/>

而deviceService bean 有这个代码：

 public Message udpMessage(Message message) {
        String response = "KO";
        try {
            response = process(message);
        } catch (Throwable th) {
            //do something
        }
        Message msg = MessageBuilder.withPayload(response.getBytes()).copyHeaders(message.getHeaders()).build();
        sendReply(msg);
        return null;
    }
    private void sendReply(Message message) {
        try {
            int port = (Integer) message.getHeaders().get("ip_port");
            String ip = (String) message.getHeaders().get("ip_address");
            InetAddress IPAddress = InetAddress.getByName(ip);

            byte[] sentence = (byte[]) message.getPayload();
            byte[] sendData = new byte[sentence.length];
            byte[] receiveData = new byte[1024];
            sendData = sentence;
            DatagramPacket sendPacket = new DatagramPacket(sendData, sendData.length, IPAddress, port);
            DatagramSocket clientSocket = new DatagramSocket();
            clientSocket.send(sendPacket);
            clientSocket.close();
        } catch (Exception ex) {
            throw new RuntimeException("KO");
        }
    }

这在我的个人计算机上运行良好，但是当我在 Amazon AWS 上部署此代码时，它无法正常工作。我发现即使程序正在侦听 UDP 9000 端口，但在服务器端没有收到通信。尽管我已允许该 AWS 实例的所有 UDP 端口，但这可能是 AWS 设置的问题。

关于如何在 Amazon AWS 上进行这项工作的任何提示？

Answer 2

我在近两年前开通了一个新功能 JIRA

https://jira.springsource.org/browse/INT-1809

但它没有收到投票或观察者，所以我关闭了它。

随意添加评论，我们可以重新打开它。

发件人的ip在邮件头中，但不是他的端口...

            message = MessageBuilder.withPayload(payload)
                    .setHeader(IpHeaders.HOSTNAME, hostName)
                    .setHeader(IpHeaders.IP_ADDRESS, hostAddress)
                    .build();

如果您想为此打开一个新的 JIRA，只需进行很小的更改即可进入 2.2。即将发布的版本（但网关不会发布 2.2）。

https://jira.springsource.org/browse/INT

Answer 3

谢谢你，加里！

我注意到远程 IP 地址在标头中，但远程端口不在，所以我做了一些更改来添加它。

我已扩展 UnicastReceivingChannelAdapter 以覆盖构造 Message 的 asyncSendMessage。我也必须扩展 DatagramPacketMessageMapper。如果我可以在 UnicastReceivingChannelAdapter 中注入我自己的 DatagramPacketMessageMapper 实现会更容易，但这是不可能的。

然后在 spring 配置中我删除了

<ip:udp-inbound-channel-adapter id="ChannelReceiver"...>

定义并添加如下

<bean id="udpInboundChannelAdapterBean" 
          class="me.spring.integration.udp.UnicastReceivingChannelAdapterExt" init-method="run">
          <constructor-arg name="port" value="5555" />
          <property name="poolSize" value="10"/>
          <property name="lookupHost" value="false"/>
          <property name="outputChannel" ref="ChannelIn" />

这似乎可行，尽管我还没有时间对其进行全面测试。

我的想法是使用远程 IP 和端口以及来自 udpInboundChannelAdapterBean 的 DatagramSocket 引用来创建一个新的激活服务，它只会将响应写入套接字。

我会按照你的建议为远程端口打开一个新的 jira。