You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by al...@apache.org on 2016/08/17 17:59:17 UTC
[03/14] nifi git commit: NIFI-2574 Changed NiFiProperties to avoid
static initializer and updated all references to it.
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarThreadContextClassLoader.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarThreadContextClassLoader.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarThreadContextClassLoader.java
index 381b54b..827abdd 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarThreadContextClassLoader.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/main/java/org/apache/nifi/nar/NarThreadContextClassLoader.java
@@ -34,11 +34,14 @@ import org.apache.nifi.reporting.ReportingTask;
import java.io.IOException;
import java.io.InputStream;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
import java.net.URL;
import java.net.URLClassLoader;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.List;
+import org.apache.nifi.util.NiFiProperties;
/**
* THREAD SAFE
@@ -165,7 +168,22 @@ public class NarThreadContextClassLoader extends URLClassLoader {
}
}
- public static <T> T createInstance(final String implementationClassName, final Class<T> typeDefinition) throws InstantiationException, IllegalAccessException, ClassNotFoundException {
+ /**
+ * Constructs an instance of the given type using either default no args
+ * constructor or a constructor which takes a NiFiProperties object
+ * (preferred).
+ *
+ * @param <T> type
+ * @param implementationClassName class
+ * @param typeDefinition def
+ * @param nifiProperties props
+ * @return constructed instance
+ * @throws InstantiationException ex
+ * @throws IllegalAccessException ex
+ * @throws ClassNotFoundException ex
+ */
+ public static <T> T createInstance(final String implementationClassName, final Class<T> typeDefinition, final NiFiProperties nifiProperties)
+ throws InstantiationException, IllegalAccessException, ClassNotFoundException {
final ClassLoader originalClassLoader = Thread.currentThread().getContextClassLoader();
Thread.currentThread().setContextClassLoader(NarThreadContextClassLoader.getInstance());
try {
@@ -181,7 +199,16 @@ public class NarThreadContextClassLoader extends URLClassLoader {
Thread.currentThread().setContextClassLoader(detectedClassLoaderForType);
final Class<?> desiredClass = rawClass.asSubclass(typeDefinition);
- return typeDefinition.cast(desiredClass.newInstance());
+ if (nifiProperties == null) {
+ return typeDefinition.cast(desiredClass.newInstance());
+ }
+ Constructor<?> constructor = null;
+ try {
+ constructor = desiredClass.getConstructor(NiFiProperties.class);
+ return typeDefinition.cast(constructor.newInstance(nifiProperties));
+ } catch (final NoSuchMethodException | InvocationTargetException ex) {
+ return typeDefinition.cast(desiredClass.newInstance());
+ }
} finally {
Thread.currentThread().setContextClassLoader(originalClassLoader);
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java
index 882c8c6..0fc8f4d 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-nar-utils/src/test/java/org/apache/nifi/nar/NarUnpackerTest.java
@@ -21,11 +21,8 @@ import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
-import java.io.BufferedInputStream;
import java.io.File;
-import java.io.FileInputStream;
import java.io.IOException;
-import java.io.InputStream;
import java.net.URISyntaxException;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
@@ -37,6 +34,9 @@ import java.util.HashSet;
import java.util.Set;
import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
@@ -81,7 +81,7 @@ public class NarUnpackerTest {
@Test
public void testUnpackNars() {
- NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
+ NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties", Collections.EMPTY_MAP);
assertEquals("./target/NarUnpacker/lib/",
properties.getProperty("nifi.nar.library.directory"));
@@ -112,14 +112,14 @@ public class NarUnpackerTest {
@Test
public void testUnpackNarsFromEmptyDir() throws IOException {
- NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
-
final File emptyDir = new File("./target/empty/dir");
emptyDir.delete();
emptyDir.deleteOnExit();
assertTrue(emptyDir.mkdirs());
- properties.setProperty("nifi.nar.library.directory.alt", emptyDir.toString());
+ final Map<String, String> others = new HashMap<>();
+ others.put("nifi.nar.library.directory.alt", emptyDir.toString());
+ NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties", others);
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
@@ -141,8 +141,9 @@ public class NarUnpackerTest {
nonExistantDir.delete();
nonExistantDir.deleteOnExit();
- NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
- properties.setProperty("nifi.nar.library.directory.alt", nonExistantDir.toString());
+ final Map<String, String> others = new HashMap<>();
+ others.put("nifi.nar.library.directory.alt", nonExistantDir.toString());
+ NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties", others);
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
@@ -165,15 +166,16 @@ public class NarUnpackerTest {
nonDir.createNewFile();
nonDir.deleteOnExit();
- NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
- properties.setProperty("nifi.nar.library.directory.alt", nonDir.toString());
+ final Map<String, String> others = new HashMap<>();
+ others.put("nifi.nar.library.directory.alt", nonDir.toString());
+ NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties", others);
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
assertNull(extensionMapping);
}
- private NiFiProperties loadSpecifiedProperties(String propertiesFile) {
+ private NiFiProperties loadSpecifiedProperties(final String propertiesFile, final Map<String, String> others) {
String filePath;
try {
filePath = NarUnpackerTest.class.getResource(propertiesFile).toURI().getPath();
@@ -181,34 +183,6 @@ public class NarUnpackerTest {
throw new RuntimeException("Cannot load properties file due to "
+ ex.getLocalizedMessage(), ex);
}
- System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, filePath);
-
- NiFiProperties properties = NiFiProperties.getInstance();
-
- // clear out existing properties
- for (String prop : properties.stringPropertyNames()) {
- properties.remove(prop);
- }
-
- InputStream inStream = null;
- try {
- inStream = new BufferedInputStream(new FileInputStream(filePath));
- properties.load(inStream);
- } catch (final Exception ex) {
- throw new RuntimeException("Cannot load properties file due to "
- + ex.getLocalizedMessage(), ex);
- } finally {
- if (null != inStream) {
- try {
- inStream.close();
- } catch (final Exception ex) {
- /**
- * do nothing *
- */
- }
- }
- }
-
- return properties;
+ return NiFiProperties.createBasicNiFiProperties(filePath, others);
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/NiFi.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/NiFi.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/NiFi.java
index a426c40..6d0fa97 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/NiFi.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-runtime/src/main/java/org/apache/nifi/NiFi.java
@@ -224,7 +224,7 @@ public class NiFi {
public static void main(String[] args) {
logger.info("Launching NiFi...");
try {
- new NiFi(NiFiProperties.getInstance());
+ new NiFi(NiFiProperties.createBasicNiFiProperties(null, null));
} catch (final Throwable t) {
logger.error("Failure to launch NiFi due to " + t, t);
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslServerSocketFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslServerSocketFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslServerSocketFactory.java
index 59ea312..458157c 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslServerSocketFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslServerSocketFactory.java
@@ -36,8 +36,8 @@ public class SslServerSocketFactory extends SSLServerSocketFactory {
private SSLServerSocketFactory sslServerSocketFactory;
- public SslServerSocketFactory() {
- final SSLContext sslCtx = SslContextFactory.createSslContext(NiFiProperties.getInstance());
+ public SslServerSocketFactory(final NiFiProperties nifiProperties) {
+ final SSLContext sslCtx = SslContextFactory.createSslContext(nifiProperties);
if (sslCtx == null) {
try {
sslServerSocketFactory = SSLContext.getDefault().getServerSocketFactory();
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslSocketFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslSocketFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslSocketFactory.java
index da0e7fb..fa6de56 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslSocketFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-security/src/main/java/org/apache/nifi/framework/security/util/SslSocketFactory.java
@@ -36,8 +36,8 @@ public class SslSocketFactory extends SSLSocketFactory {
private final SSLSocketFactory sslSocketFactory;
- public SslSocketFactory() {
- final SSLContext sslCtx = SslContextFactory.createSslContext(NiFiProperties.getInstance());
+ public SslSocketFactory(final NiFiProperties nifiProperties) {
+ final SSLContext sslCtx = SslContextFactory.createSslContext(nifiProperties);
if (sslCtx == null) {
try {
sslSocketFactory = SSLContext.getDefault().getSocketFactory();
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/HttpRemoteSiteListener.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/HttpRemoteSiteListener.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/HttpRemoteSiteListener.java
index b335f48..7a001ab 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/HttpRemoteSiteListener.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/HttpRemoteSiteListener.java
@@ -51,7 +51,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
private ProcessGroup rootGroup;
private ScheduledFuture<?> transactionMaintenanceTask;
- private HttpRemoteSiteListener() {
+ private HttpRemoteSiteListener(final NiFiProperties nifiProperties) {
super();
taskExecutor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
private final ThreadFactory defaultFactory = Executors.defaultThreadFactory();
@@ -65,10 +65,9 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
}
});
- NiFiProperties properties = NiFiProperties.getInstance();
int txTtlSec;
try {
- final String snapshotFrequency = properties.getProperty(SITE_TO_SITE_HTTP_TRANSACTION_TTL, DEFAULT_SITE_TO_SITE_HTTP_TRANSACTION_TTL);
+ final String snapshotFrequency = nifiProperties.getProperty(SITE_TO_SITE_HTTP_TRANSACTION_TTL, DEFAULT_SITE_TO_SITE_HTTP_TRANSACTION_TTL);
txTtlSec = (int) FormatUtils.getTimeDuration(snapshotFrequency, TimeUnit.SECONDS);
} catch (final Exception e) {
txTtlSec = (int) FormatUtils.getTimeDuration(DEFAULT_SITE_TO_SITE_HTTP_TRANSACTION_TTL, TimeUnit.SECONDS);
@@ -78,11 +77,11 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
transactionTtlSec = txTtlSec;
}
- public static HttpRemoteSiteListener getInstance() {
+ public static HttpRemoteSiteListener getInstance(final NiFiProperties nifiProperties) {
if (instance == null) {
synchronized (HttpRemoteSiteListener.class) {
if (instance == null) {
- instance = new HttpRemoteSiteListener();
+ instance = new HttpRemoteSiteListener(nifiProperties);
}
}
}
@@ -90,6 +89,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
}
private class TransactionWrapper {
+
private final FlowFileTransaction transaction;
private final HandshakeProperties handshakenProperties;
private long lastCommunicationAt;
@@ -129,7 +129,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
try {
Set<String> transactionIds = transactions.keySet().stream().collect(Collectors.toSet());
transactionIds.stream().filter(tid -> !isTransactionActive(tid))
- .forEach(tid -> cancelTransaction(tid));
+ .forEach(tid -> cancelTransaction(tid));
} catch (Exception e) {
// Swallow exception so that this thread can keep working.
logger.error("An exception occurred while maintaining transactions", e);
@@ -146,7 +146,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
} else {
logger.debug("Cancel a transaction. transactionId={}", transactionId);
FlowFileTransaction t = wrapper.transaction;
- if(t != null && t.getSession() != null){
+ if (t != null && t.getSession() != null) {
logger.info("Cancel a transaction, rollback its session. transactionId={}", transactionId);
try {
t.getSession().rollback();
@@ -158,10 +158,9 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
}
}
-
@Override
public void stop() {
- if(transactionMaintenanceTask != null) {
+ if (transactionMaintenanceTask != null) {
logger.debug("Stopping transactionMaintenanceTask...");
transactionMaintenanceTask.cancel(true);
}
@@ -191,10 +190,9 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
/**
* @param transactionId transactionId to check
- * @return Returns a HandshakeProperties instance which is created when this transaction is started,
- only if the transaction is active,
- and it holds a HandshakeProperties,
- otherwise return null
+ * @return Returns a HandshakeProperties instance which is created when this
+ * transaction is started, only if the transaction is active, and it holds a
+ * HandshakeProperties, otherwise return null
*/
public HandshakeProperties getHandshakenProperties(final String transactionId) {
TransactionWrapper transaction = transactions.get(transactionId);
@@ -205,7 +203,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
}
public void holdTransaction(final String transactionId, final FlowFileTransaction transaction,
- final HandshakeProperties handshakenProperties) throws IllegalStateException {
+ final HandshakeProperties handshakenProperties) throws IllegalStateException {
// We don't check expiration of the transaction here, to support large file transport or slow network.
// The availability of current transaction is already checked when the HTTP request was received at SiteToSiteResource.
TransactionWrapper currentTransaction = transactions.remove(transactionId);
@@ -224,7 +222,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
}
public FlowFileTransaction finalizeTransaction(final String transactionId) throws IllegalStateException {
- if (!isTransactionActive(transactionId)){
+ if (!isTransactionActive(transactionId)) {
throw new IllegalStateException("Transaction was not found or not active anymore. transactionId=" + transactionId);
}
TransactionWrapper transaction = transactions.remove(transactionId);
@@ -239,7 +237,7 @@ public class HttpRemoteSiteListener implements RemoteSiteListener {
}
public void extendTransaction(final String transactionId) throws IllegalStateException {
- if (!isTransactionActive(transactionId)){
+ if (!isTransactionActive(transactionId)) {
throw new IllegalStateException("Transaction was not found or not active anymore. transactionId=" + transactionId);
}
TransactionWrapper transaction = transactions.get(transactionId);
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/RemoteResourceFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/RemoteResourceFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/RemoteResourceFactory.java
index b0ce357..0e29aef 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/RemoteResourceFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/RemoteResourceFactory.java
@@ -122,8 +122,8 @@ public class RemoteResourceFactory extends RemoteResourceInitiator {
}
public static <T extends VersionedRemoteResource> T
- receiveResourceNegotiation(final Class<T> cls, final DataInputStream dis, final DataOutputStream dos, final Class<?>[] constructorArgClasses, final Object[] constructorArgs)
- throws IOException, HandshakeException {
+ receiveResourceNegotiation(final Class<T> cls, final DataInputStream dis, final DataOutputStream dos, final Class<?>[] constructorArgClasses, final Object[] constructorArgs)
+ throws IOException, HandshakeException {
final String resourceClassName = dis.readUTF();
final T resource;
try {
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/SocketRemoteSiteListener.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/SocketRemoteSiteListener.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/SocketRemoteSiteListener.java
index bd9d204..9f9e3aa 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/SocketRemoteSiteListener.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/SocketRemoteSiteListener.java
@@ -46,6 +46,8 @@ import java.util.Arrays;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import org.apache.nifi.remote.cluster.ClusterNodeInformation;
+import org.apache.nifi.util.NiFiProperties;
public class SocketRemoteSiteListener implements RemoteSiteListener {
@@ -55,18 +57,20 @@ public class SocketRemoteSiteListener implements RemoteSiteListener {
private final SSLContext sslContext;
private final NodeInformant nodeInformant;
private final AtomicReference<ProcessGroup> rootGroup = new AtomicReference<>();
+ private final NiFiProperties nifiProperties;
private final AtomicBoolean stopped = new AtomicBoolean(false);
private static final Logger LOG = LoggerFactory.getLogger(SocketRemoteSiteListener.class);
- public SocketRemoteSiteListener(final int socketPort, final SSLContext sslContext) {
- this(socketPort, sslContext, null);
+ public SocketRemoteSiteListener(final int socketPort, final SSLContext sslContext, final NiFiProperties nifiProperties) {
+ this(socketPort, sslContext, nifiProperties, null);
}
- public SocketRemoteSiteListener(final int socketPort, final SSLContext sslContext, final NodeInformant nodeInformant) {
+ public SocketRemoteSiteListener(final int socketPort, final SSLContext sslContext, final NiFiProperties nifiProperties, final NodeInformant nodeInformant) {
this.socketPort = socketPort;
this.sslContext = sslContext;
+ this.nifiProperties = nifiProperties;
this.nodeInformant = nodeInformant;
}
@@ -267,7 +271,14 @@ public class SocketRemoteSiteListener implements RemoteSiteListener {
protocol.getPort().receiveFlowFiles(peer, protocol);
break;
case REQUEST_PEER_LIST:
- protocol.sendPeerList(peer, nodeInformant == null ? Optional.empty() : Optional.of(nodeInformant.getNodeInformation()));
+ final Optional<ClusterNodeInformation> nodeInfo = (nodeInformant == null) ? Optional.empty() : Optional.of(nodeInformant.getNodeInformation());
+ protocol.sendPeerList(
+ peer,
+ nodeInfo,
+ nifiProperties.getRemoteInputHost(),
+ nifiProperties.getRemoteInputPort(),
+ nifiProperties.getRemoteInputHttpPort(),
+ nifiProperties.isSiteToSiteSecure());
break;
case SHUTDOWN:
protocol.shutdown(peer);
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java
index 8f115f7..1996357 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/StandardRemoteGroupPort.java
@@ -74,14 +74,17 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
private final AtomicBoolean targetRunning = new AtomicBoolean(true);
private final SSLContext sslContext;
private final TransferDirection transferDirection;
+ private final NiFiProperties nifiProperties;
private final AtomicReference<SiteToSiteClient> clientRef = new AtomicReference<>();
+
SiteToSiteClient getSiteToSiteClient() {
return clientRef.get();
}
public StandardRemoteGroupPort(final String id, final String name, final ProcessGroup processGroup, final RemoteProcessGroup remoteGroup,
- final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler) {
+ final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler,
+ final NiFiProperties nifiProperties) {
// remote group port id needs to be unique but cannot just be the id of the port
// in the remote group instance. this supports referencing the same remote
// instance more than once.
@@ -90,11 +93,12 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
this.remoteGroup = remoteGroup;
this.transferDirection = direction;
this.sslContext = sslContext;
+ this.nifiProperties = nifiProperties;
setScheduldingPeriod(MINIMUM_SCHEDULING_NANOS + " nanos");
}
- private static File getPeerPersistenceFile(final String portId) {
- final File stateDir = NiFiProperties.getInstance().getPersistentStateDirectory();
+ private static File getPeerPersistenceFile(final String portId, final NiFiProperties nifiProperties) {
+ final File stateDir = nifiProperties.getPersistentStateDirectory();
return new File(stateDir, portId + ".peers");
}
@@ -138,17 +142,17 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
final long penalizationMillis = FormatUtils.getTimeDuration(remoteGroup.getYieldDuration(), TimeUnit.MILLISECONDS);
final SiteToSiteClient client = new SiteToSiteClient.Builder()
- .url(remoteGroup.getTargetUri().toString())
- .portIdentifier(getIdentifier())
- .sslContext(sslContext)
- .useCompression(isUseCompression())
- .eventReporter(remoteGroup.getEventReporter())
- .peerPersistenceFile(getPeerPersistenceFile(getIdentifier()))
- .nodePenalizationPeriod(penalizationMillis, TimeUnit.MILLISECONDS)
- .timeout(remoteGroup.getCommunicationsTimeout(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS)
- .transportProtocol(remoteGroup.getTransportProtocol())
- .httpProxy(new HttpProxy(remoteGroup.getProxyHost(), remoteGroup.getProxyPort(), remoteGroup.getProxyUser(), remoteGroup.getProxyPassword()))
- .build();
+ .url(remoteGroup.getTargetUri().toString())
+ .portIdentifier(getIdentifier())
+ .sslContext(sslContext)
+ .useCompression(isUseCompression())
+ .eventReporter(remoteGroup.getEventReporter())
+ .peerPersistenceFile(getPeerPersistenceFile(getIdentifier(), nifiProperties))
+ .nodePenalizationPeriod(penalizationMillis, TimeUnit.MILLISECONDS)
+ .timeout(remoteGroup.getCommunicationsTimeout(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS)
+ .transportProtocol(remoteGroup.getTransportProtocol())
+ .httpProxy(new HttpProxy(remoteGroup.getProxyHost(), remoteGroup.getProxyPort(), remoteGroup.getProxyUser(), remoteGroup.getProxyPassword()))
+ .build();
clientRef.set(client);
}
@@ -306,7 +310,7 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
final String flowFileDescription = (flowFilesSent.size() < 20) ? flowFilesSent.toString() : flowFilesSent.size() + " FlowFiles";
logger.info("{} Successfully sent {} ({}) to {} in {} milliseconds at a rate of {}", new Object[]{
- this, flowFileDescription, dataSize, transaction.getCommunicant().getUrl(), uploadMillis, uploadDataRate});
+ this, flowFileDescription, dataSize, transaction.getCommunicant().getUrl(), uploadMillis, uploadDataRate});
return flowFilesSent.size();
} catch (final Exception e) {
@@ -364,7 +368,7 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
final long uploadMillis = stopWatch.getDuration(TimeUnit.MILLISECONDS);
final String dataSize = FormatUtils.formatDataSize(bytesReceived);
logger.info("{} Successfully receveied {} ({}) from {} in {} milliseconds at a rate of {}", new Object[]{
- this, flowFileDescription, dataSize, transaction.getCommunicant().getUrl(), uploadMillis, uploadDataRate});
+ this, flowFileDescription, dataSize, transaction.getCommunicant().getUrl(), uploadMillis, uploadDataRate});
}
return flowFilesReceived.size();
@@ -386,16 +390,16 @@ public class StandardRemoteGroupPort extends RemoteGroupPort {
ValidationResult error = null;
if (!targetExists.get()) {
error = new ValidationResult.Builder()
- .explanation(String.format("Remote instance indicates that port '%s' no longer exists.", getName()))
- .subject(String.format("Remote port '%s'", getName()))
- .valid(false)
- .build();
+ .explanation(String.format("Remote instance indicates that port '%s' no longer exists.", getName()))
+ .subject(String.format("Remote port '%s'", getName()))
+ .valid(false)
+ .build();
} else if (getConnectableType() == ConnectableType.REMOTE_OUTPUT_PORT && getConnections(Relationship.ANONYMOUS).isEmpty()) {
error = new ValidationResult.Builder()
- .explanation(String.format("Port '%s' has no outbound connections", getName()))
- .subject(String.format("Remote port '%s'", getName()))
- .valid(false)
- .build();
+ .explanation(String.format("Port '%s' has no outbound connections", getName()))
+ .subject(String.format("Remote port '%s'", getName()))
+ .valid(false)
+ .build();
}
if (error != null) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/AbstractFlowFileServerProtocol.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/AbstractFlowFileServerProtocol.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/AbstractFlowFileServerProtocol.java
index 8600368..fe324ad 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/AbstractFlowFileServerProtocol.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/AbstractFlowFileServerProtocol.java
@@ -201,7 +201,7 @@ public abstract class AbstractFlowFileServerProtocol implements ServerProtocol {
handshakeCompleted = true;
}
- abstract protected HandshakeProperties doHandshake(final Peer peer) throws IOException, HandshakeException;
+ abstract protected HandshakeProperties doHandshake(final Peer peer) throws IOException, HandshakeException;
@Override
public int transferFlowFiles(final Peer peer, final ProcessContext context, final ProcessSession session, final FlowFileCodec codec) throws IOException, ProtocolException {
@@ -394,9 +394,10 @@ public abstract class AbstractFlowFileServerProtocol implements ServerProtocol {
protected final void writeTransactionResponse(boolean isTransfer, ResponseCode response, CommunicationsSession commsSession) throws IOException {
writeTransactionResponse(isTransfer, response, commsSession, null);
}
+
protected void writeTransactionResponse(boolean isTransfer, ResponseCode response, CommunicationsSession commsSession, String explanation) throws IOException {
final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream());
- if(explanation == null){
+ if (explanation == null) {
response.writeResponse(dos);
} else {
response.writeResponse(dos, explanation);
@@ -436,7 +437,7 @@ public abstract class AbstractFlowFileServerProtocol implements ServerProtocol {
final CheckedInputStream checkedInputStream = new CheckedInputStream(flowFileInputStream, crc);
final DataPacket dataPacket = codec.decode(checkedInputStream);
- if(dataPacket == null){
+ if (dataPacket == null) {
logger.debug("{} Received null dataPacket indicating the end of transaction from {}", this, peer);
break;
}
@@ -528,7 +529,7 @@ public abstract class AbstractFlowFileServerProtocol implements ServerProtocol {
final long uploadMillis = stopWatch.getDuration(TimeUnit.MILLISECONDS);
final String dataSize = FormatUtils.formatDataSize(bytesReceived);
logger.info("{} Successfully received {} ({}) from {} in {} milliseconds at a rate of {}", new Object[]{
- this, flowFileDescription, dataSize, peer, uploadMillis, uploadDataRate});
+ this, flowFileDescription, dataSize, peer, uploadMillis, uploadDataRate});
return flowFilesReceived.size();
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/FlowFileTransaction.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/FlowFileTransaction.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/FlowFileTransaction.java
index 560cbaf..4f14fbb 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/FlowFileTransaction.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/FlowFileTransaction.java
@@ -33,7 +33,7 @@ public class FlowFileTransaction {
private final String calculatedCRC;
public FlowFileTransaction() {
- this(null, null, new StopWatch(true), 0, null, null);
+ this(null, null, new StopWatch(true), 0, null, null);
}
public FlowFileTransaction(ProcessSession session, ProcessContext context, StopWatch stopWatch, long bytesSent, Set<FlowFile> flowFilesSent, String calculatedCRC) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/HandshakeProperties.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/HandshakeProperties.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/HandshakeProperties.java
index c4538da..b12a5b5 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/HandshakeProperties.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/HandshakeProperties.java
@@ -28,7 +28,6 @@ public class HandshakeProperties {
private long batchBytes = 0L;
private long batchDurationNanos = 0L;
-
public String getCommsIdentifier() {
return commsIdentifier;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java
index b2171df..e71ac1d 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/http/StandardHttpFlowFileServerProtocol.java
@@ -40,6 +40,7 @@ import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Optional;
+import org.apache.nifi.util.NiFiProperties;
public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerProtocol implements HttpFlowFileServerProtocol {
@@ -47,11 +48,12 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr
private final FlowFileCodec codec = new StandardFlowFileCodec();
private final VersionNegotiator versionNegotiator;
- private final HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance();
+ private final HttpRemoteSiteListener transactionManager;
- public StandardHttpFlowFileServerProtocol(final VersionNegotiator versionNegotiator) {
+ public StandardHttpFlowFileServerProtocol(final VersionNegotiator versionNegotiator, final NiFiProperties nifiProperties) {
super();
this.versionNegotiator = versionNegotiator;
+ this.transactionManager = HttpRemoteSiteListener.getInstance(nifiProperties);
}
@Override
@@ -91,7 +93,7 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr
HttpServerCommunicationsSession commSession = (HttpServerCommunicationsSession) commsSession;
commSession.setResponseCode(response);
- if(isTransfer){
+ if (isTransfer) {
switch (response) {
case NO_MORE_DATA:
logger.debug("{} There's no data to send.", this);
@@ -136,8 +138,8 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr
ByteArrayOutputStream bos = new ByteArrayOutputStream();
Transaction.TransactionState currentStatus = commSession.getStatus();
- if(isTransfer){
- switch (currentStatus){
+ if (isTransfer) {
+ switch (currentStatus) {
case DATA_EXCHANGED:
String clientChecksum = commSession.getChecksum();
logger.debug("readTransactionResponse. clientChecksum={}", clientChecksum);
@@ -149,7 +151,7 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr
break;
}
} else {
- switch (currentStatus){
+ switch (currentStatus) {
case TRANSACTION_STARTED:
logger.debug("readTransactionResponse. returning CONTINUE_TRANSACTION.");
// We don't know if there's more data to receive, so just continue it.
@@ -159,7 +161,7 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr
// Checksum was successfully validated at client side, or BAD_CHECKSUM is returned.
ResponseCode responseCode = commSession.getResponseCode();
logger.debug("readTransactionResponse. responseCode={}", responseCode);
- if(responseCode.containsMessage()){
+ if (responseCode.containsMessage()) {
responseCode.writeResponse(new DataOutputStream(bos), "");
} else {
responseCode.writeResponse(new DataOutputStream(bos));
@@ -226,7 +228,8 @@ public class StandardHttpFlowFileServerProtocol extends AbstractFlowFileServerPr
}
@Override
- public void sendPeerList(final Peer peer, final Optional<ClusterNodeInformation> clusterNodeInformation) throws IOException {
+ public void sendPeerList(Peer peer, Optional<ClusterNodeInformation> clusterNodeInfo, String remoteInputHost,
+ int remoteInputPort, int remoteInputHttpPort, boolean isSiteToSiteSecure) throws IOException {
}
@Override
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/socket/SocketFlowFileServerProtocol.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/socket/SocketFlowFileServerProtocol.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/socket/SocketFlowFileServerProtocol.java
index 574c726..e965cf4 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/socket/SocketFlowFileServerProtocol.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/main/java/org/apache/nifi/remote/protocol/socket/SocketFlowFileServerProtocol.java
@@ -30,7 +30,6 @@ import org.apache.nifi.remote.protocol.CommunicationsSession;
import org.apache.nifi.remote.protocol.HandshakeProperties;
import org.apache.nifi.remote.protocol.RequestType;
import org.apache.nifi.remote.protocol.ResponseCode;
-import org.apache.nifi.util.NiFiProperties;
import java.io.DataInputStream;
import java.io.DataOutputStream;
@@ -84,7 +83,7 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol
validateHandshakeRequest(confirmed, peer, properties);
} catch (HandshakeException e) {
ResponseCode handshakeResult = e.getResponseCode();
- if(handshakeResult.containsMessage()){
+ if (handshakeResult.containsMessage()) {
handshakeResult.writeResponse(dos, e.getMessage());
} else {
handshakeResult.writeResponse(dos);
@@ -136,7 +135,6 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol
}
}
-
@Override
public RequestType getRequestType(final Peer peer) throws IOException {
if (!handshakeCompleted) {
@@ -154,7 +152,13 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol
}
@Override
- public void sendPeerList(final Peer peer, final Optional<ClusterNodeInformation> clusterNodeInfo) throws IOException {
+ public void sendPeerList(
+ final Peer peer,
+ final Optional<ClusterNodeInformation> clusterNodeInfo,
+ final String remoteInputHost,
+ final int remoteInputPort,
+ final int remoteInputHttpPort,
+ final boolean isSiteToSiteSecure) throws IOException {
if (!handshakeCompleted) {
throw new IllegalStateException("Handshake has not been completed");
}
@@ -166,11 +170,9 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol
final CommunicationsSession commsSession = peer.getCommunicationsSession();
final DataOutputStream dos = new DataOutputStream(commsSession.getOutput().getOutputStream());
- final NiFiProperties properties = NiFiProperties.getInstance();
-
- String remoteInputHost = properties.getRemoteInputHost();
- if (remoteInputHost == null) {
- remoteInputHost = InetAddress.getLocalHost().getHostName();
+ String remoteInputHostVal = remoteInputHost;
+ if (remoteInputHostVal == null) {
+ remoteInputHostVal = InetAddress.getLocalHost().getHostName();
}
logger.debug("{} Advertising Remote Input host name {}", this, peer);
@@ -178,8 +180,8 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol
if (clusterNodeInfo.isPresent()) {
nodeInfos = new ArrayList<>(clusterNodeInfo.get().getNodeInformation());
} else {
- final NodeInformation self = new NodeInformation(remoteInputHost, properties.getRemoteInputPort(), properties.getRemoteInputHttpPort(), properties.getRemoteInputHttpPort(),
- properties.isSiteToSiteSecure(), 0);
+ final NodeInformation self = new NodeInformation(remoteInputHostVal, remoteInputPort, remoteInputHttpPort, remoteInputHttpPort,
+ isSiteToSiteSecure, 0);
nodeInfos = Collections.singletonList(self);
}
@@ -212,7 +214,6 @@ public class SocketFlowFileServerProtocol extends AbstractFlowFileServerProtocol
return RESOURCE_NAME;
}
-
@Override
public VersionNegotiator getVersionNegotiator() {
return versionNegotiator;
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestHttpRemoteSiteListener.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestHttpRemoteSiteListener.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestHttpRemoteSiteListener.java
index e485095..58d4d26 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestHttpRemoteSiteListener.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestHttpRemoteSiteListener.java
@@ -39,7 +39,7 @@ public class TestHttpRemoteSiteListener {
@Test
public void testNormalTransactionProgress() {
- HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance();
+ HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null));
String transactionId = transactionManager.createTransaction();
assertTrue("Transaction should be active.", transactionManager.isTransactionActive(transactionId));
@@ -59,7 +59,7 @@ public class TestHttpRemoteSiteListener {
@Test
public void testDuplicatedTransactionId() {
- HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance();
+ HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null));
String transactionId = transactionManager.createTransaction();
assertTrue("Transaction should be active.", transactionManager.isTransactionActive(transactionId));
@@ -78,7 +78,7 @@ public class TestHttpRemoteSiteListener {
@Test
public void testNoneExistingTransaction() {
- HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance();
+ HttpRemoteSiteListener transactionManager = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null));
String transactionId = "does-not-exist-1";
assertFalse("Transaction should not be active.", transactionManager.isTransactionActive(transactionId));
@@ -88,8 +88,8 @@ public class TestHttpRemoteSiteListener {
try {
transactionManager.holdTransaction(transactionId, transaction, null);
} catch (IllegalStateException e) {
- fail("Transaction can be held even if the transaction id is not valid anymore," +
- " in order to support large file or slow network.");
+ fail("Transaction can be held even if the transaction id is not valid anymore,"
+ + " in order to support large file or slow network.");
}
transactionId = "does-not-exist-2";
@@ -100,5 +100,4 @@ public class TestHttpRemoteSiteListener {
}
}
-
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java
index 4209c93..b44f118 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/TestStandardRemoteGroupPort.java
@@ -43,6 +43,7 @@ import java.net.URI;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Map;
+import org.apache.nifi.util.NiFiProperties;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.eq;
@@ -70,17 +71,18 @@ public class TestStandardRemoteGroupPort {
@BeforeClass
public static void setup() throws Exception {
+ System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties");
System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi.remote", "DEBUG");
}
private void setupMock(final SiteToSiteTransportProtocol protocol,
- final TransferDirection direction) throws Exception {
+ final TransferDirection direction) throws Exception {
setupMock(protocol, direction, mock(Transaction.class));
}
private void setupMock(final SiteToSiteTransportProtocol protocol,
- final TransferDirection direction,
- final Transaction transaction) throws Exception {
+ final TransferDirection direction,
+ final Transaction transaction) throws Exception {
processGroup = null;
remoteGroup = mock(RemoteProcessGroup.class);
scheduler = null;
@@ -102,7 +104,7 @@ public class TestStandardRemoteGroupPort {
break;
}
port = spy(new StandardRemoteGroupPort(ID, NAME,
- processGroup, remoteGroup, direction, connectableType, null, scheduler));
+ processGroup, remoteGroup, direction, connectableType, null, scheduler, NiFiProperties.createBasicNiFiProperties(null, null)));
doReturn(true).when(remoteGroup).isTransmitting();
doReturn(protocol).when(remoteGroup).getTransportProtocol();
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/TestSocketChannelStreams.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/TestSocketChannelStreams.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/TestSocketChannelStreams.java
index 03f8190..27a8807 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/TestSocketChannelStreams.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/TestSocketChannelStreams.java
@@ -76,7 +76,7 @@ package org.apache.nifi.remote.io.socket;
// dos.flush();
//
// final EventReporter eventReporter = Mockito.mock(EventReporter.class);
-// final StandardSiteToSiteProtocol proposedProtocol = new StandardSiteToSiteProtocol(commsSession, eventReporter, NiFiProperties.getInstance());
+// final StandardSiteToSiteProtocol proposedProtocol = new StandardSiteToSiteProtocol(commsSession, eventReporter, nifiProperties);
//
// final StandardSiteToSiteProtocol negotiatedProtocol = (StandardSiteToSiteProtocol) RemoteResourceFactory.initiateResourceNegotiation(proposedProtocol, dis, dos);
// System.out.println(negotiatedProtocol);
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/ssl/TestSSLSocketChannel.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/ssl/TestSSLSocketChannel.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/ssl/TestSSLSocketChannel.java
index 8fe7149..2ead857 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/ssl/TestSSLSocketChannel.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/io/socket/ssl/TestSSLSocketChannel.java
@@ -67,7 +67,7 @@ package org.apache.nifi.remote.io.socket.ssl;
// public void testSendingToLocalInstance() throws IOException, InterruptedException, HandshakeException, UnknownPortException, PortNotRunningException, URISyntaxException {
// System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties");
//
-// final NiFiProperties properties = NiFiProperties.getInstance();
+// final NiFiProperties properties;
// final SSLContext sslContext = SslContextFactory.createSslContext(properties);
//
// final SSLSocketChannel channel = new SSLSocketChannel(sslContext, "localhost", 5000, true);
@@ -84,7 +84,7 @@ package org.apache.nifi.remote.io.socket.ssl;
// dos.flush();
//
// final EventReporter eventReporter = Mockito.mock(EventReporter.class);
-// final StandardSiteToSiteProtocol proposedProtocol = new StandardSiteToSiteProtocol(commsSession, eventReporter, NiFiProperties.getInstance());
+// final StandardSiteToSiteProtocol proposedProtocol = new StandardSiteToSiteProtocol(commsSession, eventReporter, nifiProperties);
// final StandardSiteToSiteProtocol negotiatedProtocol = (StandardSiteToSiteProtocol) RemoteResourceFactory.initiateResourceNegotiation(proposedProtocol, dis, dos);
// System.out.println(negotiatedProtocol);
//
@@ -104,7 +104,7 @@ package org.apache.nifi.remote.io.socket.ssl;
// public void testWithSimpleSSLSocket() throws IOException, InterruptedException {
// System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties");
//
-// final NiFiProperties properties = NiFiProperties.getInstance();
+// final NiFiProperties properties;
// final SSLContext sslContext = SslContextFactory.createSslContext(properties);
//
// final ServerThread server = new ServerThread(sslContext);
@@ -138,7 +138,7 @@ package org.apache.nifi.remote.io.socket.ssl;
// public void testDirectChannelComms() throws IOException, InterruptedException {
// System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties");
//
-// final NiFiProperties properties = NiFiProperties.getInstance();
+// final NiFiProperties properties;
// final SSLContext sslContext = SslContextFactory.createSslContext(properties);
//
// final ServerThread server = new ServerThread(sslContext);
@@ -193,7 +193,7 @@ package org.apache.nifi.remote.io.socket.ssl;
// public void testWriteTimesOut() throws IOException, InterruptedException {
// System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties");
//
-// final NiFiProperties properties = NiFiProperties.getInstance();
+// final NiFiProperties properties;
// final SSLContext sslContext = SslContextFactory.createSslContext(properties);
//
// final ServerThread server = new ServerThread(sslContext);
@@ -238,7 +238,7 @@ package org.apache.nifi.remote.io.socket.ssl;
// public void testInputOutputStreams() throws IOException, InterruptedException {
// System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, "src/test/resources/nifi.properties");
//
-// final NiFiProperties properties = NiFiProperties.getInstance();
+// final NiFiProperties properties;
// final SSLContext sslContext = SslContextFactory.createSslContext(properties);
//
// final ServerThread server = new ServerThread(sslContext);
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/protocol/http/TestHttpFlowFileServerProtocol.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/protocol/http/TestHttpFlowFileServerProtocol.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/protocol/http/TestHttpFlowFileServerProtocol.java
index 7c9d30b..f5e803d 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/protocol/http/TestHttpFlowFileServerProtocol.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-site-to-site/src/test/java/org/apache/nifi/remote/protocol/http/TestHttpFlowFileServerProtocol.java
@@ -90,7 +90,7 @@ public class TestHttpFlowFileServerProtocol {
private HttpFlowFileServerProtocol getDefaultHttpFlowFileServerProtocol() {
final StandardVersionNegotiator versionNegotiator = new StandardVersionNegotiator(5, 4, 3, 2, 1);
- return new StandardHttpFlowFileServerProtocol(versionNegotiator);
+ return new StandardHttpFlowFileServerProtocol(versionNegotiator, NiFiProperties.createBasicNiFiProperties(null, null));
}
@Test
@@ -294,7 +294,7 @@ public class TestHttpFlowFileServerProtocol {
}
private Peer transferOneFile(final HttpFlowFileServerProtocol serverProtocol, final String transactionId) throws IOException {
- final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance();
+ final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null));
final Peer peer = getDefaultPeer(transactionId);
final HttpServerCommunicationsSession commsSession = (HttpServerCommunicationsSession) peer.getCommunicationsSession();
final String endpointUri = "https://peer-host:8443/nifi-api/output-ports/port-id/transactions/"
@@ -338,7 +338,7 @@ public class TestHttpFlowFileServerProtocol {
@Test
public void testTransferTwoFiles() throws Exception {
- final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance();
+ final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null));
final String transactionId = "testTransferTwoFiles";
final Peer peer = getDefaultPeer(transactionId);
@@ -470,7 +470,7 @@ public class TestHttpFlowFileServerProtocol {
}
private void receiveOneFile(final HttpFlowFileServerProtocol serverProtocol, final String transactionId, final Peer peer) throws IOException {
- final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance();
+ final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null));
final String endpointUri = "https://peer-host:8443/nifi-api/input-ports/port-id/transactions/"
+ transactionId + "/flow-files";
final HttpServerCommunicationsSession commsSession = (HttpServerCommunicationsSession) peer.getCommunicationsSession();
@@ -530,7 +530,7 @@ public class TestHttpFlowFileServerProtocol {
@Test
public void testReceiveTwoFiles() throws Exception {
- final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance();
+ final HttpRemoteSiteListener remoteSiteListener = HttpRemoteSiteListener.getInstance(NiFiProperties.createBasicNiFiProperties(null, null));
final String transactionId = "testReceiveTwoFile";
final String endpointUri = "https://peer-host:8443/nifi-api/input-ports/port-id/transactions/"
http://git-wip-us.apache.org/repos/asf/nifi/blob/7d7401ad/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/JettyServerTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/JettyServerTest.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/JettyServerTest.java
index 314e331..f4f0bf8 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/JettyServerTest.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-jetty/src/test/java/org/apache/nifi/web/server/JettyServerTest.java
@@ -17,13 +17,13 @@
package org.apache.nifi.web.server;
+import java.lang.reflect.InvocationTargetException;
+import java.util.HashMap;
+import java.util.Map;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.apache.nifi.util.NiFiProperties;
import org.junit.Test;
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
-
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -31,17 +31,17 @@ import static org.mockito.Mockito.verify;
public class JettyServerTest {
@Test
- public void testConfigureSslContextFactoryWithKeystorePasswordAndKeyPassword() throws InvocationTargetException, NoSuchMethodException, InstantiationException, IllegalAccessException {
+ public void testConfigureSslContextFactoryWithKeystorePasswordAndKeyPassword() {
// Expect that if we set both passwords, KeyStore password is used for KeyStore, Key password is used for Key Manager
String testKeystorePassword = "testKeystorePassword";
String testKeyPassword = "testKeyPassword";
- NiFiProperties nifiProperties = createNifiProperties();
+ final Map<String, String> addProps = new HashMap<>();
+ addProps.put(NiFiProperties.SECURITY_KEYSTORE_PASSWD, testKeystorePassword);
+ addProps.put(NiFiProperties.SECURITY_KEY_PASSWD, testKeyPassword);
+ NiFiProperties nifiProperties = NiFiProperties.createBasicNiFiProperties(null, addProps);
SslContextFactory contextFactory = mock(SslContextFactory.class);
- nifiProperties.setProperty(NiFiProperties.SECURITY_KEYSTORE_PASSWD, testKeystorePassword);
- nifiProperties.setProperty(NiFiProperties.SECURITY_KEY_PASSWD, testKeyPassword);
-
JettyServer.configureSslContextFactory(contextFactory, nifiProperties);
verify(contextFactory).setKeyStorePassword(testKeystorePassword);
@@ -53,11 +53,11 @@ public class JettyServerTest {
// Expect that with no KeyStore password, we will only need to set Key Manager Password
String testKeyPassword = "testKeyPassword";
- NiFiProperties nifiProperties = createNifiProperties();
+ final Map<String, String> addProps = new HashMap<>();
+ addProps.put(NiFiProperties.SECURITY_KEY_PASSWD, testKeyPassword);
+ NiFiProperties nifiProperties = NiFiProperties.createBasicNiFiProperties(null, addProps);
SslContextFactory contextFactory = mock(SslContextFactory.class);
- nifiProperties.setProperty(NiFiProperties.SECURITY_KEY_PASSWD, testKeyPassword);
-
JettyServer.configureSslContextFactory(contextFactory, nifiProperties);
verify(contextFactory).setKeyManagerPassword(testKeyPassword);
@@ -69,20 +69,15 @@ public class JettyServerTest {
// Expect that with no KeyPassword, we use the same one from the KeyStore
String testKeystorePassword = "testKeystorePassword";
- NiFiProperties nifiProperties = createNifiProperties();
+ final Map<String, String> addProps = new HashMap<>();
+ addProps.put(NiFiProperties.SECURITY_KEYSTORE_PASSWD, testKeystorePassword);
+ NiFiProperties nifiProperties = NiFiProperties.createBasicNiFiProperties(null, addProps);
SslContextFactory contextFactory = mock(SslContextFactory.class);
- nifiProperties.setProperty(NiFiProperties.SECURITY_KEYSTORE_PASSWD, testKeystorePassword);
-
JettyServer.configureSslContextFactory(contextFactory, nifiProperties);
verify(contextFactory).setKeyStorePassword(testKeystorePassword);
verify(contextFactory).setKeyManagerPassword(testKeystorePassword);
}
- private NiFiProperties createNifiProperties() throws NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException {
- Constructor<NiFiProperties> constructor = NiFiProperties.class.getDeclaredConstructor();
- constructor.setAccessible(true);
- return constructor.newInstance();
- }
}