content: /*
   * Copyright 2017-2025 Hyland Software, Inc. and its affiliates.
   *
   * 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
   *
   *     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.activiti.cloud.common.messaging.config.test;
  
  import static org.activiti.cloud.common.messaging.config.AbstractFunctionalBindingConfiguration.getInBinding;
  import static org.activiti.cloud.common.messaging.config.InputBindingConfiguration.INPUT_BINDING;
  import static org.activiti.cloud.common.messaging.config.OutputBindingConfiguration.OUTPUT_BINDING;
  import static org.activiti.cloud.common.messaging.config.test.TestBindingsChannels.AUDIT_CONSUMER;
  import static org.activiti.cloud.common.messaging.config.test.TestBindingsChannels.COMMAND_CONSUMER;
  import static org.activiti.cloud.common.messaging.config.test.TestBindingsChannels.QUERY_CONSUMER;
  import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
  import static org.awaitility.Awaitility.await;
  import static org.springframework.cloud.function.context.FunctionRegistration.REGISTRATION_NAME_SUFFIX;
  
  import java.util.Arrays;
  import java.util.function.Consumer;
  import java.util.function.Function;
  import org.activiti.cloud.common.messaging.config.FunctionBindingConfiguration.BindingResolver;
  import org.activiti.cloud.common.messaging.config.FunctionBindingPropertySource;
  import org.activiti.cloud.common.messaging.functional.FunctionBinding;
  import org.assertj.core.api.Assertions;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Test;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.boot.test.context.SpringBootTest;
  import org.springframework.boot.test.context.TestConfiguration;
  import org.springframework.cloud.function.context.FunctionRegistry;
  import org.springframework.cloud.stream.binder.test.InputDestination;
  import org.springframework.cloud.stream.binder.test.OutputDestination;
  import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
  import org.springframework.cloud.stream.config.BindingServiceProperties;
  import org.springframework.cloud.stream.function.StreamFunctionProperties;
  import org.springframework.context.ConfigurableApplicationContext;
  import org.springframework.context.annotation.Bean;
  import org.springframework.context.annotation.Import;
  import org.springframework.messaging.Message;
  import org.springframework.messaging.MessageChannel;
  import org.springframework.messaging.support.MessageBuilder;
  
  @SpringBootTest(
      properties = {
          "activiti.cloud.application.name=foo",
          "spring.application.name=bar",
          "activiti.cloud.messaging.destination-transformers-enabled=false",
          "spring.cloud.stream.bindings.commandConsumer.destination=commandConsumer",
          "spring.cloud.stream.bindings.commandConsumer.group=${spring.application.name}",
          "spring.cloud.stream.bindings.auditProducer.destination=engineEvents",
          "spring.cloud.stream.bindings.auditConsumer.destination=engineEvents",
          "spring.cloud.stream.bindings.queryConsumer.destination=engineEvents",
          "spring.cloud.stream.bindings.commandResults.destination=commandResults",
      }
  )
  @Import({ TestChannelBinderConfiguration.class, TestBindingsChannelsConfiguration.class })
  public class FunctionBindingConfigurationIT {
  
      private static final String FUNCTION_HANDLER_NAME = "queryConsumerHandler";
      private static final String FUNCTION_PROCESSOR_NAME = "commandProcessorHandler";
      private static final String FUNCTION_AUDIT_SUPPLIER_NAME = "auditProducer" + OUTPUT_BINDING;
      private static final String FUNCTION_COMMAND_SUPPLIER_NAME = "commandResults" + OUTPUT_BINDING;
      private static final String FUNCTION_COMMAND_CONSUMER_NAME = "commandConsumer" + INPUT_BINDING;
      private static final String FUNCTION_AUDIT_CONSUMER_NAME = "auditConsumer" + INPUT_BINDING;
      private static final String FUNCTION_QUERY_CONSUMER_NAME = "queryConsumer" + INPUT_BINDING;
  
      private static Message<?> consumerMessage = null;
  
      @Autowired
      private TestBindingsChannels channels;
  
      @Autowired
      private FunctionBindingPropertySource functionBindingPropertySource;
  
      @Autowired
      private FunctionRegistry functionRegistry;
  
      @Autowired
      private StreamFunctionProperties streamFunctionProperties;
  
      @Autowired
      private BindingResolver bindingResolver;
  
      @Autowired
      private BindingServiceProperties bindingServiceProperties;
  
      @Autowired
      private ConfigurableApplicationContext context;
  
      @Autowired
      private InputDestination input;
  
      @Autowired
      private OutputDestination output;
  
      @TestConfiguration
      static class ApplicationConfig {
  
          @Bean(FUNCTION_HANDLER_NAME)
          @FunctionBinding(input = QUERY_CONSUMER)
          public Consumer<Message<?>> queryConsumerHandler() {
              return message -> {
                  consumerMessage = message;
              };
          }
  
          @Bean(FUNCTION_PROCESSOR_NAME)
          @FunctionBinding(input = COMMAND_CONSUMER, output = TestBindingsChannels.COMMAND_RESULTS)
          public Function<Message<?>, Message<?>> commandProcessorHandler(TestBindingsChannels channels) {
              return message -> {
                  assertThat(message).isNotNull();
                  Message outMessage = MessageBuilder
                      .withPayload(message.getPayload())
                      .setHeader("type", "Test Send")
                      .build();
                  channels.auditProducer().send(outMessage);
                  return MessageBuilder.withPayload(message.getPayload()).setHeader("type", "Test Reply").build();
              };
          }
      }
  
      @BeforeEach
      public void setUp() {
          consumerMessage = null;
          output.clear();
      }
  
      @Test
      public void testFunctionDefinitions() {
          // given
          String functionDefinitions = (String) functionBindingPropertySource.getProperty(
              FunctionBindingPropertySource.SPRING_CLOUD_FUNCTION_DEFINITION
          );
  
          assertThat(functionDefinitions).isNotNull();
  
          String[] functions = functionDefinitions.split(";");
  
          // then
          assertThat(functions)
              .doesNotContain(
                  FUNCTION_AUDIT_SUPPLIER_NAME,
                  FUNCTION_COMMAND_SUPPLIER_NAME,
                  FUNCTION_HANDLER_NAME,
                  FUNCTION_PROCESSOR_NAME
              );
      }
  
      @Test
      public void testOutputBindingsDefinitions() {
          // then
          assertThat(context.getBean(TestBindingsChannels.AUDIT_PRODUCER, MessageChannel.class)).isNotNull();
          assertThat(bindingServiceProperties.getOutputBindings()).contains(FUNCTION_AUDIT_SUPPLIER_NAME);
          assertThat(streamFunctionProperties.getOutputBindings(FUNCTION_AUDIT_SUPPLIER_NAME))
              .isEqualTo(Arrays.asList(TestBindingsChannels.AUDIT_PRODUCER));
  
          assertThat(context.getBean(TestBindingsChannels.COMMAND_RESULTS, MessageChannel.class)).isNotNull();
          assertThat(bindingServiceProperties.getOutputBindings()).contains(FUNCTION_COMMAND_SUPPLIER_NAME);
          assertThat(streamFunctionProperties.getOutputBindings(FUNCTION_COMMAND_SUPPLIER_NAME))
              .isEqualTo(Arrays.asList(TestBindingsChannels.COMMAND_RESULTS));
      }
  
      @Test
      void testInputBindingsDefinitions() {
          Assertions.assertThat(context.getBean(COMMAND_CONSUMER, MessageChannel.class)).isNotNull();
          Assertions.assertThat(bindingServiceProperties.getInputBindings()).contains(FUNCTION_COMMAND_CONSUMER_NAME);
          Assertions
              .assertThat(streamFunctionProperties.getBindings().get(getInBinding(FUNCTION_COMMAND_CONSUMER_NAME)))
              .isEqualTo(COMMAND_CONSUMER);
  
          Assertions.assertThat(context.getBean(AUDIT_CONSUMER, MessageChannel.class)).isNotNull();
          Assertions.assertThat(bindingServiceProperties.getInputBindings()).contains(FUNCTION_AUDIT_CONSUMER_NAME);
          Assertions
              .assertThat(streamFunctionProperties.getBindings().get(getInBinding(FUNCTION_AUDIT_CONSUMER_NAME)))
              .isEqualTo(AUDIT_CONSUMER);
  
          Assertions.assertThat(context.getBean(QUERY_CONSUMER, MessageChannel.class)).isNotNull();
          Assertions.assertThat(bindingServiceProperties.getInputBindings()).contains(FUNCTION_QUERY_CONSUMER_NAME);
          Assertions
              .assertThat(streamFunctionProperties.getBindings().get(getInBinding(FUNCTION_QUERY_CONSUMER_NAME)))
              .isEqualTo(QUERY_CONSUMER);
      }
  
      @Test
      public void testFunctionRegistry() {
          assertThat(functionRegistry.<Object>lookup(FUNCTION_HANDLER_NAME + REGISTRATION_NAME_SUFFIX)).isNotNull();
          assertThat(functionRegistry.<Object>lookup(FUNCTION_PROCESSOR_NAME + REGISTRATION_NAME_SUFFIX)).isNotNull();
          assertThat(functionRegistry.<Object>lookup(FUNCTION_AUDIT_SUPPLIER_NAME)).isNull();
          assertThat(functionRegistry.<Object>lookup(FUNCTION_COMMAND_SUPPLIER_NAME)).isNull();
      }
  
      @Test
      public void testStreamBindings() {
          assertThat(streamFunctionProperties.getInputBindings(FUNCTION_HANDLER_NAME))
              .matches(bindings -> bindings == null || bindings.isEmpty());
          assertThat(streamFunctionProperties.getOutputBindings(FUNCTION_HANDLER_NAME))
              .matches(bindings -> bindings == null || bindings.isEmpty());
  
          assertThat(streamFunctionProperties.getInputBindings(FUNCTION_PROCESSOR_NAME))
              .matches(bindings -> bindings == null || bindings.isEmpty());
          assertThat(streamFunctionProperties.getOutputBindings(FUNCTION_PROCESSOR_NAME))
              .matches(bindings -> bindings == null || bindings.isEmpty());
  
          assertThat(streamFunctionProperties.getInputBindings(FUNCTION_AUDIT_SUPPLIER_NAME))
              .matches(bindings -> bindings == null || bindings.isEmpty());
          assertThat(streamFunctionProperties.getOutputBindings(FUNCTION_AUDIT_SUPPLIER_NAME))
              .matches(bindings -> bindings.size() == 1 && bindings.contains(TestBindingsChannels.AUDIT_PRODUCER));
  
          assertThat(streamFunctionProperties.getInputBindings(FUNCTION_COMMAND_SUPPLIER_NAME))
              .matches(bindings -> bindings == null || bindings.isEmpty());
          assertThat(streamFunctionProperties.getOutputBindings(FUNCTION_COMMAND_SUPPLIER_NAME))
              .matches(bindings -> bindings.size() == 1 && bindings.contains(TestBindingsChannels.COMMAND_RESULTS));
      }
  
      @Test
      public void testConsumerBindings() {
          // given
          Message<String> message = MessageBuilder.withPayload("Test").setHeader("type", "Test Consumer").build();
  
          // when
          input.send(message, "engineEvents");
  
          // then
          assertThat(consumerMessage).isNotNull();
          assertThat(consumerMessage.getHeaders().get("type", String.class)).isEqualTo("Test Consumer");
      }
  
      @Test
      public void testFunctionBindings() throws InterruptedException {
          // given
          Message<String> message = MessageBuilder.withPayload("Test").setHeader("type", "Test Consumer").build();
  
          // when
          channels.commandConsumer().send(message);
  
          // then
          assertThat(consumerMessage).isNotNull();
          assertThat(consumerMessage.getHeaders().get("type", String.class)).isEqualTo("Test Send");
  
          await()
              .untilAsserted(() -> {
                  Message<?> outputMessage = output.receive(
                      1000,
                      bindingResolver.getBindingDestination(TestBindingsChannels.COMMAND_RESULTS)
                  );
                  assertThat(outputMessage).isNotNull();
                  assertThat(outputMessage.getHeaders().get("type", String.class)).isEqualTo("Test Reply");
              });
      }
  }
  