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/05/02 22:23:53 UTC
[1/2] incubator-nifi git commit: NIFI-557 fixed the correct test and
removed extraneous/duplicative ones - build now works
Repository: incubator-nifi
Updated Branches:
refs/heads/develop 997ed9465 -> 4f35637a2
NIFI-557 fixed the correct test and removed extraneous/duplicative ones - build now works
Signed-off-by: joewitt <jo...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/4f35637a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/4f35637a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/4f35637a
Branch: refs/heads/develop
Commit: 4f35637a297640e6274771e23c1960f748196350
Parents: 7853e3c
Author: joewitt <jo...@apache.org>
Authored: Fri May 1 16:17:47 2015 -0400
Committer: joewitt <jo...@apache.org>
Committed: Sat May 2 16:03:28 2015 -0400
----------------------------------------------------------------------
.../protocol/impl/ClusterManagerProtocolSenderImplTest.java | 9 +++------
.../cluster/protocol/impl/SocketProtocolListenerTest.java | 4 ++--
2 files changed, 5 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4f35637a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterManagerProtocolSenderImplTest.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/ClusterManagerProtocolSenderImplTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterManagerProtocolSenderImplTest.java
index ab10b01..5bfb5ed 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterManagerProtocolSenderImplTest.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterManagerProtocolSenderImplTest.java
@@ -41,9 +41,6 @@ import static org.mockito.Mockito.when;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
-/**
- * @author unattributed
- */
public class ClusterManagerProtocolSenderImplTest {
private InetAddress address;
@@ -90,7 +87,7 @@ public class ClusterManagerProtocolSenderImplTest {
when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
when(mockHandler.handle(any(ProtocolMessage.class))).thenReturn(new FlowResponseMessage());
FlowRequestMessage request = new FlowRequestMessage();
- request.setNodeId(new NodeIdentifier("id", "api-address", 1, address.getHostAddress(), port));
+ request.setNodeId(new NodeIdentifier("id", "api-address", 1, "localhost", port));
FlowResponseMessage response = sender.requestFlow(request);
assertNotNull(response);
}
@@ -101,7 +98,7 @@ public class ClusterManagerProtocolSenderImplTest {
when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
when(mockHandler.handle(any(ProtocolMessage.class))).thenReturn(new PingMessage());
FlowRequestMessage request = new FlowRequestMessage();
- request.setNodeId(new NodeIdentifier("id", "api-address", 1, address.getHostAddress(), port));
+ request.setNodeId(new NodeIdentifier("id", "api-address", 1, "localhost", port));
try {
sender.requestFlow(request);
fail("failed to throw exception");
@@ -125,7 +122,7 @@ public class ClusterManagerProtocolSenderImplTest {
}
});
FlowRequestMessage request = new FlowRequestMessage();
- request.setNodeId(new NodeIdentifier("id", "api-address", 1, address.getHostAddress(), port));
+ request.setNodeId(new NodeIdentifier("id", "api-address", 1, "localhost", port));
try {
sender.requestFlow(request);
fail("failed to throw exception");
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/4f35637a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListenerTest.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/SocketProtocolListenerTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListenerTest.java
index 7a91c29..bc10c20 100644
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListenerTest.java
+++ b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListenerTest.java
@@ -27,8 +27,8 @@ 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.cluster.protocol.testutils.DelayedProtocolHandler;
-import org.apache.nifi.cluster.protocol.testutils.ReflexiveProtocolHandler;
+import org.apache.nifi.cluster.protocol.impl.testutils.DelayedProtocolHandler;
+import org.apache.nifi.cluster.protocol.impl.testutils.ReflexiveProtocolHandler;
import org.apache.nifi.io.socket.ServerSocketConfiguration;
import org.apache.nifi.io.socket.SocketConfiguration;
import org.apache.nifi.io.socket.SocketUtils;
[2/2] incubator-nifi git commit: NIFI-557 fixed the correct test and
removed extraneous/duplicative ones - build now works
Posted by jo...@apache.org.
NIFI-557 fixed the correct test and removed extraneous/duplicative ones - build now works
Signed-off-by: joewitt <jo...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/7853e3c1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/7853e3c1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/7853e3c1
Branch: refs/heads/develop
Commit: 7853e3c165bbbf0d4c3bcb116bd9d9512f5e3578
Parents: 997ed94
Author: joewitt <jo...@apache.org>
Authored: Fri May 1 16:17:25 2015 -0400
Committer: joewitt <jo...@apache.org>
Committed: Sat May 2 16:03:28 2015 -0400
----------------------------------------------------------------------
.../impl/SocketProtocolListenerTest.java | 133 ++++++++++++
.../ClusterManagerProtocolSenderImplTest.java | 144 -------------
.../impl/ClusterServiceLocatorTest.java | 121 -----------
.../impl/ClusterServicesBroadcasterTest.java | 132 ------------
.../impl/MulticastProtocolListenerTest.java | 171 ----------------
.../impl/NodeProtocolSenderImplTest.java | 202 -------------------
.../impl/SocketProtocolListenerTest.java | 133 ------------
.../testutils/DelayedProtocolHandler.java | 57 ------
.../testutils/ReflexiveProtocolHandler.java | 47 -----
9 files changed, 133 insertions(+), 1007 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7853e3c1/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListenerTest.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/SocketProtocolListenerTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/test/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListenerTest.java
new file mode 100644
index 0000000..7a91c29
--- /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/SocketProtocolListenerTest.java
@@ -0,0 +1,133 @@
+/*
+ * 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.IOException;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import org.apache.nifi.cluster.protocol.ProtocolContext;
+import org.apache.nifi.cluster.protocol.ProtocolMessageMarshaller;
+import org.apache.nifi.cluster.protocol.ProtocolMessageUnmarshaller;
+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.cluster.protocol.testutils.DelayedProtocolHandler;
+import org.apache.nifi.cluster.protocol.testutils.ReflexiveProtocolHandler;
+import org.apache.nifi.io.socket.ServerSocketConfiguration;
+import org.apache.nifi.io.socket.SocketConfiguration;
+import org.apache.nifi.io.socket.SocketUtils;
+import org.junit.After;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+import org.junit.Before;
+import org.junit.Test;
+
+/**
+ * @author unattributed
+ */
+public class SocketProtocolListenerTest {
+
+ private SocketProtocolListener listener;
+
+ private Socket socket;
+
+ private ProtocolMessageMarshaller<ProtocolMessage> marshaller;
+
+ private ProtocolMessageUnmarshaller<ProtocolMessage> unmarshaller;
+
+ @Before
+ public void setup() throws Exception {
+
+ final ProtocolContext protocolContext = new JaxbProtocolContext(JaxbProtocolUtils.JAXB_CONTEXT);
+ marshaller = protocolContext.createMarshaller();
+ unmarshaller = protocolContext.createUnmarshaller();
+
+ ServerSocketConfiguration configuration = new ServerSocketConfiguration();
+ configuration.setSocketTimeout(1000);
+
+ listener = new SocketProtocolListener(5, 0, configuration, protocolContext);
+ listener.start();
+
+ int port = listener.getPort();
+
+ SocketConfiguration config = new SocketConfiguration();
+ config.setReuseAddress(true);
+ config.setSocketTimeout(1000);
+ socket = SocketUtils.createSocket(new InetSocketAddress("localhost", port), config);
+ }
+
+ @After
+ public void teardown() throws IOException {
+ try {
+ if (listener.isRunning()) {
+ listener.stop();
+ }
+ } finally {
+ SocketUtils.closeQuietly(socket);
+ }
+ }
+
+ @Test
+ public void testBadRequest() throws Exception {
+ DelayedProtocolHandler handler = new DelayedProtocolHandler(0);
+ listener.addHandler(handler);
+ socket.getOutputStream().write(5);
+ Thread.sleep(250);
+ assertEquals(0, handler.getMessages().size());
+ }
+
+ @Test
+ public void testRequest() throws Exception {
+ ProtocolMessage msg = new PingMessage();
+
+ ReflexiveProtocolHandler handler = new ReflexiveProtocolHandler();
+ listener.addHandler(handler);
+
+ // marshal message to output stream
+ marshaller.marshal(msg, socket.getOutputStream());
+
+ // unmarshall response and return
+ ProtocolMessage response = unmarshaller.unmarshal(socket.getInputStream());
+ assertEquals(msg.getType(), response.getType());
+
+ assertEquals(1, handler.getMessages().size());
+ assertEquals(msg.getType(), handler.getMessages().get(0).getType());
+ }
+
+ @Test
+ public void testDelayedRequest() throws Exception {
+ ProtocolMessage msg = new PingMessage();
+
+ DelayedProtocolHandler handler = new DelayedProtocolHandler(2000);
+ listener.addHandler(handler);
+
+ // marshal message to output stream
+ marshaller.marshal(msg, socket.getOutputStream());
+
+ try {
+ socket.getInputStream().read();
+ fail("Socket timeout not received.");
+ } catch (SocketTimeoutException ste) {
+ }
+
+ assertEquals(1, handler.getMessages().size());
+ assertEquals(msg.getType(), handler.getMessages().get(0).getType());
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7853e3c1/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterManagerProtocolSenderImplTest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterManagerProtocolSenderImplTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterManagerProtocolSenderImplTest.java
deleted file mode 100644
index 9e4237f..0000000
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterManagerProtocolSenderImplTest.java
+++ /dev/null
@@ -1,144 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.cluster.protocol.impl;
-
-import java.io.IOException;
-import java.net.InetAddress;
-
-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.jaxb.JaxbProtocolContext;
-import org.apache.nifi.cluster.protocol.jaxb.message.JaxbProtocolUtils;
-import org.apache.nifi.cluster.protocol.message.FlowRequestMessage;
-import org.apache.nifi.cluster.protocol.message.FlowResponseMessage;
-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.junit.After;
-
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.fail;
-
-import org.junit.Before;
-import org.junit.Test;
-
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-/**
- * @author unattributed
- */
-public class ClusterManagerProtocolSenderImplTest {
-
- private InetAddress address;
-
- private int port;
-
- private SocketProtocolListener listener;
-
- private ClusterManagerProtocolSenderImpl sender;
-
- private ProtocolHandler mockHandler;
-
- @Before
- public void setup() throws IOException, InterruptedException {
-
- address = InetAddress.getLocalHost();
- final ServerSocketConfiguration serverSocketConfiguration = new ServerSocketConfiguration();
-
- mockHandler = mock(ProtocolHandler.class);
-
- final ProtocolContext protocolContext = new JaxbProtocolContext(JaxbProtocolUtils.JAXB_CONTEXT);
-
- listener = new SocketProtocolListener(5, 0, serverSocketConfiguration, protocolContext);
- listener.addHandler(mockHandler);
- listener.start();
-
- // Need to be sure that we give the listener plenty of time to startup. Otherwise, we get intermittent
- // test failures because the Thread started by listener.start() isn't ready to accept connections
- // before we make them.
- Thread.sleep(1000L);
-
- port = listener.getPort();
-
- final SocketConfiguration socketConfiguration = new SocketConfiguration();
- sender = new ClusterManagerProtocolSenderImpl(socketConfiguration, protocolContext);
- }
-
- @After
- public void teardown() throws IOException {
- if (listener.isRunning()) {
- listener.stop();
- }
- }
-
- @Test
- public void testRequestFlow() throws Exception {
- when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
- when(mockHandler.handle(any(ProtocolMessage.class))).thenReturn(new FlowResponseMessage());
- final FlowRequestMessage request = new FlowRequestMessage();
- request.setNodeId(new NodeIdentifier("id", "api-address", 1, "localhost", port));
- final FlowResponseMessage response = sender.requestFlow(request);
- assertNotNull(response);
- }
-
- @Test
- public void testRequestFlowWithBadResponseMessage() throws Exception {
-
- when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
- when(mockHandler.handle(any(ProtocolMessage.class))).thenReturn(new PingMessage());
- final FlowRequestMessage request = new FlowRequestMessage();
- request.setNodeId(new NodeIdentifier("id", "api-address", 1, "localhost", port));
- try {
- sender.requestFlow(request);
- fail("failed to throw exception");
- } catch (final ProtocolException pe) {
- }
-
- }
-
- @Test
- public void testRequestFlowDelayedResponse() throws Exception {
-
- final int time = 250;
- sender.getSocketConfiguration().setSocketTimeout(time);
-
- when(mockHandler.canHandle(any(ProtocolMessage.class))).thenReturn(Boolean.TRUE);
- when(mockHandler.handle(any(ProtocolMessage.class))).thenAnswer(new Answer<FlowResponseMessage>() {
- @Override
- public FlowResponseMessage answer(final InvocationOnMock invocation) throws Throwable {
- Thread.sleep(time * 3);
- return new FlowResponseMessage();
- }
- });
- final FlowRequestMessage request = new FlowRequestMessage();
- request.setNodeId(new NodeIdentifier("id", "api-address", 1, "localhost", port));
- try {
- sender.requestFlow(request);
- fail("failed to throw exception");
- } catch (final ProtocolException pe) {
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7853e3c1/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterServiceLocatorTest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterServiceLocatorTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterServiceLocatorTest.java
deleted file mode 100644
index ea40150..0000000
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterServiceLocatorTest.java
+++ /dev/null
@@ -1,121 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.cluster.protocol.impl;
-
-import java.net.InetSocketAddress;
-import java.util.concurrent.TimeUnit;
-import org.apache.nifi.io.socket.multicast.DiscoverableService;
-import org.apache.nifi.io.socket.multicast.DiscoverableServiceImpl;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import org.junit.Before;
-import org.junit.Test;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-import org.mockito.stubbing.OngoingStubbing;
-
-public class ClusterServiceLocatorTest {
-
- private ClusterServiceDiscovery mockServiceDiscovery;
-
- private int fixedPort;
-
- private DiscoverableService fixedService;
-
- private ClusterServiceLocator serviceDiscoveryLocator;
-
- private ClusterServiceLocator serviceDiscoveryFixedPortLocator;
-
- private ClusterServiceLocator fixedServiceLocator;
-
- @Before
- public void setup() throws Exception {
-
- fixedPort = 1;
- mockServiceDiscovery = mock(ClusterServiceDiscovery.class);
- fixedService = new DiscoverableServiceImpl("some-service", InetSocketAddress.createUnresolved("some-host", 20));
-
- serviceDiscoveryLocator = new ClusterServiceLocator(mockServiceDiscovery);
- serviceDiscoveryFixedPortLocator = new ClusterServiceLocator(mockServiceDiscovery, fixedPort);
- fixedServiceLocator = new ClusterServiceLocator(fixedService);
-
- }
-
- @Test
- public void getServiceWhenServiceDiscoveryNotStarted() {
- assertNull(serviceDiscoveryLocator.getService());
- }
-
- @Test
- public void getServiceWhenServiceDiscoveryFixedPortNotStarted() {
- assertNull(serviceDiscoveryLocator.getService());
- }
-
- @Test
- public void getServiceWhenFixedServiceNotStarted() {
- assertEquals(fixedService, fixedServiceLocator.getService());
- }
-
- @Test
- public void getServiceNotOnFirstAttempt() {
-
- ClusterServiceLocator.AttemptsConfig config = new ClusterServiceLocator.AttemptsConfig();
- config.setNumAttempts(2);
- config.setTimeBetweenAttempsUnit(TimeUnit.SECONDS);
- config.setTimeBetweenAttempts(1);
-
- serviceDiscoveryLocator.setAttemptsConfig(config);
-
- OngoingStubbing<DiscoverableService> stubbing = null;
- for (int i = 0; i < config.getNumAttempts() - 1; i++) {
- if (stubbing == null) {
- stubbing = when(mockServiceDiscovery.getService()).thenReturn(null);
- } else {
- stubbing.thenReturn(null);
- }
- }
- stubbing.thenReturn(fixedService);
-
- assertEquals(fixedService, serviceDiscoveryLocator.getService());
-
- }
-
- @Test
- public void getServiceNotOnFirstAttemptWithFixedPort() {
-
- ClusterServiceLocator.AttemptsConfig config = new ClusterServiceLocator.AttemptsConfig();
- config.setNumAttempts(2);
- config.setTimeBetweenAttempsUnit(TimeUnit.SECONDS);
- config.setTimeBetweenAttempts(1);
-
- serviceDiscoveryFixedPortLocator.setAttemptsConfig(config);
-
- OngoingStubbing<DiscoverableService> stubbing = null;
- for (int i = 0; i < config.getNumAttempts() - 1; i++) {
- if (stubbing == null) {
- stubbing = when(mockServiceDiscovery.getService()).thenReturn(null);
- } else {
- stubbing.thenReturn(null);
- }
- }
- stubbing.thenReturn(fixedService);
-
- InetSocketAddress resultAddress = InetSocketAddress.createUnresolved(fixedService.getServiceAddress().getHostName(), fixedPort);
- DiscoverableService resultService = new DiscoverableServiceImpl(fixedService.getServiceName(), resultAddress);
- assertEquals(resultService, serviceDiscoveryFixedPortLocator.getService());
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7853e3c1/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterServicesBroadcasterTest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterServicesBroadcasterTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterServicesBroadcasterTest.java
deleted file mode 100644
index 0f834fc..0000000
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/ClusterServicesBroadcasterTest.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.cluster.protocol.impl;
-
-import java.net.InetSocketAddress;
-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.jaxb.JaxbProtocolContext;
-import org.apache.nifi.cluster.protocol.jaxb.message.JaxbProtocolUtils;
-import org.apache.nifi.cluster.protocol.message.ProtocolMessage;
-import org.apache.nifi.cluster.protocol.message.ServiceBroadcastMessage;
-import org.apache.nifi.io.socket.multicast.DiscoverableService;
-import org.apache.nifi.io.socket.multicast.DiscoverableServiceImpl;
-import org.apache.nifi.io.socket.multicast.MulticastConfiguration;
-import org.junit.After;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Test;
-
-/**
- * @author unattributed
- */
-public class ClusterServicesBroadcasterTest {
-
- private ClusterServicesBroadcaster broadcaster;
-
- private MulticastProtocolListener listener;
-
- private DummyProtocolHandler handler;
-
- private InetSocketAddress multicastAddress;
-
- private DiscoverableService broadcastedService;
-
- private ProtocolContext protocolContext;
-
- private MulticastConfiguration configuration;
-
- @Before
- public void setup() throws Exception {
-
- broadcastedService = new DiscoverableServiceImpl("some-service", new InetSocketAddress("localhost", 11111));
-
- multicastAddress = new InetSocketAddress("225.1.1.1", 22222);
-
- configuration = new MulticastConfiguration();
-
- protocolContext = new JaxbProtocolContext(JaxbProtocolUtils.JAXB_CONTEXT);
-
- broadcaster = new ClusterServicesBroadcaster(multicastAddress, configuration, protocolContext, "500 ms");
- broadcaster.addService(broadcastedService);
-
- handler = new DummyProtocolHandler();
- listener = new MulticastProtocolListener(5, multicastAddress, configuration, protocolContext);
- listener.addHandler(handler);
- }
-
- @After
- public void teardown() {
-
- if (broadcaster.isRunning()) {
- broadcaster.stop();
- }
-
- try {
- if (listener.isRunning()) {
- listener.stop();
- }
- } catch (Exception ex) {
- ex.printStackTrace(System.out);
- }
-
- }
-
- @Test
- @Ignore
- public void testBroadcastReceived() throws Exception {
-
- broadcaster.start();
- listener.start();
-
- Thread.sleep(1000);
-
- listener.stop();
-
- assertNotNull(handler.getProtocolMessage());
- assertEquals(ProtocolMessage.MessageType.SERVICE_BROADCAST, handler.getProtocolMessage().getType());
- final ServiceBroadcastMessage msg = (ServiceBroadcastMessage) handler.getProtocolMessage();
- assertEquals(broadcastedService.getServiceName(), msg.getServiceName());
- assertEquals(broadcastedService.getServiceAddress().getHostName(), msg.getAddress());
- assertEquals(broadcastedService.getServiceAddress().getPort(), msg.getPort());
- }
-
- private class DummyProtocolHandler implements ProtocolHandler {
-
- private ProtocolMessage protocolMessage;
-
- @Override
- public boolean canHandle(ProtocolMessage msg) {
- return true;
- }
-
- @Override
- public ProtocolMessage handle(ProtocolMessage msg) throws ProtocolException {
- this.protocolMessage = msg;
- return null;
- }
-
- public ProtocolMessage getProtocolMessage() {
- return protocolMessage;
- }
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7853e3c1/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/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/src/test/java/org/apache/nifi/cluster/protocol/impl/MulticastProtocolListenerTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/MulticastProtocolListenerTest.java
deleted file mode 100644
index f5037a8..0000000
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/MulticastProtocolListenerTest.java
+++ /dev/null
@@ -1,171 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-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.assertEquals;
-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("This test must be reworked. Requires an active 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());
- }
-
- @Test
- @Ignore
- 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/7853e3c1/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/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/src/test/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImplTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImplTest.java
deleted file mode 100644
index a759b86..0000000
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImplTest.java
+++ /dev/null
@@ -1,202 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.cluster.protocol.impl;
-
-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.HeartbeatPayload;
-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.Test;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-/**
- * @author unattributed
- */
-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 msg = new HeartbeatMessage();
- HeartbeatPayload hbPayload = new HeartbeatPayload();
- Heartbeat hb = new Heartbeat(new NodeIdentifier("id", "localhost", 3, "localhost", 4), false, false, hbPayload.marshal());
- msg.setHeartbeat(hb);
- sender.heartbeat(msg);
- }
-
- @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/7853e3c1/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListenerTest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListenerTest.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListenerTest.java
deleted file mode 100644
index 7a91c29..0000000
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/impl/SocketProtocolListenerTest.java
+++ /dev/null
@@ -1,133 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.cluster.protocol.impl;
-
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.net.SocketTimeoutException;
-import org.apache.nifi.cluster.protocol.ProtocolContext;
-import org.apache.nifi.cluster.protocol.ProtocolMessageMarshaller;
-import org.apache.nifi.cluster.protocol.ProtocolMessageUnmarshaller;
-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.cluster.protocol.testutils.DelayedProtocolHandler;
-import org.apache.nifi.cluster.protocol.testutils.ReflexiveProtocolHandler;
-import org.apache.nifi.io.socket.ServerSocketConfiguration;
-import org.apache.nifi.io.socket.SocketConfiguration;
-import org.apache.nifi.io.socket.SocketUtils;
-import org.junit.After;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-import org.junit.Before;
-import org.junit.Test;
-
-/**
- * @author unattributed
- */
-public class SocketProtocolListenerTest {
-
- private SocketProtocolListener listener;
-
- private Socket socket;
-
- private ProtocolMessageMarshaller<ProtocolMessage> marshaller;
-
- private ProtocolMessageUnmarshaller<ProtocolMessage> unmarshaller;
-
- @Before
- public void setup() throws Exception {
-
- final ProtocolContext protocolContext = new JaxbProtocolContext(JaxbProtocolUtils.JAXB_CONTEXT);
- marshaller = protocolContext.createMarshaller();
- unmarshaller = protocolContext.createUnmarshaller();
-
- ServerSocketConfiguration configuration = new ServerSocketConfiguration();
- configuration.setSocketTimeout(1000);
-
- listener = new SocketProtocolListener(5, 0, configuration, protocolContext);
- listener.start();
-
- int port = listener.getPort();
-
- SocketConfiguration config = new SocketConfiguration();
- config.setReuseAddress(true);
- config.setSocketTimeout(1000);
- socket = SocketUtils.createSocket(new InetSocketAddress("localhost", port), config);
- }
-
- @After
- public void teardown() throws IOException {
- try {
- if (listener.isRunning()) {
- listener.stop();
- }
- } finally {
- SocketUtils.closeQuietly(socket);
- }
- }
-
- @Test
- public void testBadRequest() throws Exception {
- DelayedProtocolHandler handler = new DelayedProtocolHandler(0);
- listener.addHandler(handler);
- socket.getOutputStream().write(5);
- Thread.sleep(250);
- assertEquals(0, handler.getMessages().size());
- }
-
- @Test
- public void testRequest() throws Exception {
- ProtocolMessage msg = new PingMessage();
-
- ReflexiveProtocolHandler handler = new ReflexiveProtocolHandler();
- listener.addHandler(handler);
-
- // marshal message to output stream
- marshaller.marshal(msg, socket.getOutputStream());
-
- // unmarshall response and return
- ProtocolMessage response = unmarshaller.unmarshal(socket.getInputStream());
- assertEquals(msg.getType(), response.getType());
-
- assertEquals(1, handler.getMessages().size());
- assertEquals(msg.getType(), handler.getMessages().get(0).getType());
- }
-
- @Test
- public void testDelayedRequest() throws Exception {
- ProtocolMessage msg = new PingMessage();
-
- DelayedProtocolHandler handler = new DelayedProtocolHandler(2000);
- listener.addHandler(handler);
-
- // marshal message to output stream
- marshaller.marshal(msg, socket.getOutputStream());
-
- try {
- socket.getInputStream().read();
- fail("Socket timeout not received.");
- } catch (SocketTimeoutException ste) {
- }
-
- assertEquals(1, handler.getMessages().size());
- assertEquals(msg.getType(), handler.getMessages().get(0).getType());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/7853e3c1/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/testutils/DelayedProtocolHandler.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/testutils/DelayedProtocolHandler.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/testutils/DelayedProtocolHandler.java
deleted file mode 100644
index 2f16777..0000000
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/testutils/DelayedProtocolHandler.java
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.cluster.protocol.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/7853e3c1/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/testutils/ReflexiveProtocolHandler.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/testutils/ReflexiveProtocolHandler.java b/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/testutils/ReflexiveProtocolHandler.java
deleted file mode 100644
index e80f52c..0000000
--- a/nifi/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/test/java/org/apache/nifi/cluster/protocol/testutils/ReflexiveProtocolHandler.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.nifi.cluster.protocol.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;
- }
-
-}