如何使用Java注释设置Spring Integration协作通道适配器？

Question

我看过其他与我的问题相关的帖子，但没有一个答案帮助我解决了我的问题。

我试图遵循下面的示例：https://github.com/garyrussell/spring-integration-samples/tree/master/medition/tcp-client-server-multiplex

显然，我的理解是错误的，因为我没有看到响应如何返回到网关，而且它不起作用。

我可以看到套接字正在打开，但它在消息发送后立即被关闭，因此反序列化器返回一个EOF：null错误。

>

是否设置了TcpReceivingChannelAdapter错误？

@EnableIntegration
@IntegrationComponentScan
@Configuration
public class TcpMultiPlexConfig implements ApplicationListener<TcpConnectionEvent> {

    protected final static Logger LOGGER = LoggerFactory.getLogger(TcpMultiPlexConfig.class);

    @Value("${engine.port}")
    private int port;// = 55001;
    @Value("${engine.address}")
    private String ipAddress;// = "192.168.1.1";
    @Value("${engine.timeout}")
    private int timeout;

    @Override
    public void onApplicationEvent(TcpConnectionEvent tcpEvent) {
        TcpConnection source = (TcpConnection) tcpEvent.getSource();
        if (tcpEvent instanceof TcpConnectionOpenEvent) {
            LOGGER.info("********* Socket Opened " + source.getConnectionId());
        } else if (tcpEvent instanceof TcpConnectionCloseEvent) {
            LOGGER.info("*********** Socket Closed " + source.getConnectionId());
        }
    }

    @MessagingGateway(defaultRequestChannel="input")
    public interface MultiPlexGateway {

        String send(@Payload String in, @Header("CORRELATION_ID") String transactionId);

    }
    // TODO the request and response are being put together
    @Bean
    @ServiceActivator(inputChannel = "input")
    public BridgeHandler bridge() {
        BridgeHandler bridge = new BridgeHandler();
        bridge.setOutputChannelName("toAggregatorClient");
        bridge.setOrder(1);
        return bridge;
    }

    @Bean
    public PublishSubscribeChannel input() {
        return new PublishSubscribeChannel();
    }

    @Bean
    public DirectChannel toAggregatorClient() {
        return new DirectChannel();
    }

    @Bean
    public DirectChannel noResponseChannel() {
        return new DirectChannel();
    }

    @Bean
    public DirectChannel toTransformerClient() {
        return new DirectChannel();
    }

    @Bean
    public TcpReceivingChannelAdapter inAdapterClient() {
        TcpReceivingChannelAdapter receivingAdapter = new TcpReceivingChannelAdapter();
        receivingAdapter.setConnectionFactory(clientConnectionFactory());
        receivingAdapter.setOutputChannel(toAggregatorClient());
        receivingAdapter.setClientMode(true);
        return receivingAdapter;
    }


    @Bean
    @ServiceActivator(inputChannel = "input")
    public TcpSendingMessageHandler outAdapterClient() {
        TcpSendingMessageHandler outAdapter = new TcpSendingMessageHandler();
        outAdapter.setOrder(2);
        outAdapter.setConnectionFactory(clientConnectionFactory());
        outAdapter.setClientMode(true);
        return outAdapter;
    }

    @Bean(name ="clientCFMP")
    public AbstractClientConnectionFactory clientConnectionFactory() {
        TcpNetClientConnectionFactory tcp = new TcpNetClientConnectionFactory(this.ipAddress , this.port);
        tcp.setSerializer(new DefaultSerializer()); // out
//      byte delimeter = "\n".getBytes()[0];
//      ElasticByteArrayRawSingleTerminatorSerializer deserializer = new ElasticByteArrayRawSingleTerminatorSerializer(delimeter);
//      DefaultDeserializer deserializer = new DefaultDeserializer();
        MyDefaultDeserializer deserializer = new MyDefaultDeserializer();
        tcp.setDeserializer(deserializer);

        tcp.setSoTimeout(timeout);
        tcp.setSingleUse(false);
        MapMessageConverter mc = new MapMessageConverter();
        mc.setHeaderNames("CORRELATION_ID");
        tcp.setMapper(new MessageConvertingTcpMessageMapper(mc));

        return tcp;
    }


    @MessageEndpoint
    public static class MyConverters {

        @Transformer(inputChannel="toTransformerClient", outputChannel = "resultToString")
        public byte[] getResponse(MessageGroup payload) {
//          byte[] result = null;
            List<Message<?>>list = new ArrayList<>(payload.getMessages());
            byte[] result = (byte[]) list.get(1).getPayload();
//          LOGGER.info(result);
            return result;
        }

        @Transformer(inputChannel="resultToString")
        public String convertResult(byte[] bytes) {
            String result = new String(bytes);
            LOGGER.info("*********** RESULT => " + result);
            return result;
        }

        @ServiceActivator(inputChannel = "noResponseChannel")
        public MessageTimeoutException  noResponse(String input) {
            throw new MessageTimeoutException("****** No response received for => " + input);
        }

    }



    @Bean
    @ServiceActivator(inputChannel = "toAggregatorClient", outputChannel = "toTransformerClient")
    public FactoryBean<MessageHandler>  aggregatorFactoryBean() {
        AggregatorFactoryBean  afb = new AggregatorFactoryBean ();
        afb.setExpireGroupsUponCompletion(true);
        afb.setExpireGroupsUponTimeout(true);
        afb.setGroupTimeoutExpression(new ValueExpression<>(this.timeout));
        afb.setCorrelationStrategy(new HeaderAttributeCorrelationStrategy("CORRELATION_ID"));
        afb.setReleaseStrategy(new MessageCountReleaseStrategy(2));
        afb.setProcessorBean(new DefaultAggregatingMessageGroupProcessor());
        afb.setSendPartialResultOnExpiry(false);
        afb.setMessageStore(new SimpleMessageStore());
        afb.setDiscardChannel(noResponseChannel());
        return afb;
    }

@Service
public class MultiPlexGatewayTransmission <T extends EngineData> extends AbstractMultiPlexEngineTransmission {

    public MultiPlexGatewayTransmission(MultiPlexGateway gateway) {
        super(gateway);
    }

    @Override
    public T request(EngineData request, Class<? extends EngineData> clazz) {
        String response = gateway.send(JaxbUtils.marshall(request), request.getApi().getMessageId());
        gateway.send(JaxbUtils.marshall(request), request.getApi().getMessageId());
        if(response == null || response.isEmpty()) {
            return null;
        }

        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("MPGateway response ::: " + response.trim());
        }

        @SuppressWarnings("unchecked")
        T clientResponse = (T) JaxbUtils.unmarshall(response, clazz);
        if (LOGGER.isDebugEnabled()) {
//          LOGGER.debug("*** Unmarshall response ::: " + clientResponse);
        }
        return clientResponse;
    }

测试用例：

@RunWith(SpringRunner.class)
@SpringBootTest
@ActiveProfiles("test")
public class ITGetClientsTest extends AbstractEngineTest {

    private final Logger LOGGER = LoggerFactory.getLogger(this.getClass());

//  @Autowired
//  private GatewayTransmission<ClientsResponse> transmission;

    @Autowired
    private MultiPlexGatewayTransmission<ClientsResponse> transmission;

    @Test
    public void testGetClients() {
        LOGGER.info("Gateway test testGetClients... ");

        Api api = new Api();
        api.setIp("192.168.1.1");
        api.setMessageId(UUID.randomUUID().toString());
        api.setVersion("1.0");      
        api.setUserToken(token);

        ClientsRequest request = new ClientsRequest();
        request.setApi(api);

        ClientsResponse response = (ClientsResponse) transmission.request(request, ClientsResponse.class);
        Assert.assertTrue(response != null);
        Assert.assertTrue(!response.getClient().isEmpty());

        LOGGER.info(Arrays.deepToString(response.getClient().toArray()));
    }



}

Answer 1

我没有详细看你的代码；现在已经很晚了，而且是一个周末，但是请参阅下面的答案，了解使用入站/出站连接ID来关联请求/应答的更简单的技术。