You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jp...@apache.org on 2016/11/08 20:07:40 UTC
[1/3] nifi-minifi git commit: MINIFI-107 - Process group support
Repository: nifi-minifi
Updated Branches:
refs/heads/master 1bbeedf6f -> 31855bbc7
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.xml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.xml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.xml
new file mode 100644
index 0000000..8cd6f1b
--- /dev/null
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.xml
@@ -0,0 +1,648 @@
+<?xml version="1.0" ?>
+<!--
+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.
+ -->
+<template encoding-version="1.0">
+ <description></description>
+ <groupId>21a36442-0158-1000-dec0-d25353fe677c</groupId>
+ <name>ProcessGroupsAndRemoteProcessGroups</name>
+ <snippet>
+ <connections>
+ <id>2077bf8f-0158-1000-0000-000000000000</id>
+ <parentGroupId>21a36442-0158-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>21a36442-0158-1000-0000-000000000000</groupId>
+ <id>2077ab1e-0158-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <selectedRelationships>success</selectedRelationships>
+ <source>
+ <groupId>21a36442-0158-1000-0000-000000000000</groupId>
+ <id>207748d1-0158-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <connections>
+ <id>2079cf6f-0158-1000-0000-000000000000</id>
+ <parentGroupId>21a36442-0158-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>207888b1-0158-1000-0000-000000000000</groupId>
+ <id>2078c936-0158-1000-0000-000000000000</id>
+ <type>INPUT_PORT</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <selectedRelationships>success</selectedRelationships>
+ <source>
+ <groupId>21a36442-0158-1000-0000-000000000000</groupId>
+ <id>2077ab1e-0158-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <connections>
+ <id>2079faa0-0158-1000-0000-000000000000</id>
+ <parentGroupId>21a36442-0158-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>21a36442-0158-1000-0000-000000000000</groupId>
+ <id>2079e8bd-0158-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <source>
+ <groupId>207888b1-0158-1000-0000-000000000000</groupId>
+ <id>2079b327-0158-1000-0000-000000000000</id>
+ <type>OUTPUT_PORT</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <processGroups>
+ <id>207888b1-0158-1000-0000-000000000000</id>
+ <parentGroupId>21a36442-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>0.0</x>
+ <y>447.0719299316406</y>
+ </position>
+ <comments></comments>
+ <contents>
+ <connections>
+ <id>20792ec2-0158-1000-0000-000000000000</id>
+ <parentGroupId>207888b1-0158-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>207888b1-0158-1000-0000-000000000000</groupId>
+ <id>2078f34e-0158-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <source>
+ <groupId>207888b1-0158-1000-0000-000000000000</groupId>
+ <id>2078c936-0158-1000-0000-000000000000</id>
+ <type>INPUT_PORT</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <connections>
+ <id>207b0eb1-0158-1000-0000-000000000000</id>
+ <parentGroupId>207888b1-0158-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>20794cd4-0158-1000-0000-000000000000</groupId>
+ <id>207a5f50-0158-1000-0000-000000000000</id>
+ <type>INPUT_PORT</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <selectedRelationships>success</selectedRelationships>
+ <source>
+ <groupId>207888b1-0158-1000-0000-000000000000</groupId>
+ <id>2078f34e-0158-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <connections>
+ <id>207b1880-0158-1000-0000-000000000000</id>
+ <parentGroupId>207888b1-0158-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>207888b1-0158-1000-0000-000000000000</groupId>
+ <id>2079b327-0158-1000-0000-000000000000</id>
+ <type>OUTPUT_PORT</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <source>
+ <groupId>20794cd4-0158-1000-0000-000000000000</groupId>
+ <id>207a6d92-0158-1000-0000-000000000000</id>
+ <type>OUTPUT_PORT</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <connections>
+ <id>21a5b1f1-0158-1000-0000-000000000000</id>
+ <parentGroupId>207888b1-0158-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>21a4e2f0-0158-1000-0000-000000000000</groupId>
+ <id>21a2fb5e-0158-1000-3b5e-5a7d3aaee01b</id>
+ <type>REMOTE_INPUT_PORT</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <selectedRelationships>success</selectedRelationships>
+ <source>
+ <groupId>207888b1-0158-1000-0000-000000000000</groupId>
+ <id>2078f34e-0158-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <inputPorts>
+ <id>2078c936-0158-1000-0000-000000000000</id>
+ <parentGroupId>207888b1-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>433.0000001775595</x>
+ <y>46.99999825784364</y>
+ </position>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <name>input</name>
+ <state>STOPPED</state>
+ <type>INPUT_PORT</type>
+ </inputPorts>
+ <outputPorts>
+ <id>2079b327-0158-1000-0000-000000000000</id>
+ <parentGroupId>207888b1-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>445.68731178433825</x>
+ <y>666.1011812611806</y>
+ </position>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <name>output</name>
+ <state>STOPPED</state>
+ <type>OUTPUT_PORT</type>
+ </outputPorts>
+ <processGroups>
+ <id>20794cd4-0158-1000-0000-000000000000</id>
+ <parentGroupId>207888b1-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>371.304682389807</x>
+ <y>397.8012321556296</y>
+ </position>
+ <comments></comments>
+ <contents>
+ <connections>
+ <id>207aca0d-0158-1000-0000-000000000000</id>
+ <parentGroupId>20794cd4-0158-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>20794cd4-0158-1000-0000-000000000000</groupId>
+ <id>207a89ba-0158-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <source>
+ <groupId>20794cd4-0158-1000-0000-000000000000</groupId>
+ <id>207a5f50-0158-1000-0000-000000000000</id>
+ <type>INPUT_PORT</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <connections>
+ <id>207ad5e9-0158-1000-0000-000000000000</id>
+ <parentGroupId>20794cd4-0158-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>20794cd4-0158-1000-0000-000000000000</groupId>
+ <id>207a6d92-0158-1000-0000-000000000000</id>
+ <type>OUTPUT_PORT</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <selectedRelationships>success</selectedRelationships>
+ <source>
+ <groupId>20794cd4-0158-1000-0000-000000000000</groupId>
+ <id>207a89ba-0158-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <connections>
+ <id>21a6abb9-0158-1000-0000-000000000000</id>
+ <parentGroupId>20794cd4-0158-1000-0000-000000000000</parentGroupId>
+ <backPressureDataSizeThreshold>1 GB</backPressureDataSizeThreshold>
+ <backPressureObjectThreshold>10000</backPressureObjectThreshold>
+ <destination>
+ <groupId>21a651bb-0158-1000-0000-000000000000</groupId>
+ <id>21a39aba-0158-1000-a1a0-1b55bcddcd72</id>
+ <type>REMOTE_INPUT_PORT</type>
+ </destination>
+ <flowFileExpiration>0 sec</flowFileExpiration>
+ <labelIndex>1</labelIndex>
+ <name></name>
+ <selectedRelationships>success</selectedRelationships>
+ <source>
+ <groupId>20794cd4-0158-1000-0000-000000000000</groupId>
+ <id>207a89ba-0158-1000-0000-000000000000</id>
+ <type>PROCESSOR</type>
+ </source>
+ <zIndex>0</zIndex>
+ </connections>
+ <inputPorts>
+ <id>207a5f50-0158-1000-0000-000000000000</id>
+ <parentGroupId>20794cd4-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>570.9999962395234</x>
+ <y>37.999998611138864</y>
+ </position>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <name>input</name>
+ <state>STOPPED</state>
+ <type>INPUT_PORT</type>
+ </inputPorts>
+ <outputPorts>
+ <id>207a6d92-0158-1000-0000-000000000000</id>
+ <parentGroupId>20794cd4-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>567.9999962395234</x>
+ <y>343.99999861113884</y>
+ </position>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <name>output</name>
+ <state>STOPPED</state>
+ <type>OUTPUT_PORT</type>
+ </outputPorts>
+ <processors>
+ <id>207a89ba-0158-1000-0000-000000000000</id>
+ <parentGroupId>20794cd4-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>509.9999962395234</x>
+ <y>146.99999861113886</y>
+ </position>
+ <config>
+ <bulletinLevel>WARN</bulletinLevel>
+ <comments></comments>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <descriptors>
+ <entry>
+ <key>Delete Attributes Expression</key>
+ <value>
+ <name>Delete Attributes Expression</name>
+ </value>
+ </entry>
+ <entry>
+ <key>bottom</key>
+ <value>
+ <name>bottom</name>
+ </value>
+ </entry>
+ </descriptors>
+ <lossTolerant>false</lossTolerant>
+ <penaltyDuration>30 sec</penaltyDuration>
+ <properties>
+ <entry>
+ <key>Delete Attributes Expression</key>
+ </entry>
+ <entry>
+ <key>bottom</key>
+ <value>bottom</value>
+ </entry>
+ </properties>
+ <runDurationMillis>0</runDurationMillis>
+ <schedulingPeriod>0 sec</schedulingPeriod>
+ <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
+ <yieldDuration>1 sec</yieldDuration>
+ </config>
+ <name>UpdateAttribute</name>
+ <relationships>
+ <autoTerminate>false</autoTerminate>
+ <name>success</name>
+ </relationships>
+ <style></style>
+ <type>org.apache.nifi.processors.attributes.UpdateAttribute</type>
+ </processors>
+ <remoteProcessGroups>
+ <id>21a651bb-0158-1000-0000-000000000000</id>
+ <parentGroupId>20794cd4-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>1140.0000482871</x>
+ <y>130.0000033911915</y>
+ </position>
+ <communicationsTimeout>30 sec</communicationsTimeout>
+ <contents>
+ <inputPorts>
+ <comments></comments>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <connected>true</connected>
+ <exists>true</exists>
+ <id>21a39aba-0158-1000-a1a0-1b55bcddcd72</id>
+ <name>input2</name>
+ <targetRunning>true</targetRunning>
+ <transmitting>false</transmitting>
+ <useCompression>false</useCompression>
+ </inputPorts>
+ </contents>
+ <proxyHost></proxyHost>
+ <proxyUser></proxyUser>
+ <targetUri>http://localhost:9091/nifi</targetUri>
+ <transportProtocol>RAW</transportProtocol>
+ <yieldDuration>10 sec</yieldDuration>
+ </remoteProcessGroups>
+ </contents>
+ <name>bottom</name>
+ </processGroups>
+ <processors>
+ <id>2078f34e-0158-1000-0000-000000000000</id>
+ <parentGroupId>207888b1-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>377.497680841622</x>
+ <y>186.3363935215155</y>
+ </position>
+ <config>
+ <bulletinLevel>WARN</bulletinLevel>
+ <comments></comments>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <descriptors>
+ <entry>
+ <key>Delete Attributes Expression</key>
+ <value>
+ <name>Delete Attributes Expression</name>
+ </value>
+ </entry>
+ <entry>
+ <key>middle</key>
+ <value>
+ <name>middle</name>
+ </value>
+ </entry>
+ </descriptors>
+ <lossTolerant>false</lossTolerant>
+ <penaltyDuration>30 sec</penaltyDuration>
+ <properties>
+ <entry>
+ <key>Delete Attributes Expression</key>
+ </entry>
+ <entry>
+ <key>middle</key>
+ <value>middle</value>
+ </entry>
+ </properties>
+ <runDurationMillis>0</runDurationMillis>
+ <schedulingPeriod>0 sec</schedulingPeriod>
+ <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
+ <yieldDuration>1 sec</yieldDuration>
+ </config>
+ <name>UpdateAttribute</name>
+ <relationships>
+ <autoTerminate>false</autoTerminate>
+ <name>success</name>
+ </relationships>
+ <style></style>
+ <type>org.apache.nifi.processors.attributes.UpdateAttribute</type>
+ </processors>
+ <remoteProcessGroups>
+ <id>21a4e2f0-0158-1000-0000-000000000000</id>
+ <parentGroupId>207888b1-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>1004.6132741275832</x>
+ <y>175.0721786721599</y>
+ </position>
+ <communicationsTimeout>30 sec</communicationsTimeout>
+ <contents>
+ <inputPorts>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <connected>true</connected>
+ <exists>true</exists>
+ <id>21a2fb5e-0158-1000-3b5e-5a7d3aaee01b</id>
+ <name>input</name>
+ <targetRunning>true</targetRunning>
+ <transmitting>false</transmitting>
+ <useCompression>false</useCompression>
+ </inputPorts>
+ </contents>
+ <proxyHost></proxyHost>
+ <proxyUser></proxyUser>
+ <targetUri>http://localhost:9090/nifi</targetUri>
+ <transportProtocol>RAW</transportProtocol>
+ <yieldDuration>10 sec</yieldDuration>
+ </remoteProcessGroups>
+ </contents>
+ <name>middle</name>
+ </processGroups>
+ <processors>
+ <id>207748d1-0158-1000-0000-000000000000</id>
+ <parentGroupId>21a36442-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>8.7503662109375</x>
+ <y>0.0</y>
+ </position>
+ <config>
+ <bulletinLevel>WARN</bulletinLevel>
+ <comments></comments>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <descriptors>
+ <entry>
+ <key>File Size</key>
+ <value>
+ <name>File Size</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Batch Size</key>
+ <value>
+ <name>Batch Size</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Data Format</key>
+ <value>
+ <name>Data Format</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Unique FlowFiles</key>
+ <value>
+ <name>Unique FlowFiles</name>
+ </value>
+ </entry>
+ </descriptors>
+ <lossTolerant>false</lossTolerant>
+ <penaltyDuration>30 sec</penaltyDuration>
+ <properties>
+ <entry>
+ <key>File Size</key>
+ <value>1 b</value>
+ </entry>
+ <entry>
+ <key>Batch Size</key>
+ <value>1</value>
+ </entry>
+ <entry>
+ <key>Data Format</key>
+ <value>Binary</value>
+ </entry>
+ <entry>
+ <key>Unique FlowFiles</key>
+ <value>false</value>
+ </entry>
+ </properties>
+ <runDurationMillis>0</runDurationMillis>
+ <schedulingPeriod>0 sec</schedulingPeriod>
+ <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
+ <yieldDuration>1 sec</yieldDuration>
+ </config>
+ <name>GenerateFlowFile</name>
+ <relationships>
+ <autoTerminate>false</autoTerminate>
+ <name>success</name>
+ </relationships>
+ <style></style>
+ <type>org.apache.nifi.processors.standard.GenerateFlowFile</type>
+ </processors>
+ <processors>
+ <id>2077ab1e-0158-1000-0000-000000000000</id>
+ <parentGroupId>21a36442-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>15.7503662109375</x>
+ <y>230.0</y>
+ </position>
+ <config>
+ <bulletinLevel>WARN</bulletinLevel>
+ <comments></comments>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <descriptors>
+ <entry>
+ <key>Delete Attributes Expression</key>
+ <value>
+ <name>Delete Attributes Expression</name>
+ </value>
+ </entry>
+ <entry>
+ <key>top</key>
+ <value>
+ <name>top</name>
+ </value>
+ </entry>
+ </descriptors>
+ <lossTolerant>false</lossTolerant>
+ <penaltyDuration>30 sec</penaltyDuration>
+ <properties>
+ <entry>
+ <key>Delete Attributes Expression</key>
+ </entry>
+ <entry>
+ <key>top</key>
+ <value>top</value>
+ </entry>
+ </properties>
+ <runDurationMillis>0</runDurationMillis>
+ <schedulingPeriod>0 sec</schedulingPeriod>
+ <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
+ <yieldDuration>1 sec</yieldDuration>
+ </config>
+ <name>UpdateAttribute</name>
+ <relationships>
+ <autoTerminate>false</autoTerminate>
+ <name>success</name>
+ </relationships>
+ <style></style>
+ <type>org.apache.nifi.processors.attributes.UpdateAttribute</type>
+ </processors>
+ <processors>
+ <id>2079e8bd-0158-1000-0000-000000000000</id>
+ <parentGroupId>21a36442-0158-1000-0000-000000000000</parentGroupId>
+ <position>
+ <x>20.064770209559924</x>
+ <y>701.7270693739383</y>
+ </position>
+ <config>
+ <bulletinLevel>WARN</bulletinLevel>
+ <comments></comments>
+ <concurrentlySchedulableTaskCount>1</concurrentlySchedulableTaskCount>
+ <descriptors>
+ <entry>
+ <key>Log Level</key>
+ <value>
+ <name>Log Level</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Log Payload</key>
+ <value>
+ <name>Log Payload</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Attributes to Log</key>
+ <value>
+ <name>Attributes to Log</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Attributes to Ignore</key>
+ <value>
+ <name>Attributes to Ignore</name>
+ </value>
+ </entry>
+ <entry>
+ <key>Log prefix</key>
+ <value>
+ <name>Log prefix</name>
+ </value>
+ </entry>
+ </descriptors>
+ <lossTolerant>false</lossTolerant>
+ <penaltyDuration>30 sec</penaltyDuration>
+ <properties>
+ <entry>
+ <key>Log Level</key>
+ <value>info</value>
+ </entry>
+ <entry>
+ <key>Log Payload</key>
+ <value>false</value>
+ </entry>
+ <entry>
+ <key>Attributes to Log</key>
+ </entry>
+ <entry>
+ <key>Attributes to Ignore</key>
+ </entry>
+ <entry>
+ <key>Log prefix</key>
+ </entry>
+ </properties>
+ <runDurationMillis>0</runDurationMillis>
+ <schedulingPeriod>0 sec</schedulingPeriod>
+ <schedulingStrategy>TIMER_DRIVEN</schedulingStrategy>
+ <yieldDuration>1 sec</yieldDuration>
+ </config>
+ <name>LogAttribute</name>
+ <relationships>
+ <autoTerminate>true</autoTerminate>
+ <name>success</name>
+ </relationships>
+ <style></style>
+ <type>org.apache.nifi.processors.standard.LogAttribute</type>
+ </processors>
+ </snippet>
+ <timestamp>11/01/2016 16:49:02 EDT</timestamp>
+</template>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.yml
new file mode 100644
index 0000000..101a360
--- /dev/null
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ProcessGroupsAndRemoteProcessGroups.yml
@@ -0,0 +1,276 @@
+# 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.
+
+MiNiFi Config Version: 2
+Flow Controller:
+ name: ProcessGroupsAndRemoteProcessGroups
+ comment: ''
+Core Properties:
+ flow controller graceful shutdown period: 10 sec
+ flow service write delay interval: 500 ms
+ administrative yield duration: 30 sec
+ bored yield duration: 10 millis
+ max concurrent threads: 1
+FlowFile Repository:
+ partitions: 256
+ checkpoint interval: 2 mins
+ always sync: false
+ Swap:
+ threshold: 20000
+ in period: 5 sec
+ in threads: 1
+ out period: 5 sec
+ out threads: 4
+Content Repository:
+ content claim max appendable size: 10 MB
+ content claim max flow files: 100
+ always sync: false
+Provenance Repository:
+ provenance rollover time: 1 min
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+Security Properties:
+ keystore: ''
+ keystore type: ''
+ keystore password: ''
+ key password: ''
+ truststore: ''
+ truststore type: ''
+ truststore password: ''
+ ssl protocol: ''
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+Processors:
+- id: 207748d1-0158-1000-0000-000000000000
+ name: GenerateFlowFile
+ class: org.apache.nifi.processors.standard.GenerateFlowFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list: []
+ Properties:
+ Batch Size: '1'
+ Data Format: Binary
+ File Size: 1 b
+ Unique FlowFiles: 'false'
+- id: 2079e8bd-0158-1000-0000-000000000000
+ name: LogAttribute
+ class: org.apache.nifi.processors.standard.LogAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ - success
+ Properties:
+ Attributes to Ignore:
+ Attributes to Log:
+ Log Level: info
+ Log Payload: 'false'
+ Log prefix:
+- id: 2077ab1e-0158-1000-0000-000000000000
+ name: UpdateAttribute
+ class: org.apache.nifi.processors.attributes.UpdateAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list: []
+ Properties:
+ Delete Attributes Expression:
+ top: top
+Process Groups:
+- id: 207888b1-0158-1000-0000-000000000000
+ name: middle
+ Processors:
+ - id: 2078f34e-0158-1000-0000-000000000000
+ name: UpdateAttribute
+ class: org.apache.nifi.processors.attributes.UpdateAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list: []
+ Properties:
+ Delete Attributes Expression:
+ middle: middle
+ Process Groups:
+ - id: 20794cd4-0158-1000-0000-000000000000
+ name: bottom
+ Processors:
+ - id: 207a89ba-0158-1000-0000-000000000000
+ name: UpdateAttribute
+ class: org.apache.nifi.processors.attributes.UpdateAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list: []
+ Properties:
+ Delete Attributes Expression:
+ bottom: bottom
+ Process Groups: []
+ Input Ports:
+ - id: 207a5f50-0158-1000-0000-000000000000
+ name: input
+ Output Ports:
+ - id: 207a6d92-0158-1000-0000-000000000000
+ name: output
+ Connections:
+ - id: 21a6abb9-0158-1000-0000-000000000000
+ name: UpdateAttribute/success/21a39aba-0158-1000-a1a0-1b55bcddcd72
+ source id: 207a89ba-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 21a39aba-0158-1000-a1a0-1b55bcddcd72
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ - id: 207ad5e9-0158-1000-0000-000000000000
+ name: UpdateAttribute/success/output
+ source id: 207a89ba-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 207a6d92-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ - id: 207aca0d-0158-1000-0000-000000000000
+ name: input//UpdateAttribute
+ source id: 207a5f50-0158-1000-0000-000000000000
+ source relationship names: []
+ destination id: 207a89ba-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ Remote Processing Groups:
+ - name: http://localhost:9091/nifi
+ url: http://localhost:9091/nifi
+ comment: ''
+ timeout: 30 sec
+ yield period: 10 sec
+ Input Ports:
+ - id: 21a39aba-0158-1000-a1a0-1b55bcddcd72
+ name: input2
+ comment: ''
+ max concurrent tasks: 1
+ use compression: false
+ Input Ports:
+ - id: 2078c936-0158-1000-0000-000000000000
+ name: input
+ Output Ports:
+ - id: 2079b327-0158-1000-0000-000000000000
+ name: output
+ Connections:
+ - id: 21a5b1f1-0158-1000-0000-000000000000
+ name: UpdateAttribute/success/21a2fb5e-0158-1000-3b5e-5a7d3aaee01b
+ source id: 2078f34e-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 21a2fb5e-0158-1000-3b5e-5a7d3aaee01b
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ - id: 207b0eb1-0158-1000-0000-000000000000
+ name: UpdateAttribute/success/input
+ source id: 2078f34e-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 207a5f50-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ - id: 20792ec2-0158-1000-0000-000000000000
+ name: input//UpdateAttribute
+ source id: 2078c936-0158-1000-0000-000000000000
+ source relationship names: []
+ destination id: 2078f34e-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ - id: 207b1880-0158-1000-0000-000000000000
+ name: output//output
+ source id: 207a6d92-0158-1000-0000-000000000000
+ source relationship names: []
+ destination id: 2079b327-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ Remote Processing Groups:
+ - name: http://localhost:9090/nifi
+ url: http://localhost:9090/nifi
+ comment: ''
+ timeout: 30 sec
+ yield period: 10 sec
+ Input Ports:
+ - id: 21a2fb5e-0158-1000-3b5e-5a7d3aaee01b
+ name: input
+ comment: ''
+ max concurrent tasks: 1
+ use compression: false
+Input Ports: []
+Output Ports: []
+Connections:
+- id: 2077bf8f-0158-1000-0000-000000000000
+ name: GenerateFlowFile/success/UpdateAttribute
+ source id: 207748d1-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 2077ab1e-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+- id: 2079cf6f-0158-1000-0000-000000000000
+ name: UpdateAttribute/success/input
+ source id: 2077ab1e-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 2078c936-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+- id: 2079faa0-0158-1000-0000-000000000000
+ name: output//LogAttribute
+ source id: 2079b327-0158-1000-0000-000000000000
+ source relationship names: []
+ destination id: 2079e8bd-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+Remote Processing Groups: []
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ReplaceTextExpressionLanguageCSVReformatting.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ReplaceTextExpressionLanguageCSVReformatting.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ReplaceTextExpressionLanguageCSVReformatting.yml
index d740cf2..36bdcd8 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ReplaceTextExpressionLanguageCSVReformatting.yml
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/ReplaceTextExpressionLanguageCSVReformatting.yml
@@ -123,6 +123,9 @@ Processors:
2006,10-01-2004,10may2004
2007,15-05-2006,10jun2005
2009,8-8-2008,10aug2008
+Process Groups: []
+Input Ports: []
+Output Ports: []
Connections:
- name: Generate Empty File/success/Set CSV Content
id: ca71a875-0ff5-41ef-bbe0-da5de0ca1e08
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/StressTestFramework.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/StressTestFramework.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/StressTestFramework.yml
index be3d200..d135be8 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/StressTestFramework.yml
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/StressTestFramework.yml
@@ -98,6 +98,9 @@ Processors:
Delete Attributes Expression:
property 1: value 1
property 2: value 2 ${nextInt()}
+Process Groups: []
+Input Ports: []
+Output Ports: []
Connections:
- name: GenerateFlowFile/success/UpdateAttribute
id: 0e6873cc-cb9d-4e98-92aa-a3319f4c1b02
[3/3] nifi-minifi git commit: MINIFI-107 - Process group support
Posted by jp...@apache.org.
MINIFI-107 - Process group support
This closes #50
Signed-off-by: Joseph Percivall <jo...@yahoo.com>
Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi/commit/31855bbc
Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi/tree/31855bbc
Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi/diff/31855bbc
Branch: refs/heads/master
Commit: 31855bbc7ee7016b460226d24790760523e07885
Parents: 1bbeedf
Author: Bryan Rosander <br...@apache.org>
Authored: Tue Nov 1 12:54:17 2016 -0400
Committer: Joseph Percivall <jo...@yahoo.com>
Committed: Tue Nov 8 15:06:47 2016 -0500
----------------------------------------------------------------------
.../bootstrap/util/ConfigTransformer.java | 183 ++++--
.../bootstrap/util/ParentGroupIdResolver.java | 87 +++
.../bootstrap/util/ConfigTransformerTest.java | 166 ++++-
.../util/ParentGroupIdResolverTest.java | 129 ++++
.../bootstrap/util/TestConfigTransformer.java | 3 -
.../test/resources/config-process-groups.yml | 276 ++++++++
.../minifi/commons/schema/ConfigSchema.java | 124 ++--
.../minifi/commons/schema/ConnectionSchema.java | 41 +-
.../nifi/minifi/commons/schema/PortSchema.java | 31 +
.../commons/schema/ProcessGroupSchema.java | 153 +++++
.../minifi/commons/schema/ProcessorSchema.java | 26 +-
.../commons/schema/RemoteInputPortSchema.java | 13 +-
.../schema/RemoteProcessingGroupSchema.java | 15 +-
.../commons/schema/common/BaseSchema.java | 52 +-
.../schema/common/BaseSchemaWithIdAndName.java | 20 +-
.../commons/schema/common/CollectionUtil.java | 39 ++
.../schema/common/CommonPropertyKeys.java | 1 +
.../commons/schema/common/StringUtil.java | 34 +-
.../exception/SchemaInstantiatonException.java | 30 +
.../commons/schema/ProcessGroupSchemaTest.java | 64 ++
.../schema/serialization/SchemaLoaderTest.java | 4 +-
.../commons/schema/v1/ConfigSchemaV1Test.java | 6 +-
.../schema/v1/ConnectionSchemaV1Test.java | 2 +-
.../src/main/markdown/System_Admin_Guide.md | 35 +-
.../src/main/resources/conf/config.yml | 3 +
.../toolkit/configuration/ConfigMain.java | 135 ++--
.../configuration/dto/ConfigSchemaFunction.java | 67 +-
.../dto/ConnectionSchemaFunction.java | 6 +-
.../configuration/dto/PortSchemaFunction.java | 46 ++
.../dto/ProcessorSchemaFunction.java | 6 +-
.../toolkit/configuration/ConfigMainTest.java | 30 +-
.../dto/PortSchemaFunctionTest.java | 74 +++
.../src/test/resources/CsvToJson.yml | 3 +
.../resources/DecompressionCircularFlow.yml | 3 +
.../resources/InvokeHttpMiNiFiTemplateTest.yml | 3 +
.../test/resources/MultipleRelationships.yml | 3 +
.../ProcessGroupsAndRemoteProcessGroups.xml | 648 +++++++++++++++++++
.../ProcessGroupsAndRemoteProcessGroups.yml | 276 ++++++++
...aceTextExpressionLanguageCSVReformatting.yml | 3 +
.../src/test/resources/StressTestFramework.yml | 3 +
40 files changed, 2532 insertions(+), 311 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformer.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformer.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformer.java
index 9794415..9fa7f05 100644
--- a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformer.java
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformer.java
@@ -27,11 +27,15 @@ import org.apache.nifi.minifi.commons.schema.ContentRepositorySchema;
import org.apache.nifi.minifi.commons.schema.CorePropertiesSchema;
import org.apache.nifi.minifi.commons.schema.FlowControllerSchema;
import org.apache.nifi.minifi.commons.schema.FlowFileRepositorySchema;
+import org.apache.nifi.minifi.commons.schema.PortSchema;
+import org.apache.nifi.minifi.commons.schema.ProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
import org.apache.nifi.minifi.commons.schema.ProvenanceReportingSchema;
import org.apache.nifi.minifi.commons.schema.ProvenanceRepositorySchema;
import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessingGroupSchema;
+import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema;
+import org.apache.nifi.minifi.commons.schema.common.Schema;
import org.apache.nifi.minifi.commons.schema.common.StringUtil;
import org.apache.nifi.minifi.commons.schema.serialization.SchemaLoader;
import org.apache.nifi.minifi.commons.schema.SecurityPropertiesSchema;
@@ -68,13 +72,13 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Optional;
import java.util.stream.Collectors;
import java.util.zip.GZIPOutputStream;
public final class ConfigTransformer {
// Underlying version of NIFI will be using
public static final String NIFI_VERSION = "0.6.1";
+ public static final String ROOT_GROUP = "Root-Group";
// Final util classes should have private constructor
private ConfigTransformer() {
@@ -88,22 +92,25 @@ public final class ConfigTransformer {
}
public static void transformConfigFile(InputStream sourceStream, String destPath) throws Exception {
- ConfigSchema configSchema = SchemaLoader.loadConfigSchemaFromYaml(sourceStream);
- if (!configSchema.isValid()) {
- throw new InvalidConfigurationException("Failed to transform config file due to:["
- + configSchema.getValidationIssues().stream().sorted().collect(Collectors.joining("], [")) + "]");
- }
+ ConvertableSchema<ConfigSchema> convertableSchema = throwIfInvalid(SchemaLoader.loadConvertableSchemaFromYaml(sourceStream));
+ ConfigSchema configSchema = throwIfInvalid(convertableSchema.convert());
// Create nifi.properties and flow.xml.gz in memory
ByteArrayOutputStream nifiPropertiesOutputStream = new ByteArrayOutputStream();
writeNiFiProperties(configSchema, nifiPropertiesOutputStream);
- DOMSource flowXml = createFlowXml(configSchema);
+ writeFlowXmlFile(configSchema, destPath);
// Write nifi.properties and flow.xml.gz
writeNiFiPropertiesFile(nifiPropertiesOutputStream, destPath);
+ }
- writeFlowXmlFile(flowXml, destPath);
+ private static <T extends Schema> T throwIfInvalid(T schema) throws InvalidConfigurationException {
+ if (!schema.isValid()) {
+ throw new InvalidConfigurationException("Failed to transform config file due to:["
+ + schema.getValidationIssues().stream().sorted().collect(Collectors.joining("], [")) + "]");
+ }
+ return schema;
}
protected static void writeNiFiPropertiesFile(ByteArrayOutputStream nifiPropertiesOutputStream, String destPath) throws IOException {
@@ -118,10 +125,8 @@ public final class ConfigTransformer {
}
}
- protected static void writeFlowXmlFile(DOMSource domSource, String path) throws IOException, TransformerException {
- final OutputStream fileOut = Files.newOutputStream(Paths.get(path, "flow.xml.gz"));
- final OutputStream outStream = new GZIPOutputStream(fileOut);
- final StreamResult streamResult = new StreamResult(outStream);
+ protected static void writeFlowXmlFile(ConfigSchema configSchema, OutputStream outputStream) throws TransformerException, ConfigTransformerException, ConfigurationChangeException, IOException {
+ final StreamResult streamResult = new StreamResult(outputStream);
// configure the transformer and convert the DOM
final TransformerFactory transformFactory = TransformerFactory.newInstance();
@@ -130,9 +135,15 @@ public final class ConfigTransformer {
transformer.setOutputProperty(OutputKeys.INDENT, "yes");
// transform the document to byte stream
- transformer.transform(domSource, streamResult);
- outStream.flush();
- outStream.close();
+ transformer.transform(createFlowXml(configSchema), streamResult);
+ }
+
+ protected static void writeFlowXmlFile(ConfigSchema configSchema, String path) throws IOException, TransformerException, ConfigurationChangeException, ConfigTransformerException {
+ try (OutputStream fileOut = Files.newOutputStream(Paths.get(path, "flow.xml.gz"))) {
+ try (OutputStream outStream = new GZIPOutputStream(fileOut)) {
+ writeFlowXmlFile(configSchema, outStream);
+ }
+ }
}
protected static void writeNiFiProperties(ConfigSchema configSchema, OutputStream outputStream) throws FileNotFoundException, UnsupportedEncodingException, ConfigurationChangeException {
@@ -280,7 +291,18 @@ public final class ConfigTransformer {
CorePropertiesSchema coreProperties = configSchema.getCoreProperties();
addTextElement(rootNode, "maxTimerDrivenThreadCount", String.valueOf(coreProperties.getMaxConcurrentThreads()));
addTextElement(rootNode, "maxEventDrivenThreadCount", String.valueOf(coreProperties.getMaxConcurrentThreads()));
- addProcessGroup(rootNode, configSchema, "rootGroup");
+
+ FlowControllerSchema flowControllerProperties = configSchema.getFlowControllerProperties();
+
+ final Element element = doc.createElement("rootGroup");
+ rootNode.appendChild(element);
+
+ ProcessGroupSchema processGroupSchema = configSchema.getProcessGroupSchema();
+ processGroupSchema.setId(ROOT_GROUP);
+ processGroupSchema.setName(flowControllerProperties.getName());
+ processGroupSchema.setComment(flowControllerProperties.getComment());
+
+ addProcessGroup(doc, element, processGroupSchema, new ParentGroupIdResolver(processGroupSchema));
SecurityPropertiesSchema securityProperties = configSchema.getSecurityProperties();
if (securityProperties.useSSL()) {
@@ -331,37 +353,38 @@ public final class ConfigTransformer {
}
}
- protected static void addProcessGroup(final Element parentElement, ConfigSchema configSchema, final String elementName) throws ConfigurationChangeException {
+ protected static void addProcessGroup(Document doc, Element element, ProcessGroupSchema processGroupSchema, ParentGroupIdResolver parentGroupIdResolver) throws ConfigurationChangeException {
try {
- FlowControllerSchema flowControllerProperties = configSchema.getFlowControllerProperties();
-
- final Document doc = parentElement.getOwnerDocument();
- final Element element = doc.createElement(elementName);
- parentElement.appendChild(element);
- addTextElement(element, "id", "Root-Group");
- addTextElement(element, "name", flowControllerProperties.getName());
+ String processGroupId = processGroupSchema.getId();
+ addTextElement(element, "id", processGroupId);
+ addTextElement(element, "name", processGroupSchema.getName());
addPosition(element);
- addTextElement(element, "comment", flowControllerProperties.getComment());
+ addTextElement(element, "comment", processGroupSchema.getComment());
- List<ProcessorSchema> processors = configSchema.getProcessors();
- if (processors != null) {
- for (ProcessorSchema processorConfig : processors) {
- addProcessor(element, processorConfig);
- }
+ for (ProcessorSchema processorConfig : processGroupSchema.getProcessors()) {
+ addProcessor(element, processorConfig);
}
- List<RemoteProcessingGroupSchema> remoteProcessingGroups = configSchema.getRemoteProcessingGroups();
- if (remoteProcessingGroups != null) {
- for (RemoteProcessingGroupSchema remoteProcessingGroupSchema : remoteProcessingGroups) {
- addRemoteProcessGroup(element, remoteProcessingGroupSchema);
- }
+ for (RemoteProcessingGroupSchema remoteProcessingGroupSchema : processGroupSchema.getRemoteProcessingGroups()) {
+ addRemoteProcessGroup(element, remoteProcessingGroupSchema);
}
- List<ConnectionSchema> connections = configSchema.getConnections();
- if (connections != null) {
- for (ConnectionSchema connectionConfig : connections) {
- addConnection(element, connectionConfig, configSchema);
- }
+ for (PortSchema portSchema : processGroupSchema.getInputPortSchemas()) {
+ addPort(doc, element, portSchema, "inputPort");
+ }
+
+ for (PortSchema portSchema : processGroupSchema.getOutputPortSchemas()) {
+ addPort(doc, element, portSchema, "outputPort");
+ }
+
+ for (ProcessGroupSchema child : processGroupSchema.getProcessGroupSchemas()) {
+ Element processGroups = doc.createElement("processGroup");
+ element.appendChild(processGroups);
+ addProcessGroup(doc, processGroups, child, parentGroupIdResolver);
+ }
+
+ for (ConnectionSchema connectionConfig : processGroupSchema.getConnections()) {
+ addConnection(element, connectionConfig, parentGroupIdResolver);
}
} catch (ConfigurationChangeException e) {
throw e;
@@ -370,6 +393,19 @@ public final class ConfigTransformer {
}
}
+ protected static void addPort(Document doc, Element parentElement, PortSchema portSchema, String tag) {
+ Element element = doc.createElement(tag);
+ parentElement.appendChild(element);
+
+ addTextElement(element, "id", portSchema.getId());
+ addTextElement(element, "name", portSchema.getName());
+
+ addPosition(element);
+ addTextElement(element, "comments", null);
+
+ addTextElement(element, "scheduledState", "RUNNING");
+ }
+
protected static void addProcessor(final Element parentElement, ProcessorSchema processorConfig) throws ConfigurationChangeException {
try {
final Document doc = parentElement.getOwnerDocument();
@@ -511,7 +547,7 @@ public final class ConfigTransformer {
}
}
- protected static void addConnection(final Element parentElement, ConnectionSchema connectionProperties, ConfigSchema configSchema) throws ConfigurationChangeException {
+ protected static void addConnection(final Element parentElement, ConnectionSchema connectionProperties, ParentGroupIdResolver parentGroupIdResolver) throws ConfigurationChangeException {
try {
final Document doc = parentElement.getOwnerDocument();
final Element element = doc.createElement("connection");
@@ -526,23 +562,16 @@ public final class ConfigTransformer {
addTextElement(element, "labelIndex", "1");
addTextElement(element, "zIndex", "0");
- addTextElement(element, "sourceId", connectionProperties.getSourceId());
- addTextElement(element, "sourceGroupId", "Root-Group");
- addTextElement(element, "sourceType", "PROCESSOR");
+ addConnectionSourceOrDestination(element, "source", connectionProperties.getSourceId(), parentGroupIdResolver);
+ addConnectionSourceOrDestination(element, "destination", connectionProperties.getDestinationId(), parentGroupIdResolver);
- final String connectionDestinationId = connectionProperties.getDestinationId();
- addTextElement(element, "destinationId", connectionDestinationId);
- final Optional<String> parentGroup = findInputPortParentGroup(connectionDestinationId, configSchema);
- if (parentGroup.isPresent()) {
- addTextElement(element, "destinationGroupId", parentGroup.get());
- addTextElement(element, "destinationType", "REMOTE_INPUT_PORT");
+ List<String> sourceRelationshipNames = connectionProperties.getSourceRelationshipNames();
+ if (sourceRelationshipNames.isEmpty()) {
+ addTextElement(element, "relationship", null);
} else {
- addTextElement(element, "destinationGroupId", "Root-Group");
- addTextElement(element, "destinationType", "PROCESSOR");
- }
-
- for (String relationshipName : connectionProperties.getSourceRelationshipNames()) {
- addTextElement(element, "relationship", relationshipName);
+ for (String relationshipName : sourceRelationshipNames) {
+ addTextElement(element, "relationship", relationshipName);
+ }
}
addTextElement(element, "maxWorkQueueSize", String.valueOf(connectionProperties.getMaxWorkQueueSize()));
@@ -557,22 +586,36 @@ public final class ConfigTransformer {
}
}
- // Locate the associated parent group for a given input port by its id
- protected static Optional<String> findInputPortParentGroup(String inputPortId, ConfigSchema configSchema) {
- final List<RemoteProcessingGroupSchema> remoteProcessingGroups = configSchema.getRemoteProcessingGroups();
- if (remoteProcessingGroups != null) {
- for (final RemoteProcessingGroupSchema remoteProcessingGroupSchema : remoteProcessingGroups) {
- final List<RemoteInputPortSchema> remoteInputPorts = remoteProcessingGroupSchema.getInputPorts();
- for (final RemoteInputPortSchema remoteInputPortSchema : remoteInputPorts) {
- if (remoteInputPortSchema != null && inputPortId.equals(remoteInputPortSchema.getId())) {
- return Optional.of(remoteProcessingGroupSchema.getName());
-
- }
+ protected static void addConnectionSourceOrDestination(Element element, String sourceOrDestination, String id, ParentGroupIdResolver parentGroupIdResolver) {
+ String idTag = sourceOrDestination + "Id";
+ String groupIdTag = sourceOrDestination + "GroupId";
+ String typeTag = sourceOrDestination + "Type";
+
+ String parentId = parentGroupIdResolver.getRemoteInputPortParentId(id);
+ String type;
+
+ if (parentId != null) {
+ type = "REMOTE_INPUT_PORT";
+ } else {
+ parentId = parentGroupIdResolver.getInputPortParentId(id);
+ if (parentId != null) {
+ type = "INPUT_PORT";
+ } else {
+ parentId = parentGroupIdResolver.getOutputPortParentId(id);
+ if (parentId != null) {
+ type = "OUTPUT_PORT";
+ } else {
+ parentId = parentGroupIdResolver.getProcessorParentId(id);
+ type = "PROCESSOR";
}
}
}
- return Optional.empty();
+ addTextElement(element, idTag, id);
+ if (parentId != null) {
+ addTextElement(element, groupIdTag, parentId);
+ }
+ addTextElement(element, typeTag, type);
}
protected static void addPosition(final Element parentElement) {
@@ -583,9 +626,7 @@ public final class ConfigTransformer {
}
protected static void addTextElementIfNotNullOrEmpty(final Element element, final String name, final String value) {
- if (!StringUtil.isNullOrEmpty(value)) {
- addTextElement(element, name, value);
- }
+ StringUtil.doIfNotNullOrEmpty(value, s -> addTextElement(element, name, value));
}
protected static void addTextElement(final Element element, final String name, final String value) {
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolver.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolver.java b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolver.java
new file mode 100644
index 0000000..71088ee
--- /dev/null
+++ b/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolver.java
@@ -0,0 +1,87 @@
+/*
+ *
+ * * 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.minifi.bootstrap.util;
+
+import org.apache.nifi.minifi.commons.schema.ProcessGroupSchema;
+import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemoteProcessingGroupSchema;
+import org.apache.nifi.minifi.commons.schema.common.BaseSchemaWithIdAndName;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+
+public class ParentGroupIdResolver {
+ private final Map<String, String> processorIdToParentIdMap;
+ private final Map<String, String> inputPortIdToParentIdMap;
+ private final Map<String, String> outputPortIdToParentIdMap;
+ private final Map<String, String> remoteInputPortIdToParentIdMap;
+
+ public ParentGroupIdResolver(ProcessGroupSchema processGroupSchema) {
+ this.processorIdToParentIdMap = getParentIdMap(processGroupSchema, ProcessGroupSchema::getProcessors);
+ this.inputPortIdToParentIdMap = getParentIdMap(processGroupSchema, ProcessGroupSchema::getInputPortSchemas);
+ this.outputPortIdToParentIdMap = getParentIdMap(processGroupSchema, ProcessGroupSchema::getOutputPortSchemas);
+ this.remoteInputPortIdToParentIdMap = getRemoteInputPortParentIdMap(processGroupSchema);
+ }
+
+ protected static Map<String, String> getParentIdMap(ProcessGroupSchema processGroupSchema, Function<ProcessGroupSchema, Collection<? extends BaseSchemaWithIdAndName>> schemaAccessor) {
+ Map<String, String> map = new HashMap<>();
+ getParentIdMap(processGroupSchema, map, schemaAccessor);
+ return map;
+ }
+
+ protected static void getParentIdMap(ProcessGroupSchema processGroupSchema, Map<String, String> output, Function<ProcessGroupSchema,
+ Collection<? extends BaseSchemaWithIdAndName>> schemaAccessor) {
+ schemaAccessor.apply(processGroupSchema).forEach(p -> output.put(p.getId(), processGroupSchema.getId()));
+ processGroupSchema.getProcessGroupSchemas().forEach(p -> getParentIdMap(p, output, schemaAccessor));
+ }
+
+ protected static Map<String, String> getRemoteInputPortParentIdMap(ProcessGroupSchema processGroupSchema) {
+ Map<String, String> result = new HashMap<>();
+ getRemoteInputPortParentIdMap(processGroupSchema, result);
+ return result;
+ }
+
+ protected static void getRemoteInputPortParentIdMap(ProcessGroupSchema processGroupSchema, Map<String, String> output) {
+ for (RemoteProcessingGroupSchema remoteProcessingGroupSchema : processGroupSchema.getRemoteProcessingGroups()) {
+ for (RemoteInputPortSchema remoteInputPortSchema : remoteProcessingGroupSchema.getInputPorts()) {
+ output.put(remoteInputPortSchema.getId(), remoteProcessingGroupSchema.getName());
+ }
+ }
+ processGroupSchema.getProcessGroupSchemas().forEach(p -> getRemoteInputPortParentIdMap(p, output));
+ }
+
+ public String getRemoteInputPortParentId(String id) {
+ return remoteInputPortIdToParentIdMap.get(id);
+ }
+
+ public String getInputPortParentId(String id) {
+ return inputPortIdToParentIdMap.get(id);
+ }
+
+ public String getOutputPortParentId(String id) {
+ return outputPortIdToParentIdMap.get(id);
+ }
+
+ public String getProcessorParentId(String id) {
+ return processorIdToParentIdMap.get(id);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformerTest.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformerTest.java b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformerTest.java
index 40dbe10..cac9d16 100644
--- a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformerTest.java
+++ b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ConfigTransformerTest.java
@@ -20,20 +20,35 @@ package org.apache.nifi.minifi.bootstrap.util;
import org.apache.nifi.minifi.bootstrap.configuration.ConfigurationChangeException;
import org.apache.nifi.minifi.commons.schema.ConfigSchema;
import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
+import org.apache.nifi.minifi.commons.schema.PortSchema;
+import org.apache.nifi.minifi.commons.schema.ProcessGroupSchema;
+import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
+import org.apache.nifi.minifi.commons.schema.RemoteInputPortSchema;
+import org.apache.nifi.minifi.commons.schema.RemoteProcessingGroupSchema;
+import org.apache.nifi.minifi.commons.schema.common.StringUtil;
+import org.apache.nifi.minifi.commons.schema.serialization.SchemaLoader;
import org.apache.nifi.prioritizer.FirstInFirstOutPrioritizer;
import org.junit.Before;
import org.junit.Test;
+import org.w3c.dom.Document;
import org.w3c.dom.Element;
+import org.w3c.dom.NodeList;
+import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.xpath.XPath;
import javax.xml.xpath.XPathConstants;
import javax.xml.xpath.XPathExpressionException;
import javax.xml.xpath.XPathFactory;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
@@ -41,17 +56,21 @@ import static org.junit.Assert.assertNull;
public class ConfigTransformerTest {
private XPathFactory xPathFactory;
+ private Document document;
private Element config;
+ private DocumentBuilder documentBuilder;
@Before
public void setup() throws ParserConfigurationException {
- config = DocumentBuilderFactory.newInstance().newDocumentBuilder().newDocument().createElement("config");
+ documentBuilder = DocumentBuilderFactory.newInstance().newDocumentBuilder();
+ document = documentBuilder.newDocument();
+ config = document.createElement("config");
xPathFactory = XPathFactory.newInstance();
}
@Test
public void testNullQueuePrioritizerNotWritten() throws ConfigurationChangeException, XPathExpressionException {
- ConfigTransformer.addConnection(config, new ConnectionSchema(Collections.emptyMap()), new ConfigSchema(Collections.emptyMap()));
+ ConfigTransformer.addConnection(config, new ConnectionSchema(Collections.emptyMap()), new ParentGroupIdResolver(new ProcessGroupSchema(Collections.emptyMap(), ConfigSchema.TOP_LEVEL_NAME)));
XPath xpath = xPathFactory.newXPath();
String expression = "connection/queuePrioritizerClass";
assertNull(xpath.evaluate(expression, config, XPathConstants.NODE));
@@ -62,7 +81,7 @@ public class ConfigTransformerTest {
Map<String, Object> map = new HashMap<>();
map.put(ConnectionSchema.QUEUE_PRIORITIZER_CLASS_KEY, "");
- ConfigTransformer.addConnection(config, new ConnectionSchema(map), new ConfigSchema(Collections.emptyMap()));
+ ConfigTransformer.addConnection(config, new ConnectionSchema(map), new ParentGroupIdResolver(new ProcessGroupSchema(Collections.emptyMap(), ConfigSchema.TOP_LEVEL_NAME)));
XPath xpath = xPathFactory.newXPath();
String expression = "connection/queuePrioritizerClass";
assertNull(xpath.evaluate(expression, config, XPathConstants.NODE));
@@ -73,9 +92,148 @@ public class ConfigTransformerTest {
Map<String, Object> map = new HashMap<>();
map.put(ConnectionSchema.QUEUE_PRIORITIZER_CLASS_KEY, FirstInFirstOutPrioritizer.class.getCanonicalName());
- ConfigTransformer.addConnection(config, new ConnectionSchema(map), new ConfigSchema(Collections.emptyMap()));
+ ConfigTransformer.addConnection(config, new ConnectionSchema(map), new ParentGroupIdResolver(new ProcessGroupSchema(Collections.emptyMap(), ConfigSchema.TOP_LEVEL_NAME)));
XPath xpath = xPathFactory.newXPath();
String expression = "connection/queuePrioritizerClass/text()";
assertEquals(FirstInFirstOutPrioritizer.class.getCanonicalName(), xpath.evaluate(expression, config, XPathConstants.STRING));
}
+
+ @Test
+ public void testProcessGroupsTransform() throws Exception {
+ ConfigSchema configSchema = SchemaLoader.loadConfigSchemaFromYaml(ConfigTransformerTest.class.getClassLoader().getResourceAsStream("config-process-groups.yml"));
+
+ ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ ConfigTransformer.writeFlowXmlFile(configSchema, outputStream);
+ Document document = documentBuilder.parse(new ByteArrayInputStream(outputStream.toByteArray()));
+
+ testProcessGroup((Element) xPathFactory.newXPath().evaluate("flowController/rootGroup", document, XPathConstants.NODE), configSchema.getProcessGroupSchema());
+ }
+
+ private void testProcessGroup(Element element, ProcessGroupSchema processGroupSchema) throws XPathExpressionException {
+ assertEquals(processGroupSchema.getId(), getText(element, "id"));
+ assertEquals(processGroupSchema.getName(), getText(element, "name"));
+ assertEquals(nullToEmpty(processGroupSchema.getComment()), nullToEmpty(getText(element, "comment")));
+
+ NodeList processorElements = (NodeList) xPathFactory.newXPath().evaluate("processor", element, XPathConstants.NODESET);
+ assertEquals(processGroupSchema.getProcessors().size(), processorElements.getLength());
+ for (int i = 0; i < processorElements.getLength(); i++) {
+ testProcessor((Element) processorElements.item(i), processGroupSchema.getProcessors().get(i));
+ }
+
+ NodeList remoteProcessGroupElements = (NodeList) xPathFactory.newXPath().evaluate("remoteProcessGroup", element, XPathConstants.NODESET);
+ assertEquals(processGroupSchema.getRemoteProcessingGroups().size(), remoteProcessGroupElements.getLength());
+ for (int i = 0; i < remoteProcessGroupElements.getLength(); i++) {
+ testRemoteProcessGroups((Element) remoteProcessGroupElements.item(i), processGroupSchema.getRemoteProcessingGroups().get(i));
+ }
+
+ NodeList inputPortElements = (NodeList) xPathFactory.newXPath().evaluate("inputPort", element, XPathConstants.NODESET);
+ assertEquals(processGroupSchema.getInputPortSchemas().size(), inputPortElements.getLength());
+ for (int i = 0; i < inputPortElements.getLength(); i++) {
+ testPort((Element) inputPortElements.item(i), processGroupSchema.getInputPortSchemas().get(i));
+ }
+
+ NodeList outputPortElements = (NodeList) xPathFactory.newXPath().evaluate("outputPort", element, XPathConstants.NODESET);
+ assertEquals(processGroupSchema.getOutputPortSchemas().size(), outputPortElements.getLength());
+ for (int i = 0; i < outputPortElements.getLength(); i++) {
+ testPort((Element) outputPortElements.item(i), processGroupSchema.getOutputPortSchemas().get(i));
+ }
+
+ NodeList processGroupElements = (NodeList) xPathFactory.newXPath().evaluate("processGroup", element, XPathConstants.NODESET);
+ assertEquals(processGroupSchema.getProcessGroupSchemas().size(), processGroupElements.getLength());
+ for (int i = 0; i < processGroupElements.getLength(); i++) {
+ testProcessGroup((Element) processGroupElements.item(i), processGroupSchema.getProcessGroupSchemas().get(i));
+ }
+
+ NodeList connectionElements = (NodeList) xPathFactory.newXPath().evaluate("connection", element, XPathConstants.NODESET);
+ assertEquals(processGroupSchema.getConnections().size(), connectionElements.getLength());
+ for (int i = 0; i < connectionElements.getLength(); i++) {
+ testConnection((Element) connectionElements.item(i), processGroupSchema.getConnections().get(i));
+ }
+ }
+
+ private void testProcessor(Element element, ProcessorSchema processorSchema) throws XPathExpressionException {
+ assertEquals(processorSchema.getId(), getText(element, "id"));
+ assertEquals(processorSchema.getName(), getText(element, "name"));
+ assertEquals(processorSchema.getProcessorClass(), getText(element, "class"));
+ assertEquals(processorSchema.getMaxConcurrentTasks().toString(), getText(element, "maxConcurrentTasks"));
+ assertEquals(processorSchema.getSchedulingPeriod(), getText(element, "schedulingPeriod"));
+ assertEquals(processorSchema.getPenalizationPeriod(), getText(element, "penalizationPeriod"));
+ assertEquals(processorSchema.getYieldPeriod(), getText(element, "yieldPeriod"));
+ assertEquals(processorSchema.getSchedulingStrategy(), getText(element, "schedulingStrategy"));
+ assertEquals(processorSchema.getRunDurationNanos().toString(), getText(element, "runDurationNanos"));
+
+ testProperties(element, processorSchema.getProperties());
+ }
+
+ private void testRemoteProcessGroups(Element element, RemoteProcessingGroupSchema remoteProcessingGroupSchema) throws XPathExpressionException {
+ assertEquals(remoteProcessingGroupSchema.getName(), getText(element, "id"));
+ assertEquals(remoteProcessingGroupSchema.getName(), getText(element, "name"));
+ assertEquals(remoteProcessingGroupSchema.getComment(), getText(element, "comment"));
+ assertEquals(remoteProcessingGroupSchema.getUrl(), getText(element, "url"));
+ assertEquals(remoteProcessingGroupSchema.getTimeout(), getText(element, "timeout"));
+ assertEquals(remoteProcessingGroupSchema.getYieldPeriod(), getText(element, "yieldPeriod"));
+
+
+ NodeList inputPortElements = (NodeList) xPathFactory.newXPath().evaluate("inputPort", element, XPathConstants.NODESET);
+ assertEquals(remoteProcessingGroupSchema.getInputPorts().size(), inputPortElements.getLength());
+ for (int i = 0; i < inputPortElements.getLength(); i++) {
+ testRemoteInputPort((Element) inputPortElements.item(i), remoteProcessingGroupSchema.getInputPorts().get(i));
+ }
+ }
+
+ private void testRemoteInputPort(Element element, RemoteInputPortSchema remoteInputPortSchema) throws XPathExpressionException {
+ assertEquals(remoteInputPortSchema.getId(), getText(element, "id"));
+ assertEquals(remoteInputPortSchema.getName(), getText(element, "name"));
+ assertEquals(remoteInputPortSchema.getComment(), getText(element, "comment"));
+ assertEquals(remoteInputPortSchema.getMax_concurrent_tasks().toString(), getText(element, "maxConcurrentTasks"));
+ assertEquals(remoteInputPortSchema.getUseCompression(), Boolean.parseBoolean(getText(element, "useCompression")));
+ }
+
+ private void testPort(Element element, PortSchema portSchema) throws XPathExpressionException {
+ assertEquals(portSchema.getId(), getText(element, "id"));
+ assertEquals(portSchema.getName(), getText(element, "name"));
+ assertEquals("RUNNING", getText(element, "scheduledState"));
+ }
+
+ private void testConnection(Element element, ConnectionSchema connectionSchema) throws XPathExpressionException {
+ assertEquals(connectionSchema.getId(), getText(element, "id"));
+ assertEquals(connectionSchema.getName(), getText(element, "name"));
+
+ assertEquals(connectionSchema.getSourceId(), getText(element, "sourceId"));
+ assertEquals(connectionSchema.getDestinationId(), getText(element, "destinationId"));
+
+ NodeList relationshipNodes = (NodeList) xPathFactory.newXPath().evaluate("relationship", element, XPathConstants.NODESET);
+ Set<String> sourceRelationships = new HashSet<>();
+ for (int i = 0; i < relationshipNodes.getLength(); i++) {
+ String textContent = relationshipNodes.item(i).getTextContent();
+ if (!StringUtil.isNullOrEmpty(textContent)) {
+ sourceRelationships.add(textContent);
+ }
+ }
+
+ assertEquals(new HashSet<>(connectionSchema.getSourceRelationshipNames()), sourceRelationships);
+
+ assertEquals(connectionSchema.getMaxWorkQueueSize().toString(), getText(element, "maxWorkQueueSize"));
+ assertEquals(connectionSchema.getMaxWorkQueueDataSize(), getText(element, "maxWorkQueueDataSize"));
+ assertEquals(connectionSchema.getFlowfileExpiration(), getText(element, "flowFileExpiration"));
+ assertEquals(connectionSchema.getQueuePrioritizerClass(), getText(element, "queuePrioritizerClass"));
+ }
+
+ private void testProperties(Element element, Map<String, Object> expected) throws XPathExpressionException {
+ NodeList propertyElements = (NodeList) xPathFactory.newXPath().evaluate("property", element, XPathConstants.NODESET);
+ Map<String, String> properties = new HashMap<>();
+ for (int i = 0; i < propertyElements.getLength(); i++) {
+ Element item = (Element) propertyElements.item(i);
+ properties.put(getText(item, "name"), getText(item, "value"));
+ }
+ assertEquals(expected.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> nullToEmpty(e.getValue()))), properties);
+ }
+
+ private String getText(Element element, String path) throws XPathExpressionException {
+ return (String) xPathFactory.newXPath().evaluate(path + "/text()", element, XPathConstants.STRING);
+ }
+
+ private String nullToEmpty(Object val) {
+ return val == null ? "" : val.toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolverTest.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolverTest.java b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolverTest.java
new file mode 100644
index 0000000..6f557b7
--- /dev/null
+++ b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/ParentGroupIdResolverTest.java
@@ -0,0 +1,129 @@
+/*
+ *
+ * * 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.minifi.bootstrap.util;
+
+import org.apache.nifi.minifi.commons.schema.exception.SchemaLoaderException;
+import org.apache.nifi.minifi.commons.schema.serialization.SchemaLoader;
+import org.junit.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class ParentGroupIdResolverTest {
+
+ @Test
+ public void testRemoteInputPortParentId() throws IOException, SchemaLoaderException {
+ List<String> configLines = new ArrayList<>();
+ configLines.add("MiNiFi Config Version: 2");
+ configLines.add("Remote Processing Groups:");
+ configLines.add("- name: rpgOne");
+ configLines.add(" Input Ports:");
+ configLines.add(" - id: one");
+ configLines.add("Process Groups:");
+ configLines.add("- Remote Processing Groups:");
+ configLines.add(" - name: rpgTwo");
+ configLines.add(" Input Ports:");
+ configLines.add(" - id: two");
+ ParentGroupIdResolver parentGroupIdResolver = createParentGroupIdResolver(configLines);
+ assertEquals("rpgOne", parentGroupIdResolver.getRemoteInputPortParentId("one"));
+ assertEquals("rpgTwo", parentGroupIdResolver.getRemoteInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getInputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getOutputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getOutputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getProcessorParentId("one"));
+ assertNull(parentGroupIdResolver.getProcessorParentId("two"));
+ }
+
+ @Test
+ public void testInputPortParentId() throws IOException, SchemaLoaderException {
+ List<String> configLines = new ArrayList<>();
+ configLines.add("MiNiFi Config Version: 2");
+ configLines.add("Input Ports:");
+ configLines.add("- id: one");
+ configLines.add("Process Groups:");
+ configLines.add("- id: pgTwo");
+ configLines.add(" Input Ports:");
+ configLines.add(" - id: two");
+ ParentGroupIdResolver parentGroupIdResolver = createParentGroupIdResolver(configLines);
+ assertNull(parentGroupIdResolver.getRemoteInputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getRemoteInputPortParentId("two"));
+ assertEquals(ConfigTransformer.ROOT_GROUP, parentGroupIdResolver.getInputPortParentId("one"));
+ assertEquals("pgTwo", parentGroupIdResolver.getInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getOutputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getOutputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getProcessorParentId("one"));
+ assertNull(parentGroupIdResolver.getProcessorParentId("two"));
+ }
+
+ @Test
+ public void testOutputPortParentId() throws IOException, SchemaLoaderException {
+ List<String> configLines = new ArrayList<>();
+ configLines.add("MiNiFi Config Version: 2");
+ configLines.add("Output Ports:");
+ configLines.add("- id: one");
+ configLines.add("Process Groups:");
+ configLines.add("- id: pgTwo");
+ configLines.add(" Output Ports:");
+ configLines.add(" - id: two");
+ ParentGroupIdResolver parentGroupIdResolver = createParentGroupIdResolver(configLines);
+ assertNull(parentGroupIdResolver.getRemoteInputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getRemoteInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getInputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getInputPortParentId("two"));
+ assertEquals(ConfigTransformer.ROOT_GROUP, parentGroupIdResolver.getOutputPortParentId("one"));
+ assertEquals("pgTwo", parentGroupIdResolver.getOutputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getProcessorParentId("one"));
+ assertNull(parentGroupIdResolver.getProcessorParentId("two"));
+ }
+
+ @Test
+ public void testProcessorParentId() throws IOException, SchemaLoaderException {
+ List<String> configLines = new ArrayList<>();
+ configLines.add("MiNiFi Config Version: 2");
+ configLines.add("Processors:");
+ configLines.add("- id: one");
+ configLines.add("Process Groups:");
+ configLines.add("- id: pgTwo");
+ configLines.add(" Processors:");
+ configLines.add(" - id: two");
+ ParentGroupIdResolver parentGroupIdResolver = createParentGroupIdResolver(configLines);
+ assertNull(parentGroupIdResolver.getRemoteInputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getRemoteInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getInputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getInputPortParentId("two"));
+ assertNull(parentGroupIdResolver.getOutputPortParentId("one"));
+ assertNull(parentGroupIdResolver.getOutputPortParentId("two"));
+ assertEquals(ConfigTransformer.ROOT_GROUP, parentGroupIdResolver.getProcessorParentId("one"));
+ assertEquals("pgTwo", parentGroupIdResolver.getProcessorParentId("two"));
+ }
+
+ private ParentGroupIdResolver createParentGroupIdResolver(List<String> configLines) throws IOException, SchemaLoaderException {
+ return new ParentGroupIdResolver(SchemaLoader.loadConfigSchemaFromYaml(new ByteArrayInputStream(configLines.stream().collect(Collectors.joining("\n"))
+ .getBytes(StandardCharsets.UTF_8))).getProcessGroupSchema());
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/TestConfigTransformer.java
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/TestConfigTransformer.java b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/TestConfigTransformer.java
index 617da90..a0077fe 100644
--- a/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/TestConfigTransformer.java
+++ b/minifi-bootstrap/src/test/java/org/apache/nifi/minifi/bootstrap/util/TestConfigTransformer.java
@@ -18,8 +18,6 @@
package org.apache.nifi.minifi.bootstrap.util;
import org.apache.nifi.minifi.bootstrap.exception.InvalidConfigurationException;
-import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
-import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
import org.apache.nifi.minifi.commons.schema.exception.SchemaLoaderException;
import org.junit.Test;
@@ -217,7 +215,6 @@ public class TestConfigTransformer {
} catch (InvalidConfigurationException e){
assertEquals("Failed to transform config file due to:['class' in section 'Processors' because it was not found and it is required], " +
"['scheduling strategy' in section 'Provenance Reporting' because it is not a valid scheduling strategy], " +
- "[" + BaseSchema.getIssueText(ConnectionSchema.SOURCE_ID_KEY, "Connections", BaseSchema.IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED) + "], " +
"['source name' in section 'Connections' because it was not found and it is required]", e.getMessage());
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-bootstrap/src/test/resources/config-process-groups.yml
----------------------------------------------------------------------
diff --git a/minifi-bootstrap/src/test/resources/config-process-groups.yml b/minifi-bootstrap/src/test/resources/config-process-groups.yml
new file mode 100644
index 0000000..e0e5ef3
--- /dev/null
+++ b/minifi-bootstrap/src/test/resources/config-process-groups.yml
@@ -0,0 +1,276 @@
+# 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.
+
+MiNiFi Config Version: 2
+Flow Controller:
+ name: ProcessGroupsAndRemoteProcessGroups
+ comment: ''
+Core Properties:
+ flow controller graceful shutdown period: 10 sec
+ flow service write delay interval: 500 ms
+ administrative yield duration: 30 sec
+ bored yield duration: 10 millis
+ max concurrent threads: 1
+FlowFile Repository:
+ partitions: 256
+ checkpoint interval: 2 mins
+ always sync: false
+ Swap:
+ threshold: 20000
+ in period: 5 sec
+ in threads: 1
+ out period: 5 sec
+ out threads: 4
+Content Repository:
+ content claim max appendable size: 10 MB
+ content claim max flow files: 100
+ always sync: false
+Provenance Repository:
+ provenance rollover time: 1 min
+Component Status Repository:
+ buffer size: 1440
+ snapshot frequency: 1 min
+Security Properties:
+ keystore: ''
+ keystore type: ''
+ keystore password: ''
+ key password: ''
+ truststore: ''
+ truststore type: ''
+ truststore password: ''
+ ssl protocol: ''
+ Sensitive Props:
+ key:
+ algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
+ provider: BC
+Processors:
+- id: 207748d1-0158-1000-0000-000000000000
+ name: GenerateFlowFile
+ class: org.apache.nifi.processors.standard.GenerateFlowFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list: []
+ Properties:
+ Batch Size: '1'
+ Data Format: Binary
+ File Size: 1 b
+ Unique FlowFiles: 'false'
+- id: 2079e8bd-0158-1000-0000-000000000000
+ name: LogAttribute
+ class: org.apache.nifi.processors.standard.LogAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ - success
+ Properties:
+ Attributes to Ignore:
+ Attributes to Log:
+ Log Level: info
+ Log Payload: 'false'
+ Log prefix:
+- id: 2077ab1e-0158-1000-0000-000000000000
+ name: UpdateAttribute
+ class: org.apache.nifi.processors.attributes.UpdateAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list: []
+ Properties:
+ Delete Attributes Expression:
+ top: top
+Process Groups:
+- id: 207888b1-0158-1000-0000-000000000000
+ name: middle
+ Processors:
+ - id: 2078f34e-0158-1000-0000-000000000000
+ name: UpdateAttribute
+ class: org.apache.nifi.processors.attributes.UpdateAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list: []
+ Properties:
+ Delete Attributes Expression:
+ middle: middle
+ Process Groups:
+ - id: 20794cd4-0158-1000-0000-000000000000
+ name: bottom
+ Processors:
+ - id: 207a89ba-0158-1000-0000-000000000000
+ name: UpdateAttribute
+ class: org.apache.nifi.processors.attributes.UpdateAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 0 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list: []
+ Properties:
+ Delete Attributes Expression:
+ bottom: bottom
+ Process Groups: []
+ Input Ports:
+ - id: 207a5f50-0158-1000-0000-000000000000
+ name: input
+ Output Ports:
+ - id: 207a6d92-0158-1000-0000-000000000000
+ name: output
+ Connections:
+ - id: 21a6abb9-0158-1000-0000-000000000000
+ name: UpdateAttribute/success/21a39aba-0158-1000-a1a0-1b55bcddcd72
+ source id: 207a89ba-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 21a39aba-0158-1000-a1a0-1b55bcddcd72
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ - id: 207ad5e9-0158-1000-0000-000000000000
+ name: UpdateAttribute/success/null
+ source id: 207a89ba-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 207a6d92-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ - id: 207aca0d-0158-1000-0000-000000000000
+ name: null//UpdateAttribute
+ source id: 207a5f50-0158-1000-0000-000000000000
+ source relationship names: []
+ destination id: 207a89ba-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ Remote Processing Groups:
+ - name: http://localhost:9091/nifi
+ url: http://localhost:9091/nifi
+ comment: ''
+ timeout: 30 sec
+ yield period: 10 sec
+ Input Ports:
+ - id: 21a39aba-0158-1000-a1a0-1b55bcddcd72
+ name: input2
+ comment: ''
+ max concurrent tasks: 1
+ use compression: false
+ Input Ports:
+ - id: 2078c936-0158-1000-0000-000000000000
+ name: input
+ Output Ports:
+ - id: 2079b327-0158-1000-0000-000000000000
+ name: output
+ Connections:
+ - id: 21a5b1f1-0158-1000-0000-000000000000
+ name: UpdateAttribute/success/21a2fb5e-0158-1000-3b5e-5a7d3aaee01b
+ source id: 2078f34e-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 21a2fb5e-0158-1000-3b5e-5a7d3aaee01b
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ - id: 207b0eb1-0158-1000-0000-000000000000
+ name: UpdateAttribute/success/null
+ source id: 2078f34e-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 207a5f50-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ - id: 20792ec2-0158-1000-0000-000000000000
+ name: null//UpdateAttribute
+ source id: 2078c936-0158-1000-0000-000000000000
+ source relationship names: []
+ destination id: 2078f34e-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ - id: 207b1880-0158-1000-0000-000000000000
+ name: null//null
+ source id: 207a6d92-0158-1000-0000-000000000000
+ source relationship names: []
+ destination id: 2079b327-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+ Remote Processing Groups:
+ - name: http://localhost:9090/nifi
+ url: http://localhost:9090/nifi
+ comment: ''
+ timeout: 30 sec
+ yield period: 10 sec
+ Input Ports:
+ - id: 21a2fb5e-0158-1000-3b5e-5a7d3aaee01b
+ name: input
+ comment: ''
+ max concurrent tasks: 1
+ use compression: false
+Input Ports: []
+Output Ports: []
+Connections:
+- id: 2077bf8f-0158-1000-0000-000000000000
+ name: GenerateFlowFile/success/UpdateAttribute
+ source id: 207748d1-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 2077ab1e-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+- id: 2079cf6f-0158-1000-0000-000000000000
+ name: UpdateAttribute/success/null
+ source id: 2077ab1e-0158-1000-0000-000000000000
+ source relationship names:
+ - success
+ destination id: 2078c936-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+- id: 2079faa0-0158-1000-0000-000000000000
+ name: null//LogAttribute
+ source id: 2079b327-0158-1000-0000-000000000000
+ source relationship names: []
+ destination id: 2079e8bd-0158-1000-0000-000000000000
+ max work queue size: 10000
+ max work queue data size: 1 GB
+ flowfile expiration: 0 sec
+ queue prioritizer class: ''
+Remote Processing Groups: []
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConfigSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConfigSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConfigSchema.java
index 8dfd9d4..abd6a6c 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConfigSchema.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConfigSchema.java
@@ -23,6 +23,8 @@ import org.apache.nifi.minifi.commons.schema.common.StringUtil;
import org.apache.nifi.minifi.commons.schema.common.WritableSchema;
import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
@@ -31,31 +33,27 @@ import java.util.Set;
import java.util.stream.Collectors;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.COMPONENT_STATUS_REPO_KEY;
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONNECTIONS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONTENT_REPO_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CORE_PROPS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.FLOWFILE_REPO_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.FLOW_CONTROLLER_PROPS_KEY;
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROCESSORS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROVENANCE_REPORTING_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROVENANCE_REPO_KEY;
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.REMOTE_PROCESSING_GROUPS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SECURITY_PROPS_KEY;
-/**
- *
- */
public class ConfigSchema extends BaseSchema implements WritableSchema, ConvertableSchema<ConfigSchema> {
- public static final String FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_IDS = "Found the following duplicate processor ids: ";
- public static final String FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS = "Found the following duplicate connection ids: ";
- public static final String FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_PROCESSING_GROUP_NAMES = "Found the following duplicate remote processing group names: ";
- public static final String FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_INPUT_PORT_IDS = "Found the following duplicate remote input port ids: ";
- public static final String FOUND_THE_FOLLOWING_DUPLICATE_IDS = "Found the following ids that occur both in Processors and Remote Input Ports: ";
public static final int CONFIG_VERSION = 2;
public static final String VERSION = "MiNiFi Config Version";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_INPUT_PORT_IDS = "Found the following duplicate remote input port ids: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_INPUT_PORT_IDS = "Found the following duplicate input port ids: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_OUTPUT_PORT_IDS = "Found the following duplicate output port ids: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_IDS = "Found the following ids that occur both in more than one Processor(s), Input Port(s), Output Port(s) and/or Remote Input Port(s): ";
public static final String CONNECTION_WITH_ID = "Connection with id ";
public static final String HAS_INVALID_SOURCE_ID = " has invalid source id ";
public static final String HAS_INVALID_DESTINATION_ID = " has invalid destination id ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_IDS = "Found the following duplicate processor ids: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS = "Found the following duplicate connection ids: ";
+ public static final String FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_PROCESSING_GROUP_NAMES = "Found the following duplicate remote processing group names: ";
public static String TOP_LEVEL_NAME = "top level";
private FlowControllerSchema flowControllerProperties;
private CorePropertiesSchema coreProperties;
@@ -63,9 +61,7 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
private ContentRepositorySchema contentRepositoryProperties;
private ComponentStatusRepositorySchema componentStatusRepositoryProperties;
private SecurityPropertiesSchema securityProperties;
- private List<ProcessorSchema> processors;
- private List<ConnectionSchema> connections;
- private List<RemoteProcessingGroupSchema> remoteProcessingGroups;
+ private ProcessGroupSchema processGroupSchema;
private ProvenanceReportingSchema provenanceReportingProperties;
private ProvenanceRepositorySchema provenanceRepositorySchema;
@@ -85,13 +81,7 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
componentStatusRepositoryProperties = getMapAsType(map, COMPONENT_STATUS_REPO_KEY, ComponentStatusRepositorySchema.class, TOP_LEVEL_NAME, false);
securityProperties = getMapAsType(map, SECURITY_PROPS_KEY, SecurityPropertiesSchema.class, TOP_LEVEL_NAME, false);
- processors = convertListToType(getOptionalKeyAsType(map, PROCESSORS_KEY, List.class, TOP_LEVEL_NAME, new ArrayList<>()), "processor", ProcessorSchema.class, PROCESSORS_KEY);
-
- remoteProcessingGroups = convertListToType(getOptionalKeyAsType(map, REMOTE_PROCESSING_GROUPS_KEY, List.class, TOP_LEVEL_NAME, new ArrayList<>()), "remote processing group",
- RemoteProcessingGroupSchema.class, REMOTE_PROCESSING_GROUPS_KEY);
-
- connections = convertListToType(getOptionalKeyAsType(map, CONNECTIONS_KEY, List.class, TOP_LEVEL_NAME, new ArrayList<>()),
- "connection", ConnectionSchema.class, CONNECTIONS_KEY);
+ processGroupSchema = new ProcessGroupSchema(map, TOP_LEVEL_NAME);
provenanceReportingProperties = getMapAsType(map, PROVENANCE_REPORTING_KEY, ProvenanceReportingSchema.class, TOP_LEVEL_NAME, false, false);
@@ -101,47 +91,63 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
addIssuesIfNotNull(contentRepositoryProperties);
addIssuesIfNotNull(componentStatusRepositoryProperties);
addIssuesIfNotNull(securityProperties);
+ addIssuesIfNotNull(processGroupSchema);
addIssuesIfNotNull(provenanceReportingProperties);
addIssuesIfNotNull(provenanceRepositorySchema);
- addIssuesIfNotNull(processors);
- addIssuesIfNotNull(connections);
- addIssuesIfNotNull(remoteProcessingGroups);
- Set<String> processorIds = new HashSet<>();
- List<String> processorIdList = processors.stream().map(ProcessorSchema::getId).collect(Collectors.toList());
- processorIds.addAll(processorIdList);
+ List<ProcessGroupSchema> allProcessGroups = getAllProcessGroups(processGroupSchema);
+ List<ConnectionSchema> allConnectionSchemas = allProcessGroups.stream().flatMap(p -> p.getConnections().stream()).collect(Collectors.toList());
+ List<RemoteProcessingGroupSchema> allRemoteProcessingGroups = allProcessGroups.stream().flatMap(p -> p.getRemoteProcessingGroups().stream()).collect(Collectors.toList());
- checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_IDS, processorIdList);
- checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS, connections.stream().map(ConnectionSchema::getId).collect(Collectors.toList()));
- checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_PROCESSING_GROUP_NAMES,
- remoteProcessingGroups.stream().map(RemoteProcessingGroupSchema::getName).collect(Collectors.toList()));
-
- Set<String> remoteInputPortIds = new HashSet<>();
- List<String> remoteInputPortIdList = remoteProcessingGroups.stream().filter(r -> r.getInputPorts() != null)
+ List<String> allProcessorIds = allProcessGroups.stream().flatMap(p -> p.getProcessors().stream()).map(ProcessorSchema::getId).collect(Collectors.toList());
+ List<String> allConnectionIds = allConnectionSchemas.stream().map(ConnectionSchema::getId).collect(Collectors.toList());
+ List<String> allRemoteProcessingGroupNames = allRemoteProcessingGroups.stream().map(RemoteProcessingGroupSchema::getName).collect(Collectors.toList());
+ List<String> allRemoteInputPortIds = allRemoteProcessingGroups.stream().filter(r -> r.getInputPorts() != null)
.flatMap(r -> r.getInputPorts().stream()).map(RemoteInputPortSchema::getId).collect(Collectors.toList());
- checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_INPUT_PORT_IDS, remoteInputPortIdList);
- remoteInputPortIds.addAll(remoteInputPortIdList);
-
- Set<String> duplicateIds = new HashSet<>(processorIds);
- duplicateIds.retainAll(remoteInputPortIds);
- if (duplicateIds.size() > 0) {
- addValidationIssue(FOUND_THE_FOLLOWING_DUPLICATE_IDS + duplicateIds.stream().sorted().collect(Collectors.joining(", ")));
+ List<String> allInputPortIds = allProcessGroups.stream().flatMap(p -> p.getInputPortSchemas().stream()).map(PortSchema::getId).collect(Collectors.toList());
+ List<String> allOutputPortIds = allProcessGroups.stream().flatMap(p -> p.getOutputPortSchemas().stream()).map(PortSchema::getId).collect(Collectors.toList());
+
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_PROCESSOR_IDS, allProcessorIds);
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS, allConnectionIds);
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_PROCESSING_GROUP_NAMES, allRemoteProcessingGroupNames);
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_REMOTE_INPUT_PORT_IDS, allRemoteInputPortIds);
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_INPUT_PORT_IDS, allInputPortIds);
+ checkForDuplicates(this::addValidationIssue, FOUND_THE_FOLLOWING_DUPLICATE_OUTPUT_PORT_IDS, allOutputPortIds);
+
+ // Potential connection sources and destinations need to have unique ids
+ OverlapResults<String> overlapResults = findOverlap(new HashSet<>(allProcessorIds), new HashSet<>(allRemoteInputPortIds), new HashSet<>(allInputPortIds), new HashSet<>(allOutputPortIds));
+ if (overlapResults.duplicates.size() > 0) {
+ addValidationIssue(FOUND_THE_FOLLOWING_DUPLICATE_IDS + overlapResults.duplicates.stream().sorted().collect(Collectors.joining(", ")));
}
- Set<String> connectableIds = new HashSet<>(processorIds);
- connectableIds.addAll(remoteInputPortIds);
- connections.forEach(c -> {
+ allConnectionSchemas.forEach(c -> {
String destinationId = c.getDestinationId();
- if (!StringUtil.isNullOrEmpty(destinationId) && !connectableIds.contains(destinationId)) {
+ if (!StringUtil.isNullOrEmpty(destinationId) && !overlapResults.seen.contains(destinationId)) {
addValidationIssue(CONNECTION_WITH_ID + c.getId() + HAS_INVALID_DESTINATION_ID + destinationId);
}
String sourceId = c.getSourceId();
- if (!StringUtil.isNullOrEmpty(sourceId) && !connectableIds.contains(sourceId)) {
+ if (!StringUtil.isNullOrEmpty(sourceId) && !overlapResults.seen.contains(sourceId)) {
addValidationIssue(CONNECTION_WITH_ID + c.getId() + HAS_INVALID_SOURCE_ID + sourceId);
}
});
}
+ protected static <T> OverlapResults<T> findOverlap(Collection<T>... collections) {
+ Set<T> seen = new HashSet<>();
+ return new OverlapResults<>(seen, Arrays.stream(collections).flatMap(c -> c.stream()).sequential().filter(s -> !seen.add(s)).collect(Collectors.toSet()));
+ }
+
+ public static List<ProcessGroupSchema> getAllProcessGroups(ProcessGroupSchema processGroupSchema) {
+ List<ProcessGroupSchema> result = new ArrayList<>();
+ addProcessGroups(processGroupSchema, result);
+ return result;
+ }
+
+ private static void addProcessGroups(ProcessGroupSchema processGroupSchema, List<ProcessGroupSchema> result) {
+ result.add(processGroupSchema);
+ processGroupSchema.getProcessGroupSchemas().forEach(p -> addProcessGroups(p, result));
+ }
+
public Map<String, Object> toMap() {
Map<String, Object> result = mapSupplier.get();
result.put(VERSION, getVersion());
@@ -152,9 +158,7 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
putIfNotNull(result, PROVENANCE_REPO_KEY, provenanceRepositorySchema);
putIfNotNull(result, COMPONENT_STATUS_REPO_KEY, componentStatusRepositoryProperties);
putIfNotNull(result, SECURITY_PROPS_KEY, securityProperties);
- putListIfNotNull(result, PROCESSORS_KEY, processors);
- putListIfNotNull(result, CONNECTIONS_KEY, connections);
- putListIfNotNull(result, REMOTE_PROCESSING_GROUPS_KEY, remoteProcessingGroups);
+ result.putAll(processGroupSchema.toMap());
putIfNotNull(result, PROVENANCE_REPORTING_KEY, provenanceReportingProperties);
return result;
}
@@ -179,16 +183,8 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
return securityProperties;
}
- public List<ProcessorSchema> getProcessors() {
- return processors;
- }
-
- public List<ConnectionSchema> getConnections() {
- return connections;
- }
-
- public List<RemoteProcessingGroupSchema> getRemoteProcessingGroups() {
- return remoteProcessingGroups;
+ public ProcessGroupSchema getProcessGroupSchema() {
+ return processGroupSchema;
}
public ProvenanceReportingSchema getProvenanceReportingProperties() {
@@ -212,4 +208,14 @@ public class ConfigSchema extends BaseSchema implements WritableSchema, Converta
public ConfigSchema convert() {
return this;
}
+
+ private static class OverlapResults<T> {
+ private final Set<T> seen;
+ private final Set<T> duplicates;
+
+ private OverlapResults(Set<T> seen, Set<T> duplicates) {
+ this.seen = seen;
+ this.duplicates = duplicates;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConnectionSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConnectionSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConnectionSchema.java
index 768213c..47a87d9 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConnectionSchema.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ConnectionSchema.java
@@ -25,8 +25,6 @@ import java.util.Collections;
import java.util.List;
import java.util.Map;
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONNECTIONS_KEY;
-
public class ConnectionSchema extends BaseSchemaWithIdAndName {
public static final String SOURCE_ID_KEY = "source id";
public static final String SOURCE_RELATIONSHIP_NAMES_KEY = "source relationship names";
@@ -41,6 +39,7 @@ public class ConnectionSchema extends BaseSchemaWithIdAndName {
public static final String DEFAULT_FLOWFILE_EXPIRATION = "0 sec";
private String sourceId;
+ private boolean needsSourceRelationships = true;
private List<String> sourceRelationshipNames;
private String destinationId;
@@ -50,21 +49,19 @@ public class ConnectionSchema extends BaseSchemaWithIdAndName {
private String queuePrioritizerClass;
public ConnectionSchema(Map map) {
- super(map, CONNECTIONS_KEY);
+ super(map, "Connection(id: {id}, name: {name})");
+ String wrapperName = getWrapperName();
// In case of older version, these may not be available until after construction, validated in getValidationIssues()
- sourceId = getOptionalKeyAsType(map, SOURCE_ID_KEY, String.class, CONNECTIONS_KEY, "");
- destinationId = getOptionalKeyAsType(map, DESTINATION_ID_KEY, String.class, CONNECTIONS_KEY, "");
-
- sourceRelationshipNames = getOptionalKeyAsType(map, SOURCE_RELATIONSHIP_NAMES_KEY, List.class, CONNECTIONS_KEY, new ArrayList<>());
- if (sourceRelationshipNames.isEmpty()) {
- addValidationIssue("Expected at least one value in " + SOURCE_RELATIONSHIP_NAMES_KEY + " for " + CONNECTIONS_KEY + " " + getName());
- }
-
- maxWorkQueueSize = getOptionalKeyAsType(map, MAX_WORK_QUEUE_SIZE_KEY, Number.class, CONNECTIONS_KEY, DEFAULT_MAX_WORK_QUEUE_SIZE);
- maxWorkQueueDataSize = getOptionalKeyAsType(map, MAX_WORK_QUEUE_DATA_SIZE_KEY, String.class, CONNECTIONS_KEY, DEFAULT_MAX_QUEUE_DATA_SIZE);
- flowfileExpiration = getOptionalKeyAsType(map, FLOWFILE_EXPIRATION__KEY, String.class, CONNECTIONS_KEY, DEFAULT_FLOWFILE_EXPIRATION);
- queuePrioritizerClass = getOptionalKeyAsType(map, QUEUE_PRIORITIZER_CLASS_KEY, String.class, CONNECTIONS_KEY, "");
+ sourceId = getOptionalKeyAsType(map, SOURCE_ID_KEY, String.class, wrapperName, "");
+ destinationId = getOptionalKeyAsType(map, DESTINATION_ID_KEY, String.class, wrapperName, "");
+
+ // This could be empty if the source is a port.
+ sourceRelationshipNames = getOptionalKeyAsType(map, SOURCE_RELATIONSHIP_NAMES_KEY, List.class, wrapperName, new ArrayList<>());
+ maxWorkQueueSize = getOptionalKeyAsType(map, MAX_WORK_QUEUE_SIZE_KEY, Number.class, wrapperName, DEFAULT_MAX_WORK_QUEUE_SIZE);
+ maxWorkQueueDataSize = getOptionalKeyAsType(map, MAX_WORK_QUEUE_DATA_SIZE_KEY, String.class, wrapperName, DEFAULT_MAX_QUEUE_DATA_SIZE);
+ flowfileExpiration = getOptionalKeyAsType(map, FLOWFILE_EXPIRATION__KEY, String.class, wrapperName, DEFAULT_FLOWFILE_EXPIRATION);
+ queuePrioritizerClass = getOptionalKeyAsType(map, QUEUE_PRIORITIZER_CLASS_KEY, String.class, wrapperName, "");
}
@Override
@@ -117,14 +114,18 @@ public class ConnectionSchema extends BaseSchemaWithIdAndName {
return queuePrioritizerClass;
}
+ public void setNeedsSourceRelationships(boolean needsSourceRelationships) {
+ this.needsSourceRelationships = needsSourceRelationships;
+ }
+
@Override
public List<String> getValidationIssues() {
+ String wrapperName = getWrapperName();
List<String> validationIssues = super.getValidationIssues();
- if (StringUtil.isNullOrEmpty(getSourceId())) {
- validationIssues.add(getIssueText(SOURCE_ID_KEY, CONNECTIONS_KEY, IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED));
- }
- if (StringUtil.isNullOrEmpty(getDestinationId())) {
- validationIssues.add(getIssueText(DESTINATION_ID_KEY, CONNECTIONS_KEY, IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED));
+ StringUtil.doIfNullOrEmpty(getSourceId(), id -> validationIssues.add(getIssueText(SOURCE_ID_KEY, wrapperName, IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED)));
+ StringUtil.doIfNullOrEmpty(getDestinationId(), id -> validationIssues.add(getIssueText(DESTINATION_ID_KEY, wrapperName, IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED)));
+ if (needsSourceRelationships && sourceRelationshipNames.isEmpty()) {
+ validationIssues.add("Expected at least one value in " + SOURCE_RELATIONSHIP_NAMES_KEY + " for " + wrapperName + " " + getName());
}
return Collections.unmodifiableList(validationIssues);
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/PortSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/PortSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/PortSchema.java
new file mode 100644
index 0000000..27c6afc
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/PortSchema.java
@@ -0,0 +1,31 @@
+/*
+ *
+ * * 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.minifi.commons.schema;
+
+import org.apache.nifi.minifi.commons.schema.common.BaseSchemaWithIdAndName;
+
+import java.util.Map;
+
+public class PortSchema extends BaseSchemaWithIdAndName {
+
+ public PortSchema(Map map, String wrapperName) {
+ super(map, wrapperName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchema.java
new file mode 100644
index 0000000..88688fc
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchema.java
@@ -0,0 +1,153 @@
+/*
+ *
+ * * 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.minifi.commons.schema;
+
+import org.apache.nifi.minifi.commons.schema.common.BaseSchemaWithIdAndName;
+import org.apache.nifi.minifi.commons.schema.common.ConvertableSchema;
+import org.apache.nifi.minifi.commons.schema.common.StringUtil;
+import org.apache.nifi.minifi.commons.schema.common.WritableSchema;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONNECTIONS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ID_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.INPUT_PORTS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.OUTPUT_PORTS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROCESSORS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.REMOTE_PROCESSING_GROUPS_KEY;
+
+public class ProcessGroupSchema extends BaseSchemaWithIdAndName implements WritableSchema, ConvertableSchema<ProcessGroupSchema> {
+
+ public static final String PROCESS_GROUPS_KEY = "Process Groups";
+ public static final String ID_DEFAULT = "Root-Group";
+
+ private String comment;
+ private List<ProcessorSchema> processors;
+ private List<ConnectionSchema> connections;
+ private List<RemoteProcessingGroupSchema> remoteProcessingGroups;
+ private List<ProcessGroupSchema> processGroupSchemas;
+ private List<PortSchema> inputPortSchemas;
+ private List<PortSchema> outputPortSchemas;
+
+ public ProcessGroupSchema(Map map, String wrapperName) {
+ super(map, wrapperName);
+
+ processors = getOptionalKeyAsList(map, PROCESSORS_KEY, ProcessorSchema::new, wrapperName);
+ remoteProcessingGroups = getOptionalKeyAsList(map, REMOTE_PROCESSING_GROUPS_KEY, RemoteProcessingGroupSchema::new, wrapperName);
+ connections = getOptionalKeyAsList(map, CONNECTIONS_KEY, ConnectionSchema::new, wrapperName);
+ inputPortSchemas = getOptionalKeyAsList(map, INPUT_PORTS_KEY, m -> new PortSchema(m, "InputPort(id: {id}, name: {name})"), wrapperName);
+ outputPortSchemas = getOptionalKeyAsList(map, OUTPUT_PORTS_KEY, m -> new PortSchema(m, "OutputPort(id: {id}, name: {name})"), wrapperName);
+ processGroupSchemas = getOptionalKeyAsList(map, PROCESS_GROUPS_KEY, m -> new ProcessGroupSchema(m, "ProcessGroup(id: {id}, name: {name})"), wrapperName);
+
+ if (ConfigSchema.TOP_LEVEL_NAME.equals(wrapperName)) {
+ if (inputPortSchemas.size() > 0) {
+ addValidationIssue(INPUT_PORTS_KEY, wrapperName, "must be empty in root group as external input/output ports are currently unsupported");
+ }
+ if (outputPortSchemas.size() > 0) {
+ addValidationIssue(OUTPUT_PORTS_KEY, wrapperName, "must be empty in root group as external input/output ports are currently unsupported");
+ }
+ } else if (ID_DEFAULT.equals(getId())) {
+ addValidationIssue(ID_KEY, wrapperName, "must be set to a value not " + ID_DEFAULT + " if not in root group");
+ }
+
+ Set<String> portIds = getPortIds();
+ connections.stream().filter(c -> portIds.contains(c.getSourceId())).forEachOrdered(c -> c.setNeedsSourceRelationships(false));
+
+ addIssuesIfNotNull(processors);
+ addIssuesIfNotNull(remoteProcessingGroups);
+ addIssuesIfNotNull(processGroupSchemas);
+ addIssuesIfNotNull(connections);
+ }
+
+ public Map<String, Object> toMap() {
+ Map<String, Object> result = mapSupplier.get();
+ String id = getId();
+ if (!ID_DEFAULT.equals(id)) {
+ result.put(ID_KEY, id);
+ }
+ StringUtil.doIfNotNullOrEmpty(getName(), name -> result.put(NAME_KEY, name));
+ putListIfNotNull(result, PROCESSORS_KEY, processors);
+ putListIfNotNull(result, PROCESS_GROUPS_KEY, processGroupSchemas);
+ putListIfNotNull(result, INPUT_PORTS_KEY, inputPortSchemas);
+ putListIfNotNull(result, OUTPUT_PORTS_KEY, outputPortSchemas);
+ putListIfNotNull(result, CONNECTIONS_KEY, connections);
+ putListIfNotNull(result, REMOTE_PROCESSING_GROUPS_KEY, remoteProcessingGroups);
+ return result;
+ }
+
+ public List<ProcessorSchema> getProcessors() {
+ return processors;
+ }
+
+ public List<ConnectionSchema> getConnections() {
+ return connections;
+ }
+
+ public List<RemoteProcessingGroupSchema> getRemoteProcessingGroups() {
+ return remoteProcessingGroups;
+ }
+
+ public List<ProcessGroupSchema> getProcessGroupSchemas() {
+ return processGroupSchemas;
+ }
+
+ public Set<String> getPortIds() {
+ Set<String> result = new HashSet<>();
+ inputPortSchemas.stream().map(PortSchema::getId).forEachOrdered(result::add);
+ outputPortSchemas.stream().map(PortSchema::getId).forEachOrdered(result::add);
+ processGroupSchemas.stream().flatMap(p -> p.getPortIds().stream()).forEachOrdered(result::add);
+ return result;
+ }
+
+ public String getComment() {
+ return comment;
+ }
+
+ public void setComment(String comment) {
+ this.comment = comment;
+ }
+
+ @Override
+ protected String getId(Map map, String wrapperName) {
+ return getOptionalKeyAsType(map, ID_KEY, String.class, wrapperName, ID_DEFAULT);
+ }
+
+ @Override
+ public ProcessGroupSchema convert() {
+ return this;
+ }
+
+ @Override
+ public int getVersion() {
+ return ConfigSchema.CONFIG_VERSION;
+ }
+
+ public List<PortSchema> getOutputPortSchemas() {
+ return outputPortSchemas;
+ }
+
+ public List<PortSchema> getInputPortSchemas() {
+ return inputPortSchemas;
+ }
+}
[2/3] nifi-minifi git commit: MINIFI-107 - Process group support
Posted by jp...@apache.org.
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessorSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessorSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessorSchema.java
index 048027c..6f2ff8e 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessorSchema.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/ProcessorSchema.java
@@ -26,7 +26,6 @@ import java.util.Map;
import java.util.TreeMap;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.PROCESSORS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_PERIOD_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_STRATEGY_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.YIELD_PERIOD_KEY;
@@ -59,22 +58,23 @@ public class ProcessorSchema extends BaseSchemaWithIdAndName {
private String annotationData = "";
public ProcessorSchema(Map map) {
- super(map, PROCESSORS_KEY);
- processorClass = getRequiredKeyAsType(map, CLASS_KEY, String.class, PROCESSORS_KEY);
- schedulingStrategy = getRequiredKeyAsType(map, SCHEDULING_STRATEGY_KEY, String.class, PROCESSORS_KEY);
+ super(map, "Processor(id: {id}, name: {name})");
+ String wrapperName = getWrapperName();
+ processorClass = getRequiredKeyAsType(map, CLASS_KEY, String.class, wrapperName);
+ schedulingStrategy = getRequiredKeyAsType(map, SCHEDULING_STRATEGY_KEY, String.class, wrapperName);
if (schedulingStrategy != null && !isSchedulingStrategy(schedulingStrategy)) {
- addValidationIssue(SCHEDULING_STRATEGY_KEY, PROCESSORS_KEY, IT_IS_NOT_A_VALID_SCHEDULING_STRATEGY);
+ addValidationIssue(SCHEDULING_STRATEGY_KEY, wrapperName, IT_IS_NOT_A_VALID_SCHEDULING_STRATEGY);
}
- schedulingPeriod = getRequiredKeyAsType(map, SCHEDULING_PERIOD_KEY, String.class, PROCESSORS_KEY);
+ schedulingPeriod = getRequiredKeyAsType(map, SCHEDULING_PERIOD_KEY, String.class, wrapperName);
- maxConcurrentTasks = getOptionalKeyAsType(map, MAX_CONCURRENT_TASKS_KEY, Number.class, PROCESSORS_KEY, DEFAULT_MAX_CONCURRENT_TASKS);
- penalizationPeriod = getOptionalKeyAsType(map, PENALIZATION_PERIOD_KEY, String.class, PROCESSORS_KEY, DEFAULT_PENALIZATION_PERIOD);
- yieldPeriod = getOptionalKeyAsType(map, YIELD_PERIOD_KEY, String.class, PROCESSORS_KEY, DEFAULT_YIELD_DURATION);
- runDurationNanos = getOptionalKeyAsType(map, RUN_DURATION_NANOS_KEY, Number.class, PROCESSORS_KEY, DEFAULT_RUN_DURATION_NANOS);
- autoTerminatedRelationshipsList = getOptionalKeyAsType(map, AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, List.class, PROCESSORS_KEY, DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST);
- properties = getOptionalKeyAsType(map, PROCESSOR_PROPS_KEY, Map.class, PROCESSORS_KEY, DEFAULT_PROPERTIES);
+ maxConcurrentTasks = getOptionalKeyAsType(map, MAX_CONCURRENT_TASKS_KEY, Number.class, wrapperName, DEFAULT_MAX_CONCURRENT_TASKS);
+ penalizationPeriod = getOptionalKeyAsType(map, PENALIZATION_PERIOD_KEY, String.class, wrapperName, DEFAULT_PENALIZATION_PERIOD);
+ yieldPeriod = getOptionalKeyAsType(map, YIELD_PERIOD_KEY, String.class, wrapperName, DEFAULT_YIELD_DURATION);
+ runDurationNanos = getOptionalKeyAsType(map, RUN_DURATION_NANOS_KEY, Number.class, wrapperName, DEFAULT_RUN_DURATION_NANOS);
+ autoTerminatedRelationshipsList = getOptionalKeyAsType(map, AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, List.class, wrapperName, DEFAULT_AUTO_TERMINATED_RELATIONSHIPS_LIST);
+ properties = getOptionalKeyAsType(map, PROCESSOR_PROPS_KEY, Map.class, wrapperName, DEFAULT_PROPERTIES);
- annotationData = getOptionalKeyAsType(map, ANNOTATION_DATA_KEY, String.class, PROCESSORS_KEY, "");
+ annotationData = getOptionalKeyAsType(map, ANNOTATION_DATA_KEY, String.class, wrapperName, "");
}
public static boolean isSchedulingStrategy(String string) {
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteInputPortSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteInputPortSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteInputPortSchema.java
index 6ff8648..736c63f 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteInputPortSchema.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteInputPortSchema.java
@@ -22,13 +22,9 @@ import org.apache.nifi.minifi.commons.schema.common.BaseSchemaWithIdAndName;
import java.util.Map;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.COMMENT_KEY;
-import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.INPUT_PORTS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.MAX_CONCURRENT_TASKS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.USE_COMPRESSION_KEY;
-/**
- *
- */
public class RemoteInputPortSchema extends BaseSchemaWithIdAndName {
public static final String DEFAULT_COMMENT = "";
public static final int DEFAULT_MAX_CONCURRENT_TASKS = 1;
@@ -39,11 +35,12 @@ public class RemoteInputPortSchema extends BaseSchemaWithIdAndName {
private Boolean useCompression = DEFAULT_USE_COMPRESSION;
public RemoteInputPortSchema(Map map) {
- super(map, INPUT_PORTS_KEY);
+ super(map, "RemoteInputPort(id: {id}, name: {name})");
+ String wrapperName = getWrapperName();
- comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, INPUT_PORTS_KEY, DEFAULT_COMMENT);
- maxConcurrentTasks = getOptionalKeyAsType(map, MAX_CONCURRENT_TASKS_KEY, Number.class, INPUT_PORTS_KEY, DEFAULT_MAX_CONCURRENT_TASKS);
- useCompression = getOptionalKeyAsType(map, USE_COMPRESSION_KEY, Boolean.class, INPUT_PORTS_KEY, DEFAULT_USE_COMPRESSION);
+ comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, wrapperName, DEFAULT_COMMENT);
+ maxConcurrentTasks = getOptionalKeyAsType(map, MAX_CONCURRENT_TASKS_KEY, Number.class, wrapperName, DEFAULT_MAX_CONCURRENT_TASKS);
+ useCompression = getOptionalKeyAsType(map, USE_COMPRESSION_KEY, Boolean.class, wrapperName, DEFAULT_USE_COMPRESSION);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteProcessingGroupSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteProcessingGroupSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteProcessingGroupSchema.java
index c1d318e..6d2bb20 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteProcessingGroupSchema.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/RemoteProcessingGroupSchema.java
@@ -18,6 +18,7 @@
package org.apache.nifi.minifi.commons.schema;
import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
+import org.apache.nifi.minifi.commons.schema.common.StringUtil;
import org.apache.nifi.minifi.commons.schema.common.WritableSchema;
import java.util.List;
@@ -29,9 +30,6 @@ import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NA
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.REMOTE_PROCESSING_GROUPS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.YIELD_PERIOD_KEY;
-/**
- *
- */
public class RemoteProcessingGroupSchema extends BaseSchema implements WritableSchema {
public static final String URL_KEY = "url";
public static final String TIMEOUT_KEY = "timeout";
@@ -50,17 +48,18 @@ public class RemoteProcessingGroupSchema extends BaseSchema implements WritableS
public RemoteProcessingGroupSchema(Map map) {
name = getRequiredKeyAsType(map, NAME_KEY, String.class, REMOTE_PROCESSING_GROUPS_KEY);
- url = getRequiredKeyAsType(map, URL_KEY, String.class, REMOTE_PROCESSING_GROUPS_KEY);
- inputPorts = convertListToType(getRequiredKeyAsType(map, INPUT_PORTS_KEY, List.class, REMOTE_PROCESSING_GROUPS_KEY), "input port", RemoteInputPortSchema.class, INPUT_PORTS_KEY);
+ String wrapperName = new StringBuilder("RemoteProcessingGroup(name: ").append(StringUtil.isNullOrEmpty(name) ? "unknown" : name).append(")").toString();
+ url = getRequiredKeyAsType(map, URL_KEY, String.class, wrapperName);
+ inputPorts = convertListToType(getRequiredKeyAsType(map, INPUT_PORTS_KEY, List.class, wrapperName), "input port", RemoteInputPortSchema.class, INPUT_PORTS_KEY);
if (inputPorts != null) {
for (RemoteInputPortSchema remoteInputPortSchema: inputPorts) {
addIssuesIfNotNull(remoteInputPortSchema);
}
}
- comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, REMOTE_PROCESSING_GROUPS_KEY, DEFAULT_COMMENT);
- timeout = getOptionalKeyAsType(map, TIMEOUT_KEY, String.class, REMOTE_PROCESSING_GROUPS_KEY, DEFAULT_TIMEOUT);
- yieldPeriod = getOptionalKeyAsType(map, YIELD_PERIOD_KEY, String.class, REMOTE_PROCESSING_GROUPS_KEY, DEFAULT_YIELD_PERIOD);
+ comment = getOptionalKeyAsType(map, COMMENT_KEY, String.class, wrapperName, DEFAULT_COMMENT);
+ timeout = getOptionalKeyAsType(map, TIMEOUT_KEY, String.class, wrapperName, DEFAULT_TIMEOUT);
+ yieldPeriod = getOptionalKeyAsType(map, YIELD_PERIOD_KEY, String.class, wrapperName, DEFAULT_YIELD_PERIOD);
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/BaseSchema.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/BaseSchema.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/BaseSchema.java
index 7ba322a..7cd82f7 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/BaseSchema.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/BaseSchema.java
@@ -17,11 +17,12 @@
package org.apache.nifi.minifi.commons.schema.common;
+import org.apache.nifi.minifi.commons.schema.exception.SchemaInstantiatonException;
+
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
@@ -29,6 +30,7 @@ import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.function.Consumer;
+import java.util.function.Function;
import java.util.function.Supplier;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
@@ -138,6 +140,40 @@ public abstract class BaseSchema implements Schema {
return result;
}
+ public <T> List<T> getOptionalKeyAsList(Map valueMap, String key, Function<Map, T> conversionFunction, String wrapperName) {
+ return convertListToType(Map.class, (List<Map>) valueMap.get(key), key, conversionFunction, wrapperName, null);
+ }
+
+ public <InputT, OutputT> List<OutputT> convertListToType(Class<InputT> inputType, List<InputT> list, String simpleListType, Function<InputT, OutputT> conversionFunction,
+ String wrapperName, Supplier<OutputT> instantiator) {
+ if (list == null) {
+ return new ArrayList<>();
+ }
+ List<OutputT> result = new ArrayList<>(list.size());
+ for (int i = 0; i < list.size(); i++) {
+ try {
+ OutputT val = interpretValueAsType(inputType, list.get(i), conversionFunction, instantiator);
+ if (val != null) {
+ result.add(val);
+ }
+ } catch (SchemaInstantiatonException e) {
+ addValidationIssue(simpleListType + " number " + i, wrapperName, e.getMessage());
+ }
+ }
+ return result;
+ }
+
+ private <InputT, OutputT> OutputT interpretValueAsType(Class<InputT> inputType, InputT input, Function<InputT, OutputT> conversionFunction, Supplier<OutputT> instantiator)
+ throws SchemaInstantiatonException {
+ if (input == null && instantiator != null) {
+ return instantiator.get();
+ }
+ if (!inputType.isInstance(input)) {
+ throw new SchemaInstantiatonException("was expecting object of type " + inputType + " but was " + input.getClass());
+ }
+ return conversionFunction.apply(input);
+ }
+
private <T> T interpretValueAsType(Object obj, String key, Class targetClass, String wrapperName, boolean required, boolean instantiateIfNull) {
if (obj == null) {
if (required){
@@ -182,20 +218,6 @@ public abstract class BaseSchema implements Schema {
}
}
- public static <T> List<T> nullToEmpty(List<T> list) {
- return list == null ? Collections.emptyList() : list;
- }
-
- public static <T> Set<T> nullToEmpty(Set<T> set) {
- return set == null ? Collections.emptySet() : set;
- }
-
- public static <K, V> Map<K, V> nullToEmpty(Map<K, V> map) {
- return map == null ? Collections.emptyMap() : map;
- }
-
-
-
public static void checkForDuplicates(Consumer<String> duplicateMessageConsumer, String errorMessagePrefix, List<String> strings) {
if (strings != null) {
Set<String> seen = new HashSet<>();
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/BaseSchemaWithIdAndName.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/BaseSchemaWithIdAndName.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/BaseSchemaWithIdAndName.java
index 9ab6718..a1f7bb5 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/BaseSchemaWithIdAndName.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/BaseSchemaWithIdAndName.java
@@ -28,26 +28,22 @@ import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NA
public abstract class BaseSchemaWithIdAndName extends BaseSchema implements WritableSchema {
public static final Pattern VALID_ID_PATTERN = Pattern.compile("[A-Za-z0-9_-]+");
- public static final String ID_DOES_NOT_MATCH_VALID_ID_PATTERN = "Id does not match valid pattern (" + VALID_ID_PATTERN + "): ";
+ public static final String ID_DOES_NOT_MATCH_VALID_ID_PATTERN = "Id does not match valid pattern (" + VALID_ID_PATTERN + ")";
private final String wrapperName;
private String id;
private String name;
public BaseSchemaWithIdAndName(Map map, String wrapperName) {
- id = getId(map, wrapperName);
- name = getName(map, wrapperName);
this.wrapperName = wrapperName;
+ id = getId(map, getWrapperName());
+ name = getOptionalKeyAsType(map, NAME_KEY, String.class, getWrapperName(), "");
}
protected String getId(Map map, String wrapperName) {
return getOptionalKeyAsType(map, ID_KEY, String.class, wrapperName, "");
}
- protected String getName(Map map, String wrapperName) {
- return getOptionalKeyAsType(map, NAME_KEY, String.class, wrapperName, "");
- }
-
public String getId() {
return id;
}
@@ -60,10 +56,14 @@ public abstract class BaseSchemaWithIdAndName extends BaseSchema implements Writ
return name;
}
- protected void setName(String name) {
+ public void setName(String name) {
this.name = name;
}
+ public String getWrapperName() {
+ return wrapperName.replace("{id}", StringUtil.isNullOrEmpty(id) ? "unkown" : id).replace("{name}", StringUtil.isNullOrEmpty(name) ? "unkown" : name);
+ }
+
@Override
public Map<String, Object> toMap() {
Map<String, Object> map = mapSupplier.get();
@@ -76,9 +76,9 @@ public abstract class BaseSchemaWithIdAndName extends BaseSchema implements Writ
public List<String> getValidationIssues() {
List<String> validationIssues = super.getValidationIssues();
if (StringUtil.isNullOrEmpty(id)) {
- validationIssues.add(getIssueText(CommonPropertyKeys.ID_KEY, wrapperName, IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED));
+ validationIssues.add(getIssueText(CommonPropertyKeys.ID_KEY, getWrapperName(), IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED));
} else if (!VALID_ID_PATTERN.matcher(id).matches()) {
- validationIssues.add(ID_DOES_NOT_MATCH_VALID_ID_PATTERN + id);
+ validationIssues.add(getIssueText(CommonPropertyKeys.ID_KEY, getWrapperName(), ID_DOES_NOT_MATCH_VALID_ID_PATTERN));
}
return validationIssues;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CollectionUtil.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CollectionUtil.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CollectionUtil.java
new file mode 100644
index 0000000..5fb9549
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CollectionUtil.java
@@ -0,0 +1,39 @@
+/*
+ *
+ * * 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.minifi.commons.schema.common;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class CollectionUtil {
+ public static <T> List<T> nullToEmpty(List<T> list) {
+ return list == null ? Collections.emptyList() : list;
+ }
+
+ public static <T> Set<T> nullToEmpty(Set<T> set) {
+ return set == null ? Collections.emptySet() : set;
+ }
+
+ public static <K, V> Map<K, V> nullToEmpty(Map<K, V> map) {
+ return map == null ? Collections.emptyMap() : map;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CommonPropertyKeys.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CommonPropertyKeys.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CommonPropertyKeys.java
index 5c3a432..a603f3e 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CommonPropertyKeys.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/CommonPropertyKeys.java
@@ -31,6 +31,7 @@ public class CommonPropertyKeys {
public static final String PROVENANCE_REPORTING_KEY = "Provenance Reporting";
public static final String REMOTE_PROCESSING_GROUPS_KEY = "Remote Processing Groups";
public static final String INPUT_PORTS_KEY = "Input Ports";
+ public static final String OUTPUT_PORTS_KEY = "Output Ports";
public static final String PROVENANCE_REPO_KEY = "Provenance Repository";
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/StringUtil.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/StringUtil.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/StringUtil.java
index ae7165e..68b4cc7 100644
--- a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/StringUtil.java
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/common/StringUtil.java
@@ -17,8 +17,40 @@
package org.apache.nifi.minifi.commons.schema.common;
+import java.util.function.Consumer;
+
public class StringUtil {
- public static boolean isNullOrEmpty(final String string) {
+ /**
+ * Returns true if the string is null or empty
+ *
+ * @param string the string
+ * @return true if the string is null or empty
+ */
+ public static boolean isNullOrEmpty(String string) {
return string == null || string.isEmpty();
}
+
+ /**
+ * Passes the string to the consumer if it is neither null nor empty
+ *
+ * @param string the input
+ * @param consumer the action to perform
+ */
+ public static void doIfNotNullOrEmpty(String string, Consumer<String> consumer) {
+ if (!isNullOrEmpty(string)) {
+ consumer.accept(string);
+ }
+ }
+
+ /**
+ * Passes the string to the consumer if it is either null nor empty
+ *
+ * @param string the input
+ * @param consumer the action to perform
+ */
+ public static void doIfNullOrEmpty(String string, Consumer<String> consumer) {
+ if (isNullOrEmpty(string)) {
+ consumer.accept(string);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/exception/SchemaInstantiatonException.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/exception/SchemaInstantiatonException.java b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/exception/SchemaInstantiatonException.java
new file mode 100644
index 0000000..d659df8
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/main/java/org/apache/nifi/minifi/commons/schema/exception/SchemaInstantiatonException.java
@@ -0,0 +1,30 @@
+/*
+ *
+ * * 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.minifi.commons.schema.exception;
+
+public class SchemaInstantiatonException extends RuntimeException {
+ public SchemaInstantiatonException(String message) {
+ super(message);
+ }
+
+ public SchemaInstantiatonException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchemaTest.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchemaTest.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchemaTest.java
new file mode 100644
index 0000000..53cf32a
--- /dev/null
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/ProcessGroupSchemaTest.java
@@ -0,0 +1,64 @@
+/*
+ *
+ * * 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.minifi.commons.schema;
+
+import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+
+public class ProcessGroupSchemaTest {
+ @Test
+ public void testNoPortsRootGroup() {
+ validateIssuesNumMatches(0, new ProcessGroupSchema(new HashMap<>(), ConfigSchema.TOP_LEVEL_NAME));
+ }
+
+ @Test
+ public void testInputPortsRootGroup() {
+ Map<String, Object> map = new HashMap<>();
+ map.put(CommonPropertyKeys.INPUT_PORTS_KEY, Arrays.asList(createPortSchema("testId", "testName", ConfigSchema.TOP_LEVEL_NAME).toMap()));
+ validateIssuesNumMatches(1, new ProcessGroupSchema(map, ConfigSchema.TOP_LEVEL_NAME));
+ }
+
+ @Test
+ public void testOutputPortsRootGroup() {
+ Map<String, Object> map = new HashMap<>();
+ map.put(CommonPropertyKeys.OUTPUT_PORTS_KEY, Arrays.asList(createPortSchema("testId", "testName", ConfigSchema.TOP_LEVEL_NAME).toMap()));
+ validateIssuesNumMatches(1, new ProcessGroupSchema(map, ConfigSchema.TOP_LEVEL_NAME));
+ }
+
+ private PortSchema createPortSchema(String id, String name, String wrapperName) {
+ Map<String, Object> map = new HashMap<>();
+ map.put(CommonPropertyKeys.ID_KEY, id);
+ map.put(CommonPropertyKeys.NAME_KEY, name);
+ return new PortSchema(map, wrapperName);
+ }
+
+ private void validateIssuesNumMatches(int expected, ProcessGroupSchema processGroupSchema) {
+ int actual = processGroupSchema.getValidationIssues().size();
+ String issues = "[" + System.lineSeparator() + processGroupSchema.getValidationIssues().stream().collect(Collectors.joining("," + System.lineSeparator())) + "]";
+ assertEquals("Expected " + expected + " issue(s), got " + actual + ": " + issues, expected, actual);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
index 5b602ac..928bc03 100644
--- a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/serialization/SchemaLoaderTest.java
@@ -79,12 +79,12 @@ public class SchemaLoaderTest {
private void validateMinimalConfigVersion1Parse(ConfigSchema configSchema) {
assertTrue(configSchema instanceof ConfigSchema);
- List<ConnectionSchema> connections = configSchema.getConnections();
+ List<ConnectionSchema> connections = configSchema.getProcessGroupSchema().getConnections();
assertNotNull(connections);
assertEquals(1, connections.size());
assertNotNull(connections.get(0).getId());
- List<ProcessorSchema> processors = configSchema.getProcessors();
+ List<ProcessorSchema> processors = configSchema.getProcessGroupSchema().getProcessors();
assertNotNull(processors);
assertEquals(2, processors.size());
processors.forEach(p -> assertNotNull(p.getId()));
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1Test.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1Test.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1Test.java
index 56a8103..7ce0587 100644
--- a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1Test.java
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConfigSchemaV1Test.java
@@ -69,8 +69,10 @@ public class ConfigSchemaV1Test {
ConfigSchema configSchema = new ConfigSchemaV1(yamlAsMap).convert();
List<String> validationIssues = configSchema.getValidationIssues();
assertEquals(4, validationIssues.size());
- assertEquals(BaseSchema.getIssueText(ConnectionSchema.DESTINATION_ID_KEY, CommonPropertyKeys.CONNECTIONS_KEY, BaseSchema.IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED), validationIssues.get(0));
- assertEquals(BaseSchema.getIssueText(ConnectionSchema.SOURCE_ID_KEY, CommonPropertyKeys.CONNECTIONS_KEY, BaseSchema.IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED), validationIssues.get(1));
+ assertEquals(BaseSchema.getIssueText(ConnectionSchema.DESTINATION_ID_KEY, "Connection(id: TailToSplit, name: TailToSplit)", BaseSchema.IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED),
+ validationIssues.get(0));
+ assertEquals(BaseSchema.getIssueText(ConnectionSchema.SOURCE_ID_KEY, "Connection(id: TailToSplit, name: TailToSplit)", BaseSchema.IT_WAS_NOT_FOUND_AND_IT_IS_REQUIRED),
+ validationIssues.get(1));
assertEquals(ConfigSchemaV1.CONNECTION_WITH_NAME + connection.get(NAME_KEY) + ConfigSchemaV1.HAS_INVALID_DESTINATION_NAME + fakeDestination, validationIssues.get(2));
assertEquals(ConfigSchemaV1.CONNECTION_WITH_NAME + connection.get(NAME_KEY) + ConfigSchemaV1.HAS_INVALID_SOURCE_NAME + fakeSource, validationIssues.get(3));
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1Test.java
----------------------------------------------------------------------
diff --git a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1Test.java b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1Test.java
index e9acf4a..28b000c 100644
--- a/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1Test.java
+++ b/minifi-commons/minifi-commons-schema/src/test/java/org/apache/nifi/minifi/commons/schema/v1/ConnectionSchemaV1Test.java
@@ -179,7 +179,7 @@ public class ConnectionSchemaV1Test {
ConfigSchema configSchema = new ConfigSchemaV1(Collections.singletonMap(CommonPropertyKeys.CONNECTIONS_KEY, listWithKeyValues)).convert();
assertMessageDoesNotExist(configSchema, ConfigSchema.FOUND_THE_FOLLOWING_DUPLICATE_CONNECTION_IDS);
- List<ConnectionSchema> connections = configSchema.getConnections();
+ List<ConnectionSchema> connections = configSchema.getProcessGroupSchema().getConnections();
assertEquals(5, connections.size());
// Generated unique ids
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-docs/src/main/markdown/System_Admin_Guide.md
----------------------------------------------------------------------
diff --git a/minifi-docs/src/main/markdown/System_Admin_Guide.md b/minifi-docs/src/main/markdown/System_Admin_Guide.md
index 38c9588..0836443 100644
--- a/minifi-docs/src/main/markdown/System_Admin_Guide.md
+++ b/minifi-docs/src/main/markdown/System_Admin_Guide.md
@@ -248,7 +248,7 @@ parses and upconverts to the current version without issue.
1. Use ids instead of names for processors, connections.
2. Allow multiple source relationships for connections.
-
+3. Support process groups, input ports, output ports
## Flow Controller
@@ -391,6 +391,39 @@ Within the Processor Configuration section, there is the `Properties` subsection
State File: ./conf/state/tail-file
Initial Start Position: Beginning of File
+## Process Groups
+
+Process groups can be nested from the top level. They can contain other process groups as well and can be used to logically group related operations.
+
+*Property* | *Description*
+----------------------------------- | -------------
+name | The name of what this process group will do.
+id | The id of this process group. This needs to be set to a unique filesystem-friendly value (regex: [A-Za-z0-9_-]+)
+Processors | The processors contained in this Process Group. (Defined above)
+Remote Processing Groups | The remote processing groups contained in this Process Group. (Defined below)
+Connections | The connections contained in this Process Group. (Defined below)
+Input Ports | The input ports contained in this Process Group. (Defined below)
+Output Ports | The output ports contained in this Process Group. (Defined below)
+Process Groups | The child Process Groups contained in this Process Group.
+
+## Input Ports
+
+These ports provide input to the Process Group they reside on. (Currently only for internal Input ports.)
+
+*Property* | *Description*
+-------------------- | -------------
+name | The name of what this input port will do.
+id | The id of this input port. This needs to be set to a unique filesystem-friendly value (regex: [A-Za-z0-9_-]+)
+
+## Output Ports
+
+These ports provide output from the Process Group they reside on. (Currently only for internal Output ports.)
+
+*Property* | *Description*
+-------------------- | -------------
+name | The name of what this output port will do.
+id | The id of this output port. This needs to be set to a unique filesystem-friendly value (regex: [A-Za-z0-9_-]+)
+
## Connections
There can be multiple connections in this version of MiNiFi. The "Connections" subsection is a list of connections. Each connection must specify these properties.
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-resources/src/main/resources/conf/config.yml
----------------------------------------------------------------------
diff --git a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-resources/src/main/resources/conf/config.yml b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-resources/src/main/resources/conf/config.yml
index 44d8d4e..c88e47e 100644
--- a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-resources/src/main/resources/conf/config.yml
+++ b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-resources/src/main/resources/conf/config.yml
@@ -56,5 +56,8 @@ Security Properties:
algorithm: PBEWITHMD5AND256BITAES-CBC-OPENSSL
provider: BC
Processors: []
+Process Groups: []
+Input Ports: []
+Output Ports: []
Connections: []
Remote Processing Groups: []
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMain.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMain.java b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMain.java
index 251e9a3..e62392d 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMain.java
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMain.java
@@ -28,6 +28,8 @@ import org.apache.nifi.web.api.dto.ConnectableDTO;
import org.apache.nifi.web.api.dto.ConnectionDTO;
import org.apache.nifi.web.api.dto.FlowSnippetDTO;
import org.apache.nifi.web.api.dto.NiFiComponentDTO;
+import org.apache.nifi.web.api.dto.PortDTO;
+import org.apache.nifi.web.api.dto.ProcessGroupDTO;
import org.apache.nifi.web.api.dto.ProcessorDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupContentsDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
@@ -42,14 +44,19 @@ import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
+import java.util.ArrayList;
import java.util.Collection;
-import java.util.HashMap;
+import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.apache.nifi.minifi.commons.schema.common.CollectionUtil.nullToEmpty;
public class ConfigMain {
public static final int ERR_INVALID_ARGS = 1;
@@ -134,80 +141,82 @@ public class ConfigMain {
System.out.println();
}
- private static void enrichTemplateDTO(TemplateDTO templateDTO) {
- FlowSnippetDTO flowSnippetDTO = templateDTO.getSnippet();
+ private static void enrichFlowSnippetDTO(FlowSnippetDTO flowSnippetDTO) {
+ List<FlowSnippetDTO> allFlowSnippets = getAllFlowSnippets(flowSnippetDTO);
- Set<RemoteProcessGroupDTO> remoteProcessGroups = flowSnippetDTO.getRemoteProcessGroups();
- if (remoteProcessGroups != null) {
- for (RemoteProcessGroupDTO remoteProcessGroupDTO : remoteProcessGroups) {
- if (StringUtil.isNullOrEmpty(remoteProcessGroupDTO.getName())) {
- remoteProcessGroupDTO.setName(remoteProcessGroupDTO.getTargetUri());
- }
- }
+ Set<RemoteProcessGroupDTO> remoteProcessGroups = getAll(allFlowSnippets, FlowSnippetDTO::getRemoteProcessGroups).collect(Collectors.toSet());
+
+ // RPGs with no name get Target URI as name
+ remoteProcessGroups.stream().filter(r -> StringUtil.isNullOrEmpty(r.getName())).forEach(r -> r.setName(r.getTargetUri()));
+
+ Map<String, String> connectableNameMap = getAll(allFlowSnippets, FlowSnippetDTO::getProcessors).collect(Collectors.toMap(NiFiComponentDTO::getId, ProcessorDTO::getName));
+
+ for (RemoteProcessGroupDTO remoteProcessGroupDTO : remoteProcessGroups) {
+ RemoteProcessGroupContentsDTO contents = remoteProcessGroupDTO.getContents();
+ addConnectables(connectableNameMap, nullToEmpty(contents.getInputPorts()), RemoteProcessGroupPortDTO::getId, RemoteProcessGroupPortDTO::getId);
+ addConnectables(connectableNameMap, nullToEmpty(contents.getOutputPorts()), RemoteProcessGroupPortDTO::getId, RemoteProcessGroupPortDTO::getId);
}
- Set<ConnectionDTO> connections = flowSnippetDTO.getConnections();
- if (connections != null) {
- Map<String, String> connectableNameMap = new HashMap<>();
- Set<ProcessorDTO> processorDTOs = flowSnippetDTO.getProcessors();
- if (processorDTOs != null) {
- connectableNameMap.putAll(processorDTOs.stream().collect(Collectors.toMap(NiFiComponentDTO::getId, ProcessorDTO::getName)));
- }
- if (remoteProcessGroups != null) {
- for (RemoteProcessGroupDTO remoteProcessGroupDTO : remoteProcessGroups) {
- RemoteProcessGroupContentsDTO contents = remoteProcessGroupDTO.getContents();
- addRemoteProcessGroupPortDTOs(connectableNameMap, contents.getInputPorts());
- addRemoteProcessGroupPortDTOs(connectableNameMap, contents.getOutputPorts());
+ addConnectables(connectableNameMap, getAll(allFlowSnippets, FlowSnippetDTO::getInputPorts).collect(Collectors.toList()), PortDTO::getId, PortDTO::getName);
+ addConnectables(connectableNameMap, getAll(allFlowSnippets, FlowSnippetDTO::getOutputPorts).collect(Collectors.toList()), PortDTO::getId, PortDTO::getName);
+
+ Set<ConnectionDTO> connections = getAll(allFlowSnippets, FlowSnippetDTO::getConnections).collect(Collectors.toSet());
+ for (ConnectionDTO connection : connections) {
+ setName(connectableNameMap, connection.getSource());
+ setName(connectableNameMap, connection.getDestination());
+ }
+
+ for (ConnectionDTO connection : connections) {
+ if (StringUtil.isNullOrEmpty(connection.getName())) {
+ StringBuilder name = new StringBuilder();
+ ConnectableDTO connectionSource = connection.getSource();
+ if (connectionSource != null) {
+ name.append(connectionSource.getName());
}
- }
- for (ConnectionDTO connection : connections) {
- setName(connectableNameMap, connection.getSource());
- setName(connectableNameMap, connection.getDestination());
- }
- for (ConnectionDTO connection : connections) {
- if (StringUtil.isNullOrEmpty(connection.getName())) {
- StringBuilder name = new StringBuilder();
- ConnectableDTO connectionSource = connection.getSource();
- if (connectionSource != null) {
- name.append(connectionSource.getName());
- }
- name.append("/");
- if (connection.getSelectedRelationships() != null && connection.getSelectedRelationships().size() > 0) {
- name.append(connection.getSelectedRelationships().iterator().next());
- }
- name.append("/");
- ConnectableDTO connectionDestination = connection.getDestination();
- if (connectionDestination != null) {
- name.append(connectionDestination.getName());
- }
- connection.setName(name.toString());
+ name.append("/");
+ if (connection.getSelectedRelationships() != null && connection.getSelectedRelationships().size() > 0) {
+ name.append(connection.getSelectedRelationships().iterator().next());
+ }
+ name.append("/");
+ ConnectableDTO connectionDestination = connection.getDestination();
+ if (connectionDestination != null) {
+ name.append(connectionDestination.getName());
}
+ connection.setName(name.toString());
}
}
+ nullToEmpty(flowSnippetDTO.getProcessGroups()).stream().map(ProcessGroupDTO::getContents).forEach(ConfigMain::enrichFlowSnippetDTO);
}
- public static ConfigSchema transformTemplateToSchema(InputStream source) throws JAXBException, SchemaLoaderException {
- TemplateDTO templateDTO = (TemplateDTO) JAXBContext.newInstance(TemplateDTO.class).createUnmarshaller().unmarshal(source);
+ private static <T> Stream<T> getAll(List<FlowSnippetDTO> allFlowSnippets, Function<FlowSnippetDTO, Collection<T>> accessor) {
+ return allFlowSnippets.stream().flatMap(f -> accessor.apply(f).stream()).filter(Objects::nonNull);
+ }
- if (templateDTO.getSnippet().getProcessGroups().size() != 0){
- throw new SchemaLoaderException("Process Groups are not currently supported in MiNiFi. Please remove any from the template and try again.");
- }
+ private static List<FlowSnippetDTO> getAllFlowSnippets(FlowSnippetDTO flowSnippetDTO) {
+ List<FlowSnippetDTO> result = new ArrayList<>();
+ getAllFlowSnippets(flowSnippetDTO, result);
+ return result;
+ }
- if (templateDTO.getSnippet().getOutputPorts().size() != 0){
- throw new SchemaLoaderException("Output Ports are not currently supported in MiNiFi. Please remove any from the template and try again.");
- }
+ private static void getAllFlowSnippets(FlowSnippetDTO flowSnippetDTO, List<FlowSnippetDTO> result) {
+ result.add(flowSnippetDTO);
+ nullToEmpty(flowSnippetDTO.getProcessGroups()).stream().map(ProcessGroupDTO::getContents).forEach(f -> getAllFlowSnippets(f, result));
+ }
- if (templateDTO.getSnippet().getInputPorts().size() != 0){
- throw new SchemaLoaderException("Input Ports are not currently supported in MiNiFi. Please remove any from the template and try again.");
- }
+ public static ConfigSchema transformTemplateToSchema(InputStream source) throws JAXBException, IOException, SchemaLoaderException {
+ try {
+ TemplateDTO templateDTO = (TemplateDTO) JAXBContext.newInstance(TemplateDTO.class).createUnmarshaller().unmarshal(source);
- if (templateDTO.getSnippet().getFunnels().size() != 0){
- throw new SchemaLoaderException("Funnels are not currently supported in MiNiFi. Please remove any from the template and try again.");
- }
+ if (templateDTO.getSnippet().getFunnels().size() != 0){
+ throw new SchemaLoaderException("Funnels are not currently supported in MiNiFi. Please remove any from the template and try again.");
+ }
- enrichTemplateDTO(templateDTO);
- ConfigSchema configSchema = new ConfigSchemaFunction().apply(templateDTO);
- return configSchema;
+ enrichFlowSnippetDTO(templateDTO.getSnippet());
+ ConfigSchema configSchema = new ConfigSchemaFunction().apply(templateDTO);
+ return configSchema;
+ } finally {
+ source.close();
+ }
}
private static void setName(Map<String, String> connectableNameMap, ConnectableDTO connectableDTO) {
@@ -219,10 +228,6 @@ public class ConfigMain {
}
}
- private static void addRemoteProcessGroupPortDTOs(Map<String, String> connectableNameMap, Collection<RemoteProcessGroupPortDTO> ports) {
- addConnectables(connectableNameMap, ports, RemoteProcessGroupPortDTO::getId, RemoteProcessGroupPortDTO::getId);
- }
-
private static <T> void addConnectables(Map<String, String> connectableNameMap, Collection<T> hasIdAndNames, Function<T, String> idGetter, Function<T, String> nameGetter) {
if (hasIdAndNames != null) {
for (T hasIdAndName : hasIdAndNames) {
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConfigSchemaFunction.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConfigSchemaFunction.java b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConfigSchemaFunction.java
index 4fc9f5a..9cdccf5 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConfigSchemaFunction.java
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConfigSchemaFunction.java
@@ -19,10 +19,12 @@ package org.apache.nifi.minifi.toolkit.configuration.dto;
import org.apache.nifi.minifi.commons.schema.ConfigSchema;
import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
+import org.apache.nifi.minifi.commons.schema.PortSchema;
+import org.apache.nifi.minifi.commons.schema.ProcessGroupSchema;
import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
import org.apache.nifi.minifi.commons.schema.RemoteProcessingGroupSchema;
-import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
+import org.apache.nifi.minifi.commons.schema.common.StringUtil;
import org.apache.nifi.web.api.dto.FlowSnippetDTO;
import org.apache.nifi.web.api.dto.TemplateDTO;
@@ -32,51 +34,96 @@ import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
+import static org.apache.nifi.minifi.commons.schema.common.CollectionUtil.nullToEmpty;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ID_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.INPUT_PORTS_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.OUTPUT_PORTS_KEY;
+
public class ConfigSchemaFunction implements Function<TemplateDTO, ConfigSchema> {
private final FlowControllerSchemaFunction flowControllerSchemaFunction;
private final ProcessorSchemaFunction processorSchemaFunction;
private final ConnectionSchemaFunction connectionSchemaFunction;
private final RemoteProcessingGroupSchemaFunction remoteProcessingGroupSchemaFunction;
+ private final PortSchemaFunction inputPortSchemaFunction;
+ private final PortSchemaFunction outputPortSchemaFunction;
public ConfigSchemaFunction() {
- this(new FlowControllerSchemaFunction(), new ProcessorSchemaFunction(), new ConnectionSchemaFunction(), new RemoteProcessingGroupSchemaFunction(new RemoteInputPortSchemaFunction()));
+ this(new FlowControllerSchemaFunction(), new ProcessorSchemaFunction(), new ConnectionSchemaFunction(), new RemoteProcessingGroupSchemaFunction(new RemoteInputPortSchemaFunction()),
+ new PortSchemaFunction(INPUT_PORTS_KEY), new PortSchemaFunction(OUTPUT_PORTS_KEY));
}
- public ConfigSchemaFunction(FlowControllerSchemaFunction flowControllerSchemaFunction, ProcessorSchemaFunction processorSchemaFunction,
- ConnectionSchemaFunction connectionSchemaFunction, RemoteProcessingGroupSchemaFunction remoteProcessingGroupSchemaFunction) {
+ public ConfigSchemaFunction(FlowControllerSchemaFunction flowControllerSchemaFunction, ProcessorSchemaFunction processorSchemaFunction, ConnectionSchemaFunction connectionSchemaFunction,
+ RemoteProcessingGroupSchemaFunction remoteProcessingGroupSchemaFunction, PortSchemaFunction inputPortSchemaFunction, PortSchemaFunction outputPortSchemaFunction) {
this.flowControllerSchemaFunction = flowControllerSchemaFunction;
this.processorSchemaFunction = processorSchemaFunction;
this.connectionSchemaFunction = connectionSchemaFunction;
this.remoteProcessingGroupSchemaFunction = remoteProcessingGroupSchemaFunction;
+ this.inputPortSchemaFunction = inputPortSchemaFunction;
+ this.outputPortSchemaFunction = outputPortSchemaFunction;
}
@Override
public ConfigSchema apply(TemplateDTO templateDTO) {
Map<String, Object> map = new HashMap<>();
+ map.put(CommonPropertyKeys.FLOW_CONTROLLER_PROPS_KEY, flowControllerSchemaFunction.apply(templateDTO).toMap());
+
FlowSnippetDTO snippet = templateDTO.getSnippet();
- map.put(CommonPropertyKeys.FLOW_CONTROLLER_PROPS_KEY, flowControllerSchemaFunction.apply(templateDTO).toMap());
+ addSnippet(map, snippet);
+
+ return new ConfigSchema(map);
+ }
+
+ protected void addSnippet(Map<String, Object> map, FlowSnippetDTO snippet) {
+ addSnippet(map, null, null, snippet);
+ }
+
+ protected Map<String, Object> addSnippet(Map<String, Object> map, String id, String name, FlowSnippetDTO snippet) {
+ if (!StringUtil.isNullOrEmpty(id)) {
+ map.put(ID_KEY, id);
+ }
- map.put(CommonPropertyKeys.PROCESSORS_KEY, BaseSchema.nullToEmpty(snippet.getProcessors()).stream()
+ if (!StringUtil.isNullOrEmpty(name)) {
+ map.put(NAME_KEY, name);
+ }
+
+ map.put(CommonPropertyKeys.PROCESSORS_KEY, nullToEmpty(snippet.getProcessors()).stream()
.map(processorSchemaFunction)
.sorted(Comparator.comparing(ProcessorSchema::getName))
.map(ProcessorSchema::toMap)
.collect(Collectors.toList()));
- map.put(CommonPropertyKeys.CONNECTIONS_KEY, BaseSchema.nullToEmpty(snippet.getConnections()).stream()
+
+
+ map.put(CommonPropertyKeys.CONNECTIONS_KEY, nullToEmpty(snippet.getConnections()).stream()
.map(connectionSchemaFunction)
.sorted(Comparator.comparing(ConnectionSchema::getName))
.map(ConnectionSchema::toMap)
.collect(Collectors.toList()));
- map.put(CommonPropertyKeys.REMOTE_PROCESSING_GROUPS_KEY, BaseSchema.nullToEmpty(snippet.getRemoteProcessGroups()).stream()
+ map.put(CommonPropertyKeys.REMOTE_PROCESSING_GROUPS_KEY, nullToEmpty(snippet.getRemoteProcessGroups()).stream()
.map(remoteProcessingGroupSchemaFunction)
.sorted(Comparator.comparing(RemoteProcessingGroupSchema::getName))
.map(RemoteProcessingGroupSchema::toMap)
.collect(Collectors.toList()));
- ConfigSchema configSchema = new ConfigSchema(map);
- return configSchema;
+ map.put(INPUT_PORTS_KEY, nullToEmpty(snippet.getInputPorts()).stream()
+ .map(inputPortSchemaFunction)
+ .sorted(Comparator.comparing(PortSchema::getName))
+ .map(PortSchema::toMap)
+ .collect(Collectors.toList()));
+
+ map.put(OUTPUT_PORTS_KEY, nullToEmpty(snippet.getOutputPorts()).stream()
+ .map(outputPortSchemaFunction)
+ .sorted(Comparator.comparing(PortSchema::getName))
+ .map(PortSchema::toMap)
+ .collect(Collectors.toList()));
+
+ map.put(ProcessGroupSchema.PROCESS_GROUPS_KEY, nullToEmpty(snippet.getProcessGroups()).stream()
+ .map(p -> addSnippet(new HashMap<>(), p.getId(), p.getName(), p.getContents())).collect(Collectors.toList()));
+
+ return map;
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConnectionSchemaFunction.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConnectionSchemaFunction.java b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConnectionSchemaFunction.java
index d3d71e6..7acab41 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConnectionSchemaFunction.java
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ConnectionSchemaFunction.java
@@ -19,7 +19,6 @@ package org.apache.nifi.minifi.toolkit.configuration.dto;
import org.apache.nifi.connectable.ConnectableType;
import org.apache.nifi.minifi.commons.schema.ConnectionSchema;
-import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
import org.apache.nifi.web.api.dto.ConnectionDTO;
import java.util.HashMap;
@@ -29,6 +28,7 @@ import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;
+import static org.apache.nifi.minifi.commons.schema.common.CollectionUtil.nullToEmpty;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.CONNECTIONS_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ID_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
@@ -40,14 +40,14 @@ public class ConnectionSchemaFunction implements Function<ConnectionDTO, Connect
map.put(ID_KEY, connectionDTO.getId());
map.put(NAME_KEY, connectionDTO.getName());
map.put(ConnectionSchema.SOURCE_ID_KEY, connectionDTO.getSource().getId());
- Set<String> selectedRelationships = BaseSchema.nullToEmpty(connectionDTO.getSelectedRelationships());
+ Set<String> selectedRelationships = nullToEmpty(connectionDTO.getSelectedRelationships());
map.put(ConnectionSchema.SOURCE_RELATIONSHIP_NAMES_KEY, selectedRelationships.stream().sorted().collect(Collectors.toList()));
map.put(ConnectionSchema.DESTINATION_ID_KEY, connectionDTO.getDestination().getId());
map.put(ConnectionSchema.MAX_WORK_QUEUE_SIZE_KEY, connectionDTO.getBackPressureObjectThreshold());
map.put(ConnectionSchema.MAX_WORK_QUEUE_DATA_SIZE_KEY, connectionDTO.getBackPressureDataSizeThreshold());
map.put(ConnectionSchema.FLOWFILE_EXPIRATION__KEY, connectionDTO.getFlowFileExpiration());
- List<String> queuePrioritizers = BaseSchema.nullToEmpty(connectionDTO.getPrioritizers());
+ List<String> queuePrioritizers = nullToEmpty(connectionDTO.getPrioritizers());
if (queuePrioritizers.size() > 0) {
map.put(ConnectionSchema.QUEUE_PRIORITIZER_CLASS_KEY, queuePrioritizers.get(0));
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/PortSchemaFunction.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/PortSchemaFunction.java b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/PortSchemaFunction.java
new file mode 100644
index 0000000..29efac3
--- /dev/null
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/PortSchemaFunction.java
@@ -0,0 +1,46 @@
+/*
+ *
+ * * 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.minifi.toolkit.configuration.dto;
+
+import org.apache.nifi.minifi.commons.schema.PortSchema;
+import org.apache.nifi.web.api.dto.PortDTO;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Function;
+
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ID_KEY;
+import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
+
+public class PortSchemaFunction implements Function<PortDTO, PortSchema> {
+ private final String wrapperName;
+
+ public PortSchemaFunction(String wrapperName) {
+ this.wrapperName = wrapperName;
+ }
+
+ @Override
+ public PortSchema apply(PortDTO portDTO) {
+ Map<String, Object> map = new HashMap<>();
+ map.put(ID_KEY, portDTO.getId());
+ map.put(NAME_KEY, portDTO.getName());
+ return new PortSchema(map, wrapperName);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ProcessorSchemaFunction.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ProcessorSchemaFunction.java b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ProcessorSchemaFunction.java
index 21b3345..89097cc 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ProcessorSchemaFunction.java
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/main/java/org/apache/nifi/minifi/toolkit/configuration/dto/ProcessorSchemaFunction.java
@@ -18,7 +18,6 @@
package org.apache.nifi.minifi.toolkit.configuration.dto;
import org.apache.nifi.minifi.commons.schema.ProcessorSchema;
-import org.apache.nifi.minifi.commons.schema.common.BaseSchema;
import org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys;
import org.apache.nifi.web.api.dto.ProcessorConfigDTO;
import org.apache.nifi.web.api.dto.ProcessorDTO;
@@ -29,6 +28,7 @@ import java.util.Map;
import java.util.function.Function;
import java.util.stream.Collectors;
+import static org.apache.nifi.minifi.commons.schema.common.CollectionUtil.nullToEmpty;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.ID_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.NAME_KEY;
import static org.apache.nifi.minifi.commons.schema.common.CommonPropertyKeys.SCHEDULING_PERIOD_KEY;
@@ -53,11 +53,11 @@ public class ProcessorSchemaFunction implements Function<ProcessorDTO, Processor
if (runDurationMillis != null) {
map.put(ProcessorSchema.RUN_DURATION_NANOS_KEY, runDurationMillis * 1000);
}
- map.put(ProcessorSchema.AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, BaseSchema.nullToEmpty(processorDTO.getRelationships()).stream()
+ map.put(ProcessorSchema.AUTO_TERMINATED_RELATIONSHIPS_LIST_KEY, nullToEmpty(processorDTO.getRelationships()).stream()
.filter(RelationshipDTO::isAutoTerminate)
.map(RelationshipDTO::getName)
.collect(Collectors.toList()));
- map.put(ProcessorSchema.PROCESSOR_PROPS_KEY, new HashMap<>(BaseSchema.nullToEmpty(processorDTOConfig.getProperties())));
+ map.put(ProcessorSchema.PROCESSOR_PROPS_KEY, new HashMap<>(nullToEmpty(processorDTOConfig.getProperties())));
String annotationData = processorDTOConfig.getAnnotationData();
if(annotationData != null && !annotationData.isEmpty()) {
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMainTest.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMainTest.java b/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMainTest.java
index cfdc48f..d61a641 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMainTest.java
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/ConfigMainTest.java
@@ -148,7 +148,7 @@ public class ConfigMainTest {
@Test
public void testTransformErrorTransformingTemplate() throws FileNotFoundException {
when(pathInputStreamFactory.create(testInput)).thenAnswer(invocation ->
- ConfigMainTest.class.getClassLoader().getResourceAsStream("TemplateWithInputPort.xml"));
+ ConfigMainTest.class.getClassLoader().getResourceAsStream("TemplateWithFunnel.xml"));
assertEquals(ConfigMain.ERR_UNABLE_TO_TRANSFORM_TEMPLATE, configMain.execute(new String[]{ConfigMain.TRANSFORM, testInput, testOutput}));
}
@@ -190,18 +190,23 @@ public class ConfigMainTest {
transformRoundTrip("MultipleRelationships");
}
- @Test(expected = SchemaLoaderException.class)
- public void testFailToTransformProcessGroup() throws IOException, JAXBException, SchemaLoaderException {
+ @Test
+ public void testTransformRoundTripProcessGroupsAndRemoteProcessGroups() throws IOException, JAXBException, SchemaLoaderException {
+ transformRoundTrip("ProcessGroupsAndRemoteProcessGroups");
+ }
+
+ @Test
+ public void testSuccessTransformProcessGroup() throws IOException, JAXBException, SchemaLoaderException {
ConfigMain.transformTemplateToSchema(getClass().getClassLoader().getResourceAsStream("TemplateWithProcessGroup.xml")).toMap();
}
- @Test(expected = SchemaLoaderException.class)
- public void testFailToTransformInputPort() throws IOException, JAXBException, SchemaLoaderException {
+ @Test
+ public void testSuccessTransformInputPort() throws IOException, JAXBException, SchemaLoaderException {
ConfigMain.transformTemplateToSchema(getClass().getClassLoader().getResourceAsStream("TemplateWithOutputPort.xml")).toMap();
}
- @Test(expected = SchemaLoaderException.class)
- public void testFailToTransformOutputPort() throws IOException, JAXBException, SchemaLoaderException {
+ @Test
+ public void testSuccessTransformOutputPort() throws IOException, JAXBException, SchemaLoaderException {
ConfigMain.transformTemplateToSchema(getClass().getClassLoader().getResourceAsStream("TemplateWithInputPort.xml")).toMap();
}
@@ -278,8 +283,8 @@ public class ConfigMainTest {
assertTrue(configSchemaUpgradedFromV1.isValid());
assertEquals(configSchemaConvertableSchema, configSchemaUpgradedFromV1);
ConfigSchema configSchemaFromCurrent = new ConfigSchema(yamlMap);
- List<ProcessorSchema> currentProcessors = configSchemaFromCurrent.getProcessors();
- List<ProcessorSchema> v1Processors = configSchemaUpgradedFromV1.getProcessors();
+ List<ProcessorSchema> currentProcessors = configSchemaFromCurrent.getProcessGroupSchema().getProcessors();
+ List<ProcessorSchema> v1Processors = configSchemaUpgradedFromV1.getProcessGroupSchema().getProcessors();
assertEquals(currentProcessors.size(), v1Processors.size());
// V1 doesn't have ids so we need to map the autogenerated ones to the ones from the template
@@ -291,11 +296,12 @@ public class ConfigMainTest {
v1IdToCurrentIdMap.put(v1Processor.getId(), currentProcessor.getId());
v1Processor.setId(currentProcessor.getId());
}
- configSchemaUpgradedFromV1.getRemoteProcessingGroups().stream().flatMap(g -> g.getInputPorts().stream()).map(RemoteInputPortSchema::getId).sequential()
+
+ configSchemaUpgradedFromV1.getProcessGroupSchema().getRemoteProcessingGroups().stream().flatMap(g -> g.getInputPorts().stream()).map(RemoteInputPortSchema::getId).sequential()
.forEach(id -> v1IdToCurrentIdMap.put(id, id));
- List<ConnectionSchema> currentConnections = configSchemaFromCurrent.getConnections();
- List<ConnectionSchema> v1Connections = configSchemaUpgradedFromV1.getConnections();
+ List<ConnectionSchema> currentConnections = configSchemaFromCurrent.getProcessGroupSchema().getConnections();
+ List<ConnectionSchema> v1Connections = configSchemaUpgradedFromV1.getProcessGroupSchema().getConnections();
// Update source and dest ids, can set connection id equal because it isn't referenced elsewhere
assertEquals(currentConnections.size(), v1Connections.size());
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/PortSchemaFunctionTest.java
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/PortSchemaFunctionTest.java b/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/PortSchemaFunctionTest.java
new file mode 100644
index 0000000..05c83cb
--- /dev/null
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/java/org/apache/nifi/minifi/toolkit/configuration/dto/PortSchemaFunctionTest.java
@@ -0,0 +1,74 @@
+/*
+ *
+ * * 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.minifi.toolkit.configuration.dto;
+
+import org.apache.nifi.minifi.commons.schema.PortSchema;
+import org.apache.nifi.web.api.dto.PortDTO;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class PortSchemaFunctionTest {
+ private String testId;
+ private String testName;
+ private String testWrapperName;
+ private PortDTO portDTO;
+ private PortSchemaFunction portSchemaFunction;
+
+ @Before
+ public void setup() {
+ testId = "testId";
+ testName = "testName";
+ testWrapperName = "testWrapperName";
+ portDTO = new PortDTO();
+ portDTO.setId(testId);
+ portDTO.setName(testName);
+ portSchemaFunction = new PortSchemaFunction(testWrapperName);
+ }
+
+ @Test
+ public void testFullMap() {
+ PortSchema portSchema = portSchemaFunction.apply(portDTO);
+ assertEquals(testId, portSchema.getId());
+ assertEquals(testName, portSchema.getName());
+ assertTrue(portSchema.isValid());
+ }
+
+ @Test
+ public void testNoId() {
+ portDTO.setId(null);
+ PortSchema portSchema = portSchemaFunction.apply(portDTO);
+ assertEquals("", portSchema.getId());
+ assertEquals(testName, portSchema.getName());
+ assertFalse(portSchema.isValid());
+ }
+
+ @Test
+ public void testNoName() {
+ portDTO.setName(null);
+ PortSchema portSchema = portSchemaFunction.apply(portDTO);
+ assertEquals(testId, portSchema.getId());
+ assertEquals("", portSchema.getName());
+ assertTrue(portSchema.isValid());
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/CsvToJson.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/CsvToJson.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/CsvToJson.yml
index b4993e3..d03bd16 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/CsvToJson.yml
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/CsvToJson.yml
@@ -148,6 +148,9 @@ Processors:
- success
Properties:
Delete Attributes Expression:
+Process Groups: []
+Input Ports: []
+Output Ports: []
Connections:
- name: ExtractText/matched/ReplaceText2
id: 56ef3e2e-ee35-4598-9fbe-ae86050960b0
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/DecompressionCircularFlow.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/DecompressionCircularFlow.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/DecompressionCircularFlow.yml
index fdec427..d2f90b2 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/DecompressionCircularFlow.yml
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/DecompressionCircularFlow.yml
@@ -187,6 +187,9 @@ Processors:
Compression Level:
Mode: decompress
Update Filename:
+Process Groups: []
+Input Ports: []
+Output Ports: []
Connections:
- name: Compressed?/gzip/Uncompress GZIP
id: 5de215d5-9f7e-414b-98aa-2edaa0514d99
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/InvokeHttpMiNiFiTemplateTest.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/InvokeHttpMiNiFiTemplateTest.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/InvokeHttpMiNiFiTemplateTest.yml
index dcd75be..aa7d6d5 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/InvokeHttpMiNiFiTemplateTest.yml
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/InvokeHttpMiNiFiTemplateTest.yml
@@ -166,6 +166,9 @@ Processors:
Properties:
Delete Attributes Expression:
q: nifi
+Process Groups: []
+Input Ports: []
+Output Ports: []
Connections:
- name: Route On Status Code/200/LogAttribute
id: 3039718a-bb40-4811-9b74-ecbe926daae8
http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/31855bbc/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/MultipleRelationships.yml
----------------------------------------------------------------------
diff --git a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/MultipleRelationships.yml b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/MultipleRelationships.yml
index 2850e67..12ed7e1 100644
--- a/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/MultipleRelationships.yml
+++ b/minifi-toolkit/minifi-toolkit-configuration/src/test/resources/MultipleRelationships.yml
@@ -114,6 +114,9 @@ Processors:
Properties:
Delete Attributes Expression:
filename: abc
+Process Groups: []
+Input Ports: []
+Output Ports: []
Connections:
- name: GenerateFlowFile/success/UpdateAttribute
id: 7c79cce3-0157-1000-0000-000000000000