Spring統合DSLカスタムエラーチャネルが動作しない

Question

私はSpring IntegrationのDSL実装を使用します。 私は以下のコードを持っており、カスタムエラーフローを使用することはできません。認証メソッドがランタイム例外をスローすると、errorChannelが処理を開始します。カスタムエラーフローを使用するためにヘッダーを充実させていますが、使用しません。

// In Class - 1 
@Bean 
    public MarshallingWebServiceInboundGateway marshallingWebServiceInboundGateway(BeanFactoryChannelResolver channelResolver, Jaxb2Marshaller marshaller) { 

     MarshallingWebServiceInboundGateway wsInboundGateway = new MarshallingWebServiceInboundGateway(); 
     wsInboundGateway.setRequestChannel(channelResolver.resolveDestination("incomingRequest.input")); 
     wsInboundGateway.setReplyChannel(channelResolver.resolveDestination("outgoingResponse.input")); 
     wsInboundGateway.setErrorChannel(channelResolver.resolveDestination("errorChannel")); 
     wsInboundGateway.setMarshaller(marshaller); 
     wsInboundGateway.setUnmarshaller(marshaller); 
     return wsInboundGateway; 
    } 


// In Class - 2 
@Bean 
    public IntegrationFlow incomingRequest() { 
     return f -> f.<Object, Class<?>>route(t -> t.getClass(), 
       mapping -> mapping.subFlowMapping(payloadType1(), 
         sf -> sf.gateway("type1.input", ConsumerEndpointSpec::transactional)) 
         .subFlowMapping(payloadType2(), 
           sf -> sf.gateway("type2.input", ConsumerEndpointSpec::transactional)), 
         conf -> conf.id("router:Incoming request router")); 
    } 

// In Class - 3 
    @Bean 
    public IntegrationFlow type1() { 
     IntegrationFlow integrationFlow = f -> f 
       .enrichHeaders(h -> h.header(MessageHeaders.ERROR_CHANNEL, "error222", true)) 
       .<Type1>handle((p, h) -> authentication.authenticate(p), 
         conf -> conf.id("service-activator:Authenticate")) 
       .transform(transformer::transformType1MsgToDataX, 
         conf -> conf.id("transform:Unmarshall type1 Message")) 
       .enrichHeaders(h -> h.headerExpression(TypeDataIntegrationMessageHeaderAccessor.MESSAGE_ID, "payload.id") 
         .headerExpression(TypeDataIntegrationMessageHeaderAccessor.MESSAGE_TYPE, "payload.messageType")) 
       .handle((GenericHandler<DataX>) repository::successResponseMessage, 
         conf -> conf.id("service-activator:return success")) 
       .channel("outgoingResponse.input") 
       ; 

     return integrationFlow; 
    } 

// In Class - 3 
@Bean 
    public IntegrationFlow error222Flow() { 

     return IntegrationFlows.from("error222").handle("repository", "failureResponseMessage").get() 

       ; 

    } 

EDIT：アルテムの答えの後

、以下のように私のコード。しかし、私はエラーフローのヘッダパラメータにアクセスすることはできません。私はエラーを取得する - 「いいえチャネルがルータによって解決しない 『ルータ：エラー応答が準備』」

// In Class - 1 
@Bean 
    public MarshallingWebServiceInboundGateway marshallingWebServiceInboundGateway(BeanFactoryChannelResolver channelResolver, Jaxb2Marshaller marshaller) { 

     MarshallingWebServiceInboundGateway wsInboundGateway = new MarshallingWebServiceInboundGateway(); 
     wsInboundGateway.setRequestChannel(channelResolver.resolveDestination("incomingRequest.input")); 
     wsInboundGateway.setReplyChannel(channelResolver.resolveDestination("outgoingResponse.input")); 
     wsInboundGateway.setErrorChannel(channelResolver.resolveDestination("errorResponse.input")); 
     wsInboundGateway.setMarshaller(marshaller); 
     wsInboundGateway.setUnmarshaller(marshaller); 
     return wsInboundGateway; 
    } 


// In Class - 2 
@Bean 
    public IntegrationFlow incomingRequest() { 
     return f -> f.<Object, Class<?>>route(t -> t.getClass(), 
       mapping -> mapping.subFlowMapping(payloadType1(), 
         sf -> sf.gateway("type1.input", ConsumerEndpointSpec::transactional)) 
         .subFlowMapping(payloadType2(), 
           sf -> sf.gateway("type2.input", ConsumerEndpointSpec::transactional)), 
         conf -> conf.id("router:Incoming request router")); 
    } 

// In Class - 2 
@Bean 
public IntegrationFlow errorResponse(){ 
    return f -> f.<MessageHandlingException, Object>route(t -> t.getFailedMessage().getHeaders().get("ABCDEF"), 
         mapping -> mapping.subFlowMapping("ABCDEF", 
           sf -> sf.gateway("customError.input", ConsumerEndpointSpec::transactional)), 
           conf -> conf.id("router:error response prepare")); 
} 

// In Class - 3 
    @Bean 
    public IntegrationFlow type1() { 
     IntegrationFlow integrationFlow = f -> f 
       .enrichHeaders(h -> h.header("ABCDEF", "ABCDEF", true)) 
       .<Type1>handle((p, h) -> authentication.authenticate(p), 
         conf -> conf.id("service-activator:Authenticate")) 
       .transform(transformer::transformType1MsgToDataX, 
         conf -> conf.id("transform:Unmarshall type1 Message")) 
       .enrichHeaders(h -> h.headerExpression(TypeDataIntegrationMessageHeaderAccessor.MESSAGE_ID, "payload.id") 
         .headerExpression(TypeDataIntegrationMessageHeaderAccessor.MESSAGE_TYPE, "payload.messageType")) 
       .handle((GenericHandler<DataX>) repository::successResponseMessage, 
         conf -> conf.id("service-activator:return success")) 
       .channel("outgoingResponse.input") 
       ; 

     return integrationFlow; 
    } 

// In Class - 3 
@Bean 
    public IntegrationFlow customError(){ 
     return f -> f.handle((GenericHandler<MessageHandlingException>)eventRepository::failureResponseMessage, 
           conf -> conf.id("service-activator:return failure")); 
    } 

にEDIT - 2：

を、私はそれがこのシナリオで動作し、アルテムのテストコードを試してみてください。下のようにtype1フローをサブフローマッピングに変換すると（サブフローコードブロックが疑わしいので）、エラーフローはABCDEFパラメータ値を出力できません。 その後、別のヘッダー（XYZTWR）をサブフローマッピングに追加しますが、印刷することはできません。

@Bean 
public IntegrationFlow type1() { 
    return f -> f.<String, String>route(t -> t.toString(), mapping -> mapping.subFlowMapping("foo", 
      sf -> sf.gateway("fooFlow.input", ConsumerEndpointSpec::transactional).enrichHeaders(h -> h.header("XYZTRW", "XYZTRW", true)))); 
} 

@Bean 
public IntegrationFlow fooFlow() { 
    return f -> f.enrichHeaders(h -> h.header("ABCDEF", "ABCDEF", true)) 
      .handle((p, h) -> { 
       throw new RuntimeException("intentional"); 
      }); 
} 

私S.OUTです：私たちは別のスレッドの実行者またはキューチャネルにメッセージをシフトするとき

GenericMessage [payload=foo, headers={history=testGateway,type1.input, id=1fad7a65-4abe-c41d-0b22-36839a103269, timestamp=1503029553071}] 

Answer 1

errorChannelヘッダが作業を開始します。それ以外の場合は、標準のthrowとtry...catchは同じ呼び出しスタックで動作します。

あなたのケースでは、認証例外は発信者 - WS受信ゲートウェイにスローされます。ここではグローバルエラーチャネルを設定しました。

私はこのテストでした：

@Configuration 
@EnableIntegration 
@IntegrationComponentScan 
public static class ContextConfiguration { 

    @Bean 
    public IntegrationFlow errorResponse() { 
     return IntegrationFlows.from(errorChannel()) 
        .<MessagingException, Message<?>>transform(MessagingException::getFailedMessage, 
          e -> e.poller(p -> p.fixedDelay(100))) 
        .get(); 
    } 

    @Bean 
    public IntegrationFlow type1() { 
      return f -> f 
        .enrichHeaders(h -> h.header("ABCDEF", "ABCDEF", true)) 
        .handle((p, h) -> { throw new RuntimeException("intentional"); }); 
    } 

    @Bean 
    public PollableChannel errorChannel() { 
     return new QueueChannel(); 
    } 
} 

@MessagingGateway(errorChannel = "errorChannel", defaultRequestChannel = "type1.input") 
public interface TestGateway { 

    Message<?> sendTest(String payload); 

} 

... 

@Autowired 
private TestGateway testGateway; 

@Test 
public void testErrorChannel() { 
    Message<?> message = this.testGateway.sendTest("foo"); 
    System.out.println(message); 
} 

そして、私のSOUTは私を示しています

GenericMessage [payload=foo, headers={ABCDEF=ABCDEF, id=ae5d2d44-46b7-912d-17d4-bf2ee656140a, timestamp=1502999446725}] 

してください、org.springframework.integrationカテゴリのデバッグログレベルを作成し、観察あなたのメッセージが希望のヘッダーを失っているステップします。

UPDATE

OK。私はあなたの問題を見る。 sf -> sf.gateway("fooFlow.input", ConsumerEndpointSpec::transactional)を使用しているため、ゲートウェイを経由してダウンストリームを呼び出すと、そこで行ったことのすべてがドアの後ろにあり、ゲートウェイの要求メッセージにエラーが発生した場合にエラーを返すことができます。ダウンストリームfailedMessageはデフォルトでは飲み込まれます。

問題を解決するには、の追加番号errorChannel()を考慮し、その下流のエラーを処理する必要があります。または、ルータのサブフローで.gateway()を使用しないでください。単純なchannelマッピング。

.transactional()は、.handle()でも設定できます。