Spring Cloud Stream SubscribableChannelの受信メッセージを傍受する

Question

方法postReceiveがorg.springframework.messaging.support.ChannelInterceptorであるため、org.springframework.messaging.SubscribableChannelでは呼び出されません。 @StreamListener(Sink.INPUT)の注釈付きメソッドの受信メッセージをすべて傍受する方法はありますか？例えば

：メッセージインターセプト

は前handle方法以下

@StreamListener(Sink.INPUT) 
public void handle(Foo foo) { 
    // ... 
} 

春クラウドストリームバインダーは、あなたのリスナーを上呼び出すDirectChannelで

public interface EventSink { 

    String INPUT1 = "input1"; 
    String INPUT2 = "input2"; 

    @Input(INPUT1) 
    SubscribableChannel input1(); 

    @Input(INPUT2) 
    SubscribableChannel input2(); 
} 

public interface EventSource { 

    String OUTPUT1 = "output1"; 
    String OUTPUT2 = "output2"; 

    @Output(OUTPUT1) 
    MessageChannel output1(); 

    @Output(OUTPUT2) 
    MessageChannel output2()'; 
} 

spring: 
    cloud: 
    stream: 
     bindings: 
     input1: 
      destination: input1 
     input2: 
      destination: input2  
     output1: 
      destination: output1 
     output2: 
      destination: output2 

public class EventHandler { 

    @StreamListener(EventSink.INPUT1) 
    public void handle(Foo1 foo) { 
     // ... 
    } 

    @StreamListener(EventSink.INPUT2) 
    public void handle(Foo2 foo) { 
     // ... 
    } 

} 

@Service 
public class Bar1Service { 

    @Autowired 
    private EventSource source; 

    public void bar1() { 
     source.output1().send(MessageBuilder.withPayload("bar1").build()); 
    } 

} 

@Service 
public class Bar2Service { 

    @Autowired 
    private EventSource source; 

    public void bar2() { 
     source.output2().send(MessageBuilder.withPayload("bar2").build()); 
    } 

} 

Answer 1

のための私のセットアップであるに入ります同じスレッドなのでpreSendがここで適切です。あなたがやりたいん何

しかし、あなたはThreadLocalを台無しにする必要はありません、あなたはメソッドのシグネチャを使用してヘッダにアクセスすることができます...

@StreamListener(Processor.INPUT) 
public void handle(Foo foo, @Header("bar") String bar) { 
    ... 
} 

EDIT

@EnableBinding(Processor.class) 
public class So41459187Application { 

    public static void main(String[] args) { 
     SpringApplication.run(So41459187Application.class, args); 
    } 

    @StreamListener(Processor.INPUT) 
    @SendTo(Processor.OUTPUT) 
    public String handle(String in) { 
     return in.toUpperCase(); 
    } 

    @Configuration 
    public static class Config { 

     @Bean 
     public BeanPostProcessor channelConfigurer() { 
      return new BeanPostProcessor() { 

       @Override 
       public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { 
        return bean; 
       } 

       @Override 
       public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { 
        if ("input".equals(beanName)) { 
         ((AbstractMessageChannel) bean).addInterceptor(new ChannelInterceptorAdapter() { 

          @Override 
          public Message<?> preSend(Message<?> message, MessageChannel channel) { 
           System.out.println("PreSend on INPUT: " + message); 
           return message; 
          } 

         }); 
        } 
        else if ("output".equals(beanName)) { 
         ((AbstractMessageChannel) bean).addInterceptor(new ChannelInterceptorAdapter() { 

          @Override 
          public Message<?> preSend(Message<?> message, MessageChannel channel) { 
           System.out.println("PreSend on OUTPUT: " + message); 
           return message; 
          } 

         }); 
        } 
        return bean; 
       } 

      }; 
     } 
    } 

}