Spring Integration version 3.0: Разделение сообщения потока TCP на несколько сообщений на основе содержимого

Question

Я подключаюсь к некоторому устаревшему серверному сокету, используя платформу интеграции с Spring.

Ниже мой клиент завод и адаптер:

<int-ip:tcp-connection-factory id="client" 
           type="client" 
           host="${tcpServer}" 
           port="${tcpPort}" 
           single-use="false" 
           using-nio="false" /> 

<int-ip:tcp-inbound-channel-adapter id="inboundServer" 
            client-mode="true" 
            channel="inputStream" 
            error-channel="errorChannel" 
            retry-interval="${retryInterval}" 
            connection-factory="client" /> 

И ниже потока в струнный преобразователь:

<int:transformer id="clientBytes2String" 
       input-channel="inputStream" 
       output-channel="inputString" 
       expression="new String(payload)" /> 

       <int:channel id="inputString" /> 

И следующий раздел пуст, так как я не уверен, что для реализации здесь, так что он может вызвать мой маршрутизатор, и маршрутизатор сделает это.

1. ---

Я попытался с разветвителем, он сделал работу, если поток идет в желаемом формате «ABCD EFGH WXYZ» или «ABCD», но если поток идет, как «ABCD XXXX EFGH WXYZ ", то его провал. Желаемые результаты - это обработать 3 сообщения и 1 ошибку. Но вместо этого он обрабатывал 1 сообщение, а все остальные игнорируются.

Код ниже:

<int:splitter input-channel="inputString" 
       output-channel="preRouter2"     
       method="splitMessage" 
       ref="messageSplitterBean"/> 

И MessageSpliterBean класс следующим образом:

@Splitter 
public List<Message<?>> splitMessage(Message<?> message) { 

    List<Message<?>> msgFragments = new ArrayList<Message<?>>(); 

    //Let say I am assuming message will be coming ABCD EFGH WXYZ 
    String str[] = message.getPayload().toString().split(" "); 
    int counter = 1; 
    for (String s : str) { 

     Message<String> resultMessage = MessageBuilder.withPayload(s) 
       .copyHeaders(message.getHeaders()) 
       .build(); 
     msgFragments.add(resultMessage); 
    } 

    return msgFragments; 
} 

И следующий будет мой маршрутизатор, который направит на соответствующий канал на основе некоторого выражения:

<int:recipient-list-router id="customRouter" input-channel="preRouter2"> 
<int:recipient channel="input1" selector-expression="payload.toString().startsWith('ABCD')"/> 
<int:recipient channel="input2" selector-expression="payload.toString().startsWith('EFGH')"/> 
<int:recipient channel="input3" selector-expression="payload.toString().startsWith('WXYZ')"/> 

Нужен ваш экспертный взгляд на это: на то, что я делаю неправильно, или что будет лучшим подходом.

Вход сокета сервера будет в потоке данных с фиксированной длиной и пространством в качестве разделителя. И каждая фиксированная длина мне нужно преобразовать в сообщение и отправить в соответствующий канал.

С уважением.

Answer 1

Прежде всего, нет причин реализовать свой собственный вариант Splitter, потому что один по умолчанию имеет :

Другой мой пункт о избыточном <int:transformer> для byte[] -> String. Весенняя интеграция предоставляет вам ObjectToStringTransformer из коробки.

И ваша проблема <int:recipient-list-router>. Как вы говорите, у вас может быть что-то не так в ваших данных, и ваш router не готов обработать такое сообщение, и он не подходит для вашего. Это только потому, что (AbstractMessageRouter):

else { 
    throw new MessageDeliveryException(message, "No channel resolved by router '" + this.getComponentName() 
       + "' and no 'defaultOutputChannel' defined."); 
} 

Что происходит, когда никто selector-expression не примет ваше неправильное сообщение.

В этом случае он отправляется на error-channel="errorChannel" на ваш <int-ip:tcp-inbound-channel-adapter> и останавливает дальнейший процесс только потому, что MessageDeliveryException.

Как вы видите, похоже, чтобы исправить вашу проблему, вы должны добавить default-output-channel в конфигурацию recipient-list-router.