Как я могу правильно управлять потоком в subFlowMapping

Это основной вопрос, но некуда спросить. (Я только что начал весеннюю интеграцию в прошлый четверг.) Я должен использовать subFlowMapping. Один файл XML может содержать несколько данных домена. Базовая фильтрация имени файла сообщает мне только, какие данные содержатся в XML-файле.

Это рабочая версия. 2 постоянных канала хорошо выполняются и сохраняют данные.

channel("updatingPartsInventoryFlow.input")

channel("updatingTransactionCompleteLogFlow.input")

.subFlowMapping(PayloadType.STOCK, sf -> sf
                    .transform(unmarshallingTransformer)
                    .<Stock, List<PartsInventory>>transform(s -> createPartsInventories(s))
                    .publishSubscribeChannel(c1 -> c1
                        .subscribe(s1 -> s1
                            .channel("updatingPartsInventoryFlow.input")))
                            .publishSubscribeChannel(c2 -> c2
                        .subscribe(s2 -> s2
                            .handle((payload, headers) -> {
                                return createPartsInventoryTransactionLog((List<PartsInventory>) payload,                                                   (String) headers.get("StartTime"));
                            })
                            .channel("updatingTransactionCompleteLogFlow.input"))))

Они плохо работают.

channel("updatingPartsInventoryFlow.input") Данные не сохраняются всегда

channel("updatingTransactionCompleteLogFlow.input")Журнал сохраняется всегда

.subFlowMapping(PayloadType.STOCK, sf -> sf
                    .transform(unmarshallingTransformer)
                    .<Stock, List<PartsInventory>>transform(s -> createPartsInventories(s))
                    .channel("updatingPartsInventoryFlow.input")
                    .handle((payload, headers) -> {
                        return createPartsInventoryTransactionLog((List<PartsInventory>) payload, 
                                (String) headers.get("StartTime"));
                    })
                    .channel("updatingTransactionCompleteLogFlow.input"))

.subFlowMapping(PayloadType.STOCK, sf -> sf
                    .publishSubscribeChannel(c1 -> c1
                    .subscribe(s1 -> s1
                        .transform(unmarshallingTransformer)
                        .<Stock, List<PartsInventory>>transform(s -> createPartsInventories(s))
                        .channel("updatingPartsInventoryFlow.input")))
                        .handle((payload, headers) -> {
                            return createPartsInventoryTransactionLog((List<PartsInventory>) payload, 
                                            (String) headers.get("StartTime"));
                        })
                        .channel("updatingTransactionCompleteLogFlow.input"))))

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

Заранее спасибо.