You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jo...@apache.org on 2015/03/17 03:48:42 UTC
[07/20] incubator-nifi git commit: NIFI-410 pushed RAT exclusions
down to the relevant modules and stopped doing broad test resource exclusion
- is now specific
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/MulticastProtocolListenerTest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/MulticastProtocolListenerTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/MulticastProtocolListenerTest.java
new file mode 100644
index 0000000..4233d88
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/MulticastProtocolListenerTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.cluster.protocol.impl;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.InetSocketAddress;
+import java.net.MulticastSocket;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.nifi.cluster.protocol.ProtocolContext;
+import org.apache.nifi.cluster.protocol.ProtocolException;
+import org.apache.nifi.cluster.protocol.ProtocolHandler;
+import org.apache.nifi.cluster.protocol.ProtocolMessageMarshaller;
+import org.apache.nifi.cluster.protocol.jaxb.JaxbProtocolContext;
+import org.apache.nifi.cluster.protocol.jaxb.message.JaxbProtocolUtils;
+import org.apache.nifi.cluster.protocol.message.MulticastProtocolMessage;
+import org.apache.nifi.cluster.protocol.message.PingMessage;
+import org.apache.nifi.cluster.protocol.message.ProtocolMessage;
+import org.apache.nifi.io.socket.multicast.MulticastConfiguration;
+import org.apache.nifi.io.socket.multicast.MulticastUtils;
+import org.junit.After;
+import static org.junit.Assert.*;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+
+/**
+ * @author unattributed
+ */
+public class MulticastProtocolListenerTest {
+
+ private MulticastProtocolListener listener;
+
+ private MulticastSocket socket;
+
+ private InetSocketAddress address;
+
+ private MulticastConfiguration configuration;
+
+ private ProtocolContext protocolContext;
+
+ @Before
+ public void setup() throws Exception {
+
+ address = new InetSocketAddress("226.1.1.1", 60000);
+ configuration = new MulticastConfiguration();
+
+ protocolContext = new JaxbProtocolContext(JaxbProtocolUtils.JAXB_CONTEXT);
+
+ listener = new MulticastProtocolListener(5, address, configuration, protocolContext);
+ listener.start();
+
+ socket = MulticastUtils.createMulticastSocket(address.getPort(), configuration);
+ }
+
+ @After
+ public void teardown() throws IOException {
+ try {
+ if(listener.isRunning()) {
+ listener.stop();
+ }
+ } finally {
+ MulticastUtils.closeQuietly(socket);
+ }
+ }
+
+ @Ignore("Test needs to be reworked. Fails if on a system without actiev network connection")
+ @Test
+ public void testBadRequest() throws Exception {
+ DelayedProtocolHandler handler = new DelayedProtocolHandler(0);
+ listener.addHandler(handler);
+ DatagramPacket packet = new DatagramPacket(new byte[] {5}, 1, address);
+ socket.send(packet);
+ Thread.sleep(250);
+ assertEquals(0, handler.getMessages().size());
+ }
+
+ @Ignore("this test works sometimes and fails others - needs work to be reliable")
+ @Test
+ public void testRequest() throws Exception {
+
+ ReflexiveProtocolHandler handler = new ReflexiveProtocolHandler();
+ listener.addHandler(handler);
+
+ ProtocolMessage msg = new PingMessage();
+ MulticastProtocolMessage multicastMsg = new MulticastProtocolMessage("some-id", msg);
+
+ // marshal message to output stream
+ ProtocolMessageMarshaller marshaller = protocolContext.createMarshaller();
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ marshaller.marshal(multicastMsg, baos);
+ byte[] requestPacketBytes = baos.toByteArray();
+ DatagramPacket packet = new DatagramPacket(requestPacketBytes, requestPacketBytes.length, address);
+ socket.send(packet);
+
+ Thread.sleep(250);
+ assertEquals(1, handler.getMessages().size());
+ assertEquals(msg.getType(), handler.getMessages().get(0).getType());
+
+ }
+
+ private class ReflexiveProtocolHandler implements ProtocolHandler {
+
+ private List<ProtocolMessage> messages = new ArrayList<>();
+
+ @Override
+ public ProtocolMessage handle(ProtocolMessage msg) throws ProtocolException {
+ messages.add(msg);
+ return msg;
+ }
+
+ @Override
+ public boolean canHandle(ProtocolMessage msg) {
+ return true;
+ }
+
+ public List<ProtocolMessage> getMessages() {
+ return messages;
+ }
+
+ }
+
+ private class DelayedProtocolHandler implements ProtocolHandler {
+
+ private int delay = 0;
+
+ private List<ProtocolMessage> messages = new ArrayList<>();
+
+ public DelayedProtocolHandler(int delay) {
+ this.delay = delay;
+ }
+
+ @Override
+ public ProtocolMessage handle(ProtocolMessage msg) throws ProtocolException {
+ try {
+ messages.add(msg);
+ Thread.sleep(delay);
+ return null;
+ } catch(final InterruptedException ie) {
+ throw new ProtocolException(ie);
+ }
+
+ }
+
+ @Override
+ public boolean canHandle(ProtocolMessage msg) {
+ return true;
+ }
+
+ public List<ProtocolMessage> getMessages() {
+ return messages;
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImplTest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImplTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImplTest.java
new file mode 100644
index 0000000..1c5ba9e
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImplTest.java
@@ -0,0 +1,203 @@
+/*
+ * 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.cluster.protocol.impl;
+
+import org.apache.nifi.cluster.protocol.impl.ClusterServiceLocator;
+import org.apache.nifi.cluster.protocol.impl.SocketProtocolListener;
+import org.apache.nifi.cluster.protocol.impl.NodeProtocolSenderImpl;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.UUID;
+
+import org.apache.nifi.cluster.protocol.ConnectionRequest;
+import org.apache.nifi.cluster.protocol.ConnectionResponse;
+import org.apache.nifi.cluster.protocol.Heartbeat;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.cluster.protocol.ProtocolContext;
+import org.apache.nifi.cluster.protocol.ProtocolException;
+import org.apache.nifi.cluster.protocol.ProtocolHandler;
+import org.apache.nifi.cluster.protocol.StandardDataFlow;
+import org.apache.nifi.cluster.protocol.UnknownServiceAddressException;
+import org.apache.nifi.cluster.protocol.jaxb.JaxbProtocolContext;
+import org.apache.nifi.cluster.protocol.jaxb.message.JaxbProtocolUtils;
+import org.apache.nifi.cluster.protocol.message.ConnectionRequestMessage;
+import org.apache.nifi.cluster.protocol.message.ConnectionResponseMessage;
+import org.apache.nifi.cluster.protocol.message.ControllerStartupFailureMessage;
+import org.apache.nifi.cluster.protocol.message.HeartbeatMessage;
+import org.apache.nifi.cluster.protocol.message.PingMessage;
+import org.apache.nifi.cluster.protocol.message.ProtocolMessage;
+import org.apache.nifi.io.socket.ServerSocketConfiguration;
+import org.apache.nifi.io.socket.SocketConfiguration;
+import org.apache.nifi.io.socket.multicast.DiscoverableService;
+import org.apache.nifi.io.socket.multicast.DiscoverableServiceImpl;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+/**
+ * @author unattributed
+ */
+@Ignore("Randomly tests... probably timing-specific")
+public class NodeProtocolSenderImplTest {
+
+ private SocketProtocolListener listener;
+
+ private NodeProtocolSenderImpl sender;
+
+ private DiscoverableService service;
+
+ private ServerSocketConfiguration serverSocketConfiguration;
+
+ private ClusterServiceLocator mockServiceLocator;
+
+ private ProtocolHandler mockHandler;
+
+ private NodeIdentifier nodeIdentifier;
+
+ @Before
+ public void setup() throws IOException {
+
+ serverSocketConfiguration = new ServerSocketConfiguration();
+
+ mockServiceLocator = mock(ClusterServiceLocator.class);
+ mockHandler = mock(ProtocolHandler.class);
+
+ nodeIdentifier = new NodeIdentifier("1", "localhost", 1234, "localhost", 5678);
+
+ ProtocolContext protocolContext = new JaxbProtocolContext(JaxbProtocolUtils.JAXB_CONTEXT);
+
+ listener = new SocketProtocolListener(5, 0, serverSocketConfiguration, protocolContext);
+ listener.setShutdownListenerSeconds(3);
+ listener.addHandler(mockHandler);
+ listener.start();
+
+ service = new DiscoverableServiceImpl("some-service", new InetSocketAddress("localhost", listener.getPort()));
+
+ SocketConfiguration socketConfiguration = new SocketConfiguration();
+ socketConfiguration.setReuseAddress(true);
+ sender = new NodeProtocolSenderImpl(mockServiceLocator, socketConfiguration, protocolContext);
+ }
+
+ @After
+ public void teardown() throws IOException {
+ if(listener.isRunning()) {
+ listener.stop();
+ }
+ }
+
+ @Test
+ public void testConnect() throws Exception {
+
+ when(mockServiceLocator.getService()).thenReturn(service);
+ when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
+ ConnectionResponseMessage mockMessage = new ConnectionResponseMessage();
+ mockMessage.setConnectionResponse(new ConnectionResponse(nodeIdentifier, new StandardDataFlow("flow".getBytes("UTF-8"), new byte[0], new byte[0]), false, null, null, UUID.randomUUID().toString()));
+ when(mockHandler.handle(any(ProtocolMessage.class))).thenReturn(mockMessage);
+
+ ConnectionRequestMessage request = new ConnectionRequestMessage();
+ request.setConnectionRequest(new ConnectionRequest(nodeIdentifier));
+ ConnectionResponseMessage response = sender.requestConnection(request);
+ assertNotNull(response);
+ }
+
+ @Test(expected = UnknownServiceAddressException.class)
+ public void testConnectNoClusterManagerAddress() throws Exception {
+
+ when(mockServiceLocator.getService()).thenReturn(null);
+ when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
+ when(mockHandler.handle(any(ProtocolMessage.class))).thenReturn(new ConnectionResponseMessage());
+
+ ConnectionRequestMessage request = new ConnectionRequestMessage();
+ request.setConnectionRequest(new ConnectionRequest(nodeIdentifier));
+
+ sender.requestConnection(request);
+ fail("failed to throw exception");
+ }
+
+ @Test(expected = ProtocolException.class)
+ public void testConnectBadResponse() throws Exception {
+
+ when(mockServiceLocator.getService()).thenReturn(service);
+ when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
+ when(mockHandler.handle(any(ProtocolMessage.class))).thenReturn(new PingMessage());
+
+ ConnectionRequestMessage request = new ConnectionRequestMessage();
+ request.setConnectionRequest(new ConnectionRequest(nodeIdentifier));
+
+ sender.requestConnection(request);
+ fail("failed to throw exception");
+
+ }
+
+ @Test(expected = ProtocolException.class)
+ public void testConnectDelayedResponse() throws Exception {
+
+ final int time = 250;
+ sender.getSocketConfiguration().setSocketTimeout(time);
+ when(mockServiceLocator.getService()).thenReturn(service);
+ when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
+ when(mockHandler.handle(any(ProtocolMessage.class))).thenAnswer(new Answer<ConnectionResponseMessage>() {
+ @Override
+ public ConnectionResponseMessage answer(InvocationOnMock invocation) throws Throwable {
+ Thread.sleep(time * 3);
+ return new ConnectionResponseMessage();
+ }
+ });
+ ConnectionRequestMessage request = new ConnectionRequestMessage();
+ request.setConnectionRequest(new ConnectionRequest(nodeIdentifier));
+
+ sender.requestConnection(request);
+ fail("failed to throw exception");
+
+ }
+
+ @Test
+ public void testHeartbeat() throws Exception {
+
+ when(mockServiceLocator.getService()).thenReturn(service);
+ when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
+ when(mockHandler.handle(any(ProtocolMessage.class))).thenReturn(null);
+
+ HeartbeatMessage hb = new HeartbeatMessage();
+ hb.setHeartbeat(new Heartbeat(new NodeIdentifier("id", "localhost", 3, "localhost", 4), false, false, new byte[] {1, 2, 3}));
+ sender.heartbeat(hb);
+ }
+
+ @Test
+ public void testNotifyControllerStartupFailure() throws Exception {
+
+ when(mockServiceLocator.getService()).thenReturn(service);
+ when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
+ when(mockHandler.handle(any(ProtocolMessage.class))).thenReturn(null);
+
+ ControllerStartupFailureMessage msg = new ControllerStartupFailureMessage();
+ msg.setNodeId(new NodeIdentifier("some-id", "some-addr", 1, "some-addr", 1));
+ msg.setExceptionMessage("some exception");
+ sender.notifyControllerStartupFailure(msg);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/testutils/DelayedProtocolHandler.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/testutils/DelayedProtocolHandler.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/testutils/DelayedProtocolHandler.java
new file mode 100644
index 0000000..07ee83a
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/testutils/DelayedProtocolHandler.java
@@ -0,0 +1,57 @@
+/*
+ * 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.cluster.protocol.impl.testutils;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.nifi.cluster.protocol.ProtocolException;
+import org.apache.nifi.cluster.protocol.ProtocolHandler;
+import org.apache.nifi.cluster.protocol.message.ProtocolMessage;
+
+/**
+ * @author unattributed
+ */
+public class DelayedProtocolHandler implements ProtocolHandler {
+
+ private int delay = 0;
+ private List<ProtocolMessage> messages = new ArrayList<>();
+
+ public DelayedProtocolHandler(int delay) {
+ this.delay = delay;
+ }
+
+ @Override
+ public ProtocolMessage handle(ProtocolMessage msg) throws ProtocolException {
+ try {
+ messages.add(msg);
+ Thread.sleep(delay);
+ return null;
+ } catch (final InterruptedException ie) {
+ throw new ProtocolException(ie);
+ }
+
+ }
+
+ @Override
+ public boolean canHandle(ProtocolMessage msg) {
+ return true;
+ }
+
+ public List<ProtocolMessage> getMessages() {
+ return messages;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/testutils/ReflexiveProtocolHandler.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/testutils/ReflexiveProtocolHandler.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/testutils/ReflexiveProtocolHandler.java
new file mode 100644
index 0000000..4e3b932
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/testutils/ReflexiveProtocolHandler.java
@@ -0,0 +1,47 @@
+/*
+ * 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.cluster.protocol.impl.testutils;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.nifi.cluster.protocol.ProtocolException;
+import org.apache.nifi.cluster.protocol.ProtocolHandler;
+import org.apache.nifi.cluster.protocol.message.ProtocolMessage;
+
+/**
+ * @author unattributed
+ */
+public class ReflexiveProtocolHandler implements ProtocolHandler {
+
+ private List<ProtocolMessage> messages = new ArrayList<>();
+
+ @Override
+ public ProtocolMessage handle(ProtocolMessage msg) throws ProtocolException {
+ messages.add(msg);
+ return msg;
+ }
+
+ @Override
+ public boolean canHandle(ProtocolMessage msg) {
+ return true;
+ }
+
+ public List<ProtocolMessage> getMessages() {
+ return messages;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/.gitignore
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/.gitignore b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/.gitignore
new file mode 100755
index 0000000..ea8c4bf
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/.gitignore
@@ -0,0 +1 @@
+/target
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/pom.xml b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/pom.xml
new file mode 100644
index 0000000..18e4ba4
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/pom.xml
@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-framework</artifactId>
+ <version>0.0.3-incubating-SNAPSHOT</version>
+ </parent>
+ <artifactId>nifi-framework-cluster-web</artifactId>
+ <packaging>jar</packaging>
+ <description>The clustering software for communicating with the NiFi web api.</description>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-properties</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-web-optimistic-locking</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-administration</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-user-actions</artifactId>
+ </dependency>
+ </dependencies>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContext.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContext.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContext.java
new file mode 100644
index 0000000..44fb25a
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContext.java
@@ -0,0 +1,59 @@
+/*
+ * 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.cluster.context;
+
+import java.io.Serializable;
+import java.util.List;
+import org.apache.nifi.action.Action;
+import org.apache.nifi.web.Revision;
+
+/**
+ * Contains contextual information about clustering that may be serialized
+ * between manager and node when communicating over HTTP.
+ */
+public interface ClusterContext extends Serializable {
+
+ /**
+ * Returns a list of auditable actions. The list is modifiable
+ * and will never be null.
+ * @return a collection of actions
+ */
+ List<Action> getActions();
+
+ Revision getRevision();
+
+ void setRevision(Revision revision);
+
+ /**
+ * @return true if the request was sent by the cluster manager; false otherwise
+ */
+ boolean isRequestSentByClusterManager();
+
+ /**
+ * Sets the flag to indicate if a request was sent by the cluster manager.
+ * @param flag true if the request was sent by the cluster manager; false otherwise
+ */
+ void setRequestSentByClusterManager(boolean flag);
+
+ /**
+ * Gets an id generation seed. This is used to ensure that nodes are able to generate the
+ * same id across the cluster. This is usually handled by the cluster manager creating the
+ * id, however for some actions (snippets, templates, etc) this is not possible.
+ * @return
+ */
+ String getIdGenerationSeed();
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContextImpl.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContextImpl.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContextImpl.java
new file mode 100644
index 0000000..06907d2
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContextImpl.java
@@ -0,0 +1,69 @@
+/*
+ * 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.cluster.context;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import org.apache.nifi.action.Action;
+import org.apache.nifi.web.Revision;
+
+/**
+ * A basic implementation of the context.
+ */
+public class ClusterContextImpl implements ClusterContext, Serializable {
+
+ private final List<Action> actions = new ArrayList<>();
+
+ private Revision revision;
+
+ private boolean requestSentByClusterManager;
+
+ private final String idGenerationSeed = UUID.randomUUID().toString();
+
+ @Override
+ public List<Action> getActions() {
+ return actions;
+ }
+
+ @Override
+ public Revision getRevision() {
+ return revision;
+ }
+
+ @Override
+ public void setRevision(Revision revision) {
+ this.revision = revision;
+ }
+
+ @Override
+ public boolean isRequestSentByClusterManager() {
+ return requestSentByClusterManager;
+ }
+
+ @Override
+ public void setRequestSentByClusterManager(boolean requestSentByClusterManager) {
+ this.requestSentByClusterManager = requestSentByClusterManager;
+ }
+
+ @Override
+ public String getIdGenerationSeed() {
+ return this.idGenerationSeed;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContextThreadLocal.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContextThreadLocal.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContextThreadLocal.java
new file mode 100644
index 0000000..012e7c7
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/cluster/context/ClusterContextThreadLocal.java
@@ -0,0 +1,47 @@
+/*
+ * 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.cluster.context;
+
+/**
+ * Manages a cluster context on a threadlocal.
+ */
+public class ClusterContextThreadLocal {
+
+ private static final ThreadLocal<ClusterContext> contextHolder = new ThreadLocal<>();
+
+ public static void removeContext() {
+ contextHolder.remove();
+ }
+
+ public static ClusterContext createEmptyContext() {
+ return new ClusterContextImpl();
+ }
+
+ public static ClusterContext getContext() {
+ ClusterContext ctx = contextHolder.get();
+ if(ctx == null) {
+ ctx = createEmptyContext();
+ contextHolder.set(ctx);
+ }
+ return ctx;
+ }
+
+ public static void setContext(final ClusterContext context) {
+ contextHolder.set(context);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/web/ClusterAwareOptimisticLockingManager.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/web/ClusterAwareOptimisticLockingManager.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/web/ClusterAwareOptimisticLockingManager.java
new file mode 100644
index 0000000..90b8a37
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-web/src/main/java/org/apache/nifi/web/ClusterAwareOptimisticLockingManager.java
@@ -0,0 +1,96 @@
+/*
+ * 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.web;
+
+import org.apache.nifi.cluster.context.ClusterContext;
+import org.apache.nifi.cluster.context.ClusterContextThreadLocal;
+
+/**
+ * An optimistic locking manager that provides for optimistic locking in a clustered
+ * environment.
+ *
+ * @author unattributed
+ */
+public class ClusterAwareOptimisticLockingManager implements OptimisticLockingManager {
+
+ private final OptimisticLockingManager optimisticLockingManager;
+
+ public ClusterAwareOptimisticLockingManager(final OptimisticLockingManager optimisticLockingManager) {
+ this.optimisticLockingManager = optimisticLockingManager;
+ }
+
+ @Override
+ public Revision checkRevision(Revision revision) throws InvalidRevisionException {
+ final Revision currentRevision = getRevision();
+ if(currentRevision.equals(revision) == false) {
+ throw new InvalidRevisionException(String.format("Given revision %s does not match current revision %s.", revision, currentRevision));
+ } else {
+ return revision.increment(revision.getClientId());
+ }
+ }
+
+ @Override
+ public boolean isCurrent(Revision revision) {
+ return getRevision().equals(revision);
+ }
+
+ @Override
+ public Revision getRevision() {
+ final ClusterContext ctx = ClusterContextThreadLocal.getContext();
+ if(ctx == null || ctx.getRevision() == null) {
+ return optimisticLockingManager.getRevision();
+ } else {
+ return ctx.getRevision();
+ }
+ }
+
+ @Override
+ public void setRevision(final Revision revision) {
+ final ClusterContext ctx = ClusterContextThreadLocal.getContext();
+ if(ctx != null) {
+ ctx.setRevision(revision);
+ }
+ optimisticLockingManager.setRevision(revision);
+ }
+
+ @Override
+ public Revision incrementRevision() {
+ final Revision currentRevision = getRevision();
+ final Revision incRevision = currentRevision.increment();
+ setRevision(incRevision);
+ return incRevision;
+ }
+
+ @Override
+ public Revision incrementRevision(final String clientId) {
+ final Revision currentRevision = getRevision();
+ final Revision incRevision = currentRevision.increment(clientId);
+ setRevision(incRevision);
+ return incRevision;
+ }
+
+ @Override
+ public String getLastModifier() {
+ return optimisticLockingManager.getLastModifier();
+ }
+
+ @Override
+ public void setLastModifier(final String lastModifier) {
+ optimisticLockingManager.setLastModifier(lastModifier);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/.gitignore
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/.gitignore b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/.gitignore
new file mode 100755
index 0000000..ea8c4bf
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/.gitignore
@@ -0,0 +1 @@
+/target
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/pom.xml b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/pom.xml
new file mode 100644
index 0000000..ef927a3
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/pom.xml
@@ -0,0 +1,144 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-framework</artifactId>
+ <version>0.0.3-incubating-SNAPSHOT</version>
+ </parent>
+ <artifactId>nifi-framework-cluster</artifactId>
+ <packaging>jar</packaging>
+ <description>The clustering software for NiFi.</description>
+ <dependencies>
+ <!-- application core dependencies -->
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-properties</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-logging-utils</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-client-dto</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-framework-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-framework-core-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-framework-cluster-protocol</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-framework-cluster-web</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-web-utils</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-administration</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-site-to-site</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-site-to-site-client</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-compress</artifactId>
+ </dependency>
+ <!-- third party dependencies -->
+
+ <!-- sun dependencies -->
+ <dependency>
+ <groupId>javax.servlet</groupId>
+ <artifactId>javax.servlet-api</artifactId>
+ </dependency>
+
+ <!-- commons dependencies -->
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>commons-net</groupId>
+ <artifactId>commons-net</artifactId>
+ </dependency>
+
+ <!-- jersey dependencies -->
+ <dependency>
+ <groupId>com.sun.jersey</groupId>
+ <artifactId>jersey-client</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.sun.jersey</groupId>
+ <artifactId>jersey-server</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.sun.jersey</groupId>
+ <artifactId>jersey-json</artifactId>
+ </dependency>
+
+ <!-- spring dependencies -->
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-beans</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-context</artifactId>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <configuration>
+ <excludes>
+ <exclude>src/test/resources/org/apache/nifi/cluster/firewall/impl/empty.txt</exclude>
+ <exclude>src/test/resources/org/apache/nifi/cluster/firewall/impl/ips.txt</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/client/MulticastTestClient.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/client/MulticastTestClient.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/client/MulticastTestClient.java
new file mode 100644
index 0000000..0b70c61
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/client/MulticastTestClient.java
@@ -0,0 +1,151 @@
+/*
+ * 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.cluster.client;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.DatagramPacket;
+import java.net.InetSocketAddress;
+import java.net.MulticastSocket;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Locale;
+import java.util.Timer;
+import java.util.TimerTask;
+
+import org.apache.nifi.cluster.protocol.ProtocolContext;
+import org.apache.nifi.cluster.protocol.ProtocolException;
+import org.apache.nifi.cluster.protocol.ProtocolHandler;
+import org.apache.nifi.cluster.protocol.ProtocolMessageMarshaller;
+import org.apache.nifi.cluster.protocol.impl.MulticastProtocolListener;
+import org.apache.nifi.cluster.protocol.jaxb.JaxbProtocolContext;
+import org.apache.nifi.cluster.protocol.jaxb.message.JaxbProtocolUtils;
+import org.apache.nifi.cluster.protocol.message.PingMessage;
+import org.apache.nifi.cluster.protocol.message.ProtocolMessage;
+import org.apache.nifi.io.socket.multicast.MulticastConfiguration;
+import org.apache.nifi.io.socket.multicast.MulticastUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Simple multicast test client that sends ping messages to a group address.
+ */
+public class MulticastTestClient {
+
+ private static final Logger logger = LoggerFactory.getLogger(MulticastTestClient.class);
+
+ private static final int PING_DELAY_SECONDS = 3;
+
+ public static void main(final String... args) throws IOException {
+
+ String group = System.getProperty("group", "225.0.0.0");
+ if (group == null) {
+ System.out.println("Host system property 'group' was not given.");
+ return;
+ }
+ group = group.trim();
+ if (group.length() == 0) {
+ System.out.println("Host system property 'group' must be non-empty.");
+ return;
+ }
+
+ final String portStr = System.getProperty("port", "2222");
+ final int port;
+ try {
+ port = Integer.parseInt(portStr);
+ } catch (final NumberFormatException nfe) {
+ System.out.println("Port system property 'port' was not a valid port.");
+ return;
+ }
+
+ logger.info(String.format("Pinging every %s seconds using multicast address: %s:%s.", PING_DELAY_SECONDS, group, port));
+ logger.info("Override defaults by using system properties '-Dgroup=<Class D IP>' and '-Dport=<unused port>'.");
+ logger.info("The test client may be stopped by entering a newline at the command line.");
+
+ final InetSocketAddress addr = new InetSocketAddress(group, port);
+ final ProtocolContext<ProtocolMessage> protocolContext = new JaxbProtocolContext<ProtocolMessage>(JaxbProtocolUtils.JAXB_CONTEXT);
+ final MulticastConfiguration multicastConfig = new MulticastConfiguration();
+ multicastConfig.setReuseAddress(true);
+
+ // setup listener
+ final MulticastProtocolListener listener = new MulticastProtocolListener(1, addr, multicastConfig, protocolContext);
+ listener.addHandler(new ProtocolHandler() {
+ @Override
+ public ProtocolMessage handle(ProtocolMessage msg) throws ProtocolException {
+ final PingMessage pingMsg = (PingMessage) msg;
+ final SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss", Locale.US);
+ logger.info("Pinged at: " + sdf.format(pingMsg.getDate()));
+ return null;
+ }
+
+ @Override
+ public boolean canHandle(ProtocolMessage msg) {
+ return true;
+ }
+ });
+
+ // setup socket
+ final MulticastSocket multicastSocket = MulticastUtils.createMulticastSocket(multicastConfig);
+
+ // setup broadcaster
+ final Timer broadcaster = new Timer("Multicast Test Client", /**
+ * is daemon *
+ */
+ true);
+
+ try {
+
+ // start listening
+ listener.start();
+
+ // start broadcasting
+ broadcaster.schedule(new TimerTask() {
+
+ @Override
+ public void run() {
+ try {
+
+ final PingMessage msg = new PingMessage();
+ msg.setDate(new Date());
+
+ // marshal message to output stream
+ final ProtocolMessageMarshaller<ProtocolMessage> marshaller = protocolContext.createMarshaller();
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ marshaller.marshal(msg, baos);
+ final byte[] packetBytes = baos.toByteArray();
+
+ // send message
+ final DatagramPacket packet = new DatagramPacket(packetBytes, packetBytes.length, addr);
+ multicastSocket.send(packet);
+
+ } catch (final Exception ex) {
+ logger.warn("Failed to send message due to: " + ex, ex);
+ }
+ }
+ }, 0, PING_DELAY_SECONDS * 1000);
+
+ // block until any input is received
+ System.in.read();
+
+ } finally {
+ broadcaster.cancel();
+ if (listener.isRunning()) {
+ listener.stop();
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/Event.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/Event.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/Event.java
new file mode 100644
index 0000000..6bc5d6c
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/Event.java
@@ -0,0 +1,122 @@
+/*
+ * 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.cluster.event;
+
+import java.util.Date;
+import org.apache.commons.lang3.StringUtils;
+
+/**
+ * Events describe the occurrence of something noteworthy. They record the
+ * event's source, a timestamp, a description, and a category.
+ *
+ * @author unattributed
+ *
+ * @Immutable
+ */
+public class Event {
+
+ public static enum Category {
+
+ DEBUG,
+ INFO,
+ WARN
+ }
+
+ private final String source;
+
+ private final long timestamp;
+
+ private final Category category;
+
+ private final String message;
+
+ /**
+ * Creates an event with the current time as the timestamp and a category of
+ * "INFO".
+ *
+ * @param source the source
+ * @param message the description
+ */
+ public Event(final String source, final String message) {
+ this(source, message, Category.INFO);
+ }
+
+ /**
+ * Creates an event with the current time as the timestamp.
+ *
+ * @param source the source
+ * @param message the description
+ * @param category the event category
+ */
+ public Event(final String source, final String message, final Category category) {
+ this(source, message, category, new Date().getTime());
+ }
+
+ /**
+ * Creates an event with the a category of "INFO".
+ *
+ * @param source the source
+ * @param message the description
+ * @param timestamp the time of occurrence
+ */
+ public Event(final String source, final String message, final long timestamp) {
+ this(source, message, Category.INFO, timestamp);
+ }
+
+ /**
+ * Creates an event.
+ *
+ * @param source the source
+ * @param message the description
+ * @param category the event category
+ * @param timestamp the time of occurrence
+ */
+ public Event(final String source, final String message, final Category category, final long timestamp) {
+
+ if (StringUtils.isBlank(source)) {
+ throw new IllegalArgumentException("Source may not be empty or null.");
+ } else if (StringUtils.isBlank(message)) {
+ throw new IllegalArgumentException("Event message may not be empty or null.");
+ } else if (category == null) {
+ throw new IllegalArgumentException("Event category may not be null.");
+ } else if (timestamp < 0) {
+ throw new IllegalArgumentException("Timestamp may not be negative: " + timestamp);
+ }
+
+ this.source = source;
+ this.message = message;
+ this.category = category;
+ this.timestamp = timestamp;
+ }
+
+ public Category getCategory() {
+ return category;
+ }
+
+ public String getMessage() {
+ return message;
+ }
+
+ public String getSource() {
+ return source;
+ }
+
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/EventManager.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/EventManager.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/EventManager.java
new file mode 100644
index 0000000..f9dfb00
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/EventManager.java
@@ -0,0 +1,65 @@
+/*
+ * 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.cluster.event;
+
+import java.util.List;
+
+/**
+ * Manages an ordered list of events. The event history size dictates the total
+ * number of events to manage for a given source at a given time. When the size
+ * is exceeded, the oldest event for that source is evicted.
+ *
+ * @author unattributed
+ */
+public interface EventManager {
+
+ /**
+ * Adds an event to the manager.
+ *
+ * @param event an Event
+ */
+ void addEvent(Event event);
+
+ /**
+ * Returns a list of events for a given source sorted by the event's
+ * timestamp where the most recent event is first in the list.
+ *
+ * @param eventSource the source
+ *
+ * @return the list of events
+ */
+ List<Event> getEvents(String eventSource);
+
+ /*
+ * Returns the most recent event for the source. If no events exist, then
+ * null is returned.
+ */
+ Event getMostRecentEvent(String eventSource);
+
+ /*
+ * Clears all events for the given source.
+ */
+ void clearEventHistory(String eventSource);
+
+ /**
+ * Returns the history size.
+ *
+ * @return the history size
+ */
+ int getEventHistorySize();
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/impl/EventManagerImpl.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/impl/EventManagerImpl.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/impl/EventManagerImpl.java
new file mode 100644
index 0000000..7fadc78
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/event/impl/EventManagerImpl.java
@@ -0,0 +1,143 @@
+/*
+ * 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.cluster.event.impl;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import org.apache.nifi.cluster.event.Event;
+import org.apache.nifi.cluster.event.EventManager;
+
+/**
+ * Implements the EventManager.
+ *
+ * @author unattributed
+ */
+public class EventManagerImpl implements EventManager {
+
+ /**
+ * associates the source ID with an ordered queue of events, ordered by most
+ * recent event
+ */
+ private final Map<String, Queue<Event>> eventsMap = new HashMap<>();
+
+ /**
+ * the number of events to maintain for a given source
+ */
+ private final int eventHistorySize;
+
+ /**
+ * Creates an instance.
+ *
+ * @param eventHistorySize the number of events to manage for a given
+ * source. Value must be positive.
+ */
+ public EventManagerImpl(final int eventHistorySize) {
+ if (eventHistorySize <= 0) {
+ throw new IllegalArgumentException("Event history size must be positive: " + eventHistorySize);
+ }
+ this.eventHistorySize = eventHistorySize;
+ }
+
+ @Override
+ public void addEvent(final Event event) {
+
+ if (event == null) {
+ throw new IllegalArgumentException("Event may not be null.");
+ }
+
+ Queue<Event> events = eventsMap.get(event.getSource());
+ if (events == null) {
+ // no events from this source, so add a new queue to the map
+ events = new PriorityQueue<>(eventHistorySize, createEventComparator());
+ eventsMap.put(event.getSource(), events);
+ }
+
+ // add event
+ events.add(event);
+
+ // if we exceeded the history size, then evict the oldest event
+ if (events.size() > eventHistorySize) {
+ removeOldestEvent(events);
+ }
+
+ }
+
+ @Override
+ public List<Event> getEvents(final String eventSource) {
+ final Queue<Event> events = eventsMap.get(eventSource);
+ if (events == null) {
+ return Collections.EMPTY_LIST;
+ } else {
+ return Collections.unmodifiableList(new ArrayList<>(events));
+ }
+ }
+
+ @Override
+ public int getEventHistorySize() {
+ return eventHistorySize;
+ }
+
+ @Override
+ public Event getMostRecentEvent(final String eventSource) {
+ final Queue<Event> events = eventsMap.get(eventSource);
+ if (events == null) {
+ return null;
+ } else {
+ return events.peek();
+ }
+ }
+
+ @Override
+ public void clearEventHistory(final String eventSource) {
+ eventsMap.remove(eventSource);
+ }
+
+ private Comparator createEventComparator() {
+ return new Comparator<Event>() {
+ @Override
+ public int compare(final Event o1, final Event o2) {
+ // orders events by most recent first
+ return (int) (o2.getTimestamp() - o1.getTimestamp());
+ }
+ };
+ }
+
+ private void removeOldestEvent(final Collection<Event> events) {
+
+ if (events.isEmpty()) {
+ return;
+ }
+
+ Event oldestEvent = null;
+ for (final Event event : events) {
+ if (oldestEvent == null || oldestEvent.getTimestamp() > event.getTimestamp()) {
+ oldestEvent = event;
+ }
+ }
+
+ events.remove(oldestEvent);
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/firewall/ClusterNodeFirewall.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/firewall/ClusterNodeFirewall.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/firewall/ClusterNodeFirewall.java
new file mode 100644
index 0000000..2e3d278
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/firewall/ClusterNodeFirewall.java
@@ -0,0 +1,35 @@
+/*
+ * 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.cluster.firewall;
+
+/**
+ * Defines the interface for restricting external client connections to a set of
+ * hosts or IPs.
+ */
+public interface ClusterNodeFirewall {
+
+ /**
+ * Returns true if the given host or IP is permissible through the firewall;
+ * false otherwise.
+ *
+ * If an IP is given, then it must be formatted in dotted decimal notation.
+ * @param hostOrIp
+ * @return
+ */
+ boolean isPermissible(String hostOrIp);
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/firewall/impl/FileBasedClusterNodeFirewall.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/firewall/impl/FileBasedClusterNodeFirewall.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/firewall/impl/FileBasedClusterNodeFirewall.java
new file mode 100644
index 0000000..916ec14
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/firewall/impl/FileBasedClusterNodeFirewall.java
@@ -0,0 +1,207 @@
+/*
+ * 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.cluster.firewall.impl;
+
+import java.io.*;
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.*;
+import org.apache.commons.net.util.SubnetUtils;
+import org.apache.nifi.cluster.firewall.ClusterNodeFirewall;
+import org.apache.nifi.util.file.FileUtils;
+import org.apache.nifi.logging.NiFiLog;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A file-based implementation of the ClusterFirewall interface. The class is
+ * configured with a file. If the file is empty, then everything is permissible.
+ * Otherwise, the file should contain hostnames or IPs formatted as dotted
+ * decimals with an optional CIDR suffix. Each entry must be separated by a
+ * newline. An example configuration is given below:
+ *
+ * <code>
+ * # hash character is a comment delimiter
+ * 1.2.3.4 # exact IP
+ * some.host.name # a host name
+ * 4.5.6.7/8 # range of CIDR IPs
+ * 9.10.11.12/13 # a smaller range of CIDR IPs
+ * </code>
+ *
+ * This class allows for synchronization with an optionally configured restore
+ * directory. If configured, then at startup, if the either the config file or
+ * the restore directory's copy is missing, then the configuration file will be
+ * copied to the appropriate location. If both restore directory contains a copy
+ * that is different in content to configuration file, then an exception is
+ * thrown at construction time.
+ */
+public class FileBasedClusterNodeFirewall implements ClusterNodeFirewall {
+
+ private final File config;
+
+ private final File restoreDirectory;
+
+ private final Collection<SubnetUtils.SubnetInfo> subnetInfos = new ArrayList<>();
+
+ private static final Logger logger = new NiFiLog(LoggerFactory.getLogger(FileBasedClusterNodeFirewall.class));
+
+ public FileBasedClusterNodeFirewall(final File config) throws IOException {
+ this(config, null);
+ }
+
+ public FileBasedClusterNodeFirewall(final File config, final File restoreDirectory) throws IOException {
+
+ if (config == null) {
+ throw new IllegalArgumentException("Firewall configuration file may not be null.");
+ }
+
+ this.config = config;
+ this.restoreDirectory = restoreDirectory;
+
+ if (restoreDirectory != null) {
+ // synchronize with restore directory
+ try {
+ syncWithRestoreDirectory();
+ } catch (final IOException ioe) {
+ throw new RuntimeException(ioe);
+ }
+ }
+
+ if (!config.exists() && !config.createNewFile()) {
+ throw new IOException("Firewall configuration file did not exist and could not be created: " + config.getAbsolutePath());
+ }
+
+ logger.info("Loading cluster firewall configuration.");
+ parseConfig(config);
+ logger.info("Cluster firewall configuration loaded.");
+ }
+
+ @Override
+ public boolean isPermissible(final String hostOrIp) {
+ try {
+
+ // if no rules, then permit everything
+ if (subnetInfos.isEmpty()) {
+ return true;
+ }
+
+ final String ip;
+ try {
+ ip = InetAddress.getByName(hostOrIp).getHostAddress();
+ } catch (final UnknownHostException uhe) {
+ logger.warn("Blocking unknown host: " + hostOrIp, uhe);
+ return false;
+ }
+
+ // check each subnet to see if IP is in range
+ for (final SubnetUtils.SubnetInfo subnetInfo : subnetInfos) {
+ if (subnetInfo.isInRange(ip)) {
+ return true;
+ }
+ }
+
+ // no match
+ return false;
+
+ } catch (final IllegalArgumentException iae) {
+ return false;
+ }
+ }
+
+ private void syncWithRestoreDirectory() throws IOException {
+
+ // sanity check that restore directory is a directory, creating it if necessary
+ FileUtils.ensureDirectoryExistAndCanAccess(restoreDirectory);
+
+ // check that restore directory is not the same as the primary directory
+ if (config.getParentFile().getAbsolutePath().equals(restoreDirectory.getAbsolutePath())) {
+ throw new IllegalStateException(
+ String.format("Cluster firewall configuration file '%s' cannot be in the restore directory '%s' ",
+ config.getAbsolutePath(), restoreDirectory.getAbsolutePath()));
+ }
+
+ // the restore copy will have same file name, but reside in a different directory
+ final File restoreFile = new File(restoreDirectory, config.getName());
+
+ // sync the primary copy with the restore copy
+ FileUtils.syncWithRestore(config, restoreFile, logger);
+
+ }
+
+ private void parseConfig(final File config) throws IOException {
+
+ // clear old information
+ subnetInfos.clear();
+ try (BufferedReader br = new BufferedReader(new FileReader(config))) {
+
+ String ipOrHostLine;
+ String ipCidr;
+ int totalIpsAdded = 0;
+ while ((ipOrHostLine = br.readLine()) != null) {
+
+ // cleanup whitespace
+ ipOrHostLine = ipOrHostLine.trim();
+
+ if (ipOrHostLine.isEmpty() || ipOrHostLine.startsWith("#")) {
+ // skip empty lines or comments
+ continue;
+ } else if (ipOrHostLine.contains("#")) {
+ // parse out comments in IP containing lines
+ ipOrHostLine = ipOrHostLine.substring(0, ipOrHostLine.indexOf("#")).trim();
+ }
+
+ // if given a complete IP, then covert to CIDR
+ if (ipOrHostLine.contains("/")) {
+ ipCidr = ipOrHostLine;
+ } else if (ipOrHostLine.contains("\\")) {
+ logger.warn("CIDR IP notation uses forward slashes '/'. Replacing backslash '\\' with forward slash'/' for '" + ipOrHostLine + "'");
+ ipCidr = ipOrHostLine.replace("\\", "/");
+ } else {
+ try {
+ ipCidr = InetAddress.getByName(ipOrHostLine).getHostAddress();
+ if (!ipOrHostLine.equals(ipCidr)) {
+ logger.debug(String.format("Resolved host '%s' to ip '%s'", ipOrHostLine, ipCidr));
+ }
+ ipCidr += "/32";
+ logger.debug("Adding CIDR to exact IP: " + ipCidr);
+ } catch (final UnknownHostException uhe) {
+ logger.warn("Firewall is skipping unknown host address: " + ipOrHostLine);
+ continue;
+ }
+ }
+
+ try {
+ logger.debug("Adding CIDR IP to firewall: " + ipCidr);
+ final SubnetUtils subnetUtils = new SubnetUtils(ipCidr);
+ subnetUtils.setInclusiveHostCount(true);
+ subnetInfos.add(subnetUtils.getInfo());
+ totalIpsAdded++;
+ } catch (final IllegalArgumentException iae) {
+ logger.warn("Firewall is skipping invalid CIDR address: " + ipOrHostLine);
+ }
+
+ }
+
+ if (totalIpsAdded == 0) {
+ logger.info("No IPs added to firewall. Firewall will accept all requests.");
+ } else {
+ logger.info(String.format("Added %d IP(s) to firewall. Only requests originating from the configured IPs will be accepted.", totalIpsAdded));
+ }
+
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/ClusterDataFlow.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/ClusterDataFlow.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/ClusterDataFlow.java
new file mode 100644
index 0000000..eedb88f
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/ClusterDataFlow.java
@@ -0,0 +1,45 @@
+/*
+ * 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.cluster.flow;
+
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+import org.apache.nifi.cluster.protocol.StandardDataFlow;
+
+/**
+ * A dataflow with additional information about the cluster.
+ *
+ * @author unattributed
+ */
+public class ClusterDataFlow {
+
+ private final StandardDataFlow dataFlow;
+
+ private final NodeIdentifier primaryNodeId;
+
+ public ClusterDataFlow(final StandardDataFlow dataFlow, final NodeIdentifier primaryNodeId) {
+ this.dataFlow = dataFlow;
+ this.primaryNodeId = primaryNodeId;
+ }
+
+ public NodeIdentifier getPrimaryNodeId() {
+ return primaryNodeId;
+ }
+
+ public StandardDataFlow getDataFlow() {
+ return dataFlow;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DaoException.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DaoException.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DaoException.java
new file mode 100644
index 0000000..6ff15a7
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DaoException.java
@@ -0,0 +1,40 @@
+/*
+ * 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.cluster.flow;
+
+/**
+ * A base exception for data access exceptions.
+ *
+ * @author unattributed
+ */
+public class DaoException extends RuntimeException {
+
+ public DaoException() {
+ }
+
+ public DaoException(String msg) {
+ super(msg);
+ }
+
+ public DaoException(Throwable cause) {
+ super(cause);
+ }
+
+ public DaoException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DataFlowDao.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DataFlowDao.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DataFlowDao.java
new file mode 100644
index 0000000..a273704
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DataFlowDao.java
@@ -0,0 +1,62 @@
+/*
+ * 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.cluster.flow;
+
+/**
+ * A data access object for loading and saving the flow managed by the cluster.
+ *
+ * @author unattributed
+ */
+public interface DataFlowDao {
+
+ /**
+ * Loads the cluster's dataflow.
+ *
+ * @return the dataflow or null if no dataflow exists
+ *
+ * @throws DaoException if the dataflow was unable to be loaded
+ */
+ ClusterDataFlow loadDataFlow() throws DaoException;
+
+ /**
+ * Saves the cluster's dataflow.
+ *
+ *
+ * @param dataFlow
+ * @throws DaoException if the dataflow was unable to be saved
+ */
+ void saveDataFlow(ClusterDataFlow dataFlow) throws DaoException;
+
+ /**
+ * Sets the state of the dataflow. If the dataflow does not exist, then an
+ * exception is thrown.
+ *
+ * @param flowState the state of the dataflow
+ *
+ * @throws DaoException if the state was unable to be updated
+ */
+ void setPersistedFlowState(PersistedFlowState flowState) throws DaoException;
+
+ /**
+ * Gets the state of the dataflow.
+ *
+ * @return the state of the dataflow
+ *
+ * @throws DaoException if the state was unable to be retrieved
+ */
+ PersistedFlowState getPersistedFlowState() throws DaoException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/761e64a4/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DataFlowManagementService.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DataFlowManagementService.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DataFlowManagementService.java
new file mode 100644
index 0000000..339d904
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/flow/DataFlowManagementService.java
@@ -0,0 +1,115 @@
+/*
+ * 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.cluster.flow;
+
+import java.util.Set;
+import org.apache.nifi.cluster.protocol.NodeIdentifier;
+
+/**
+ * A service for managing the cluster's flow. The service will attempt to keep
+ * the cluster's dataflow current while respecting the value of the configured
+ * retrieval delay.
+ *
+ * The eligible retrieval time is reset with the configured delay every time the
+ * flow state is set to STALE. If the state is set to UNKNOWN or CURRENT, then
+ * the flow will not be retrieved.
+ *
+ * Clients must call start() and stop() to initialize and stop the instance.
+ *
+ * @author unattributed
+ */
+public interface DataFlowManagementService {
+
+ /**
+ * Starts the instance. Start may only be called if the instance is not
+ * running.
+ */
+ void start();
+
+ /**
+ * Stops the instance. Stop may only be called if the instance is running.
+ */
+ void stop();
+
+ /**
+ * @return true if the instance is started; false otherwise.
+ */
+ boolean isRunning();
+
+ /**
+ * Loads the dataflow.
+ *
+ * @return the dataflow or null if no dataflow exists
+ */
+ ClusterDataFlow loadDataFlow();
+
+ /**
+ * Updates the dataflow with the given primary node identifier.
+ *
+ * @param nodeId the node identifier
+ *
+ * @throws DaoException if the update failed
+ */
+ void updatePrimaryNode(NodeIdentifier nodeId) throws DaoException;
+
+ /**
+ * Sets the state of the flow.
+ *
+ * @param flowState the state
+ *
+ * @see PersistedFlowState
+ */
+ void setPersistedFlowState(PersistedFlowState flowState);
+
+ /**
+ * @return the state of the flow
+ */
+ PersistedFlowState getPersistedFlowState();
+
+ /**
+ * @return true if the flow is current; false otherwise.
+ */
+ boolean isFlowCurrent();
+
+ /**
+ * Sets the node identifiers to use when attempting to retrieve the flow.
+ *
+ * @param nodeIds the node identifiers
+ */
+ void setNodeIds(Set<NodeIdentifier> nodeIds);
+
+ /**
+ * Returns the set of node identifiers the service is using to retrieve the
+ * flow.
+ *
+ * @return the set of node identifiers the service is using to retrieve the
+ * flow.
+ */
+ Set<NodeIdentifier> getNodeIds();
+
+ /**
+ * @return the retrieval delay in seconds
+ */
+ int getRetrievalDelaySeconds();
+
+ /**
+ * Sets the retrieval delay.
+ *
+ * @param delay the retrieval delay in seconds
+ */
+ void setRetrievalDelay(String delay);
+}