截获spring云流订阅通道的传入消息

jv2fixgn 于 2021-07-09 发布在 Java

关注(0)|答案(1)|浏览(323)

自方法 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());
    }

}

Java spring-cloud-stream spring-integration spring-messaging

来源：https://stackoverflow.com/questions/41459187/intercept-incoming-message-of-spring-cloud-stream-subscribablechannel

1条答案

按热度按时间

5anewei61#

用一个 DirectChannel 绑定器在同一线程上调用侦听器，因此 preSend 在这里是合适的。
不过，你不必惹麻烦 ThreadLocal ，您可以使用方法签名访问标头。。。

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

编辑

@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;
                }

            };
        }
    }

}

赞(0）回复(0）举报 2021-07-09

我来回答

截获spring云流订阅通道的传入消息

1条答案

相关问题

热门标签

最新问答