/*
 * Copyright 2020-2023 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 org.springframework.cloud.stream.function;

import java.lang.reflect.Type;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.function.Function;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.FunctionRegistration;
import org.springframework.cloud.function.context.FunctionRegistry;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.PassThruFunction;
import org.springframework.cloud.function.context.message.MessageUtils;
import org.springframework.cloud.function.core.FunctionInvocationHelper;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.cloud.stream.binding.BindingService;
import org.springframework.cloud.stream.binding.NewDestinationBindingCallback;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.messaging.DirectWithAttributesChannel;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.ResolvableType;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.config.GlobalChannelInterceptorProcessor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.Assert;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

/**
 * A class which allows user to send data to an output binding.
 * While in a common scenario of a typical spring-cloud-stream application user rarely
 * has to manually send data, there are times when the sources of data are outside of
 * spring-cloud-stream context and therefore we need to bridge such foreign sources
 * with spring-cloud-stream.
 * <br><br>
 * This utility class allows user to do just that - <i>bridge non-spring-cloud-stream applications
 * with spring-cloud-stream</i> by providing a mechanism (bridge) to send data to an output binding while
 * maintaining the  same invocation contract (i.e., type conversion, partitioning etc) as if it was
 * done through a declared function.
 *
 * @author Oleg Zhurakousky
 * @author Soby Chacko
 * @author Byungjun You
 * @since 3.0.3
 *
 */
@SuppressWarnings("rawtypes")
public final class StreamBridge implements StreamOperations, SmartInitializingSingleton {

	private static final String STREAM_BRIDGE_FUNC_NAME = "streamBridge";

	private final Log logger = LogFactory.getLog(getClass());

	private final Map<String, MessageChannel> channelCache;

	private final FunctionCatalog functionCatalog;

	private final NewDestinationBindingCallback destinationBindingCallback;

	private final BindingServiceProperties bindingServiceProperties;

	private final ConfigurableApplicationContext applicationContext;

	private boolean initialized;

	private final BindingService bindingService;

	private final Map<Integer, FunctionInvocationWrapper> streamBridgeFunctionCache;

	private final FunctionInvocationHelper<?> functionInvocationHelper;

	/**
	 *
	 * @param functionCatalog instance of {@link FunctionCatalog}
	 * @param bindingServiceProperties instance of {@link BindingServiceProperties}
	 * @param applicationContext instance of {@link ConfigurableApplicationContext}
	 */
	@SuppressWarnings("serial")
	StreamBridge(FunctionCatalog functionCatalog, BindingServiceProperties bindingServiceProperties,
		ConfigurableApplicationContext applicationContext, @Nullable NewDestinationBindingCallback destinationBindingCallback) {
		Assert.notNull(functionCatalog, "'functionCatalog' must not be null");
		Assert.notNull(applicationContext, "'applicationContext' must not be null");
		Assert.notNull(bindingServiceProperties, "'bindingServiceProperties' must not be null");
		this.bindingService = applicationContext.getBean(BindingService.class);
		this.functionCatalog = functionCatalog;
		this.applicationContext = applicationContext;
		this.bindingServiceProperties = bindingServiceProperties;
		this.destinationBindingCallback = destinationBindingCallback;
		this.channelCache = new LinkedHashMap<String, MessageChannel>() {
			@Override
			protected boolean removeEldestEntry(Map.Entry<String, MessageChannel> eldest) {
				boolean remove = size() > bindingServiceProperties.getDynamicDestinationCacheSize();
				if (remove) {
					if (logger.isDebugEnabled()) {
						logger.debug("Removing message channel from cache " + eldest.getKey());
					}
					bindingService.unbindProducers(eldest.getKey());
				}
				return remove;
			}
		};
		this.functionInvocationHelper = applicationContext.getBean(FunctionInvocationHelper.class);
		this.streamBridgeFunctionCache = new HashMap<>();
	}

	@Override
	public boolean send(String bindingName, Object data) {
		BindingProperties bindingProperties = this.bindingServiceProperties.getBindingProperties(bindingName);
		MimeType contentType = StringUtils.hasText(bindingProperties.getContentType()) ? MimeType.valueOf(bindingProperties.getContentType()) : MimeTypeUtils.APPLICATION_JSON;
		return this.send(bindingName, data, contentType);
	}

	@Override
	public boolean send(String bindingName, Object data, MimeType outputContentType) {
		return this.send(bindingName, null, data, outputContentType);
	}
	@Override
	public boolean send(String bindingName, @Nullable String binderName, Object data) {
		return this.send(bindingName, binderName, data, MimeTypeUtils.APPLICATION_JSON);
	}

	@Override
	@SuppressWarnings({ "unchecked"})
	public boolean send(String bindingName, @Nullable String binderName, Object data, MimeType outputContentType) {
		if (!this.initialized) {
			this.afterSingletonsInstantiated();
		}
		ProducerProperties producerProperties = this.bindingServiceProperties.getProducerProperties(bindingName);
		MessageChannel messageChannel = this.resolveDestination(bindingName, producerProperties, binderName);

		Function functionToInvoke = this.getStreamBridgeFunction(outputContentType.toString(), producerProperties);

		if (producerProperties != null && producerProperties.isPartitioned()) {
			functionToInvoke = new PartitionAwareFunctionWrapper(functionToInvoke, this.applicationContext, producerProperties);
		}

		String targetType = this.resolveBinderTargetType(bindingName, binderName, MessageChannel.class,
			this.applicationContext.getBean(BinderFactory.class));

		Message<?> messageToSend = data instanceof Message messageData
				? MessageBuilder.fromMessage(messageData).setHeaderIfAbsent(MessageUtils.TARGET_PROTOCOL, targetType).build()
						: new GenericMessage<>(data, Collections.singletonMap(MessageUtils.TARGET_PROTOCOL, targetType));

		Message<?> resultMessage;
		synchronized (this) {
			resultMessage = (Message<byte[]>) functionToInvoke.apply(messageToSend);
		}

		if (resultMessage == null
			&& ((Message) messageToSend).getPayload().getClass().getName().equals("org.springframework.kafka.support.KafkaNull")) {
			resultMessage = messageToSend;
		}

		resultMessage = (Message<?>) this.functionInvocationHelper.postProcessResult(resultMessage, null);

		return messageChannel.send(resultMessage);
	}

	private int hashProducerProperties(ProducerProperties producerProperties, String outputContentType) {
		int hash = outputContentType.hashCode()
				+ Boolean.hashCode(producerProperties.isUseNativeEncoding())
				+ Boolean.hashCode(producerProperties.isPartitioned())
				+ producerProperties.getPartitionCount();

		return hash;
	}

	private synchronized FunctionInvocationWrapper getStreamBridgeFunction(String outputContentType, ProducerProperties producerProperties) {
		int streamBridgeFunctionKey = this.hashProducerProperties(producerProperties, outputContentType);
		if (this.streamBridgeFunctionCache.containsKey(streamBridgeFunctionKey)) {
			return this.streamBridgeFunctionCache.get(streamBridgeFunctionKey);
		}
		else {
			FunctionInvocationWrapper functionToInvoke = this.functionCatalog.lookup(STREAM_BRIDGE_FUNC_NAME, outputContentType.toString());
			this.streamBridgeFunctionCache.put(streamBridgeFunctionKey, functionToInvoke);
			functionToInvoke.setSkipOutputConversion(producerProperties.isUseNativeEncoding());
			return functionToInvoke;
		}
	}

	@Override
	public void afterSingletonsInstantiated() {
		if (this.initialized) {
			return;
		}

		FunctionRegistration<Function<Object, Object>> fr = new FunctionRegistration<>(new PassThruFunction(), STREAM_BRIDGE_FUNC_NAME);
		fr.getProperties().put("singleton", "false");

		Type functionType = ResolvableType.forClassWithGenerics(Function.class, Object.class, Object.class).getType();
		((FunctionRegistry) this.functionCatalog).register(fr.type(functionType));
		this.initialized = true;
	}

	@SuppressWarnings({ "unchecked"})
	synchronized MessageChannel resolveDestination(String destinationName, ProducerProperties producerProperties, String binderName) {
		MessageChannel messageChannel = null;
		if (StringUtils.hasText(binderName)) {
			messageChannel = this.channelCache.get(binderName + ":" + destinationName);
		}
		else {
			messageChannel = this.channelCache.get(destinationName);
		}
		if (messageChannel == null) {
			if (this.applicationContext.containsBean(destinationName)) {
				messageChannel = this.applicationContext.getBean(destinationName, MessageChannel.class);
				String[] consumerBindingNames = this.bindingService.getConsumerBindingNames();
				if (ObjectUtils.containsElement(consumerBindingNames, destinationName)) { //GH-2563
					logger.warn("You seem to be sending data to the input binding.  It is not "
							+ "recommended, since you are bypassing the binder and this the messaging system exposed by the binder.");
				}
			}
			else {
				messageChannel = new DirectWithAttributesChannel();
				if (this.destinationBindingCallback != null) {
					Object extendedProducerProperties = this.bindingService
							.getExtendedProducerProperties(messageChannel, destinationName);
					this.destinationBindingCallback.configure(destinationName, messageChannel,
							producerProperties, extendedProducerProperties);
				}

				Binder binder = null;
				if (StringUtils.hasText(binderName)) {
					BinderFactory binderFactory = this.applicationContext.getBean(BinderFactory.class);
					binder = binderFactory.getBinder(binderName, messageChannel.getClass());
				}
				// Commenting out the following block due to this issue: https://github.com/spring-cloud/spring-cloud-stream/issues/2759
				// Once we confirm that there is no unknown regression, we will remove this block from here completely,
				// since we already perform the partition finding algorithm once via StreamBridge#send.
//				if (producerProperties != null && producerProperties.isPartitioned()) {
//					BindingProperties bindingProperties = this.bindingServiceProperties.getBindingProperties(destinationName);
//					((AbstractMessageChannel) messageChannel)
//						.addInterceptor(new DefaultPartitioningInterceptor(bindingProperties, this.applicationContext.getBeanFactory()));
//				}
				this.addInterceptors((AbstractMessageChannel) messageChannel, destinationName);

				this.bindingService.bindProducer(messageChannel, destinationName, true, binder);
				if (StringUtils.hasText(binderName)) {
					this.channelCache.put(binderName + ":" + destinationName, messageChannel);
				}
				else {
					this.channelCache.put(destinationName, messageChannel);
				}
			}
		}

		return messageChannel;
	}

	private String resolveBinderTargetType(String channelName, String binderName, Class<?> bindableType, BinderFactory binderFactory) {
		String binderConfigurationName = binderName != null ? binderName : this.bindingServiceProperties
				.getBinder(channelName);
		Binder<?, ?, ?> binder = binderFactory.getBinder(binderConfigurationName, bindableType);
		String targetProtocol = binder.getClass().getSimpleName().startsWith("Rabbit") ? "amqp" : "kafka";
		return targetProtocol;
	}

	private void addInterceptors(AbstractMessageChannel messageChannel, String destinationName) {
		final GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor =
			this.applicationContext.getBean(GlobalChannelInterceptorProcessor.class);
		globalChannelInterceptorProcessor.postProcessAfterInitialization(messageChannel, destinationName);
	}
}
