/*
 * 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.integration.inbound;

import java.util.Objects;

import com.alibaba.cloud.stream.binder.rocketmq.custom.RocketMQBeanContainerCache;
import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
import com.alibaba.cloud.stream.binder.rocketmq.utils.RocketMQUtils;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

/**
 * Extended function related to producer . eg:initial
 *
 * @author zkzlx
 */
public final class RocketMQConsumerFactory {

	private RocketMQConsumerFactory() {
	}

	private final static Logger log = LoggerFactory
			.getLogger(RocketMQConsumerFactory.class);

	public static DefaultMQPushConsumer initPushConsumer(
			ExtendedConsumerProperties<RocketMQConsumerProperties> extendedConsumerProperties) {
		RocketMQConsumerProperties consumerProperties = extendedConsumerProperties
				.getExtension();
		Assert.notNull(consumerProperties.getGroup(),
				"Property 'group' is required - consumerGroup");
		Assert.notNull(consumerProperties.getNameServer(),
				"Property 'nameServer' is required");
		AllocateMessageQueueStrategy allocateMessageQueueStrategy = RocketMQBeanContainerCache
				.getBean(consumerProperties.getAllocateMessageQueueStrategy(),
						AllocateMessageQueueStrategy.class,
						new AllocateMessageQueueAveragely());
		RPCHook rpcHook = null;
		if (StringUtils.hasLength(consumerProperties.getAccessKey())
				&& StringUtils.hasLength(consumerProperties.getSecretKey())) {
			rpcHook = new AclClientRPCHook(
					new SessionCredentials(consumerProperties.getAccessKey(),
							consumerProperties.getSecretKey()));
		}
		RPCHook actualRpcHook = rpcHook != null ? rpcHook
				: new AclClientRPCHook(new SessionCredentials("", ""));
		String group = Objects.requireNonNull(consumerProperties.getGroup());
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group,
				actualRpcHook, allocateMessageQueueStrategy,
				consumerProperties.getEnableMsgTrace(),
				consumerProperties.getCustomizedTraceTopic());
		consumer.setVipChannelEnabled(
				null == rpcHook && consumerProperties.getVipChannelEnabled());
		consumer.setInstanceName(
				RocketMQUtils.getInstanceName(actualRpcHook, group));
		consumer.setNamespace(consumerProperties.getNamespace());
		consumer.setNamespaceV2(consumerProperties.getNamespaceV2());
		consumer.setNamesrvAddr(consumerProperties.getNameServer());
		consumer.setMessageModel(getMessageModel(consumerProperties.getMessageModel()));
		consumer.setUseTLS(consumerProperties.getUseTLS());
		consumer.setPullTimeDelayMillsWhenException(
				consumerProperties.getPullTimeDelayMillsWhenException());
		consumer.setPullBatchSize(consumerProperties.getPullBatchSize());
		consumer.setConsumeFromWhere(consumerProperties.getConsumeFromWhere());
		consumer.setHeartbeatBrokerInterval(
				consumerProperties.getHeartbeatBrokerInterval());
		consumer.setPersistConsumerOffsetInterval(
				consumerProperties.getPersistConsumerOffsetInterval());
		consumer.setPullInterval(consumerProperties.getPush().getPullInterval());
		consumer.setConsumeThreadMin(extendedConsumerProperties.getConcurrency());
		consumer.setConsumeThreadMax(extendedConsumerProperties.getConcurrency());
		consumer.setUnitName(consumerProperties.getUnitName());
		consumer.setMaxReconsumeTimes(
				consumerProperties.getPush().getMaxReconsumeTimes());
		consumer.setConsumeTimeout(consumerProperties.getPush().getConsumeTimeout());
		consumer.setAccessChannel(AccessChannel.valueOf(consumerProperties.getAccessChannel()));
		return consumer;
	}

	/**
	 * todo Compatible with versions less than 4.6 ?
	 * @param topic consumer topic.
	 * @param extendedConsumerProperties extendedConsumerProperties
	 * @return DefaultLitePullConsumer
	 */
	public static DefaultLitePullConsumer initPullConsumer(
			String topic,
			ExtendedConsumerProperties<RocketMQConsumerProperties> extendedConsumerProperties) {
		RocketMQConsumerProperties consumerProperties = extendedConsumerProperties
				.getExtension();
		boolean anonymous = !StringUtils.hasLength(consumerProperties.getGroup());
		/***
		 * 	When using DLQ, at least the group property must be provided for proper naming of the DLQ destination
		 *  According to https://docs.spring.io/spring-cloud-stream/docs/3.2.1/reference/html/spring-cloud-stream.html#spring-cloud-stream-reference
		 */
		if (anonymous && NamespaceUtil.isDLQTopic(topic)) {
			throw new RuntimeException(
					"group must be configured for DLQ" + topic);
		}
		if (anonymous) {
			consumerProperties.setGroup(RocketMQUtils.anonymousGroup(topic));
		}
		String consumerGroup = Objects.requireNonNull(consumerProperties.getGroup());

		Assert.notNull(consumerProperties.getNameServer(),
				"Property 'nameServer' is required");
		AllocateMessageQueueStrategy allocateMessageQueueStrategy = RocketMQBeanContainerCache
				.getBean(consumerProperties.getAllocateMessageQueueStrategy(),
						AllocateMessageQueueStrategy.class);

		RPCHook rpcHook = null;
		if (StringUtils.hasLength(consumerProperties.getAccessKey())
				&& StringUtils.hasLength(consumerProperties.getSecretKey())) {
			rpcHook = new AclClientRPCHook(
					new SessionCredentials(consumerProperties.getAccessKey(),
							consumerProperties.getSecretKey()));
		}
		RPCHook actualRpcHook = rpcHook != null ? rpcHook
				: new AclClientRPCHook(new SessionCredentials("", ""));

		DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(
				consumerProperties.getNamespace(), consumerGroup, actualRpcHook);
		consumer.setVipChannelEnabled(
				null == rpcHook && consumerProperties.getVipChannelEnabled());
		consumer.setInstanceName(
				RocketMQUtils.getInstanceName(actualRpcHook, consumerGroup));
		if (null != allocateMessageQueueStrategy) {
			consumer.setAllocateMessageQueueStrategy(allocateMessageQueueStrategy);
		}
		consumer.setNamesrvAddr(consumerProperties.getNameServer());
		consumer.setMessageModel(getMessageModel(consumerProperties.getMessageModel()));
		consumer.setNamespaceV2(consumerProperties.getNamespaceV2());
		consumer.setUseTLS(consumerProperties.getUseTLS());
		consumer.setPullTimeDelayMillsWhenException(
				consumerProperties.getPullTimeDelayMillsWhenException());
		consumer.setConsumerTimeoutMillisWhenSuspend(
				consumerProperties.getPull().getConsumerTimeoutMillisWhenSuspend());
		consumer.setPullBatchSize(consumerProperties.getPullBatchSize());
		consumer.setConsumeFromWhere(consumerProperties.getConsumeFromWhere());
		consumer.setHeartbeatBrokerInterval(
				consumerProperties.getHeartbeatBrokerInterval());
		consumer.setPersistConsumerOffsetInterval(
				consumerProperties.getPersistConsumerOffsetInterval());
		consumer.setPollTimeoutMillis(
				consumerProperties.getPull().getPollTimeoutMillis());
		consumer.setPullThreadNums(extendedConsumerProperties.getConcurrency());
		// The internal queues are cached by a maximum of 1000
		consumer.setPullThresholdForAll(extendedConsumerProperties.getExtension()
				.getPull().getPullThresholdForAll());
		consumer.setUnitName(consumerProperties.getUnitName());
		consumer.setAccessChannel(AccessChannel.valueOf(consumerProperties.getAccessChannel()));
		return consumer;
	}

	private static MessageModel getMessageModel(String messageModel) {
		for (MessageModel model : MessageModel.values()) {
			if (model.getModeCN().equalsIgnoreCase(messageModel)) {
				return model;
			}
		}
		return MessageModel.CLUSTERING;
	}

}
