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

import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyDescriptor.Builder;
import org.apache.nifi.connectable.Connection;
import org.apache.nifi.controller.ComponentNode;
import org.apache.nifi.controller.ProcessorNode;
import org.apache.nifi.controller.StandardSnippet;
import org.apache.nifi.controller.queue.DropFlowFileState;
import org.apache.nifi.controller.queue.DropFlowFileStatus;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.integration.FrameworkIntegrationTest;
import org.apache.nifi.integration.processor.BiConsumerProcessor;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.Revision;
import org.junit.jupiter.api.Test;

import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;

public class StandardProcessGroupIT extends FrameworkIntegrationTest {

    @Test
    public void testConflictingVersionedComponentId() {
        final ProcessorNode proc1 = createProcessorNode(BiConsumerProcessor.class, null);
        getRootGroup().addProcessor(proc1);

        final ProcessorNode proc2 = createProcessorNode(BiConsumerProcessor.class, null);
        proc2.setVersionedComponentId("aaa");
        getRootGroup().addProcessor(proc2);
        // Ensure that id didn't change
        assertEquals("aaa", proc2.getVersionedComponentId().get());

        final ProcessorNode proc3 = createProcessorNode(BiConsumerProcessor.class, null);
        proc3.setVersionedComponentId("bbb");
        getRootGroup().addProcessor(proc3);
        assertEquals("bbb", proc3.getVersionedComponentId().get());

        final ProcessorNode proc4 = createProcessorNode(BiConsumerProcessor.class, null);
        proc4.setVersionedComponentId("bbb");
        getRootGroup().addProcessor(proc4);
        // Ensure that versioned component id was nulled out
        assertFalse(proc4.getVersionedComponentId().isPresent());

        final ProcessGroup childGroup = getFlowController().getFlowManager().createProcessGroup("child");
        childGroup.setName("child");
        getRootGroup().addProcessGroup(childGroup);

        final ProcessorNode proc5 = createProcessorNode(BiConsumerProcessor.class, null);
        proc5.setVersionedComponentId("bbb");
        childGroup.addProcessor(proc5);
        assertEquals("bbb", proc5.getVersionedComponentId().get());

        // Move processor from child group to parent group.
        // This should null out the ID for proc5 and leave proc3 as is.
        final StandardSnippet snippet = new StandardSnippet();
        snippet.addProcessors(Collections.singletonMap(proc5.getIdentifier(), new Revision(0L, "abc", proc5.getIdentifier())));
        childGroup.move(snippet, getRootGroup());
        assertFalse(proc5.getVersionedComponentId().isPresent());
        assertEquals("bbb", proc3.getVersionedComponentId().get());
    }

    @Test
    public void testProcessGroupDefaults() {
        // Connect two processors with default settings of the root process group
        ProcessorNode sourceProcessor = createGenerateProcessor(1);
        ProcessorNode destinationProcessor = createProcessorNode((context, session) -> {});
        Connection connection = connect(sourceProcessor, destinationProcessor, Collections.singleton(REL_SUCCESS));


        // Verify all defaults are in place on the process group and the connection
        assertEquals("0 sec", getRootGroup().getDefaultFlowFileExpiration());
        assertEquals(NiFiProperties.DEFAULT_BACKPRESSURE_COUNT, (long) getRootGroup().getDefaultBackPressureObjectThreshold());
        assertEquals(NiFiProperties.DEFAULT_BACKPRESSURE_SIZE, getRootGroup().getDefaultBackPressureDataSizeThreshold());
        assertEquals("0 sec", connection.getFlowFileQueue().getFlowFileExpiration());
        assertEquals(NiFiProperties.DEFAULT_BACKPRESSURE_COUNT, (long) connection.getFlowFileQueue().getBackPressureObjectThreshold());
        assertEquals(NiFiProperties.DEFAULT_BACKPRESSURE_SIZE, connection.getFlowFileQueue().getBackPressureDataSizeThreshold());

        // Update default settings of the process group, and create a new connection
        getRootGroup().setDefaultFlowFileExpiration("99 min");
        getRootGroup().setDefaultBackPressureObjectThreshold(99L);
        getRootGroup().setDefaultBackPressureDataSizeThreshold("99 MB");

        ProcessorNode sourceProcessor1 = createGenerateProcessor(1);
        ProcessorNode destinationProcessor1 = createProcessorNode((context, session) -> {});
        Connection connection1 = connect(sourceProcessor1, destinationProcessor1, Collections.singleton(REL_SUCCESS));

        // Verify updated settings are in place on the process group and the connection
        assertEquals("99 min", getRootGroup().getDefaultFlowFileExpiration());
        assertEquals(99, (long) getRootGroup().getDefaultBackPressureObjectThreshold());
        assertEquals("99 MB", getRootGroup().getDefaultBackPressureDataSizeThreshold());
        assertEquals("99 min", connection1.getFlowFileQueue().getFlowFileExpiration());
        assertEquals(99, (long) connection1.getFlowFileQueue().getBackPressureObjectThreshold());
        assertEquals("99 MB", connection1.getFlowFileQueue().getBackPressureDataSizeThreshold());
    }

    @Test
    public void testDropAllFlowFilesFromOneConnection() throws Exception {
        ProcessorNode sourceProcessGroup = createGenerateProcessor(1);
        ProcessorNode destinationProcessGroup = createProcessorNode((context, session) -> {});

        Connection connection = connect(sourceProcessGroup, destinationProcessGroup, Collections.singleton(REL_SUCCESS));

        for (int i = 0; i < 5; i++) {
            triggerOnce(sourceProcessGroup);
        }
        assertEquals(5, connection.getFlowFileQueue().size().getObjectCount());

        // WHEN
        DropFlowFileStatus dropFlowFileStatus = getRootGroup().dropAllFlowFiles("requestId", "unimportant");
        while (dropFlowFileStatus.getState() != DropFlowFileState.COMPLETE) {
            TimeUnit.MILLISECONDS.sleep(10);
            dropFlowFileStatus = getRootGroup().getDropAllFlowFilesStatus("requestId");
        }

        // THEN
        assertEquals(0, connection.getFlowFileQueue().size().getObjectCount());
    }

    @Test
    public void testDropAllFlowFilesFromOneConnectionInChildProcessGroup() throws Exception {
        ProcessGroup childProcessGroup = getFlowController().getFlowManager().createProcessGroup("childProcessGroup");
        childProcessGroup.setName("ChildProcessGroup");
        getRootGroup().addProcessGroup(childProcessGroup);

        ProcessorNode sourceProcessGroup = createGenerateProcessor(1);
        moveProcessor(sourceProcessGroup, childProcessGroup);

        ProcessorNode destinationProcessGroup = createProcessorNode((context, session) -> {});
        moveProcessor(destinationProcessGroup, childProcessGroup);

        Connection connection = connect(childProcessGroup, sourceProcessGroup, destinationProcessGroup, Collections.singleton(REL_SUCCESS));

        for (int i = 0; i < 5; i++) {
            triggerOnce(sourceProcessGroup);
        }
        assertEquals(5, connection.getFlowFileQueue().size().getObjectCount());

        // WHEN
        DropFlowFileStatus dropFlowFileStatus = getRootGroup().dropAllFlowFiles("requestId", "unimportant");
        while (dropFlowFileStatus.getState() != DropFlowFileState.COMPLETE) {
            TimeUnit.MILLISECONDS.sleep(10);
            dropFlowFileStatus = childProcessGroup.getDropAllFlowFilesStatus("requestId");
        }

        // THEN
        assertEquals(0, connection.getFlowFileQueue().size().getObjectCount());
    }

    @Test
    public void testDropAllFlowFilesFromMultipleConnections() throws Exception {
        ProcessGroup childProcessGroup = getFlowController().getFlowManager().createProcessGroup("childProcessGroup");
        childProcessGroup.setName("ChildProcessGroup");
        getRootGroup().addProcessGroup(childProcessGroup);

        ProcessorNode sourceProcessGroup1 = createGenerateProcessor(4);
        moveProcessor(sourceProcessGroup1, childProcessGroup);

        ProcessorNode sourceProcessGroup2 = createGenerateProcessor(5);
        moveProcessor(sourceProcessGroup2, childProcessGroup);

        ProcessorNode destinationProcessGroup = createProcessorNode((context, session) -> {});
        moveProcessor(destinationProcessGroup, childProcessGroup);

        Connection connection1 = connect(childProcessGroup, sourceProcessGroup1, destinationProcessGroup, Collections.singleton(REL_SUCCESS));
        Connection connection2 = connect(childProcessGroup, sourceProcessGroup2, destinationProcessGroup, Collections.singleton(REL_SUCCESS));

        for (int i = 0; i < 5; i++) {
            triggerOnce(sourceProcessGroup1);
        }
        assertEquals(5, connection1.getFlowFileQueue().size().getObjectCount());

        for (int i = 0; i < 10; i++) {
            triggerOnce(sourceProcessGroup2);
        }
        assertEquals(10, connection2.getFlowFileQueue().size().getObjectCount());

        // WHEN
        DropFlowFileStatus dropFlowFileStatus = getRootGroup().dropAllFlowFiles("requestId", "unimportant");
        while (dropFlowFileStatus.getState() != DropFlowFileState.COMPLETE) {
            TimeUnit.MILLISECONDS.sleep(10);
            dropFlowFileStatus = childProcessGroup.getDropAllFlowFilesStatus("requestId");
        }

        // THEN
        assertEquals(5 + 10, dropFlowFileStatus.getOriginalSize().getObjectCount());
        assertEquals(20 + 50, dropFlowFileStatus.getOriginalSize().getByteCount());

        assertEquals(0, dropFlowFileStatus.getCurrentSize().getObjectCount());
        assertEquals(0, dropFlowFileStatus.getCurrentSize().getByteCount());

        assertEquals(5 + 10, dropFlowFileStatus.getDroppedSize().getObjectCount());
        assertEquals(20 + 50, dropFlowFileStatus.getDroppedSize().getByteCount());

        assertEquals(0, connection1.getFlowFileQueue().size().getObjectCount());
        assertEquals(0, connection2.getFlowFileQueue().size().getObjectCount());
    }

    @Test
    public void testComponentsAffectedByVariableOverridden() {
        final ProcessGroup child = getFlowController().getFlowManager().createProcessGroup("child");
        child.setName("Child");
        child.setVariables(Collections.singletonMap("number", "5"));

        getRootGroup().setVariables(Collections.singletonMap("number", "1"));
        getRootGroup().addProcessGroup(child);

        final ProcessorNode processor = createProcessorNode(NumberRefProcessor.class);
        processor.setProperties(Collections.singletonMap(NumberRefProcessor.NUMBER.getName(), "${number}"));
        moveProcessor(processor, child);

        final Set<ComponentNode> componentsAffected = child.getComponentsAffectedByVariable("number");
        assertEquals(1, componentsAffected.size());
        assertTrue(componentsAffected.contains(processor));

        final Set<ComponentNode> rootAffected = getRootGroup().getComponentsAffectedByVariable("number");
        assertTrue(rootAffected.isEmpty());

        processor.setSchedulingPeriod("1 hour");
        child.startProcessor(processor, false);

        getRootGroup().setVariables(Collections.singletonMap("number", "2"));

        assertThrows(IllegalStateException.class,
                () -> child.setVariables(Collections.singletonMap("number", "10")),
                "Updated variable that is referenced by a running processor");

        child.stopProcessor(processor);
    }


    public static class NumberRefProcessor extends AbstractProcessor {
        static final PropertyDescriptor NUMBER = new Builder()
            .name("Number")
            .displayName("Number")
            .description("A Number")
            .required(true)
            .addValidator(StandardValidators.NUMBER_VALIDATOR)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .defaultValue("1")
            .build();

        @Override
        protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
            return Collections.singletonList(NUMBER);
        }

        @Override
        public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {

        }
    }
}
