private void initRocketMQPushConsumer()

in rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/support/DefaultListenerContainer.java [253:284]

29 lines of code
6 McCabe index (conditional complexity)


    private void initRocketMQPushConsumer() {
        if (rocketMQMessageListener == null) {
            throw new IllegalArgumentException("Property 'rocketMQMessageListener' is required");
        }
        Assert.notNull(consumerGroup, "Property 'consumerGroup' is required");
        Assert.notNull(topic, "Property 'topic' is required");
        Assert.notNull(tag, "Property 'tag' is required");
        FilterExpression filterExpression = null;
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        if (StringUtils.hasLength(this.getTag())) {
            filterExpression = RocketMQUtil.createFilterExpression(this.getTag(),this.getType());
        }
        ClientConfiguration clientConfiguration = RocketMQUtil.createClientConfiguration(this.getAccessKey(), this.getSecretKey(),
                this.getEndpoints(), this.getRequestTimeout(), this.sslEnabled, this.namespace);

        PushConsumerBuilder pushConsumerBuilder = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration);
        // Set the consumer group name.
        if (StringUtils.hasLength(this.getConsumerGroup())) {
            pushConsumerBuilder.setConsumerGroup(this.getConsumerGroup());
        }
        // Set the subscription for the consumer.
        if (StringUtils.hasLength(this.getTopic()) && Objects.nonNull(filterExpression)) {
            pushConsumerBuilder.setSubscriptionExpressions(Collections.singletonMap(this.getTopic(), filterExpression));
        }
        pushConsumerBuilder
                .setConsumptionThreadCount(this.getConsumptionThreadCount())
                .setMaxCacheMessageSizeInBytes(this.getMaxCacheMessageSizeInBytes())
                .setMaxCacheMessageCount(this.getMaxCachedMessageCount())
                .setMessageListener(rocketMQListener);
        this.setPushConsumerBuilder(pushConsumerBuilder);
    }