rocketmq-v5-client-spring-boot/src/main/java/org/apache/rocketmq/client/support/DefaultListenerContainer.java - rocketmq-spring - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.rocketmq.client.support;

 import org.apache.rocketmq.client.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.client.apis.ClientConfiguration;
 import org.apache.rocketmq.client.apis.ClientServiceProvider;
 import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
 import org.apache.rocketmq.client.apis.consumer.PushConsumer;
 import org.apache.rocketmq.client.apis.consumer.PushConsumerBuilder;
 import org.apache.rocketmq.client.apis.consumer.FilterExpression;
 import org.apache.rocketmq.client.core.RocketMQListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;

 import java.io.IOException;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Objects;


 public class DefaultListenerContainer implements InitializingBean,
         RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware {
     private final static Logger log = LoggerFactory.getLogger(DefaultListenerContainer.class);

     private ApplicationContext applicationContext;

     /**
      * The name of the DefaultRocketMQListenerContainer instance
      */
     private String name;

     private boolean running;

     private PushConsumer pushConsumer;

     private PushConsumerBuilder pushConsumerBuilder;

     private RocketMQListener rocketMQListener;

     private RocketMQMessageListener rocketMQMessageListener;

     String accessKey;

     String secretKey;

     String endpoints;

     String consumerGroup;

     String tag;

     String topic;

     String type;

     FilterExpressionType filterExpressionType;

     Duration requestTimeout;

     int maxCachedMessageCount = 1024;

     int maxCacheMessageSizeInBytes = 67108864;

     int consumptionThreadCount = 20;

     Boolean sslEnabled;

     public String getName() {
         return name;
     }

     public void setName(String name) {
         this.name = name;
     }

     @Override
     public boolean isRunning() {
         return running;
     }

     public void setRunning(boolean running) {
         this.running = running;
     }

     public PushConsumer getPushConsumer() {
         return pushConsumer;
     }

     public void setPushConsumer(PushConsumer pushConsumer) {
         this.pushConsumer = pushConsumer;
     }

     public PushConsumerBuilder getPushConsumerBuilder() {
         return pushConsumerBuilder;
     }

     public void setPushConsumerBuilder(PushConsumerBuilder pushConsumerBuilder) {
         this.pushConsumerBuilder = pushConsumerBuilder;
     }

     public String getAccessKey() {
         return accessKey;
     }

     public void setAccessKey(String accessKey) {
         this.accessKey = accessKey;
     }

     public String getSecretKey() {
         return secretKey;
     }

     public void setSecretKey(String secretKey) {
         this.secretKey = secretKey;
     }

     public String getEndpoints() {
         return endpoints;
     }

     public void setEndpoints(String endpoints) {
         this.endpoints = endpoints;
     }

     public String getConsumerGroup() {
         return consumerGroup;
     }

     public void setConsumerGroup(String consumerGroup) {
         this.consumerGroup = consumerGroup;
     }

     public String getTag() {
         return tag;
     }

     public void setTag(String tag) {
         this.tag = tag;
     }

     public String getTopic() {
         return topic;
     }

     public void setTopic(String topic) {
         this.topic = topic;
     }

     public Duration getRequestTimeout() {
         return requestTimeout;
     }

     public void setRequestTimeout(Duration requestTimeout) {
         this.requestTimeout = requestTimeout;
     }

     public FilterExpressionType getFilterExpressionType() {
         return filterExpressionType;
     }

     public void setFilterExpressionType(FilterExpressionType filterExpressionType) {
         this.filterExpressionType = filterExpressionType;
     }

     public int getMaxCachedMessageCount() {
         return maxCachedMessageCount;
     }

     public void setMaxCachedMessageCount(int maxCachedMessageCount) {
         this.maxCachedMessageCount = maxCachedMessageCount;
     }

     public int getMaxCacheMessageSizeInBytes() {
         return maxCacheMessageSizeInBytes;
     }

     public void setMaxCacheMessageSizeInBytes(int maxCacheMessageSizeInBytes) {
         this.maxCacheMessageSizeInBytes = maxCacheMessageSizeInBytes;
     }

     public int getConsumptionThreadCount() {
         return consumptionThreadCount;
     }

     public void setConsumptionThreadCount(int consumptionThreadCount) {
         this.consumptionThreadCount = consumptionThreadCount;
     }

     public RocketMQListener getMessageListener() {
         return rocketMQListener;
     }

     public void setMessageListener(RocketMQListener rocketMQListener) {
         this.rocketMQListener = rocketMQListener;
     }

     public RocketMQMessageListener getRocketMQMessageListener() {
         return rocketMQMessageListener;
     }

     public void setRocketMQMessageListener(RocketMQMessageListener rocketMQMessageListener) {
         this.rocketMQMessageListener = rocketMQMessageListener;
     }

     public String getType() {
         return type;
     }

     public void setType(String type) {
         this.type = type;
     }

     public Boolean getSslEnabled() {
         return sslEnabled;
     }

     public void setSslEnabled(Boolean sslEnabled) {
         this.sslEnabled = sslEnabled;
     }

     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);

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


     @Override
     public boolean isAutoStartup() {
         return true;
     }

     @Override
     public void stop(Runnable callback) {
         stop();
         callback.run();
     }

     @Override
     public void destroy() throws Exception {
         this.setRunning(false);
         if (Objects.nonNull(pushConsumer)) {
             pushConsumer.close();
         }
         log.info("container destroyed, {}", this.toString());
     }

     @Override
     public void stop() {
         if (this.isRunning()) {
             if (Objects.nonNull(pushConsumer)) {
                 try {
                     pushConsumer.close();
                 } catch (IOException e) {
                     throw new RuntimeException(e);
                 }
             }
             setRunning(false);
         }
     }

     @Override
     public void start() {
         if (this.isRunning()) {
             throw new IllegalStateException("container already running. " + name);
         }
         if (Objects.nonNull(pushConsumer)) {
             throw new IllegalStateException("consumer has been build. " + name);
         }
         try {
             this.pushConsumer = pushConsumerBuilder.build();
         } catch (Exception e) {
             throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
         }
         this.setRunning(true);

         log.info("running container: {}", this.toString());
     }


     @Override
     public void afterPropertiesSet() throws Exception {
         initRocketMQPushConsumer();
     }

     @Override
     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
         this.applicationContext = applicationContext;
     }

     @Override
     public String toString() {
         return "DefaultListenerContainer{" +
                 "name='" + name + '\'' +
                 ", running=" + running +
                 ", accessKey='" + accessKey + '\'' +
                 ", secretKey='" + secretKey + '\'' +
                 ", endpoints='" + endpoints + '\'' +
                 ", consumerGroup='" + consumerGroup + '\'' +
                 ", tag='" + tag + '\'' +
                 ", topic='" + topic + '\'' +
                 ", type='" + type + '\'' +
                 ", filterExpressionType=" + filterExpressionType +
                 ", requestTimeout=" + requestTimeout +
                 ", maxCachedMessageCount=" + maxCachedMessageCount +
                 ", maxCacheMessageSizeInBytes=" + maxCacheMessageSizeInBytes +
                 ", consumptionThreadCount=" + consumptionThreadCount +
                 '}';
     }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.rocketmq.client.support;

	import org.apache.rocketmq.client.annotation.RocketMQMessageListener;
	import org.apache.rocketmq.client.apis.ClientConfiguration;
	import org.apache.rocketmq.client.apis.ClientServiceProvider;
	import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
	import org.apache.rocketmq.client.apis.consumer.PushConsumer;
	import org.apache.rocketmq.client.apis.consumer.PushConsumerBuilder;
	import org.apache.rocketmq.client.apis.consumer.FilterExpression;
	import org.apache.rocketmq.client.core.RocketMQListener;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;
	import org.springframework.beans.BeansException;
	import org.springframework.beans.factory.InitializingBean;
	import org.springframework.context.ApplicationContext;
	import org.springframework.context.ApplicationContextAware;
	import org.springframework.context.SmartLifecycle;
	import org.springframework.util.Assert;
	import org.springframework.util.StringUtils;

	import java.io.IOException;
	import java.time.Duration;
	import java.util.Collections;
	import java.util.Objects;


	public class DefaultListenerContainer implements InitializingBean,
	RocketMQListenerContainer, SmartLifecycle, ApplicationContextAware {
	private final static Logger log = LoggerFactory.getLogger(DefaultListenerContainer.class);

	private ApplicationContext applicationContext;

	/**
	* The name of the DefaultRocketMQListenerContainer instance
	*/
	private String name;

	private boolean running;

	private PushConsumer pushConsumer;

	private PushConsumerBuilder pushConsumerBuilder;

	private RocketMQListener rocketMQListener;

	private RocketMQMessageListener rocketMQMessageListener;

	String accessKey;

	String secretKey;

	String endpoints;

	String consumerGroup;

	String tag;

	String topic;

	String type;

	FilterExpressionType filterExpressionType;

	Duration requestTimeout;

	int maxCachedMessageCount = 1024;

	int maxCacheMessageSizeInBytes = 67108864;

	int consumptionThreadCount = 20;

	Boolean sslEnabled;

	public String getName() {
	return name;
	}

	public void setName(String name) {
	this.name = name;
	}

	@Override
	public boolean isRunning() {
	return running;
	}

	public void setRunning(boolean running) {
	this.running = running;
	}

	public PushConsumer getPushConsumer() {
	return pushConsumer;
	}

	public void setPushConsumer(PushConsumer pushConsumer) {
	this.pushConsumer = pushConsumer;
	}

	public PushConsumerBuilder getPushConsumerBuilder() {
	return pushConsumerBuilder;
	}

	public void setPushConsumerBuilder(PushConsumerBuilder pushConsumerBuilder) {
	this.pushConsumerBuilder = pushConsumerBuilder;
	}

	public String getAccessKey() {
	return accessKey;
	}

	public void setAccessKey(String accessKey) {
	this.accessKey = accessKey;
	}

	public String getSecretKey() {
	return secretKey;
	}

	public void setSecretKey(String secretKey) {
	this.secretKey = secretKey;
	}

	public String getEndpoints() {
	return endpoints;
	}

	public void setEndpoints(String endpoints) {
	this.endpoints = endpoints;
	}

	public String getConsumerGroup() {
	return consumerGroup;
	}

	public void setConsumerGroup(String consumerGroup) {
	this.consumerGroup = consumerGroup;
	}

	public String getTag() {
	return tag;
	}

	public void setTag(String tag) {
	this.tag = tag;
	}

	public String getTopic() {
	return topic;
	}

	public void setTopic(String topic) {
	this.topic = topic;
	}

	public Duration getRequestTimeout() {
	return requestTimeout;
	}

	public void setRequestTimeout(Duration requestTimeout) {
	this.requestTimeout = requestTimeout;
	}

	public FilterExpressionType getFilterExpressionType() {
	return filterExpressionType;
	}

	public void setFilterExpressionType(FilterExpressionType filterExpressionType) {
	this.filterExpressionType = filterExpressionType;
	}

	public int getMaxCachedMessageCount() {
	return maxCachedMessageCount;
	}

	public void setMaxCachedMessageCount(int maxCachedMessageCount) {
	this.maxCachedMessageCount = maxCachedMessageCount;
	}

	public int getMaxCacheMessageSizeInBytes() {
	return maxCacheMessageSizeInBytes;
	}

	public void setMaxCacheMessageSizeInBytes(int maxCacheMessageSizeInBytes) {
	this.maxCacheMessageSizeInBytes = maxCacheMessageSizeInBytes;
	}

	public int getConsumptionThreadCount() {
	return consumptionThreadCount;
	}

	public void setConsumptionThreadCount(int consumptionThreadCount) {
	this.consumptionThreadCount = consumptionThreadCount;
	}

	public RocketMQListener getMessageListener() {
	return rocketMQListener;
	}

	public void setMessageListener(RocketMQListener rocketMQListener) {
	this.rocketMQListener = rocketMQListener;
	}

	public RocketMQMessageListener getRocketMQMessageListener() {
	return rocketMQMessageListener;
	}

	public void setRocketMQMessageListener(RocketMQMessageListener rocketMQMessageListener) {
	this.rocketMQMessageListener = rocketMQMessageListener;
	}

	public String getType() {
	return type;
	}

	public void setType(String type) {
	this.type = type;
	}

	public Boolean getSslEnabled() {
	return sslEnabled;
	}

	public void setSslEnabled(Boolean sslEnabled) {
	this.sslEnabled = sslEnabled;
	}

	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);

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


	@Override
	public boolean isAutoStartup() {
	return true;
	}

	@Override
	public void stop(Runnable callback) {
	stop();
	callback.run();
	}

	@Override
	public void destroy() throws Exception {
	this.setRunning(false);
	if (Objects.nonNull(pushConsumer)) {
	pushConsumer.close();
	}
	log.info("container destroyed, {}", this.toString());
	}

	@Override
	public void stop() {
	if (this.isRunning()) {
	if (Objects.nonNull(pushConsumer)) {
	try {
	pushConsumer.close();
	} catch (IOException e) {
	throw new RuntimeException(e);
	}
	}
	setRunning(false);
	}
	}

	@Override
	public void start() {
	if (this.isRunning()) {
	throw new IllegalStateException("container already running. " + name);
	}
	if (Objects.nonNull(pushConsumer)) {
	throw new IllegalStateException("consumer has been build. " + name);
	}
	try {
	this.pushConsumer = pushConsumerBuilder.build();
	} catch (Exception e) {
	throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
	}
	this.setRunning(true);

	log.info("running container: {}", this.toString());
	}


	@Override
	public void afterPropertiesSet() throws Exception {
	initRocketMQPushConsumer();
	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
	this.applicationContext = applicationContext;
	}

	@Override
	public String toString() {
	return "DefaultListenerContainer{" +
	"name='" + name + '\'' +
	", running=" + running +
	", accessKey='" + accessKey + '\'' +
	", secretKey='" + secretKey + '\'' +
	", endpoints='" + endpoints + '\'' +
	", consumerGroup='" + consumerGroup + '\'' +
	", tag='" + tag + '\'' +
	", topic='" + topic + '\'' +
	", type='" + type + '\'' +
	", filterExpressionType=" + filterExpressionType +
	", requestTimeout=" + requestTimeout +
	", maxCachedMessageCount=" + maxCachedMessageCount +
	", maxCacheMessageSizeInBytes=" + maxCacheMessageSizeInBytes +
	", consumptionThreadCount=" + consumptionThreadCount +
	'}';
	}
	}