Перехватить входящее сообщение 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) { 
    // ... 
} 

Ниже моя установка для Spring Облако поток

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

WITH в DirectChannel связующий вызывает ваш слушателя такая же нить, поэтому 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; 
       } 

      }; 
     } 
    } 

}