删除动态生成的目标中的嵌入标头

Question

我想在使用动态生成的目的地时禁用消息中的嵌入标头，类似于来自 here 的示例（例如输出主题 = dyntopic1,dyntopic2,...）。

我已经设置了如下属性，但我仍然得到标题，如果我错过了什么，有什么建议吗？

spring.cloud.stream.bindings.output.group=test-ogroup
spring.cloud.stream.bindings.output.binder=kafka
spring.cloud.stream.bindings.output.producer.headerMode=raw
spring.cloud.stream.bindings.output.content-type=text/plain

Kafka = v0.10

spring-cloud-dependencies.version = Edgware.M1

Answer 1

Spring Cloud Stream 1.3 或更低版本无法为动态目的地设置生产者属性。

如果提前知道属性，可以在属性中设置...

spring.cloud.stream.bindings.dyntopic1.producer.headerMode=raw

该功能已被 added to master 提供，并将在 2.0 版本中提供。

编辑

只要您不介意使用反射来重置标志，就可以使用 Edgware 来完成。您必须替换通道解析器 bean。

我使用 Edgware.SR1 对此进行了测试 - 你真的不应该再使用 M1，这是一个预发布里程碑。

我不能保证这将适用于较新的版本，因为它会破坏框架内部结构。

@SpringBootApplication
@EnableBinding(Sink.class)
public class So48543143Application {

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

    @Bean
    public ApplicationRunner runner(MessageChannel routeChannel) {
        return args -> {
            routeChannel.send(new GenericMessage<>("foo"));
        };
    }

    @ServiceActivator(inputChannel = "routeChannel")
    @Bean
    public AbstractMappingMessageRouter router(MyBinderAwareChannelResolver resolver) {
        ExpressionEvaluatingRouter router = new ExpressionEvaluatingRouter(new LiteralExpression("foo"));
        router.setDefaultOutputChannelName("default");
        router.setChannelResolver(resolver);
        return router;
    }

    @Bean
    public MyBinderAwareChannelResolver binderAwareChannelResolver(BindingService bindingService,
            AbstractBindingTargetFactory<? extends MessageChannel> bindingTargetFactory,
            DynamicDestinationsBindable dynamicDestinationsBindable) {
        return new MyBinderAwareChannelResolver(bindingService, bindingTargetFactory, dynamicDestinationsBindable);
    }

    public static class MyBinderAwareChannelResolver extends BinderAwareChannelResolver {

        public MyBinderAwareChannelResolver(BindingService bindingService,
                AbstractBindingTargetFactory<? extends MessageChannel> bindingTargetFactory,
                DynamicDestinationsBindable dynamicDestinationsBindable) {
            super(bindingService, bindingTargetFactory, dynamicDestinationsBindable);
        }

        @Override
        public MessageChannel resolveDestination(String channelName) {
            MessageChannel channel = super.resolveDestination(channelName);
            DirectFieldAccessor dfa = new DirectFieldAccessor(channel);
            AbstractDispatcher dispatcher = (AbstractDispatcher) dfa.getPropertyValue("dispatcher");
            dfa = new DirectFieldAccessor(dispatcher);
            @SuppressWarnings("unchecked")
            Set<MessageHandler> handlers = (Set<MessageHandler>) dfa.getPropertyValue("handlers");
            // there should be exactly one handler
            MessageHandler handler = handlers.iterator().next();
            dfa = new DirectFieldAccessor(handler);
            dfa.setPropertyValue("embedHeaders", false);
            return channel;
        }

    }

}