You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mc...@apache.org on 2015/05/01 21:45:04 UTC
[03/10] incubator-nifi git commit: NIFI-292: Merging develop into
NIFI-292
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/PasswordBasedEncryptor.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/PasswordBasedEncryptor.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/PasswordBasedEncryptor.java
new file mode 100644
index 0000000..4f6bdc8
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/PasswordBasedEncryptor.java
@@ -0,0 +1,155 @@
+/*
+ * 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.processors.standard.util;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.security.SecureRandom;
+
+import javax.crypto.BadPaddingException;
+import javax.crypto.Cipher;
+import javax.crypto.IllegalBlockSizeException;
+import javax.crypto.SecretKey;
+import javax.crypto.SecretKeyFactory;
+import javax.crypto.spec.PBEKeySpec;
+import javax.crypto.spec.PBEParameterSpec;
+
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.io.StreamCallback;
+import org.apache.nifi.processors.standard.EncryptContent.Encryptor;
+import org.apache.nifi.stream.io.StreamUtils;
+
+public class PasswordBasedEncryptor implements Encryptor {
+
+ private Cipher cipher;
+ private int saltSize;
+ private SecretKey secretKey;
+
+ public static final String SECURE_RANDOM_ALGORITHM = "SHA1PRNG";
+ public static final int DEFAULT_SALT_SIZE = 8;
+
+ public PasswordBasedEncryptor(final String algorithm, final String providerName, final char[] password) {
+ super();
+ try {
+ // initialize cipher
+ this.cipher = Cipher.getInstance(algorithm, providerName);
+ int algorithmBlockSize = cipher.getBlockSize();
+ this.saltSize = (algorithmBlockSize > 0) ? algorithmBlockSize : DEFAULT_SALT_SIZE;
+
+ // initialize SecretKey from password
+ PBEKeySpec pbeKeySpec = new PBEKeySpec(password);
+ SecretKeyFactory factory = SecretKeyFactory.getInstance(algorithm, providerName);
+ this.secretKey = factory.generateSecret(pbeKeySpec);
+ } catch (Exception e) {
+ throw new ProcessException(e);
+ }
+ }
+
+ @Override
+ public StreamCallback getEncryptionCallback() throws ProcessException {
+ try {
+ byte[] salt = new byte[saltSize];
+ SecureRandom secureRandom = SecureRandom.getInstance(SECURE_RANDOM_ALGORITHM);
+ secureRandom.setSeed(System.currentTimeMillis());
+ secureRandom.nextBytes(salt);
+ return new EncryptCallback(salt);
+ } catch (Exception e) {
+ throw new ProcessException(e);
+ }
+ }
+
+ @Override
+ public StreamCallback getDecryptionCallback() throws ProcessException {
+ return new DecryptCallback();
+ }
+
+ private class DecryptCallback implements StreamCallback {
+
+ public DecryptCallback() {
+ }
+
+ @Override
+ public void process(final InputStream in, final OutputStream out) throws IOException {
+ final byte[] salt = new byte[saltSize];
+ try {
+ StreamUtils.fillBuffer(in, salt);
+ } catch (final EOFException e) {
+ throw new ProcessException("Cannot decrypt because file size is smaller than salt size", e);
+ }
+
+ final PBEParameterSpec parameterSpec = new PBEParameterSpec(salt, 1000);
+ try {
+ cipher.init(Cipher.DECRYPT_MODE, secretKey, parameterSpec);
+ } catch (final Exception e) {
+ throw new ProcessException(e);
+ }
+
+ final byte[] buffer = new byte[65536];
+ int len;
+ while ((len = in.read(buffer)) > 0) {
+ final byte[] decryptedBytes = cipher.update(buffer, 0, len);
+ if (decryptedBytes != null) {
+ out.write(decryptedBytes);
+ }
+ }
+
+ try {
+ out.write(cipher.doFinal());
+ } catch (final Exception e) {
+ throw new ProcessException(e);
+ }
+ }
+ }
+
+ private class EncryptCallback implements StreamCallback {
+
+ private final byte[] salt;
+
+ public EncryptCallback(final byte[] salt) {
+ this.salt = salt;
+ }
+
+ @Override
+ public void process(final InputStream in, final OutputStream out) throws IOException {
+ final PBEParameterSpec parameterSpec = new PBEParameterSpec(salt, 1000);
+ try {
+ cipher.init(Cipher.ENCRYPT_MODE, secretKey, parameterSpec);
+ } catch (final Exception e) {
+ throw new ProcessException(e);
+ }
+
+ out.write(salt);
+
+ final byte[] buffer = new byte[65536];
+ int len;
+ while ((len = in.read(buffer)) > 0) {
+ final byte[] encryptedBytes = cipher.update(buffer, 0, len);
+ if (encryptedBytes != null) {
+ out.write(encryptedBytes);
+ }
+ }
+
+ try {
+ out.write(cipher.doFinal());
+ } catch (final IllegalBlockSizeException | BadPaddingException e) {
+ throw new ProcessException(e);
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
index 7fbd781..17339bc 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/META-INF/services/org.apache.nifi.processor.Processor
@@ -18,6 +18,7 @@ org.apache.nifi.processors.standard.ControlRate
org.apache.nifi.processors.standard.ConvertCharacterSet
org.apache.nifi.processors.standard.DetectDuplicate
org.apache.nifi.processors.standard.DistributeLoad
+org.apache.nifi.processors.standard.DuplicateFlowFile
org.apache.nifi.processors.standard.EncryptContent
org.apache.nifi.processors.standard.EvaluateJsonPath
org.apache.nifi.processors.standard.EvaluateRegularExpression
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.EncryptContent/additionalDetails.html
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.EncryptContent/additionalDetails.html b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.EncryptContent/additionalDetails.html
new file mode 100644
index 0000000..b0603c6
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/resources/docs/org.apache.nifi.processors.standard.EncryptContent/additionalDetails.html
@@ -0,0 +1,30 @@
+<!DOCTYPE html>
+<html lang="en">
+ <!--
+ 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.
+ -->
+ <head>
+ <meta charset="utf-8"/>
+ <title>EvaluateJsonPath</title>
+ <link rel="stylesheet" href="../../css/component-usage.css" type="text/css"/>
+ </head>
+
+ <body>
+ <!-- Processor Documentation ================================================== -->
+ <p>
+ <strong>Note:</strong> This processor supports OpenPGP algorithms that are compatible with third party programs.
+ However, it currently cannot add a digital signature to an encrypted FlowFile.
+ </p>
+ </body>
+</html>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDetectDuplicate.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDetectDuplicate.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDetectDuplicate.java
index e8434f0..4166d94 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDetectDuplicate.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDetectDuplicate.java
@@ -57,7 +57,7 @@ public class TestDetectDuplicate {
@Test
public void testDuplicate() throws InitializationException {
- TestRunner runner = TestRunners.newTestRunner(DetectDuplicate.class);
+ final TestRunner runner = TestRunners.newTestRunner(DetectDuplicate.class);
final DistributedMapCacheClientImpl client = createClient();
final Map<String, String> clientProperties = new HashMap<>();
clientProperties.put(DistributedMapCacheClientService.HOSTNAME.getName(), "localhost");
@@ -65,7 +65,7 @@ public class TestDetectDuplicate {
runner.setProperty(DetectDuplicate.DISTRIBUTED_CACHE_SERVICE, "client");
runner.setProperty(DetectDuplicate.FLOWFILE_DESCRIPTION, "The original flow file");
runner.setProperty(DetectDuplicate.AGE_OFF_DURATION, "48 hours");
- Map<String, String> props = new HashMap<>();
+ final Map<String, String> props = new HashMap<>();
props.put("hash.value", "1000");
runner.enqueue(new byte[]{}, props);
runner.enableControllerService(client);
@@ -84,7 +84,7 @@ public class TestDetectDuplicate {
@Test
public void testDuplicateWithAgeOff() throws InitializationException, InterruptedException {
- TestRunner runner = TestRunners.newTestRunner(DetectDuplicate.class);
+ final TestRunner runner = TestRunners.newTestRunner(DetectDuplicate.class);
final DistributedMapCacheClientImpl client = createClient();
final Map<String, String> clientProperties = new HashMap<>();
clientProperties.put(DistributedMapCacheClientService.HOSTNAME.getName(), "localhost");
@@ -94,7 +94,7 @@ public class TestDetectDuplicate {
runner.setProperty(DetectDuplicate.AGE_OFF_DURATION, "2 secs");
runner.enableControllerService(client);
- Map<String, String> props = new HashMap<>();
+ final Map<String, String> props = new HashMap<>();
props.put("hash.value", "1000");
runner.enqueue(new byte[]{}, props);
@@ -114,7 +114,7 @@ public class TestDetectDuplicate {
final DistributedMapCacheClientImpl client = new DistributedMapCacheClientImpl();
final ComponentLog logger = new MockProcessorLog("client", client);
- MockControllerServiceInitializationContext clientInitContext = new MockControllerServiceInitializationContext(client, "client", logger);
+ final MockControllerServiceInitializationContext clientInitContext = new MockControllerServiceInitializationContext(client, "client", logger);
client.initialize(clientInitContext);
return client;
@@ -130,12 +130,12 @@ public class TestDetectDuplicate {
}
@Override
- public void onPropertyModified(PropertyDescriptor descriptor, String oldValue, String newValue) {
+ public void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue) {
}
@Override
protected java.util.List<PropertyDescriptor> getSupportedPropertyDescriptors() {
- List<PropertyDescriptor> props = new ArrayList<>();
+ final List<PropertyDescriptor> props = new ArrayList<>();
props.add(DistributedMapCacheClientService.HOSTNAME);
props.add(DistributedMapCacheClientService.COMMUNICATIONS_TIMEOUT);
props.add(DistributedMapCacheClientService.PORT);
@@ -144,7 +144,7 @@ public class TestDetectDuplicate {
}
@Override
- public <K, V> boolean putIfAbsent(K key, V value, Serializer<K> keySerializer, Serializer<V> valueSerializer) throws IOException {
+ public <K, V> boolean putIfAbsent(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) throws IOException {
if (exists) {
return false;
}
@@ -154,7 +154,8 @@ public class TestDetectDuplicate {
}
@Override
- public <K, V> V getAndPutIfAbsent(K key, V value, Serializer<K> keySerializer, Serializer<V> valueSerializer, Deserializer<V> valueDeserializer) throws IOException {
+ public <K, V> V getAndPutIfAbsent(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer,
+ final Deserializer<V> valueDeserializer) throws IOException {
if (exists) {
return (V) cacheValue;
}
@@ -163,20 +164,24 @@ public class TestDetectDuplicate {
}
@Override
- public <K> boolean containsKey(K key, Serializer<K> keySerializer) throws IOException {
+ public <K> boolean containsKey(final K key, final Serializer<K> keySerializer) throws IOException {
return exists;
}
@Override
- public <K, V> V get(K key, Serializer<K> keySerializer, Deserializer<V> valueDeserializer) throws IOException {
+ public <K, V> V get(final K key, final Serializer<K> keySerializer, final Deserializer<V> valueDeserializer) throws IOException {
return null;
}
@Override
- public <K> boolean remove(K key, Serializer<K> serializer) throws IOException {
+ public <K> boolean remove(final K key, final Serializer<K> serializer) throws IOException {
exists = false;
return true;
}
+
+ @Override
+ public <K, V> void put(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) throws IOException {
+ }
}
private static class StringSerializer implements Serializer<String> {
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDuplicateFlowFile.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDuplicateFlowFile.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDuplicateFlowFile.java
new file mode 100644
index 0000000..82fee1b
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestDuplicateFlowFile.java
@@ -0,0 +1,36 @@
+/*
+ * 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.processors.standard;
+
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Test;
+
+public class TestDuplicateFlowFile {
+
+ @Test
+ public void test() {
+ final TestRunner runner = TestRunners.newTestRunner(DuplicateFlowFile.class);
+ runner.setProperty(DuplicateFlowFile.NUM_COPIES, "100");
+
+ runner.enqueue("hello".getBytes());
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(DuplicateFlowFile.REL_SUCCESS, 101);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestEncryptContent.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestEncryptContent.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestEncryptContent.java
index 7340e0f..a5581b3 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestEncryptContent.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestEncryptContent.java
@@ -1,64 +1,163 @@
-/*
- * 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.processors.standard;
-
-import java.io.File;
-import java.io.IOException;
-import java.nio.file.Paths;
-
-import org.apache.nifi.security.util.EncryptionMethod;
-import org.apache.nifi.util.MockFlowFile;
-import org.apache.nifi.util.TestRunner;
-import org.apache.nifi.util.TestRunners;
-
-import org.junit.Test;
-
-public class TestEncryptContent {
-
- @Test
- public void testRoundTrip() throws IOException {
- final TestRunner testRunner = TestRunners.newTestRunner(new EncryptContent());
- testRunner.setProperty(EncryptContent.PASSWORD, "Hello, World!");
-
- for (final EncryptionMethod method : EncryptionMethod.values()) {
- if (method.isUnlimitedStrength()) {
- continue; // cannot test unlimited strength in unit tests because it's not enabled by the JVM by default.
- }
- testRunner.setProperty(EncryptContent.ENCRYPTION_ALGORITHM, method.name());
- testRunner.setProperty(EncryptContent.MODE, EncryptContent.ENCRYPT_MODE);
-
- testRunner.enqueue(Paths.get("src/test/resources/hello.txt"));
- testRunner.clearTransferState();
- testRunner.run();
-
- testRunner.assertAllFlowFilesTransferred(EncryptContent.REL_SUCCESS, 1);
-
- MockFlowFile flowFile = testRunner.getFlowFilesForRelationship(EncryptContent.REL_SUCCESS).get(0);
- testRunner.assertQueueEmpty();
-
- testRunner.setProperty(EncryptContent.MODE, EncryptContent.DECRYPT_MODE);
- testRunner.enqueue(flowFile);
- testRunner.clearTransferState();
- testRunner.run();
- testRunner.assertAllFlowFilesTransferred(EncryptContent.REL_SUCCESS, 1);
-
- flowFile = testRunner.getFlowFilesForRelationship(EncryptContent.REL_SUCCESS).get(0);
- flowFile.assertContentEquals(new File("src/test/resources/hello.txt"));
- }
- }
-
-}
+/*
+ * 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.processors.standard;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Paths;
+import java.util.Collection;
+import java.util.HashSet;
+
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.security.util.EncryptionMethod;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.MockProcessContext;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestEncryptContent {
+
+ @Test
+ public void testRoundTrip() throws IOException {
+ final TestRunner testRunner = TestRunners.newTestRunner(new EncryptContent());
+ testRunner.setProperty(EncryptContent.PASSWORD, "Hello, World!");
+
+ for (final EncryptionMethod method : EncryptionMethod.values()) {
+ if (method.isUnlimitedStrength()) {
+ continue; // cannot test unlimited strength in unit tests because it's not enabled by the JVM by default.
+ }
+ testRunner.setProperty(EncryptContent.ENCRYPTION_ALGORITHM, method.name());
+ testRunner.setProperty(EncryptContent.MODE, EncryptContent.ENCRYPT_MODE);
+
+ testRunner.enqueue(Paths.get("src/test/resources/hello.txt"));
+ testRunner.clearTransferState();
+ testRunner.run();
+
+ testRunner.assertAllFlowFilesTransferred(EncryptContent.REL_SUCCESS, 1);
+
+ MockFlowFile flowFile = testRunner.getFlowFilesForRelationship(EncryptContent.REL_SUCCESS).get(0);
+ testRunner.assertQueueEmpty();
+
+ testRunner.setProperty(EncryptContent.MODE, EncryptContent.DECRYPT_MODE);
+ testRunner.enqueue(flowFile);
+ testRunner.clearTransferState();
+ testRunner.run();
+ testRunner.assertAllFlowFilesTransferred(EncryptContent.REL_SUCCESS, 1);
+
+ flowFile = testRunner.getFlowFilesForRelationship(EncryptContent.REL_SUCCESS).get(0);
+ flowFile.assertContentEquals(new File("src/test/resources/hello.txt"));
+ }
+ }
+
+ @Test
+ public void testDecryptSmallerThanSaltSize() {
+ final TestRunner runner = TestRunners.newTestRunner(EncryptContent.class);
+ runner.setProperty(EncryptContent.PASSWORD, "Hello, World!");
+ runner.setProperty(EncryptContent.MODE, EncryptContent.DECRYPT_MODE);
+ runner.enqueue(new byte[4]);
+ runner.run();
+ runner.assertAllFlowFilesTransferred(EncryptContent.REL_FAILURE, 1);
+ }
+
+ @Test
+ public void testPGPDecrypt() throws IOException {
+ final TestRunner runner = TestRunners.newTestRunner(EncryptContent.class);
+ runner.setProperty(EncryptContent.MODE, EncryptContent.DECRYPT_MODE);
+ runner.setProperty(EncryptContent.ENCRYPTION_ALGORITHM, EncryptionMethod.PGP_ASCII_ARMOR.name());
+ runner.setProperty(EncryptContent.PASSWORD, "Hello, World!");
+
+ runner.enqueue(Paths.get("src/test/resources/TestEncryptContent/text.txt.asc"));
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(EncryptContent.REL_SUCCESS, 1);
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(EncryptContent.REL_SUCCESS).get(0);
+ flowFile.assertContentEquals(Paths.get("src/test/resources/TestEncryptContent/text.txt"));
+ }
+
+ @Test
+ public void testValidation() {
+ final TestRunner runner = TestRunners.newTestRunner(EncryptContent.class);
+ Collection<ValidationResult> results;
+ MockProcessContext pc;
+
+ results = new HashSet<>();
+ runner.enqueue(new byte[0]);
+ pc = (MockProcessContext) runner.getProcessContext();
+ results = pc.validate();
+ Assert.assertEquals(1, results.size());
+ for (final ValidationResult vr : results) {
+ Assert.assertTrue(vr.toString()
+ .contains(EncryptContent.PASSWORD.getDisplayName() + " is required when using algorithm"));
+ }
+
+ results = new HashSet<>();
+ runner.setProperty(EncryptContent.ENCRYPTION_ALGORITHM, EncryptionMethod.PGP.name());
+ runner.setProperty(EncryptContent.PUBLIC_KEYRING, "src/test/resources/TestEncryptContent/text.txt");
+ runner.enqueue(new byte[0]);
+ pc = (MockProcessContext) runner.getProcessContext();
+ results = pc.validate();
+ Assert.assertEquals(1, results.size());
+ for (final ValidationResult vr : results) {
+ Assert.assertTrue(vr.toString().contains(
+ " encryption without a " + EncryptContent.PASSWORD.getDisplayName() + " requires both "
+ + EncryptContent.PUBLIC_KEYRING.getDisplayName() + " and "
+ + EncryptContent.PUBLIC_KEY_USERID.getDisplayName()));
+ }
+
+ results = new HashSet<>();
+ runner.setProperty(EncryptContent.PUBLIC_KEY_USERID, "USERID");
+ runner.enqueue(new byte[0]);
+ pc = (MockProcessContext) runner.getProcessContext();
+ results = pc.validate();
+ Assert.assertEquals(1, results.size());
+ for (final ValidationResult vr : results) {
+ Assert.assertTrue(vr.toString().contains("does not contain user id USERID"));
+ }
+
+ runner.removeProperty(EncryptContent.PUBLIC_KEYRING);
+ runner.removeProperty(EncryptContent.PUBLIC_KEY_USERID);
+
+ results = new HashSet<>();
+ runner.setProperty(EncryptContent.MODE, EncryptContent.DECRYPT_MODE);
+ runner.setProperty(EncryptContent.PRIVATE_KEYRING, "src/test/resources/TestEncryptContent/text.txt");
+ runner.enqueue(new byte[0]);
+ pc = (MockProcessContext) runner.getProcessContext();
+ results = pc.validate();
+ Assert.assertEquals(1, results.size());
+ for (final ValidationResult vr : results) {
+ Assert.assertTrue(vr.toString().contains(
+ " decryption without a " + EncryptContent.PASSWORD.getDisplayName() + " requires both "
+ + EncryptContent.PRIVATE_KEYRING.getDisplayName() + " and "
+ + EncryptContent.PRIVATE_KEYRING_PASSPHRASE.getDisplayName()));
+
+ }
+
+ results = new HashSet<>();
+ runner.setProperty(EncryptContent.PRIVATE_KEYRING_PASSPHRASE, "PASSWORD");
+ runner.enqueue(new byte[0]);
+ pc = (MockProcessContext) runner.getProcessContext();
+ results = pc.validate();
+ Assert.assertEquals(1, results.size());
+ for (final ValidationResult vr : results) {
+ Assert.assertTrue(vr.toString().contains(
+ " could not be opened with the provided " + EncryptContent.PRIVATE_KEYRING_PASSPHRASE.getDisplayName()));
+
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
index 7529e6d..ff98dfa 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestExecuteProcess.java
@@ -20,11 +20,14 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
+import java.io.File;
import java.util.List;
+import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.util.MockFlowFile;
import org.apache.nifi.util.TestRunner;
import org.apache.nifi.util.TestRunners;
+import org.junit.Ignore;
import org.junit.Test;
public class TestExecuteProcess {
@@ -58,6 +61,7 @@ public class TestExecuteProcess {
assertEquals("good bye", twoQuotedArg.get(1));
}
+ @Ignore // won't run under Windows
@Test
public void testEcho() {
System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi", "TRACE");
@@ -75,4 +79,82 @@ public class TestExecuteProcess {
System.out.println(new String(flowFile.toByteArray()));
}
}
+
+ // @Test
+ public void testBigBinaryInputData() {
+ System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi", "TRACE");
+ System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi.processors.standard", "DEBUG");
+
+ String workingDirName = "/var/test";
+ String testFile = "eclipse-java-luna-SR2-win32.zip";
+
+ final TestRunner runner = TestRunners.newTestRunner(ExecuteProcess.class);
+ runner.setProperty(ExecuteProcess.COMMAND, "cmd");
+ runner.setProperty(ExecuteProcess.COMMAND_ARGUMENTS, " /c type " + testFile);
+ runner.setProperty(ExecuteProcess.WORKING_DIR, workingDirName);
+
+ File inFile = new File(workingDirName, testFile);
+ System.out.println(inFile.getAbsolutePath());
+
+ runner.run();
+
+ final List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ExecuteProcess.REL_SUCCESS);
+ long totalFlowFilesSize = 0;
+ for (final MockFlowFile flowFile : flowFiles) {
+ System.out.println(flowFile);
+ totalFlowFilesSize += flowFile.getSize();
+ // System.out.println(new String(flowFile.toByteArray()));
+ }
+
+ assertEquals(inFile.length(), totalFlowFilesSize);
+ }
+
+ @Test
+ public void testBigInputSplit() {
+ System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi", "TRACE");
+ System.setProperty("org.slf4j.simpleLogger.log.org.apache.nifi.processors.standard", "DEBUG");
+
+ String workingDirName = "/var/test";
+ String testFile = "Novo_dicionário_da_língua_portuguesa_by_Cândido_de_Figueiredo.txt";
+ // String testFile = "eclipse-java-luna-SR2-win32.zip";
+
+ final TestRunner runner = TestRunners.newTestRunner(ExecuteProcess.class);
+ runner.setProperty(ExecuteProcess.COMMAND, "cmd");
+ runner.setProperty(ExecuteProcess.COMMAND_ARGUMENTS, " /c type " + testFile);
+ runner.setProperty(ExecuteProcess.WORKING_DIR, workingDirName);
+ runner.setProperty(ExecuteProcess.BATCH_DURATION, "150 millis");
+
+ File inFile = new File(workingDirName, testFile);
+ System.out.println(inFile.getAbsolutePath());
+
+ // runner.run(1,false,true);
+
+ ProcessContext processContext = runner.getProcessContext();
+
+ ExecuteProcess processor = (ExecuteProcess) runner.getProcessor();
+ processor.updateScheduledTrue();
+ processor.setupExecutor(processContext);
+
+ processor.onTrigger(processContext, runner.getProcessSessionFactory());
+ processor.onTrigger(processContext, runner.getProcessSessionFactory());
+ processor.onTrigger(processContext, runner.getProcessSessionFactory());
+ processor.onTrigger(processContext, runner.getProcessSessionFactory());
+ processor.onTrigger(processContext, runner.getProcessSessionFactory());
+ processor.onTrigger(processContext, runner.getProcessSessionFactory());
+ processor.onTrigger(processContext, runner.getProcessSessionFactory());
+ processor.onTrigger(processContext, runner.getProcessSessionFactory());
+ processor.onTrigger(processContext, runner.getProcessSessionFactory());
+
+ // runner.run(5,true,false);
+
+ final List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ExecuteProcess.REL_SUCCESS);
+ long totalFlowFilesSize = 0;
+ for (final MockFlowFile flowFile : flowFiles) {
+ System.out.println(flowFile);
+ totalFlowFilesSize += flowFile.getSize();
+ // System.out.println(new String(flowFile.toByteArray()));
+ }
+
+ // assertEquals(inFile.length(), totalFlowFilesSize);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
index 688b9eb..235ec2d 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestHandleHttpRequest.java
@@ -42,8 +42,8 @@ import org.junit.Test;
public class TestHandleHttpRequest {
- @Test
- public void testRequestAddedToService() throws InitializationException, MalformedURLException, IOException {
+ @Test(timeout=10000)
+ public void testRequestAddedToService() throws InitializationException, MalformedURLException, IOException, InterruptedException {
final TestRunner runner = TestRunners.newTestRunner(HandleHttpRequest.class);
runner.setProperty(HandleHttpRequest.PORT, "0");
@@ -79,15 +79,11 @@ public class TestHandleHttpRequest {
});
httpThread.start();
- // give processor a bit to handle the http request
- try {
- Thread.sleep(1000L);
- } catch (final InterruptedException ie) {
+ while ( runner.getFlowFilesForRelationship(HandleHttpRequest.REL_SUCCESS).isEmpty() ) {
+ // process the request.
+ runner.run(1, false);
}
- // process the request.
- runner.run(1, false);
-
runner.assertAllFlowFilesTransferred(HandleHttpRequest.REL_SUCCESS, 1);
assertEquals(1, contextMap.size());
@@ -110,18 +106,18 @@ public class TestHandleHttpRequest {
private final ConcurrentMap<String, HttpServletResponse> responseMap = new ConcurrentHashMap<>();
@Override
- public boolean register(String identifier, HttpServletRequest request, HttpServletResponse response, AsyncContext context) {
+ public boolean register(final String identifier, final HttpServletRequest request, final HttpServletResponse response, final AsyncContext context) {
responseMap.put(identifier, response);
return true;
}
@Override
- public HttpServletResponse getResponse(String identifier) {
+ public HttpServletResponse getResponse(final String identifier) {
return responseMap.get(identifier);
}
@Override
- public void complete(String identifier) {
+ public void complete(final String identifier) {
responseMap.remove(identifier);
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestEncryptContent/text.txt
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestEncryptContent/text.txt b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestEncryptContent/text.txt
new file mode 100644
index 0000000..3470ef5
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestEncryptContent/text.txt
@@ -0,0 +1,17 @@
+/*
+ * 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.
+ */
+This is some clear text.
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestEncryptContent/text.txt.asc
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestEncryptContent/text.txt.asc b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestEncryptContent/text.txt.asc
new file mode 100644
index 0000000..7607f83
--- /dev/null
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestEncryptContent/text.txt.asc
@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+-----BEGIN PGP MESSAGE-----
+Version: GnuPG v2
+
+jA0EAwMC54x6GeEEmLy3ycE/pY5T6vawFMS9/Io8chEQGD8UNAbZFzqXcVEvvy5f
+wJQggyuF6Lo1RZGnhxdGoz0do1DZzR0lVaTL8dR0/jtz/kRZ1omz+OxICuo9BaRX
+M+fT2mdna5lhDGJmE7nCctDaGwXqjglEXPqOdi8j/tL225HViTKP1VmlKuu8AjiH
+lMGuC65bqILSWCaE2jewCbsmjHPgLGmH9NN6EAo2kiFEMOrA+UYo35PuShkgQsZp
+gA0m2A31JjLW28SUsNd1vLk5bWBZaIFA1UvhR7u0pagv3qtu5f8qls19nHnAT/bz
+Kh2KrnIm0peWWVEPGQkFoK3Lt9vJTjmHdHPUXQHyg+SMN1PIGA2sxwiSrkQlAyon
+uNg/I24ctydWU+qndz+ycDWR6zBziA09KHw7uKo5CDtTm+Zo3K9U9uf9y8iZ+AKd
+vgF/4Nw2lJTqQfNtkmK+N+cKEyZNmJa4r+uDzJF/dCv8R5jGj2dBYRTLxj5tgllU
+4GiwuJR6w09hK0S0oe9XTdcNciWigb12H9z6U6JOse+1S/fYoUa0CZRJg9Bgeqym
+uuT/mNSKwRVcWN27vOGy+zGf0tqw6A5idLrK+8FZzd1sgxtKsgkYz8FcZgo9rq1f
+PHR9KF4JhaGpqNJmEu/GucYIAgq3aeo9GoV/RpDZtoHAVBuqPwcDTzGHeiAoZ49U
+6Q==
+=H5nf
+-----END PGP MESSAGE-----
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-client-service-api/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClient.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-client-service-api/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClient.java b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-client-service-api/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClient.java
index d816e8c..ea3bb63 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-client-service-api/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClient.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-client-service-api/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClient.java
@@ -86,8 +86,27 @@ public interface DistributedMapCacheClient extends ControllerService {
<K> boolean containsKey(K key, Serializer<K> keySerializer) throws IOException;
/**
- * @param <K> type of key
- * @param <V> type of value
+ * Adds the specified key and value to the cache, overwriting any value that is
+ * currently set.
+ *
+ * @param <K> the key type
+ * @param <V> the value type
+ * @param key The key to set
+ * @param value The value to associate with the given Key
+ * @param keySerializer the Serializer that will be used to serialize the key into bytes
+ * @param valueSerializer the Serializer that will be used to serialize the value into bytes
+ *
+ * @throws IOException if unable to communicate with the remote instance
+ * @throws NullPointerException if the key or either serializer is null
+ */
+ <K, V> void put(K key, V value, Serializer<K> keySerializer, Serializer<V> valueSerializer) throws IOException;
+
+ /**
+ * Returns the value in the cache for the given key, if one exists;
+ * otherwise returns <code>null</code>
+ *
+ * @param <K> the key type
+ * @param <V> the value type
* @param key the key to lookup in the map
* @param keySerializer key serializer
* @param valueDeserializer value serializer
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClientService.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClientService.java b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClientService.java
index 51138b9..c03dd5a 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClientService.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedMapCacheClientService.java
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.AbstractControllerService;
@@ -42,41 +43,42 @@ import org.apache.nifi.stream.io.DataOutputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Tags({"distributed", "cache", "state", "map", "cluster"})
@SeeAlso(classNames = {"org.apache.nifi.distributed.cache.server.map.DistributedMapCacheServer", "org.apache.nifi.ssl.StandardSSLContextService"})
@CapabilityDescription("Provides the ability to communicate with a DistributedMapCacheServer. This can be used in order to share a Map "
- + "between nodes in a NiFi cluster")
+ + "between nodes in a NiFi cluster")
public class DistributedMapCacheClientService extends AbstractControllerService implements DistributedMapCacheClient {
private static final Logger logger = LoggerFactory.getLogger(DistributedMapCacheClientService.class);
public static final PropertyDescriptor HOSTNAME = new PropertyDescriptor.Builder()
- .name("Server Hostname")
- .description("The name of the server that is running the DistributedMapCacheServer service")
- .required(true)
- .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
- .build();
+ .name("Server Hostname")
+ .description("The name of the server that is running the DistributedMapCacheServer service")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
public static final PropertyDescriptor PORT = new PropertyDescriptor.Builder()
- .name("Server Port")
- .description("The port on the remote server that is to be used when communicating with the DistributedMapCacheServer service")
- .required(true)
- .addValidator(StandardValidators.PORT_VALIDATOR)
- .defaultValue("4557")
- .build();
+ .name("Server Port")
+ .description("The port on the remote server that is to be used when communicating with the DistributedMapCacheServer service")
+ .required(true)
+ .addValidator(StandardValidators.PORT_VALIDATOR)
+ .defaultValue("4557")
+ .build();
public static final PropertyDescriptor SSL_CONTEXT_SERVICE = new PropertyDescriptor.Builder()
- .name("SSL Context Service")
- .description("If specified, indicates the SSL Context Service that is used to communicate with the "
- + "remote server. If not specified, communications will not be encrypted")
- .required(false)
- .identifiesControllerService(SSLContextService.class)
- .build();
+ .name("SSL Context Service")
+ .description("If specified, indicates the SSL Context Service that is used to communicate with the "
+ + "remote server. If not specified, communications will not be encrypted")
+ .required(false)
+ .identifiesControllerService(SSLContextService.class)
+ .build();
public static final PropertyDescriptor COMMUNICATIONS_TIMEOUT = new PropertyDescriptor.Builder()
- .name("Communications Timeout")
- .description("Specifies how long to wait when communicating with the remote server before determining that "
- + "there is a communications failure if data cannot be sent or received")
- .required(true)
- .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
- .defaultValue("30 secs")
- .build();
+ .name("Communications Timeout")
+ .description("Specifies how long to wait when communicating with the remote server before determining that "
+ + "there is a communications failure if data cannot be sent or received")
+ .required(true)
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .defaultValue("30 secs")
+ .build();
private final BlockingQueue<CommsSession> queue = new LinkedBlockingQueue<>();
private volatile ConfigurationContext configContext;
@@ -117,6 +119,29 @@ public class DistributedMapCacheClientService extends AbstractControllerService
}
@Override
+ public <K, V> void put(final K key, final V value, final Serializer<K> keySerializer, final Serializer<V> valueSerializer) throws IOException {
+ withCommsSession(new CommsAction<Object>() {
+ @Override
+ public Object execute(final CommsSession session) throws IOException {
+ final DataOutputStream dos = new DataOutputStream(session.getOutputStream());
+ dos.writeUTF("put");
+
+ serialize(key, keySerializer, dos);
+ serialize(value, valueSerializer, dos);
+
+ dos.flush();
+ final DataInputStream dis = new DataInputStream(session.getInputStream());
+ final boolean success = dis.readBoolean();
+ if ( !success ) {
+ throw new IOException("Expected to receive confirmation of 'put' request but received unexpected response");
+ }
+
+ return null;
+ }
+ });
+ }
+
+ @Override
public <K> boolean containsKey(final K key, final Serializer<K> keySerializer) throws IOException {
return withCommsSession(new CommsAction<Boolean>() {
@Override
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedSetCacheClientService.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedSetCacheClientService.java b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedSetCacheClientService.java
index 63d59ca..8c95c77 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedSetCacheClientService.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedSetCacheClientService.java
@@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnEnabled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.controller.AbstractControllerService;
@@ -42,6 +43,7 @@ import org.apache.nifi.stream.io.DataOutputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+@Tags({"distributed", "cache", "state", "set", "cluster"})
@SeeAlso(classNames = {"org.apache.nifi.distributed.cache.server.DistributedSetCacheServer", "org.apache.nifi.ssl.StandardSSLContextService"})
@CapabilityDescription("Provides the ability to communicate with a DistributedSetCacheServer. This can be used in order to share a Set "
+ "between nodes in a NiFi cluster")
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/resources/docs/org.apache.nifi.distributed.cache.client.DistributedMapCacheClientService/additionalDetails.html
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/resources/docs/org.apache.nifi.distributed.cache.client.DistributedMapCacheClientService/additionalDetails.html b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/resources/docs/org.apache.nifi.distributed.cache.client.DistributedMapCacheClientService/additionalDetails.html
deleted file mode 100644
index 1568635..0000000
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/resources/docs/org.apache.nifi.distributed.cache.client.DistributedMapCacheClientService/additionalDetails.html
+++ /dev/null
@@ -1,45 +0,0 @@
-<!DOCTYPE html>
-<html lang="en">
- <!--
- 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.
- -->
- <head>
- <meta charset="utf-8" />
- <title>Distributed Map Cache Client Service</title>
- <link rel="stylesheet" href="../../css/component-usage.css" type="text/css" />
- </head>
-
- <body>
- <p>
- Below is an example of how to create a client connection to your distributed map cache server.
- Note that the identifier in this example is <code>cache-client</code>. If you are using this template
- to create your own MapCacheClient service, replace the values in this template with values that are
- suitable for your system. Possible options for <code>Server Hostname</code>, <code>Server Port</code>,
- <code>Communications Timeout</code>, and <span style="font-style: italic;">SSL Context Service</span>.
- </p>
-
- <pre>
-<?xml version="1.0" encoding="UTF-8" ?>
-<services>
- <service>
- <identifier>cache-client</identifier>
- <class>org.apache.nifi.distributed.cache.client.DistributedMapCacheClientService</class>
- <property name="Server Hostname">localhost</property>
- <property name="Server Port">4557</property>
- <property name="Communications Timeout">30 secs</property>
- </service>
-</services>
- </pre>
- </body>
-</html>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/MapCache.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/MapCache.java b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/MapCache.java
index fad0adb..e9c6f1d 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/MapCache.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/MapCache.java
@@ -23,6 +23,8 @@ public interface MapCache {
MapPutResult putIfAbsent(ByteBuffer key, ByteBuffer value) throws IOException;
+ MapPutResult put(ByteBuffer key, ByteBuffer value) throws IOException;
+
boolean containsKey(ByteBuffer key) throws IOException;
ByteBuffer get(ByteBuffer key) throws IOException;
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/MapCacheServer.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/MapCacheServer.java b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/MapCacheServer.java
index 943d6aa..13ed0df 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/MapCacheServer.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/MapCacheServer.java
@@ -55,63 +55,70 @@ public class MapCacheServer extends AbstractCacheServer {
final String action = dis.readUTF();
try {
switch (action) {
- case "close": {
- return false;
- }
- case "putIfAbsent": {
- final byte[] key = readValue(dis);
- final byte[] value = readValue(dis);
- final MapPutResult putResult = cache.putIfAbsent(ByteBuffer.wrap(key), ByteBuffer.wrap(value));
- dos.writeBoolean(putResult.isSuccessful());
- break;
- }
- case "containsKey": {
- final byte[] key = readValue(dis);
- final boolean contains = cache.containsKey(ByteBuffer.wrap(key));
- dos.writeBoolean(contains);
- break;
- }
- case "getAndPutIfAbsent": {
- final byte[] key = readValue(dis);
- final byte[] value = readValue(dis);
-
- final MapPutResult putResult = cache.putIfAbsent(ByteBuffer.wrap(key), ByteBuffer.wrap(value));
- if (putResult.isSuccessful()) {
- // Put was successful. There was no old value to get.
- dos.writeInt(0);
- } else {
- // we didn't put. Write back the previous value
- final byte[] byteArray = putResult.getExistingValue().array();
- dos.writeInt(byteArray.length);
- dos.write(byteArray);
- }
-
- break;
- }
- case "get": {
- final byte[] key = readValue(dis);
- final ByteBuffer existingValue = cache.get(ByteBuffer.wrap(key));
- if (existingValue == null) {
- // there was no existing value; we did a "put".
- dos.writeInt(0);
- } else {
- // a value already existed. we did not update the map
- final byte[] byteArray = existingValue.array();
- dos.writeInt(byteArray.length);
- dos.write(byteArray);
- }
-
- break;
- }
- case "remove": {
- final byte[] key = readValue(dis);
- final boolean removed = cache.remove(ByteBuffer.wrap(key)) != null;
- dos.writeBoolean(removed);
- break;
+ case "close": {
+ return false;
+ }
+ case "putIfAbsent": {
+ final byte[] key = readValue(dis);
+ final byte[] value = readValue(dis);
+ final MapPutResult putResult = cache.putIfAbsent(ByteBuffer.wrap(key), ByteBuffer.wrap(value));
+ dos.writeBoolean(putResult.isSuccessful());
+ break;
+ }
+ case "put": {
+ final byte[] key = readValue(dis);
+ final byte[] value = readValue(dis);
+ cache.put(ByteBuffer.wrap(key), ByteBuffer.wrap(value));
+ dos.writeBoolean(true);
+ break;
+ }
+ case "containsKey": {
+ final byte[] key = readValue(dis);
+ final boolean contains = cache.containsKey(ByteBuffer.wrap(key));
+ dos.writeBoolean(contains);
+ break;
+ }
+ case "getAndPutIfAbsent": {
+ final byte[] key = readValue(dis);
+ final byte[] value = readValue(dis);
+
+ final MapPutResult putResult = cache.putIfAbsent(ByteBuffer.wrap(key), ByteBuffer.wrap(value));
+ if (putResult.isSuccessful()) {
+ // Put was successful. There was no old value to get.
+ dos.writeInt(0);
+ } else {
+ // we didn't put. Write back the previous value
+ final byte[] byteArray = putResult.getExistingValue().array();
+ dos.writeInt(byteArray.length);
+ dos.write(byteArray);
}
- default: {
- throw new IOException("Illegal Request");
+
+ break;
+ }
+ case "get": {
+ final byte[] key = readValue(dis);
+ final ByteBuffer existingValue = cache.get(ByteBuffer.wrap(key));
+ if (existingValue == null) {
+ // there was no existing value; we did a "put".
+ dos.writeInt(0);
+ } else {
+ // a value already existed. we did not update the map
+ final byte[] byteArray = existingValue.array();
+ dos.writeInt(byteArray.length);
+ dos.write(byteArray);
}
+
+ break;
+ }
+ case "remove": {
+ final byte[] key = readValue(dis);
+ final boolean removed = cache.remove(ByteBuffer.wrap(key)) != null;
+ dos.writeBoolean(removed);
+ break;
+ }
+ default: {
+ throw new IOException("Illegal Request");
+ }
}
} finally {
dos.flush();
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/PersistentMapCache.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/PersistentMapCache.java b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/PersistentMapCache.java
index e821fbf..c2fc0d7 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/PersistentMapCache.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/PersistentMapCache.java
@@ -35,7 +35,6 @@ import org.wali.UpdateType;
import org.wali.WriteAheadRepository;
public class PersistentMapCache implements MapCache {
-
private final MapCache wrapped;
private final WriteAheadRepository<MapWaliRecord> wali;
@@ -80,6 +79,30 @@ public class PersistentMapCache implements MapCache {
}
@Override
+ public MapPutResult put(final ByteBuffer key, final ByteBuffer value) throws IOException {
+ final MapPutResult putResult = wrapped.put(key, value);
+ if ( putResult.isSuccessful() ) {
+ // The put was successful.
+ final MapWaliRecord record = new MapWaliRecord(UpdateType.CREATE, key, value);
+ final List<MapWaliRecord> records = new ArrayList<>();
+ records.add(record);
+
+ if ( putResult.getEvictedKey() != null ) {
+ records.add(new MapWaliRecord(UpdateType.DELETE, putResult.getEvictedKey(), putResult.getEvictedValue()));
+ }
+
+ wali.update(Collections.singletonList(record), false);
+
+ final long modCount = modifications.getAndIncrement();
+ if ( modCount > 0 && modCount % 100000 == 0 ) {
+ wali.checkpoint();
+ }
+ }
+
+ return putResult;
+ }
+
+ @Override
public boolean containsKey(final ByteBuffer key) throws IOException {
return wrapped.containsKey(key);
}
@@ -90,7 +113,7 @@ public class PersistentMapCache implements MapCache {
}
@Override
- public ByteBuffer remove(ByteBuffer key) throws IOException {
+ public ByteBuffer remove(final ByteBuffer key) throws IOException {
final ByteBuffer removeResult = wrapped.remove(key);
if (removeResult != null) {
final MapWaliRecord record = new MapWaliRecord(UpdateType.DELETE, key, removeResult);
@@ -139,7 +162,7 @@ public class PersistentMapCache implements MapCache {
private static class Serde implements SerDe<MapWaliRecord> {
@Override
- public void serializeEdit(MapWaliRecord previousRecordState, MapWaliRecord newRecordState, java.io.DataOutputStream out) throws IOException {
+ public void serializeEdit(final MapWaliRecord previousRecordState, final MapWaliRecord newRecordState, final java.io.DataOutputStream out) throws IOException {
final UpdateType updateType = newRecordState.getUpdateType();
if (updateType == UpdateType.DELETE) {
out.write(0);
@@ -157,7 +180,7 @@ public class PersistentMapCache implements MapCache {
}
@Override
- public void serializeRecord(MapWaliRecord record, java.io.DataOutputStream out) throws IOException {
+ public void serializeRecord(final MapWaliRecord record, final java.io.DataOutputStream out) throws IOException {
serializeEdit(null, record, out);
}
@@ -182,7 +205,7 @@ public class PersistentMapCache implements MapCache {
}
@Override
- public MapWaliRecord deserializeRecord(DataInputStream in, int version) throws IOException {
+ public MapWaliRecord deserializeRecord(final DataInputStream in, final int version) throws IOException {
return deserializeEdit(in, new HashMap<Object, MapWaliRecord>(), version);
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/SimpleMapCache.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/SimpleMapCache.java b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/SimpleMapCache.java
index 9e8bbd1..b167c62 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/SimpleMapCache.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/java/org/apache/nifi/distributed/cache/server/map/SimpleMapCache.java
@@ -107,6 +107,28 @@ public class SimpleMapCache implements MapCache {
}
}
+
+ @Override
+ public MapPutResult put(final ByteBuffer key, final ByteBuffer value) {
+ writeLock.lock();
+ try {
+ // evict if we need to in order to make room for a new entry.
+ final MapCacheRecord evicted = evict();
+
+ final MapCacheRecord record = new MapCacheRecord(key, value);
+ final MapCacheRecord existing = cache.put(key, record);
+ inverseCacheMap.put(record, key);
+
+ final ByteBuffer existingValue = (existing == null) ? null : existing.getValue();
+ final ByteBuffer evictedKey = (evicted == null) ? null : evicted.getKey();
+ final ByteBuffer evictedValue = (evicted == null) ? null : evicted.getValue();
+
+ return new MapPutResult(true, key, value, existingValue, evictedKey, evictedValue);
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
@Override
public boolean containsKey(final ByteBuffer key) {
readLock.lock();
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/resources/docs/org.apache.nifi.distributed.cache.server.map.DistributedMapCacheServer/additionalDetails.html
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/resources/docs/org.apache.nifi.distributed.cache.server.map.DistributedMapCacheServer/additionalDetails.html b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/resources/docs/org.apache.nifi.distributed.cache.server.map.DistributedMapCacheServer/additionalDetails.html
deleted file mode 100644
index 740abec..0000000
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/main/resources/docs/org.apache.nifi.distributed.cache.server.map.DistributedMapCacheServer/additionalDetails.html
+++ /dev/null
@@ -1,46 +0,0 @@
-<!DOCTYPE html>
-<html lang="en">
- <!--
- 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.
- -->
- <head>
- <meta charset="utf-8" />
- <title>Distributed Map Cache Client Service</title>
- <link rel="stylesheet" href="../../css/component-usage.css" type="text/css" />
- </head>
-
- <body>
- <p>
- Below is an example of how to create a distributed map cache server for clients to connect to.
- Note that the identifier in this example is <code>cache-server</code>. If you are using this template
- to create your own DistributedMapCache server, replace the values in this template with values that are
- suitable for your system. Possible options for <code>Port</code>, <code>Maximum Cache Entries</code>,
- <code>Eviction Strategy</code>, <span style="font-style: italic;">SSL Context Service</span>, and
- <span style="font-style: italic;">Persistence Directory</span>
- </p>
-
- <pre>
-<?xml version="1.0" encoding="UTF-8" ?>
-<services>
- <service>
- <identifier>cache-server</identifier>
- <class>org.apache.nifi.distributed.cache.client.DistributedMapCacheServer</class>
- <property name="Port">4557</property>
- <property name="Maximum Cache Entries">10000</property>
- <property name="Eviction Strategy">Least Recently Used</property>
- </service>
-</services>
- </pre>
- </body>
-</html>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/docs/org.apache.nifi.http.StandardHttpContextMap/index.html
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/docs/org.apache.nifi.http.StandardHttpContextMap/index.html b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/docs/org.apache.nifi.http.StandardHttpContextMap/index.html
deleted file mode 100644
index 774c3d9..0000000
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-http-context-map-bundle/nifi-http-context-map/src/main/resources/docs/org.apache.nifi.http.StandardHttpContextMap/index.html
+++ /dev/null
@@ -1,67 +0,0 @@
-<!DOCTYPE html>
-<html lang="en">
- <!--
- 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.
- -->
- <head>
- <meta charset="utf-8" />
- <title>StandardHttpContextMap</title>
-
- <link rel="stylesheet" href="../../css/component-usage.css" type="text/css" />
- </head>
-
- <body>
- <h2>Description:</h2>
- <p>
- This is the standard implementation of the SSL Context Map. This service is used to provide
- coordination between
- <a href="../org.apache.nifi.processors.standard.HandleHttpRequest/index.html">HandleHttpRequest</a>
- and
- <a href="../org.apache.nifi.processors.standard.HandleHttpResponse/index.html">HandleHttpResponse</a>
- Processors.
- </p>
-
- <!-- Service Documentation ================================================== -->
- <h2>Configuring the HTTP Context Map:</h2>
- <p>
- The <code>controller-services.xml</code> file is located in the NiFi <code>conf</code>
- directory. The user may set up any number of controller services within this file.
- </p>
-
- <p>
- This controller service exposes a single property named <code>Maximum Outstanding Requests</code>.
- This property determines the maximum number of HTTP requests that can be outstanding at any one time.
- Any attempt to register an additional HTTP Request will cause an error. The default value is 5000.
- Below is an example of the template for a StandardHttpContextMap controller service.
- </p>
-
- <pre>
-<?xml version="1.0" encoding="UTF-8" ?>
-<services>
- <service>
- <identifier>http-context-map</identifier>
- <class>org.apache.nifi.http.StandardHttpContextMap</class>
- <property name="Maximum Outstanding Requests">5000</property>
- </service>
-</services>
- </pre>
-
- <p>
- <strong>See Also:</strong><br />
- <a href="../org.apache.nifi.processors.standard.HandleHttpRequest/index.html">HandleHttpRequest</a><br />
- <a href="../org.apache.nifi.processors.standard.HandleHttpResponse/index.html">HandleHttpResponse</a><br />
- </p>
-
- </body>
-</html>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/nifi-standard-services/nifi-ssl-context-bundle/nifi-ssl-context-service/src/main/resources/docs/org.apache.nifi.ssl.StandardSSLContextService/additionalDetails.html
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-ssl-context-bundle/nifi-ssl-context-service/src/main/resources/docs/org.apache.nifi.ssl.StandardSSLContextService/additionalDetails.html b/nifi/nifi-nar-bundles/nifi-standard-services/nifi-ssl-context-bundle/nifi-ssl-context-service/src/main/resources/docs/org.apache.nifi.ssl.StandardSSLContextService/additionalDetails.html
deleted file mode 100644
index 525337d..0000000
--- a/nifi/nifi-nar-bundles/nifi-standard-services/nifi-ssl-context-bundle/nifi-ssl-context-service/src/main/resources/docs/org.apache.nifi.ssl.StandardSSLContextService/additionalDetails.html
+++ /dev/null
@@ -1,49 +0,0 @@
-<!DOCTYPE html>
-<html lang="en">
- <!--
- 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.
- -->
- <head>
- <meta charset="utf-8" />
- <title>StandardSSLContextService</title>
-
- <link rel="stylesheet" href="../../css/component-usage.css" type="text/css" />
- </head>
-
- <body>
- <p>
- Below is an example of the template for a SSLContext controller service. Note that the identifier
- in this example is <code>ssl-context</code>. If using this template to create your own SSLContext controller
- service, replace the property values with values that are suitable for your system. Possible options for
- <code>Keystore Type</code> and <code>Truststore Type</code> are <span style="font-style: italic;">JKS</span>
- or <span style="font-style: italic;">PKCS12</span>.
- </p>
-
- <pre>
-<?xml version="1.0" encoding="UTF-8" ?>
-<services>
- <service>
- <identifier>ssl-context</identifier>
- <class>org.apache.nifi.ssl.StandardSSLContextService</class>
- <property name="Keystore Filename">C:/testpki/localtest-ks.jks</property>
- <property name="Keystore Password">localtest</property>
- <property name="Keystore Type">JKS</property>
- <property name="Truststore Filename">C:/testpki/localtest-ts.jks</property>
- <property name="Truststore Password">localtest</property>
- <property name="Truststore Type">JKS</property>
- </service>
-</services>
- </pre>
- </body>
-</html>
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/6f5b6225/nifi/nifi-nar-bundles/pom.xml
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/pom.xml b/nifi/nifi-nar-bundles/pom.xml
index 7d80826..674bb92 100644
--- a/nifi/nifi-nar-bundles/pom.xml
+++ b/nifi/nifi-nar-bundles/pom.xml
@@ -34,6 +34,7 @@
<module>nifi-update-attribute-bundle</module>
<module>nifi-kafka-bundle</module>
<module>nifi-kite-bundle</module>
+ <module>nifi-solr-bundle</module>
<module>nifi-aws-bundle</module>
<module>nifi-social-media-bundle</module>
<module>nifi-geo-bundle</module>