/*
 * Copyright 2013-present the original author or authors.
 *
 * Licensed 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
 *
 *      https://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 com.alibaba.cloud.stream.binder.rocketmq.properties;

import java.io.Serializable;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.impl.consumer.ConsumeMessageConcurrentlyService;
import org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.jspecify.annotations.Nullable;

/**
 * Extended consumer properties for RocketMQ binder.
 *
 * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
 */
public class RocketMQConsumerProperties extends RocketMQCommonProperties {

	/**
	 * Message model defines the way how messages are delivered to each consumer clients.
	 * This field defaults to clustering.
	 */
	private String messageModel = MessageModel.CLUSTERING.getModeCN();

	/**
	 * Queue allocation algorithm specifying how message queues are allocated to each
	 * consumer clients.
	 */
	private @Nullable String allocateMessageQueueStrategy;

	/**
	 * The expressions include tags or SQL,as follow:
	 * <p>
	 * tag: {@code tag1||tag2||tag3 }; sql: {@code 'color'='blue' AND 'price'>100 } .
	 * </p>
	 * Determines whether there are specific characters "{@code ||}" in the expression to
	 * determine how the message is filtered,tags or SQL.
	 */
	private @Nullable String subscription;

	/**
	 * Delay some time when exception occur .
	 */
	private long pullTimeDelayMillsWhenException = 1000;

	/**
	 * Consuming point on consumer booting.
	 *
	 * There are three consuming points:
	 * <ul>
	 * <li><code>CONSUME_FROM_LAST_OFFSET</code>: consumer clients pick up where it
	 * stopped previously. If it were a newly booting up consumer client, according aging
	 * of the consumer group, there are two cases:
	 * <ol>
	 * <li>if the consumer group is created so recently that the earliest message being
	 * subscribed has yet expired, which means the consumer group represents a lately
	 * launched business, consuming will start from the very beginning;</li>
	 * <li>if the earliest message being subscribed has expired, consuming will start from
	 * the latest messages, meaning messages born prior to the booting timestamp would be
	 * ignored.</li>
	 * </ol>
	 * </li>
	 * <li><code>CONSUME_FROM_FIRST_OFFSET</code>: Consumer client will start from
	 * earliest messages available.</li>
	 * <li><code>CONSUME_FROM_TIMESTAMP</code>: Consumer client will start from specified
	 * timestamp, which means messages born prior to {@link #consumeTimestamp} will be
	 * ignored</li>
	 * </ul>
	 */
	private ConsumeFromWhere consumeFromWhere = ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET;

	/**
	 * Backtracking consumption time with second precision. Time format is
	 * 20131223171201<br>
	 * Implying Seventeen twelve and 01 seconds on December 23, 2013 year<br>
	 * Default backtracking consumption time Half an hour ago.
	 */
	private String consumeTimestamp = UtilAll
			.timeMillisToHumanString3(System.currentTimeMillis() - (1000 * 60 * 30));

	/**
	 * Flow control threshold on queue level, each message queue will cache at most 1000
	 * messages by default, Consider the {@link #pullBatchSize}, the instantaneous value
	 * may exceed the limit .
	 */
	private int pullThresholdForQueue = 1000;

	/**
	 * Limit the cached message size on queue level, each message queue will cache at most
	 * 100 MiB messages by default, Consider the {@link #pullBatchSize}, the instantaneous
	 * value may exceed the limit .
	 *
	 * <p>
	 * The size of a message only measured by message body, so it's not accurate
	 */
	private int pullThresholdSizeForQueue = 100;

	/**
	 * Maximum number of messages pulled each time.
	 */
	private int pullBatchSize = 10;

	/**
	 * Consume max span offset.it has no effect on sequential consumption.
	 */
	private int consumeMaxSpan = 2000;

	private Push push = new Push();

	private Pull pull = new Pull();

	public String getMessageModel() {
		return messageModel;
	}

	public RocketMQConsumerProperties setMessageModel(String messageModel) {
		this.messageModel = messageModel;
		return this;
	}

	public @Nullable String getAllocateMessageQueueStrategy() {
		return allocateMessageQueueStrategy;
	}

	public void setAllocateMessageQueueStrategy(
			@Nullable String allocateMessageQueueStrategy) {
		this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
	}

	public @Nullable String getSubscription() {
		return subscription;
	}

	public void setSubscription(@Nullable String subscription) {
		this.subscription = subscription;
	}

	public Push getPush() {
		return push;
	}

	public void setPush(Push push) {
		this.push = push;
	}

	public long getPullTimeDelayMillsWhenException() {
		return pullTimeDelayMillsWhenException;
	}

	public RocketMQConsumerProperties setPullTimeDelayMillsWhenException(
			long pullTimeDelayMillsWhenException) {
		this.pullTimeDelayMillsWhenException = pullTimeDelayMillsWhenException;
		return this;
	}

	public ConsumeFromWhere getConsumeFromWhere() {
		return consumeFromWhere;
	}

	public RocketMQConsumerProperties setConsumeFromWhere(
			ConsumeFromWhere consumeFromWhere) {
		this.consumeFromWhere = consumeFromWhere;
		return this;
	}

	public String getConsumeTimestamp() {
		return consumeTimestamp;
	}

	public RocketMQConsumerProperties setConsumeTimestamp(String consumeTimestamp) {
		this.consumeTimestamp = consumeTimestamp;
		return this;
	}

	public int getPullThresholdForQueue() {
		return pullThresholdForQueue;
	}

	public RocketMQConsumerProperties setPullThresholdForQueue(
			int pullThresholdForQueue) {
		this.pullThresholdForQueue = pullThresholdForQueue;
		return this;
	}

	public int getPullThresholdSizeForQueue() {
		return pullThresholdSizeForQueue;
	}

	public RocketMQConsumerProperties setPullThresholdSizeForQueue(
			int pullThresholdSizeForQueue) {
		this.pullThresholdSizeForQueue = pullThresholdSizeForQueue;
		return this;
	}

	public int getPullBatchSize() {
		return pullBatchSize;
	}

	public RocketMQConsumerProperties setPullBatchSize(int pullBatchSize) {
		this.pullBatchSize = pullBatchSize;
		return this;
	}

	public Pull getPull() {
		return pull;
	}

	public RocketMQConsumerProperties setPull(Pull pull) {
		this.pull = pull;
		return this;
	}

	public int getConsumeMaxSpan() {
		return consumeMaxSpan;
	}

	public RocketMQConsumerProperties setConsumeMaxSpan(int consumeMaxSpan) {
		this.consumeMaxSpan = consumeMaxSpan;
		return this;
	}

	public static class Push implements Serializable {

		private static final long serialVersionUID = -7398468554978817630L;

		/**
		 * if orderly is true, using {@link MessageListenerOrderly} else if orderly if
		 * false, using {@link MessageListenerConcurrently}.
		 */
		private boolean orderly = false;

		/**
		 * Suspending pulling time for cases requiring slow pulling like flow-control
		 * scenario. see{@link ConsumeMessageOrderlyService#processConsumeResult}.
		 * see{@link ConsumeOrderlyContext#getSuspendCurrentQueueTimeMillis}.
		 */
		private int suspendCurrentQueueTimeMillis = 1000;

		/**
		 * https://github.com/alibaba/spring-cloud-alibaba/issues/1866 Max re-consume
		 * times. -1 means 16 times. If messages are re-consumed more than
		 * {@link #maxReconsumeTimes} before success, it's be directed to a deletion queue
		 * waiting.
		 */
		private int maxReconsumeTimes;

		/**
		 * for concurrently listener. message consume retry strategy. -1 means dlq(or
		 * discard. see {@link ConsumeMessageConcurrentlyService#processConsumeResult}.
		 * see {@link ConsumeConcurrentlyContext#getDelayLevelWhenNextConsume}.
		 */
		private int delayLevelWhenNextConsume = 0;

		/**
		 * Flow control threshold on topic level, default value is -1(Unlimited)
		 * <p>
		 * The value of {@code pullThresholdForQueue} will be overwrote and calculated
		 * based on {@code pullThresholdForTopic} if it is't unlimited
		 * <p>
		 * For example, if the value of pullThresholdForTopic is 1000 and 10 message
		 * queues are assigned to this consumer, then pullThresholdForQueue will be set to
		 * 100.
		 */
		private int pullThresholdForTopic = -1;

		/**
		 * Limit the cached message size on topic level, default value is -1
		 * MiB(Unlimited)
		 * <p>
		 * The value of {@code pullThresholdSizeForQueue} will be overwrote and calculated
		 * based on {@code pullThresholdSizeForTopic} if it is't unlimited .
		 * <p>
		 * For example, if the value of pullThresholdSizeForTopic is 1000 MiB and 10
		 * message queues are assigned to this consumer, then pullThresholdSizeForQueue
		 * will be set to 100 MiB .
		 */
		private int pullThresholdSizeForTopic = -1;

		/**
		 * Message pull Interval.
		 */
		private long pullInterval = 0;

		/**
		 * Batch consumption size.
		 */
		private int consumeMessageBatchMaxSize = 1;

		/**
		 * Maximum amount of time in minutes a message may block the consuming thread.
		 * Unit: Minutes
		 */
		private long consumeTimeout = 15;

		public boolean getOrderly() {
			return orderly;
		}

		public void setOrderly(boolean orderly) {
			this.orderly = orderly;
		}

		public int getSuspendCurrentQueueTimeMillis() {
			return suspendCurrentQueueTimeMillis;
		}

		public void setSuspendCurrentQueueTimeMillis(int suspendCurrentQueueTimeMillis) {
			this.suspendCurrentQueueTimeMillis = suspendCurrentQueueTimeMillis;
		}

		public int getMaxReconsumeTimes() {
			return maxReconsumeTimes;
		}

		public void setMaxReconsumeTimes(int maxReconsumeTimes) {
			this.maxReconsumeTimes = maxReconsumeTimes;
		}

		public int getDelayLevelWhenNextConsume() {
			return delayLevelWhenNextConsume;
		}

		public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
			this.delayLevelWhenNextConsume = delayLevelWhenNextConsume;
		}

		public int getPullThresholdForTopic() {
			return pullThresholdForTopic;
		}

		public void setPullThresholdForTopic(int pullThresholdForTopic) {
			this.pullThresholdForTopic = pullThresholdForTopic;
		}

		public int getPullThresholdSizeForTopic() {
			return pullThresholdSizeForTopic;
		}

		public void setPullThresholdSizeForTopic(int pullThresholdSizeForTopic) {
			this.pullThresholdSizeForTopic = pullThresholdSizeForTopic;
		}

		public long getPullInterval() {
			return pullInterval;
		}

		public void setPullInterval(long pullInterval) {
			this.pullInterval = pullInterval;
		}

		public int getConsumeMessageBatchMaxSize() {
			return consumeMessageBatchMaxSize;
		}

		public void setConsumeMessageBatchMaxSize(int consumeMessageBatchMaxSize) {
			this.consumeMessageBatchMaxSize = consumeMessageBatchMaxSize;
		}

		public long getConsumeTimeout() {
			return consumeTimeout;
		}

		public void setConsumeTimeout(long consumeTimeout) {
			this.consumeTimeout = consumeTimeout;
		}
	}

	public static class Pull implements Serializable {

		/**
		 * The poll timeout in milliseconds.
		 */
		private long pollTimeoutMillis = 1000 * 5;

		/**
		 * Pull thread number.
		 */
		private int pullThreadNums = 20;

		/**
		 * Interval time in in milliseconds for checking changes in topic metadata.
		 */
		private long topicMetadataCheckIntervalMillis = 30 * 1000;

		/**
		 * Long polling mode, the Consumer connection timeout(must greater than
		 * brokerSuspendMaxTimeMillis), it is not recommended to modify.
		 */
		private long consumerTimeoutMillisWhenSuspend = 1000 * 30;

		/**
		 * Ack state handling, including receive, reject, and retry, when a consumption
		 * exception occurs.
		 */
		private @Nullable String errAcknowledge;

		private long pullThresholdForAll = 1000L;

		public long getPollTimeoutMillis() {
			return pollTimeoutMillis;
		}

		public void setPollTimeoutMillis(long pollTimeoutMillis) {
			this.pollTimeoutMillis = pollTimeoutMillis;
		}

		public int getPullThreadNums() {
			return pullThreadNums;
		}

		public void setPullThreadNums(int pullThreadNums) {
			this.pullThreadNums = pullThreadNums;
		}

		public long getTopicMetadataCheckIntervalMillis() {
			return topicMetadataCheckIntervalMillis;
		}

		public void setTopicMetadataCheckIntervalMillis(
				long topicMetadataCheckIntervalMillis) {
			this.topicMetadataCheckIntervalMillis = topicMetadataCheckIntervalMillis;
		}

		public long getConsumerTimeoutMillisWhenSuspend() {
			return consumerTimeoutMillisWhenSuspend;
		}

		public void setConsumerTimeoutMillisWhenSuspend(
				long consumerTimeoutMillisWhenSuspend) {
			this.consumerTimeoutMillisWhenSuspend = consumerTimeoutMillisWhenSuspend;
		}

		public @Nullable String getErrAcknowledge() {
			return errAcknowledge;
		}

		public void setErrAcknowledge(@Nullable String errAcknowledge) {
			this.errAcknowledge = errAcknowledge;
		}

		public long getPullThresholdForAll() {
			return pullThresholdForAll;
		}

		public void setPullThresholdForAll(long pullThresholdForAll) {
			this.pullThresholdForAll = pullThresholdForAll;
		}

	}

}
