Polish implement createPolledConsumerResources method of binder
@@ -34,6 +34,7 @@ import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder;
import org.springframework.cloud.stream.binder.rocketmq.consuming.RocketMQListenerBindingContainer;
import org.springframework.cloud.stream.binder.rocketmq.integration.RocketMQInboundChannelAdapter;
import org.springframework.cloud.stream.binder.rocketmq.integration.RocketMQMessageHandler;
+import org.springframework.cloud.stream.binder.rocketmq.integration.RocketMQMessageSource;
import org.springframework.cloud.stream.binder.rocketmq.metrics.InstrumentationManager;
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
import org.springframework.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
@@ -206,6 +207,17 @@ public class RocketMQMessageChannelBinder extends
return rocketInboundChannelAdapter;
}
+ @Override
+ protected PolledConsumerResources createPolledConsumerResources(String name,
+ String group, ConsumerDestination destination,
+ ExtendedConsumerProperties<RocketMQConsumerProperties> consumerProperties) {
+ RocketMQMessageSource rocketMQMessageSource = new RocketMQMessageSource(
+ rocketBinderConfigurationProperties, consumerProperties, name, group);
+ return new PolledConsumerResources(rocketMQMessageSource,
+ registerErrorInfrastructure(destination, group, consumerProperties,
+ true));
+ }
+
@Override
public RocketMQConsumerProperties getExtendedConsumerProperties(String channelName) {
return extendedBindingProperties.getExtendedConsumerProperties(channelName);
