<|start_filename|>nifi-opcua-processors/src/main/java/de/fraunhofer/fit/processors/opcua/utils/RecordAggregator.java<|end_filename|>
package de.fraunhofer.fit.processors.opcua.utils;

import java.util.*;

public class RecordAggregator {

    // This variable indicates how long a record waits for notification messages from OPC server
    private final int PUBLISH_INTERVAL_MULTIPLIER = 4;

    // Index of elements in the queue message
    private final int VARIABLE_ID_INDEX = 0;
    private final int SOURCE_TS_INDEX = 2;
    private final int VALUE_INDEX = 3;
    private final int STATUS_CODE_INDEX = 4;

    private long PUBLISH_THRESHOLD_TIME;
    private List<String> tags;
    private Map<String, Integer> tagOrderMap;
    private Map<String, Record> recordMap;

    // minPublishInterval is the minimum subscription notification publish interval from OPC UA server
    public RecordAggregator(List<String> tags, long minPublishInterval) {

        this.tags = tags;
        this.tagOrderMap = new HashMap<>();
        this.recordMap = new HashMap<>();

        PUBLISH_THRESHOLD_TIME = PUBLISH_INTERVAL_MULTIPLIER * minPublishInterval;

        // Create a map which with tag name as key and its order in list as value
        for (int i = 0; i < tags.size(); i++) {
            this.tagOrderMap.put(tags.get(i), i);
        }

    }

    public void aggregate(String rawMsg) {


            // msg has the following elements in order:
            // 1. variable ID; 2. server time stamp; 3. source time stamp; 4. value; 5. status code
            String[] msg = rawMsg.trim().split(",");

            // Ditch all messages with bad status code
            if (!msg[STATUS_CODE_INDEX].equals("0")) {
                return;
            }

            String timeStamp = msg[SOURCE_TS_INDEX];
            String variableId = msg[VARIABLE_ID_INDEX];
            String value = msg[VALUE_INDEX];

            // Check if a record is already exist for the given time stamp
            Record rec;
            if (recordMap.containsKey(timeStamp)) {
                // Get the record with the same time stamp as the message
                rec = recordMap.get(timeStamp);
            } else {
                // Create a record
                rec = new Record(timeStamp, tags.size());
                // Insert it into the map
                recordMap.put(timeStamp, rec);
            }
            // Get the index of the variable given in the message
            if (!tagOrderMap.containsKey(variableId)) {
                return;
            }
            int index = tagOrderMap.get(variableId);
            // Update the value in the record array
            rec.getRecordArray()[index] = value;
        }


    public List<String> getReadyRecords() {

        List<String> list = new ArrayList<>();
        List<String> recordKeyList = new ArrayList<>(recordMap.keySet());
        Collections.sort(recordKeyList);

        for(String key : recordKeyList) {
            Record rec = recordMap.get(key);
            if(rec.isReady(PUBLISH_THRESHOLD_TIME)) {
                list.add(key + "," +
                        String.join(",", rec.getRecordArray()) +
                        System.getProperty("line.separator"));
                recordMap.remove(key);
            }
        }

        return list;
    }


    class Record {

        private String timeStamp;
        private long createdTime; // The createdTime is used to see whether a record is ready to be published
        private String[] recordValues;

        Record(String timeStamp, int recordSize) {
            this.timeStamp = timeStamp;
            createdTime = System.currentTimeMillis();
            recordValues = new String[recordSize];
            Arrays.fill(recordValues, "");
        }

        String[] getRecordArray() {
            return recordValues;
        }

        String getTimeStamp() {
            return timeStamp;
        }

        boolean isReady(long timeThrashold) {
            long currentTime = System.currentTimeMillis();

            return (currentTime - createdTime) > timeThrashold;
        }

    }

}


<|start_filename|>nifi-opcua-processors/src/main/java/de/fraunhofer/fit/processors/opcua/ListOPCNodes.java<|end_filename|>
/*
 * 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 de.fraunhofer.fit.processors.opcua;

import de.fraunhofer.fit.opcua.OPCUAService;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.Validator;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.util.StandardValidators;

import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

@Tags({"example"})
@CapabilityDescription("Provide a description")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute = "", description = "")})
@WritesAttributes({@WritesAttribute(attribute = "", description = "")})
public class ListOPCNodes extends AbstractProcessor {

    private static String starting_node = null;
    private static String print_indentation = "No";
    private static Integer max_recursiveDepth;
    private static Integer max_reference_per_node;
    private static boolean print_non_leaf_nodes;

    public static final PropertyDescriptor OPCUA_SERVICE = new PropertyDescriptor.Builder()
            .name("OPC UA Service")
            .description("Specifies the OPC UA Service that can be used to access data")
            .required(true)
            .identifiesControllerService(OPCUAService.class)
            .build();

    public static final PropertyDescriptor STARTING_NODE = new PropertyDescriptor
            .Builder().name("Starting Nodes")
            .description("From what node should Nifi begin browsing the node tree. Default is the root node. Seperate multiple nodes with a comma (,)")
            .addValidator(StandardValidators.NON_BLANK_VALIDATOR)
            .build();

    public static final PropertyDescriptor RECURSIVE_DEPTH = new PropertyDescriptor
            .Builder().name("Recursive Depth")
            .description("Maximum depth from the starting node to read, Default is 0")
            .required(true)
            .addValidator(StandardValidators.INTEGER_VALIDATOR)
            .build();

    public static final PropertyDescriptor PRINT_INDENTATION = new PropertyDescriptor
            .Builder().name("Print Indentation")
            .description("Should Nifi add indentation to the output text")
            .required(true)
            .defaultValue("")
            .addValidator(Validator.VALID)
            .build();

    public static final PropertyDescriptor MAX_REFERENCE_PER_NODE = new PropertyDescriptor
            .Builder().name("Max References Per Node")
            .description("The number of Reference Descriptions to pull per node query.")
            .required(true)
            .defaultValue("1000")
            .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
            .build();

    public static final PropertyDescriptor PRINT_NON_LEAF_NODES = new PropertyDescriptor
            .Builder().name("Print Non Leaf Nodes")
            .description("Whether or not to print the nodes which are not leaves.")
            .required(true)
            .defaultValue("true")
            .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
            .build();

    public static final Relationship SUCCESS = new Relationship.Builder()
            .name("Success")
            .description("Successful OPC read")
            .build();

    public static final Relationship FAILURE = new Relationship.Builder()
            .name("Failure")
            .description("Failed OPC read")
            .build();

    public List<PropertyDescriptor> descriptors;

    public Set<Relationship> relationships;

    @Override
    protected void init(final ProcessorInitializationContext context) {
        final List<PropertyDescriptor> descriptors = new ArrayList<>();
        descriptors.add(OPCUA_SERVICE);
        descriptors.add(RECURSIVE_DEPTH);
        descriptors.add(STARTING_NODE);
        descriptors.add(PRINT_INDENTATION);
        descriptors.add(MAX_REFERENCE_PER_NODE);
        descriptors.add(PRINT_NON_LEAF_NODES);

        this.descriptors = Collections.unmodifiableList(descriptors);

        final Set<Relationship> relationships = new HashSet<>();
        relationships.add(SUCCESS);
        relationships.add(FAILURE);
        this.relationships = Collections.unmodifiableSet(relationships);
    }

    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

    @OnScheduled
    public void onScheduled(final ProcessContext context) {
        print_indentation = context.getProperty(PRINT_INDENTATION).getValue();
        max_recursiveDepth = Integer.valueOf(context.getProperty(RECURSIVE_DEPTH).getValue());
        starting_node = context.getProperty(STARTING_NODE).getValue();
        max_reference_per_node = Integer.valueOf(context.getProperty(MAX_REFERENCE_PER_NODE).getValue());
        print_non_leaf_nodes = Boolean.valueOf(context.getProperty(PRINT_NON_LEAF_NODES).getValue());
    }

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

        final OPCUAService opcUAService = context.getProperty(OPCUA_SERVICE)
                .asControllerService(OPCUAService.class);

        byte[] nodes = opcUAService.getNodes(print_indentation, max_recursiveDepth,
                max_reference_per_node, print_non_leaf_nodes, starting_node);

        // Write the results back out to a flow file
        FlowFile flowFile = session.create();

        if (flowFile != null) {
            try {
                flowFile = session.write(flowFile, (OutputStream out) -> {
                    out.write(nodes);
                });

                // Transfer data to flow file
                session.transfer(flowFile, SUCCESS);
            } catch (ProcessException ex) {
                getLogger().error("Unable to process", ex);
                session.transfer(flowFile, FAILURE);
            }
        }

    }
}


<|start_filename|>nifi-opcua-service/src/main/java/de/fraunhofer/fit/opcua/KeyStoreLoader.java<|end_filename|>
package de.fraunhofer.fit.opcua;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileInputStream;
import java.security.*;
import java.security.cert.X509Certificate;
import java.util.Enumeration;
import java.util.regex.Pattern;

class KeyStoreLoader {

    private static final Pattern IP_ADDR_PATTERN = Pattern.compile(
            "^(([01]?\\d\\d?|2[0-4]\\d|25[0-5])\\.){3}([01]?\\d\\d?|2[0-4]\\d|25[0-5])$");

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private X509Certificate clientCertificate;
    private KeyPair clientKeyPair;

    KeyStoreLoader load(String ksLocation, char[] password) throws Exception {

        KeyStore keyStore = KeyStore.getInstance("JKS");
        keyStore.load(new FileInputStream(ksLocation), password);

        clientCertificate = null;
        clientKeyPair = null;

        Enumeration<String> aliases = keyStore.aliases();
        // Find the first PrivateKeyEntry in the keystore, which has both PrivateKey and Certificate
        while(aliases.hasMoreElements()) {
            String alias = aliases.nextElement();
            // Key and keystore share the same password
            Key clientPrivateKey = keyStore.getKey(alias, password);
            if (clientPrivateKey != null) {
                if (clientPrivateKey instanceof PrivateKey) {
                    if (keyStore.getCertificate(alias) != null) {
                        clientCertificate = (X509Certificate) keyStore.getCertificate(alias);
                        PublicKey clientPublicKey = clientCertificate.getPublicKey();
                        clientKeyPair = new KeyPair(clientPublicKey, (PrivateKey) clientPrivateKey);
                        break;
                    }
                }
            }
        }

        if (clientCertificate == null || clientKeyPair == null) {
            throw new Exception("No keypair found in keystore " + ksLocation);
        }

        return this;
    }

    X509Certificate getClientCertificate() {
        return clientCertificate;
    }

    KeyPair getClientKeyPair() {
        return clientKeyPair;
    }

}


<|start_filename|>nifi-opcua-service/src/main/java/de/fraunhofer/fit/opcua/TrustStoreLoader.java<|end_filename|>
package de.fraunhofer.fit.opcua;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileInputStream;
import java.security.KeyStore;
import java.security.cert.X509Certificate;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Enumeration;
import java.util.List;

class TrustStoreLoader {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private KeyStore keyStore;

    TrustStoreLoader load(String tsLocation, char[] password) throws Exception {

        keyStore = KeyStore.getInstance("JKS");
        keyStore.load(new FileInputStream(tsLocation), password);

        return this;
    }

    // Only verify the first certificate, and CA certificate at the end of the chain. Intermediate certificates are not verified
    public void verify(List<X509Certificate> serverCerts) throws Exception {

        // TODO: maybe also need to verify certificate according to the application name, application uri

        if (serverCerts.size() == 0) throw new Exception("No server certificate.");

        // Get a list of certificates from trust store
        List<X509Certificate> trustedCerts = new ArrayList<>();
        Enumeration<String> aliases = keyStore.aliases();
        while (aliases.hasMoreElements()) {
            String alias = aliases.nextElement();
            X509Certificate cert = (X509Certificate) keyStore.getCertificate(alias);
            if (cert != null) {
                trustedCerts.add(cert);
            }
        }

        // Check the first certificate to see if it is trusted
        for(int i=0; i<serverCerts.size(); i++ ) {

            // Verify the current server certificate with all trusted one
            X509Certificate cert = serverCerts.get(i);
            boolean certTrusted = trustedCerts.stream()
                    .anyMatch(c -> Arrays.equals(cert.getSignature(), c.getSignature()));
            if (certTrusted) return;

            // If no match, then we check whether the current server cert is signed by the next server cert up the chain
            if (i < serverCerts.size() - 1) {
                try {
                    serverCerts.get(i).verify(serverCerts.get(i + 1).getPublicKey());
                } catch (Exception e) {
                    throw new Exception("Server certificate chain not valid.");
                }
            }

        }

        // If the program reaches this point, it means it has checked all server certs along the chain but none matches
        throw new Exception("No trusted certificate found in the server certificate chain");
    }
}


<|start_filename|>Dockerfile<|end_filename|>
# Build the nar file

FROM maven:3.5 as builder

ARG BASE_DIR=/source

COPY . ${BASE_DIR}

WORKDIR ${BASE_DIR}

RUN mvn clean package -DskipTests


# ----------------
# Build a new Nifi image with the newly generated nar file included

FROM apache/nifi:1.4.0

ARG BASE_DIR=/source

ENV NIFI_BASE_DIR /opt/nifi
ENV NIFI_HOME ${NIFI_BASE_DIR}/nifi-1.4.0

COPY --from=builder ${BASE_DIR}/nifi-opcua-nar/target/*.nar ${NIFI_HOME}/lib/nifi-opcua.nar

EXPOSE 8080 8443 10000

USER nifi

WORKDIR ${NIFI_HOME}

# Startup NiFi
ENTRYPOINT ["bin/nifi.sh"]
CMD ["run"]


<|start_filename|>nifi-opcua-service/src/test/java/de/fraunhofer/fit/opcua/TestStandardOPCUAService.java<|end_filename|>
/*
 * 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 de.fraunhofer.fit.opcua;

import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.Before;
import org.junit.Test;

import java.util.Arrays;
import java.util.List;

public class TestStandardOPCUAService {

    private final String endpoint = "opc.tcp://10.223.104.20:48010";
    private TestRunner runner;
    private StandardOPCUAService service;

    @Before
    public void init() throws InitializationException {
        runner = TestRunners.newTestRunner(TestProcessor.class);
        service = new StandardOPCUAService();
        runner.addControllerService("test-good", service);
    }

/*    @Test
    public void testServiceInitialization() {

        runner.setProperty(service, StandardOPCUAService.ENDPOINT, endpoint);
        runner.assertValid(service);

        runner.enableControllerService(service);

        runner.disableControllerService(service);

    }

    @Test
    public void testServiceGetNodes() {
        runner.setProperty(service, StandardOPCUAService.ENDPOINT, endpoint);
        runner.assertValid(service);

        runner.enableControllerService(service);

        System.out.println(new String(service.getNodes("--", 3, 10, false,
                "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars")));

        runner.disableControllerService(service);
    }

    @Test
    public void testServiceGetValues() {
        runner.setProperty(service, StandardOPCUAService.ENDPOINT, endpoint);
        runner.assertValid(service);

        runner.enableControllerService(service);

        List<String> tagList = Arrays.asList("ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_EXT",
                "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG2_EXT");

        byte[] bytes = service.getValue(tagList, "Both", true, "");
        System.out.println(new String(bytes));

        runner.disableControllerService(service);
    }

    @Test
    public void testSecurityAccess() {
        runner.setProperty(service, StandardOPCUAService.ENDPOINT, endpoint);
        runner.setProperty(service, StandardOPCUAService.SECURITY_POLICY, "Basic128Rsa15");
        runner.setProperty(service, StandardOPCUAService.SECURITY_MODE, "SignAndEncrypt");
        runner.setProperty(service, StandardOPCUAService.CLIENT_KS_LOCATION, "src/test/resources/client.jks");
        runner.setProperty(service, StandardOPCUAService.CLIENT_KS_PASSWORD, "password");
        runner.setProperty(service, StandardOPCUAService.REQUIRE_SERVER_AUTH, "true");
        runner.setProperty(service, StandardOPCUAService.TRUSTSTORE_LOCATION, "src/test/resources/trust.jks");
        runner.setProperty(service, StandardOPCUAService.TRUSTSTORE_PASSWORD, "<PASSWORD>");
        runner.setProperty(service, StandardOPCUAService.AUTH_POLICY, "Anon");

        runner.assertValid(service);

        runner.enableControllerService(service);

        List<String> tagList = Arrays.asList("ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_EXT",
                "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG2_EXT");

        byte[] bytes = service.getValue(tagList, "Both", true, "");
        System.out.println(new String(bytes));

        runner.disableControllerService(service);
    }

    @Test(expected = InitializationException.class)
    public void testSecurityAccessWrongTrustStore() throws InitializationException {
        runner.setProperty(service, StandardOPCUAService.ENDPOINT, endpoint);
        runner.setProperty(service, StandardOPCUAService.SECURITY_POLICY, "Basic128Rsa15");
        runner.setProperty(service, StandardOPCUAService.SECURITY_MODE, "SignAndEncrypt");
        runner.setProperty(service, StandardOPCUAService.CLIENT_KS_LOCATION, "src/test/resources/client.jks");
        runner.setProperty(service, StandardOPCUAService.CLIENT_KS_PASSWORD, "password");
        runner.setProperty(service, StandardOPCUAService.REQUIRE_SERVER_AUTH, "true");
        runner.setProperty(service, StandardOPCUAService.TRUSTSTORE_LOCATION, "src/test/resources/trust-wrong.jks");
        runner.setProperty(service, StandardOPCUAService.TRUSTSTORE_PASSWORD, "password");
        runner.setProperty(service, StandardOPCUAService.AUTH_POLICY, "Anon");

        runner.assertValid(service);

        try {
            runner.enableControllerService(service);
        } catch (AssertionError e) {
            throw new InitializationException("");
        }

        List<String> tagList = Arrays.asList("ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_EXT",
                "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG2_EXT");

        byte[] bytes = service.getValue(tagList, "Both", true, "");
        System.out.println(new String(bytes));

        runner.disableControllerService(service);
    }*/

    @Test
    public void testUsernameSecurityAccess() {
        runner.setProperty(service, StandardOPCUAService.ENDPOINT, endpoint);
        runner.setProperty(service, StandardOPCUAService.SECURITY_POLICY, "Basic256Sha256");
        runner.setProperty(service, StandardOPCUAService.SECURITY_MODE, "SignAndEncrypt");
        runner.setProperty(service, StandardOPCUAService.APPLICATION_URI, "urn:ibhlinkua_001151:IBHsoftec:IBHLinkUA");
        runner.setProperty(service, StandardOPCUAService.CLIENT_KS_LOCATION, "src/test/resources/client.jks");
        runner.setProperty(service, StandardOPCUAService.CLIENT_KS_PASSWORD, "password");
        runner.setProperty(service, StandardOPCUAService.REQUIRE_SERVER_AUTH, "true");
        runner.setProperty(service, StandardOPCUAService.TRUSTSTORE_LOCATION, "src/test/resources/trust.jks");
        runner.setProperty(service, StandardOPCUAService.TRUSTSTORE_PASSWORD, "<PASSWORD>");
        runner.setProperty(service, StandardOPCUAService.AUTH_POLICY, "Username");
        runner.setProperty(service, StandardOPCUAService.USERNAME, "test1");
        runner.setProperty(service, StandardOPCUAService.PASSWORD, "password");

        runner.assertValid(service);

        runner.enableControllerService(service);

        List<String> tagList = Arrays.asList("ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_EXT",
                "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG2_EXT");

        byte[] bytes = service.getValue(tagList, "Both", true, "");
        System.out.println(new String(bytes));

        runner.disableControllerService(service);
    }

}


<|start_filename|>nifi-opcua-processors/src/test/java/de/fraunhofer/fit/processors/opcua/utils/RecordAggregatorTest.java<|end_filename|>
package de.fraunhofer.fit.processors.opcua.utils;

import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.io.StringReader;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

import static org.junit.Assert.assertEquals;

public class RecordAggregatorTest {

    private BlockingQueue<String> queue;
    private List<String> tags;
    private RecordAggregator ra;


    @Before
    public void init() throws IOException {
        queue = new LinkedBlockingQueue<>();
        String tagFilePath = (new File("src\\test\\resources\\husky_tags.txt")).getAbsolutePath();
        tags = parseFile(Paths.get(tagFilePath));
        ra = new RecordAggregator(tags, 100);
    }

    @Test
    public void testUniformTimeStampNoWait() {

        String queueString = "ns=2;s=47.ProcessVariables.Shot_Length,1528285608582,1528285608582,38.71,0\n" +
                "ns=2;s=47.ProcessVariables.Shot_Size,1528285608582,1528285608582,42.543697,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Position,1528285608582,1528285608582,10.848699,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Open_Time,1528285608582,1528285608582,0.20775,0\n" +
                "ns=2;s=47.ProcessVariables.Maximum_Fill_Pressure,1528285608582,1528285608582,33.245487,0\n" +
                "ns=2;s=47.CycleCounter,1528285608582,1528285608582,2419756,0\n" +
                "ns=2;s=47.ProcessVariables.Tonnage,1528285608582,1528285608582,150.13992,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Pressure,1528285608582,1528285608582,32.43998,0\n" +
                "ns=2;s=47.ProcessVariables.Back_Pressure,1528285608582,1528285608582,4.540133,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_Run_Time,1528285608582,1528285608582,0.948,0\n" +
                "ns=2;s=47.ProcessVariables.Oil_Temperature,1528285608582,1528285608582,50.000004,0\n" +
                "ns=2;s=47.ProcessVariables.Ejector_Maximum_Forward_Position,1528285608582,1528285608582,11.0207815,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Growth,1528285608582,1528285608582,-0.07940674,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_RPM,1528285608582,1528285608582,350.05255,0";

        String[] msgs = queueString.split("\n");
        for (int i = 0; i<msgs.length; i++) {
            ra.aggregate(msgs[i]);
        }

        assertEquals(0, ra.getReadyRecords().size());
    }

    @Test
    public void testUniformTimeStampWithWait() throws InterruptedException {

        String queueString = "ns=2;s=47.ProcessVariables.Shot_Length,1528285608582,1528285608582,38.71,0\n" +
                "ns=2;s=47.ProcessVariables.Shot_Size,1528285608582,1528285608582,42.543697,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Position,1528285608582,1528285608582,10.848699,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Open_Time,1528285608582,1528285608582,0.20775,0\n" +
                "ns=2;s=47.ProcessVariables.Maximum_Fill_Pressure,1528285608582,1528285608582,33.245487,0\n" +
                "ns=2;s=47.CycleCounter,1528285608582,1528285608582,2419756,0\n" +
                "ns=2;s=47.ProcessVariables.Tonnage,1528285608582,1528285608582,150.13992,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Pressure,1528285608582,1528285608582,32.43998,0\n" +
                "ns=2;s=47.ProcessVariables.Back_Pressure,1528285608582,1528285608582,4.540133,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_Run_Time,1528285608582,1528285608582,0.948,0\n" +
                "ns=2;s=47.ProcessVariables.Oil_Temperature,1528285608582,1528285608582,50.000004,0\n" +
                "ns=2;s=47.ProcessVariables.Ejector_Maximum_Forward_Position,1528285608582,1528285608582,11.0207815,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Growth,1528285608582,1528285608582,-0.07940674,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_RPM,1528285608582,1528285608582,350.05255,0";

        String[] msgs = queueString.split("\n");
        for (int i = 0; i<msgs.length; i++) {
            ra.aggregate(msgs[i]);
        }

        Thread.sleep(500);

        assertEquals(1, ra.getReadyRecords().size());
    }


    @Test
    public void testMutipleTimeStampsNoWait() {

        String queueString = "ns=2;s=47.SerialNumber,1528285396720,1526914447403,7691403,0\n" +
                "ns=2;s=47.Description,1528285396720,1526914447403,H300-RS65/60,0\n" +
                "ns=2;s=47.MachineState,1528285396720,1528257014806,Auto Cycling,0\n" +
                "ns=2;s=47.CycleInterruption,1528285396720,1528257014806,,0\n" +
                "ns=2;s=47.CycleCounter,1528285396720,1528285390205,2419724,0\n" +
                "ns=2;s=47.CycleCounter,1528285397032,1528285397032,2419725,0\n" +
                "ns=2;s=47.ProcessVariables.Cycle_Time,1528285396720,1528285390205,6.82,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Closing_Time,1528285396720,1528285390205,0.944,0\n" +
                "ns=2;s=47.ProcessVariables.Tonnage,1528285396720,1528285390205,150.00896,0\n" +
                "ns=2;s=47.ProcessVariables.Tonnage,1528285397032,1528285397032,149.96933,0\n" +
                "ns=2;s=47.ProcessVariables.Shot_Size,1528285396720,1528285390205,42.598698,0\n" +
                "ns=2;s=47.ProcessVariables.Shot_Size,1528285397032,1528285397032,42.6787,0\n" +
                "ns=2;s=47.ProcessVariables.Cushion,1528285396720,1528285390205,3.8786993,0\n" +
                "ns=2;s=47.ProcessVariables.Cushion,1528285397032,1528285397032,3.9336991,0\n" +
                "ns=2;s=47.ProcessVariables.Shot_Length,1528285396720,1528285390205,38.719997,0\n" +
                "ns=2;s=47.ProcessVariables.Shot_Length,1528285397032,1528285397032,38.745,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Position,1528285396720,1528285390205,10.883699,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Position,1528285397032,1528285397032,10.963699,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Pressure,1528285396720,1528285390205,32.73289,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Pressure,1528285397032,1528285397032,32.513206,0\n" +
                "ns=2;s=47.ProcessVariables.Maximum_Fill_Pressure,1528285396720,1528285390205,33.172256,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_1,1528285396720,1528285390205,21.529015,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_2,1528285396720,1528285390205,18.52667,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_2,1528285397032,1528285397032,18.746353,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_3,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_4,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_10,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Back_Pressure,1528285396720,1528285390205,4.759816,0\n" +
                "ns=2;s=47.ProcessVariables.Back_Pressure,1528285397032,1528285397032,4.833044,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_Run_Time,1528285396720,1528285390205,0.952,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_Run_Time,1528285397032,1528285397032,0.932,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Opening_Time,1528285396720,1528285390205,1.12,0\n" +
                "ns=2;s=47.ProcessVariables.Ejector_Forward_Time,1528285396720,1528285390205,0.132,0\n" +
                "ns=2;s=47.ProcessVariables.Oil_Temperature,1528285396720,1528285390205,50.09346,0\n" +
                "ns=2;s=47.ProcessVariables.Oil_Temperature,1528285397032,1528285397032,50.062447,0\n" +
                "ns=2;s=47.ProcessVariables.Extruder_Temperature_-_1,1528285396720,1528285390205,195.0,0\n" +
                "ns=2;s=47.ProcessVariables.Extruder_Temperature_-_2,1528285396720,1528285390205,195.0,0\n" +
                "ns=2;s=47.ProcessVariables.Extruder_Temperature_-_3,1528285396720,1528285390205,200.0,0\n" +
                "ns=2;s=47.ProcessVariables.Barrel_Head_Temperature,1528285396720,1528285390205,210.0,0\n" +
                "ns=2;s=47.ProcessVariables.Ejector_Back_Time,1528285396720,1528285390205,0.296,0\n" +
                "ns=2;s=47.ProcessVariables.Ejector_Maximum_Forward_Position,1528285396720,1528285390205,11.003781,0\n" +
                "ns=2;s=47.ProcessVariables.Ejector_Maximum_Forward_Position,1528285397032,1528285397032,11.036781,0\n" +
                "ns=2;s=47.ProcessVariables.Maximum_Cavity_Pressure,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Cavity_Pressure_At_Transition,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Growth,1528285396720,1528285390205,-0.0803833,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Growth,1528285397032,1528285397032,-0.07940674,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Open_Time,1528285396720,1528285390205,0.208,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Open_Time,1528285397032,1528285397032,0.207,0\n" +
                "ns=2;s=47.ProcessVariables.Nozzle_Adapter_Temperature,1528285396720,1528285390205,220.1,0\n" +
                "ns=2;s=47.ProcessVariables.Nozzle_Adapter_Temperature,1528285397032,1528285397032,220.0,0\n" +
                "ns=2;s=47.ProcessVariables.Nozzle_Shutoff_Temperature,1528285396720,1528285390205,215.0,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_RPM,1528285396720,1528285390205,350.12796,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_RPM,1528285397032,1528285397032,349.8606,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_5,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_6,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_7,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_8,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_9,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Extruder_Temperature_-_4,1528285396720,1528285390205,205.0,0\n" +
                "ns=2;s=47.ProcessVariables.Cooling_Time,1528285396720,1528285390205,2.308,0\n" +
                "ns=2;s=47.ProcessVariables.Injection_Hold_Time,1528285396720,1528285390205,1.702,0";

        String[] msgs = queueString.split("\n");
        for (int i = 0; i<msgs.length; i++) {
            ra.aggregate(msgs[i]);
        }

        assertEquals(0, ra.getReadyRecords().size());
    }


    @Test
    public void testMutipleTimeStampsWithWait() throws InterruptedException {

        String queueString =
                "ns=2;s=47.CycleCounter,1528285396720,1528285390205,2419724,0\n" +
                "ns=2;s=47.CycleCounter,1528285397032,1528285397032,2419725,0\n" +
                "ns=2;s=47.ProcessVariables.Cycle_Time,1528285396720,1528285390205,6.82,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Closing_Time,1528285396720,1528285390205,0.944,0\n" +
                "ns=2;s=47.ProcessVariables.Tonnage,1528285396720,1528285390205,150.00896,0\n" +
                "ns=2;s=47.ProcessVariables.Tonnage,1528285397032,1528285397032,149.96933,0\n" +
                "ns=2;s=47.ProcessVariables.Shot_Size,1528285396720,1528285390205,42.598698,0\n" +
                "ns=2;s=47.ProcessVariables.Shot_Size,1528285397032,1528285397032,42.6787,0\n" +
                "ns=2;s=47.ProcessVariables.Cushion,1528285396720,1528285390205,3.8786993,0\n" +
                "ns=2;s=47.ProcessVariables.Cushion,1528285397032,1528285397032,3.9336991,0\n" +
                "ns=2;s=47.ProcessVariables.Shot_Length,1528285396720,1528285390205,38.719997,0\n" +
                "ns=2;s=47.ProcessVariables.Shot_Length,1528285397032,1528285397032,38.745,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Position,1528285396720,1528285390205,10.883699,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Position,1528285397032,1528285397032,10.963699,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Pressure,1528285396720,1528285390205,32.73289,0\n" +
                "ns=2;s=47.ProcessVariables.Transition_Pressure,1528285397032,1528285397032,32.513206,0\n" +
                "ns=2;s=47.ProcessVariables.Maximum_Fill_Pressure,1528285396720,1528285390205,33.172256,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_1,1528285396720,1528285390205,21.529015,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_2,1528285396720,1528285390205,18.52667,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_2,1528285397032,1528285397032,18.746353,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_3,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_4,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_10,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Back_Pressure,1528285396720,1528285390205,4.759816,0\n" +
                "ns=2;s=47.ProcessVariables.Back_Pressure,1528285397032,1528285397032,4.833044,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_Run_Time,1528285396720,1528285390205,0.952,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_Run_Time,1528285397032,1528285397032,0.932,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Opening_Time,1528285396720,1528285390205,1.12,0\n" +
                "ns=2;s=47.ProcessVariables.Ejector_Forward_Time,1528285396720,1528285390205,0.132,0\n" +
                "ns=2;s=47.ProcessVariables.Oil_Temperature,1528285396720,1528285390205,50.09346,0\n" +
                "ns=2;s=47.ProcessVariables.Oil_Temperature,1528285397032,1528285397032,50.062447,0\n" +
                "ns=2;s=47.ProcessVariables.Extruder_Temperature_-_1,1528285396720,1528285390205,195.0,0\n" +
                "ns=2;s=47.ProcessVariables.Extruder_Temperature_-_2,1528285396720,1528285390205,195.0,0\n" +
                "ns=2;s=47.ProcessVariables.Extruder_Temperature_-_3,1528285396720,1528285390205,200.0,0\n" +
                "ns=2;s=47.ProcessVariables.Barrel_Head_Temperature,1528285396720,1528285390205,210.0,0\n" +
                "ns=2;s=47.ProcessVariables.Ejector_Back_Time,1528285396720,1528285390205,0.296,0\n" +
                "ns=2;s=47.ProcessVariables.Ejector_Maximum_Forward_Position,1528285396720,1528285390205,11.003781,0\n" +
                "ns=2;s=47.ProcessVariables.Ejector_Maximum_Forward_Position,1528285397032,1528285397032,11.036781,0\n" +
                "ns=2;s=47.ProcessVariables.Maximum_Cavity_Pressure,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Cavity_Pressure_At_Transition,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Growth,1528285396720,1528285390205,-0.0803833,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Growth,1528285397032,1528285397032,-0.07940674,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Open_Time,1528285396720,1528285390205,0.208,0\n" +
                "ns=2;s=47.ProcessVariables.Mold_Open_Time,1528285397032,1528285397032,0.207,0\n" +
                "ns=2;s=47.ProcessVariables.Nozzle_Adapter_Temperature,1528285396720,1528285390205,220.1,0\n" +
                "ns=2;s=47.ProcessVariables.Nozzle_Adapter_Temperature,1528285397032,1528285397032,220.0,0\n" +
                "ns=2;s=47.ProcessVariables.Nozzle_Shutoff_Temperature,1528285396720,1528285390205,215.0,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_RPM,1528285396720,1528285390205,350.12796,0\n" +
                "ns=2;s=47.ProcessVariables.Screw_RPM,1528285397032,1528285397032,349.8606,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_5,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_6,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_7,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_8,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Hold_Pressure_Zone_-_9,1528285396720,1528285390205,0.0,0\n" +
                "ns=2;s=47.ProcessVariables.Extruder_Temperature_-_4,1528285396720,1528285390205,205.0,0\n" +
                "ns=2;s=47.ProcessVariables.Cooling_Time,1528285396720,1528285390205,2.308,0\n" +
                "ns=2;s=47.ProcessVariables.Injection_Hold_Time,1528285396720,1528285390205,1.702,0";

        String[] msgs = queueString.split("\n");
        for (int i = 0; i<msgs.length; i++) {
            ra.aggregate(msgs[i]);
        }

        Thread.sleep(500);

        assertEquals(2, ra.getReadyRecords().size());
    }


    private List<String> parseFile(Path filePath) throws IOException {
        byte[] encoded;
        encoded = Files.readAllBytes(filePath);
        String fileContent = new String(encoded, Charset.defaultCharset());
        return new BufferedReader(new StringReader(fileContent)).lines().collect(Collectors.toList());
    }


}


<|start_filename|>nifi-opcua-processors/src/test/java/de/fraunhofer/fit/processors/opcua/ListOPCNodesTest.java<|end_filename|>
/*
 * 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 de.fraunhofer.fit.processors.opcua;

import de.fraunhofer.fit.opcua.StandardOPCUAService;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.Before;
import org.junit.Test;

import java.util.List;

public class ListOPCNodesTest {

    private TestRunner testRunner;
    private final String endpoint = "opc.tcp://10.223.104.20:48010";
    private StandardOPCUAService service;

    @Before
    public void init() throws InitializationException {
        testRunner = TestRunners.newTestRunner(ListOPCNodes.class);
        service = new StandardOPCUAService();
        testRunner.addControllerService("controller", service);

        testRunner.setProperty(service, StandardOPCUAService.ENDPOINT, endpoint);
        testRunner.assertValid(service);

        testRunner.enableControllerService(service);
    }

    @Test
    public void testListingNodes() {
        testRunner.setProperty(ListOPCNodes.OPCUA_SERVICE, "controller");
        testRunner.setProperty(ListOPCNodes.MAX_REFERENCE_PER_NODE, "10");
        testRunner.setProperty(ListOPCNodes.PRINT_INDENTATION, "-");
        //testRunner.setProperty(ListOPCNodes.STARTING_NODE, "ns=4;s=S71500/ET200MP-Station_2.PLC_1");
        testRunner.setProperty(ListOPCNodes.RECURSIVE_DEPTH, "4");
        testRunner.setProperty(ListOPCNodes.PRINT_NON_LEAF_NODES, "true");

        testRunner.run();

        List<MockFlowFile> results = testRunner.getFlowFilesForRelationship(GetOPCData.SUCCESS);
        System.out.println(new String(testRunner.getContentAsByteArray(results.get(0))));
    }

}


<|start_filename|>nifi-opcua-processors/src/test/java/de/fraunhofer/fit/processors/opcua/SubscribeOPCNodesTest.java<|end_filename|>
/*
 * 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 de.fraunhofer.fit.processors.opcua;

import de.fraunhofer.fit.opcua.StandardOPCUAService;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.stubbing.Answer;

import java.io.File;
import java.util.List;
import java.util.concurrent.BlockingQueue;

import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.spy;


public class SubscribeOPCNodesTest {

    private TestRunner testRunner;
    private StandardOPCUAService service;

    @Before
    public void init() throws InitializationException {
        testRunner = TestRunners.newTestRunner(SubscribeOPCNodes.class);

        // Use partial mock
        service = spy(new StandardOPCUAService());
        Mockito.doNothing().when(service).unsubscribe(anyString());
        Mockito.doNothing().when(service).onEnabled(any());
        Mockito.doNothing().when(service).shutdown();

        testRunner.addControllerService("controller", service);

        testRunner.setProperty(service, StandardOPCUAService.ENDPOINT, "dummy endpoint");
        testRunner.assertValid(service);

        testRunner.enableControllerService(service);
    }


    @Test
    public void testAggregateRecords() throws Exception {

        String tagFilePath = (new File("src/test/resources/husky_tags.txt")).getAbsolutePath();

        String queueString =
                "ns=2;s=47.ProcessVariables.Shot_Length,1528285608582,1528285608582,38.71,0\n" +
                        "ns=2;s=47.ProcessVariables.Shot_Size,1528285608582,1528285608582,42.543697,0\n" +
                        "ns=2;s=47.ProcessVariables.Transition_Position,1528285608582,1528285608582,10.848699,0\n" +
                        "ns=2;s=47.ProcessVariables.Mold_Open_Time,1528285608582,1528285608582,0.20775,0\n" +
                        "ns=2;s=47.ProcessVariables.Maximum_Fill_Pressure,1528285608582,1528285608582,33.245487,0\n" +
                        "ns=2;s=47.CycleCounter,1528285608582,1528285608582,2419756,0\n" +
                        "ns=2;s=47.ProcessVariables.Tonnage,1528285608582,1528285608582,150.13992,0\n" +
                        "ns=2;s=47.ProcessVariables.Transition_Pressure,1528285608582,1528285608582,32.43998,0\n" +
                        "ns=2;s=47.ProcessVariables.Back_Pressure,1528285608582,1528285608582,4.540133,0\n" +
                        "ns=2;s=47.ProcessVariables.Screw_Run_Time,1528285608582,1528285608582,0.948,0\n" +
                        "ns=2;s=47.ProcessVariables.Oil_Temperature,1528285608582,1528285608582,50.000004,0\n" +
                        "ns=2;s=47.ProcessVariables.Ejector_Maximum_Forward_Position,1528285608582,1528285608582,11.0207815,0\n" +
                        "ns=2;s=47.ProcessVariables.Mold_Growth,1528285608582,1528285608582,-0.07940674,0\n" +
                        "ns=2;s=47.ProcessVariables.Screw_RPM,1528285608582,1528285608582,350.05255,0";


        Mockito.doAnswer(
                (Answer<String>) invocation -> {
                    Object[] args = invocation.getArguments();
                    populateQueue((BlockingQueue<String>) args[1], queueString);
                    return "12345678"; // random subscriber uid, doesn't matter in test
                }
        ).when(service).subscribe(any(), any(), anyBoolean(), anyLong());


        testRunner.setProperty(SubscribeOPCNodes.OPCUA_SERVICE, "controller");
        testRunner.setProperty(SubscribeOPCNodes.TAG_FILE_LOCATION, tagFilePath);
        testRunner.setProperty(SubscribeOPCNodes.AGGREGATE_RECORD, "true");
        testRunner.setProperty(SubscribeOPCNodes.MIN_PUBLISH_INTERVAL, "100");

        testRunner.run(1, false, true);
        Thread.sleep(500);
        testRunner.run(1, true, false);


        List<MockFlowFile> results = testRunner.getFlowFilesForRelationship(GetOPCData.SUCCESS);
        assertEquals(1, results.size());
        String expectedPayload = "1528285608582,,,,,2419756,,,150.13992,42.543697,,38.71,,10.848699,32.43998,33.245487,,,,,,4.540133,0.948,,,,50.000004,,,,,,11.0207815,,,-0.07940674,0.20775,,,350.05255,,,,,,,,,,,," + System.lineSeparator();
        results.get(0).assertContentEquals(expectedPayload);

    }


    @Test
    public void testReal() throws Exception {

        String tagFilePath = (new File("src/test/resources/arburg_tag.txt")).getAbsolutePath();

        StandardOPCUAService realService = new StandardOPCUAService();

        testRunner.addControllerService("real", realService);
        testRunner.setProperty(realService, StandardOPCUAService.ENDPOINT, "opc.tcp://localhost:9000");
        testRunner.setProperty(realService, StandardOPCUAService.AUTH_POLICY, "Username");
        testRunner.setProperty(realService, StandardOPCUAService.USERNAME, "host_computer");
        testRunner.setProperty(realService, StandardOPCUAService.PASSWORD, "");
        testRunner.setProperty(realService, StandardOPCUAService.USE_PROXY, "true");
        testRunner.enableControllerService(realService);

        testRunner.setProperty(SubscribeOPCNodes.OPCUA_SERVICE, "real");
        testRunner.setProperty(SubscribeOPCNodes.TAG_FILE_LOCATION, tagFilePath);
        testRunner.setProperty(SubscribeOPCNodes.AGGREGATE_RECORD, "true");
        testRunner.setProperty(SubscribeOPCNodes.MIN_PUBLISH_INTERVAL, "1000");

        testRunner.run(1, false, true);
        Thread.sleep(1000);
        testRunner.run(1, false, false);
        Thread.sleep(1000);
        testRunner.run(1, false, false);
        Thread.sleep(1000);
        testRunner.run(1, false, false);
        Thread.sleep(1000);
        testRunner.run(1, false, false);
        Thread.sleep(1000);
        testRunner.run(1, false, false);
        Thread.sleep(1000);
        testRunner.run(1, false, false);
        Thread.sleep(1000);
        testRunner.run(1, false, false);
        Thread.sleep(1000);
        testRunner.run(1, true, false);


        List<MockFlowFile> results = testRunner.getFlowFilesForRelationship(GetOPCData.SUCCESS);
        for(MockFlowFile f : results) {
            System.out.println(new String(testRunner.getContentAsByteArray(f)));
        }

        assertEquals(1, results.size());
        String expectedPayload = "1528285608582,,,,,2419756,,,150.13992,42.543697,,38.71,,10.848699,32.43998,33.245487,,,,,,4.540133,0.948,,,,50.000004,,,,,,11.0207815,,,-0.07940674,0.20775,,,350.05255,,,,,,,,,,,," + System.lineSeparator();
        results.get(0).assertContentEquals(expectedPayload);

        testRunner.disableControllerService(realService);

    }



    @After
    public void shutdown() {
        testRunner.disableControllerService(service);
    }


    private void populateQueue(BlockingQueue<String> queue, String str) {
        String[] msgs = str.split("\n");
        for (int i = 0; i < msgs.length; i++) {
            queue.offer(msgs[i] + '\n');
        }
    }

}


<|start_filename|>nifi-opcua-processors/src/main/java/de/fraunhofer/fit/processors/opcua/SubscribeOPCNodes.java<|end_filename|>
/*
 * 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 de.fraunhofer.fit.processors.opcua;

import de.fraunhofer.fit.opcua.OPCUAService;
import de.fraunhofer.fit.processors.opcua.utils.RecordAggregator;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.annotation.lifecycle.OnStopped;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.expression.ExpressionLanguageScope;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.*;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.OutputStream;
import java.io.StringReader;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.stream.Collectors;

@Tags({"opc"})
@CapabilityDescription("Subscribe to a list of nodes and output flowfiles when changes are detected.")
public class SubscribeOPCNodes extends AbstractProcessor {

    private OPCUAService opcUaService;
    private BlockingQueue<String> msgQueue;
    private List<String> tagNames;
    private String subscriberUid;
    private boolean aggregateRecord;
    private boolean tsChangedNotify;
    private long minPublishInterval;
    private RecordAggregator recordAggregator;

    public static final PropertyDescriptor OPCUA_SERVICE = new PropertyDescriptor.Builder()
            .name("OPC UA Service")
            .description("Specifies the OPC UA Service that can be used to access data")
            .required(true)
            .identifiesControllerService(OPCUAService.class)
            .build();

    public static final PropertyDescriptor TAG_FILE_LOCATION = new PropertyDescriptor
            .Builder().name("Tag List File Location")
            .description("The location of the tag list file")
            .required(true)
            .addValidator(StandardValidators.FILE_EXISTS_VALIDATOR)
            .sensitive(false)
            .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY)
            .build();

    public static final PropertyDescriptor AGGREGATE_RECORD = new PropertyDescriptor
            .Builder().name("Aggregate Records")
            .description("Whether to aggregate records. If this is set to true, then variable with the same time stamp will be merged into a single line. This is useful for batch-based data.")
            .required(true)
            .defaultValue("false")
            .allowableValues("true", "false")
            .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
            .sensitive(false)
            .build();

    public static final PropertyDescriptor TS_CHANGE_NOTIFY = new PropertyDescriptor
            .Builder().name("Notified when Timestamp changed")
            .description("Whether the data should be collected, when only the timestamp of a variable has changed, but not its value.")
            .required(true)
            .defaultValue("true")
            .allowableValues("true", "false")
            .addValidator(StandardValidators.BOOLEAN_VALIDATOR)
            .sensitive(false)
            .build();

    public static final PropertyDescriptor MIN_PUBLISH_INTERVAL = new PropertyDescriptor
            .Builder().name("Minimum publish interval of subscription notification messages")
            .description("The minimum publish interval of subscription notification messages. Set this property to a lower value so that rapid change of data can be detected.")
            .required(true)
            .defaultValue("1000")
            .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
            .build();

    public static final Relationship SUCCESS = new Relationship.Builder()
            .name("Success")
            .description("Successful OPC read")
            .build();

    public static final Relationship FAILURE = new Relationship.Builder()
            .name("Failure")
            .description("Failed OPC read")
            .build();

    public List<PropertyDescriptor> descriptors;

    public Set<Relationship> relationships;

    @Override
    protected void init(final ProcessorInitializationContext context) {
        final List<PropertyDescriptor> descriptors = new ArrayList<>();
        descriptors.add(OPCUA_SERVICE);
        descriptors.add(TAG_FILE_LOCATION);
        descriptors.add(AGGREGATE_RECORD);
        descriptors.add(TS_CHANGE_NOTIFY);
        descriptors.add(MIN_PUBLISH_INTERVAL);

        this.descriptors = Collections.unmodifiableList(descriptors);

        final Set<Relationship> relationships = new HashSet<>();
        relationships.add(SUCCESS);
        relationships.add(FAILURE);
        this.relationships = Collections.unmodifiableSet(relationships);

        msgQueue = new LinkedBlockingQueue<>();
    }

    @Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

    @OnScheduled
    public void onScheduled(final ProcessContext context) {

        opcUaService = context.getProperty(OPCUA_SERVICE)
                .asControllerService(OPCUAService.class);


        try {
            tagNames = parseFile(Paths.get(context.getProperty(TAG_FILE_LOCATION).evaluateAttributeExpressions().toString()));
        } catch (IOException e) {
            getLogger().error("Error reading tag list from local file.");
            return;
        }

        aggregateRecord = Boolean.valueOf(context.getProperty(AGGREGATE_RECORD).getValue());
        tsChangedNotify = Boolean.valueOf(context.getProperty(TS_CHANGE_NOTIFY).getValue());
        minPublishInterval = context.getProperty(MIN_PUBLISH_INTERVAL).asLong();

        subscriberUid = opcUaService.subscribe(tagNames, msgQueue, tsChangedNotify, minPublishInterval);

        recordAggregator = new RecordAggregator(tagNames, minPublishInterval);
    }

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

        if(!aggregateRecord) {
            String msg;
            while ((msg = msgQueue.poll()) != null) {
                // Write the results back out to a flow file
                FlowFile flowFile = session.create();

                byte[] outputMsgBytes = msg.getBytes();
                if (flowFile != null) {
                    try {
                        flowFile = session.write(flowFile, (OutputStream out) -> out.write(outputMsgBytes));

                        // Transfer data to flow file
                        session.transfer(flowFile, SUCCESS);
                    } catch (ProcessException ex) {
                        getLogger().error("Unable to process", ex);
                        session.transfer(flowFile, FAILURE);
                    }
                }
            }
        } else {
            String rawMsg;
            while ((rawMsg = msgQueue.poll()) != null) {
                recordAggregator.aggregate(rawMsg);
            }

            List<String> list = recordAggregator.getReadyRecords();
            for(String msg: list) {
                // Write the results back out to a flow file
                FlowFile flowFile = session.create();

                byte[] outputMsgBytes = msg.getBytes();
                if (flowFile != null) {
                    try {
                        flowFile = session.write(flowFile, (OutputStream out) -> out.write(outputMsgBytes));

                        // add header to attribute (remember to add time stamp colum to the first)
                        Map<String, String> attrMap = new HashMap<>();
                        attrMap.put("csvHeader", "timestamp," + String.join(",", tagNames) + System.getProperty("line.separator"));
                        flowFile = session.putAllAttributes(flowFile, attrMap);

                        // Transfer data to flow file
                        session.transfer(flowFile, SUCCESS);
                    } catch (ProcessException ex) {
                        getLogger().error("Unable to process", ex);
                        session.transfer(flowFile, FAILURE);
                    }
                }
            }
        }

    }

    @OnStopped
    public void onStopped(final ProcessContext context) {

        getLogger().debug("Unsubscribing from OPC Server...");
        opcUaService.unsubscribe(subscriberUid);

    }

    private List<String> parseFile(Path filePath) throws IOException {
        byte[] encoded;
        encoded = Files.readAllBytes(filePath);
        String fileContent = new String(encoded, Charset.defaultCharset());
        return new BufferedReader(new StringReader(fileContent)).lines().collect(Collectors.toList());
    }
}


<|start_filename|>nifi-opcua-processors/src/test/java/de/fraunhofer/fit/processors/opcua/GetOPCDataTest.java<|end_filename|>
/*
 * 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 de.fraunhofer.fit.processors.opcua;

import de.fraunhofer.fit.opcua.StandardOPCUAService;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;

import java.io.File;
import java.util.List;

import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.spy;


public class GetOPCDataTest {

    private TestRunner testRunner;
    private StandardOPCUAService service;

    @Before
    public void init() throws InitializationException {
        testRunner = TestRunners.newTestRunner(GetOPCData.class);

        // Use partial mock
        service = spy(new StandardOPCUAService());
        Mockito.doNothing().when(service).onEnabled(any());
        Mockito.doNothing().when(service).shutdown();

        testRunner.addControllerService("controller", service);

        testRunner.setProperty(service, StandardOPCUAService.ENDPOINT, "dummy endpoint");
        testRunner.assertValid(service);

        testRunner.enableControllerService(service);
    }

    @Test
    public void testGetData() {

        String tagFilePath = (new File("src/test/resources/tags.txt")).getAbsolutePath();

        testRunner.setProperty(GetOPCData.OPCUA_SERVICE, "controller");
        testRunner.setProperty(GetOPCData.RETURN_TIMESTAMP, "Both");
        testRunner.setProperty(GetOPCData.EXCLUDE_NULL_VALUE, "Yes");
        testRunner.setProperty(GetOPCData.TAG_LIST_SOURCE, "Local File");
        testRunner.setProperty(GetOPCData.TAG_LIST_FILE, tagFilePath);

        byte[] values = new String(
                "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_EXT,123456,123456,1,0\n" +
                "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_RET,123456,123456,2,0").getBytes();

        Mockito.doReturn(values).when(service).getValue(any(), any(), anyBoolean(), any());

        testRunner.run();

        List<MockFlowFile> results = testRunner.getFlowFilesForRelationship(GetOPCData.SUCCESS);
        assertEquals(1, results.size());
        results.get(0).assertContentEquals("ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_EXT,123456,123456,1,0\n" +
                "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_RET,123456,123456,2,0");
        //System.out.println(new String(testRunner.getContentAsByteArray(results.get(0))));

    }

    @Test
    public void testGetDataWithAggregation() {

        String tagFilePath = (new File("src/test/resources/tags.txt")).getAbsolutePath();

        testRunner.setProperty(GetOPCData.OPCUA_SERVICE, "controller");
        testRunner.setProperty(GetOPCData.RETURN_TIMESTAMP, "Both");
        testRunner.setProperty(GetOPCData.EXCLUDE_NULL_VALUE, "Yes");
        testRunner.setProperty(GetOPCData.TAG_LIST_SOURCE, "Local File");
        testRunner.setProperty(GetOPCData.TAG_LIST_FILE, tagFilePath);
        testRunner.setProperty(GetOPCData.AGGREGATE_RECORD, "true");

        byte[] values = new String(
                "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_EXT,123456,123456,1,0" + System.lineSeparator() +
                        "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_RET,123456,123456,2,0"+ System.lineSeparator() +
                        "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG2_RET,123456,123456,3,0"+ System.lineSeparator() +
                        "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG2_RET,123456,123456,4,0"+ System.lineSeparator() +
                        "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG3_RET,123456,123456,5,0"+ System.lineSeparator() +
                        "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG3_RET,123456,123456,6,0").getBytes();

        Mockito.doReturn(values).when(service).getValue(any(), any(), anyBoolean(), any());

        testRunner.run();

        List<MockFlowFile> results = testRunner.getFlowFilesForRelationship(GetOPCData.SUCCESS);
        assertEquals(1, results.size());
        results.get(0).assertContentEquals("123456,1,2,3,4,5,6");
        results.get(0).assertAttributeEquals("csvHeader",
                            "timestamp," +
                                    "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_EXT," +
                                    "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG1_RET," +
                                    "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG2_EXT," +
                                    "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG3_RET," +
                                    "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG3_EXT," +
                                    "ns=4;s=S71500/ET200MP-Station_2.PLC_1.GlobalVars.I_MAG3_RET");

    }

    @After
    public void shutdown() {
        testRunner.disableControllerService(service);
    }

}
