http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateMapJsonSerDe.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateMapJsonSerDe.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateMapJsonSerDe.java new file mode 100644 index 0000000..170e7ac --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateMapJsonSerDe.java @@ -0,0 +1,85 @@ +/* + * 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.redis.state; + +import com.fasterxml.jackson.core.JsonFactory; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.Map; + +/** + * A RedisStateMapSerDe that uses JSON as the underlying representation. + */ +public class RedisStateMapJsonSerDe implements RedisStateMapSerDe { + + public static final String FIELD_VERSION = "version"; + public static final String FIELD_ENCODING = "encodingVersion"; + public static final String FIELD_STATE_VALUES = "stateValues"; + + private final JsonFactory jsonFactory = new JsonFactory(new ObjectMapper()); + + @Override + public byte[] serialize(final RedisStateMap stateMap) throws IOException { + if (stateMap == null) { + return null; + } + + try (final ByteArrayOutputStream out = new ByteArrayOutputStream()) { + final JsonGenerator jsonGenerator = jsonFactory.createGenerator(out); + jsonGenerator.writeStartObject(); + jsonGenerator.writeNumberField(FIELD_VERSION, stateMap.getVersion()); + jsonGenerator.writeNumberField(FIELD_ENCODING, stateMap.getEncodingVersion()); + + jsonGenerator.writeObjectFieldStart(FIELD_STATE_VALUES); + for (Map.Entry<String,String> entry : stateMap.toMap().entrySet()) { + jsonGenerator.writeStringField(entry.getKey(), entry.getValue()); + } + jsonGenerator.writeEndObject(); + + jsonGenerator.writeEndObject(); + jsonGenerator.flush(); + + return out.toByteArray(); + } + } + + @Override + public RedisStateMap deserialize(final byte[] data) throws IOException { + if (data == null || data.length == 0) { + return null; + } + + final RedisStateMap.Builder builder = new RedisStateMap.Builder(); + + try (final JsonParser jsonParser = jsonFactory.createParser(data)) { + final JsonNode rootNode = jsonParser.readValueAsTree(); + builder.version(rootNode.get(FIELD_VERSION).asLong()); + builder.encodingVersion(rootNode.get(FIELD_ENCODING).asInt()); + + final JsonNode stateValuesNode = rootNode.get(FIELD_STATE_VALUES); + stateValuesNode.fields().forEachRemaining(e -> builder.stateValue(e.getKey(), e.getValue().asText())); + } + + return builder.build(); + } + +}
http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateMapSerDe.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateMapSerDe.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateMapSerDe.java new file mode 100644 index 0000000..ce3067d --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateMapSerDe.java @@ -0,0 +1,44 @@ +/* + * 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.redis.state; + +import java.io.IOException; + +/** + * Provides serialization/deserialization of a RedisStateMap. + */ +public interface RedisStateMapSerDe { + + /** + * Serializes the given RedisStateMap. + * + * @param stateMap the RedisStateMap to serialize + * @return the serialized bytes or null if stateMap is null + * @throws IOException if an error occurs when serializing + */ + byte[] serialize(RedisStateMap stateMap) throws IOException; + + /** + * Deserializes the given bytes to a RedisStateMap. + * + * @param data bytes previously stored by RedisStateProvider + * @return a RedisStateMap or null if data is null or length 0 + * @throws IOException if an error occurs when deserializing + */ + RedisStateMap deserialize(byte[] data) throws IOException; + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java new file mode 100644 index 0000000..c23bb81 --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/state/RedisStateProvider.java @@ -0,0 +1,299 @@ +/* + * 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.redis.state; + +import org.apache.nifi.components.AbstractConfigurableComponent; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.components.state.StateProvider; +import org.apache.nifi.components.state.StateProviderInitializationContext; +import org.apache.nifi.context.PropertyContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.redis.RedisType; +import org.apache.nifi.redis.util.RedisAction; +import org.apache.nifi.redis.util.RedisUtils; +import org.springframework.data.redis.connection.RedisConnection; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; + +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +/** + * A StateProvider backed by Redis. + */ +public class RedisStateProvider extends AbstractConfigurableComponent implements StateProvider { + + static final int ENCODING_VERSION = 1; + + public static final PropertyDescriptor KEY_PREFIX = new PropertyDescriptor.Builder() + .name("Key Prefix") + .displayName("Key Prefix") + .description("The prefix for each key stored by this state provider. When sharing a single Redis across multiple NiFi instances, " + + "setting a unique value for the Key Prefix will make it easier to identify which instances the keys came from.") + .required(true) + .defaultValue("nifi/components/") + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .build(); + + static final List<PropertyDescriptor> STATE_PROVIDER_PROPERTIES; + static { + final List<PropertyDescriptor> props = new ArrayList<>(RedisUtils.REDIS_CONNECTION_PROPERTY_DESCRIPTORS); + props.add(KEY_PREFIX); + STATE_PROVIDER_PROPERTIES = Collections.unmodifiableList(props); + } + + private String identifier; + private String keyPrefix; + private ComponentLog logger; + private PropertyContext context; + + private volatile boolean enabled; + private volatile JedisConnectionFactory connectionFactory; + + private final RedisStateMapSerDe serDe = new RedisStateMapJsonSerDe(); + + @Override + public final void initialize(final StateProviderInitializationContext context) throws IOException { + this.context = context; + this.identifier = context.getIdentifier(); + this.logger = context.getLogger(); + + String keyPrefix = context.getProperty(KEY_PREFIX).getValue(); + if (!keyPrefix.endsWith("/")) { + keyPrefix = keyPrefix + "/"; + } + this.keyPrefix = keyPrefix; + } + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return STATE_PROVIDER_PROPERTIES; + } + + @Override + protected Collection<ValidationResult> customValidate(ValidationContext validationContext) { + final List<ValidationResult> results = new ArrayList<>(RedisUtils.validate(validationContext)); + + final RedisType redisType = RedisType.fromDisplayName(validationContext.getProperty(RedisUtils.REDIS_MODE).getValue()); + if (redisType != null && redisType == RedisType.CLUSTER) { + results.add(new ValidationResult.Builder() + .subject(RedisUtils.REDIS_MODE.getDisplayName()) + .valid(false) + .explanation(RedisUtils.REDIS_MODE.getDisplayName() + + " is configured in clustered mode, and this service requires a non-clustered Redis") + .build()); + } + + return results; + } + + @Override + public String getIdentifier() { + return identifier; + } + + @Override + public void enable() { + enabled = true; + } + + @Override + public void disable() { + enabled = false; + } + + @Override + public boolean isEnabled() { + return enabled; + } + + @Override + public void shutdown() { + if (connectionFactory != null) { + connectionFactory.destroy(); + connectionFactory = null; + } + } + + @Override + public void setState(final Map<String, String> state, final String componentId) throws IOException { + verifyEnabled(); + + final StateMap currStateMap = getState(componentId); + + int attempted = 0; + boolean updated = false; + + while (!updated && attempted < 20) { + updated = replace(currStateMap, state, componentId, true); + attempted++; + } + + if (!updated) { + throw new IOException("Unable to update state due to concurrent modifications"); + } + } + + @Override + public StateMap getState(final String componentId) throws IOException { + return withConnection(redisConnection -> { + final byte[] key = getComponentKey(componentId).getBytes(StandardCharsets.UTF_8); + final byte[] value = redisConnection.get(key); + + final RedisStateMap stateMap = serDe.deserialize(value); + if (stateMap == null) { + return new RedisStateMap.Builder().encodingVersion(ENCODING_VERSION).build(); + } else { + return stateMap; + } + }); + } + + @Override + public boolean replace(final StateMap oldValue, final Map<String, String> newValue, final String componentId) throws IOException { + return replace(oldValue, newValue, componentId, false); + } + + private boolean replace(final StateMap oldValue, final Map<String, String> newValue, final String componentId, final boolean allowReplaceMissing) throws IOException { + return withConnection(redisConnection -> { + + boolean replaced = false; + + // start a watch on the key and retrieve the current value + final byte[] key = getComponentKey(componentId).getBytes(StandardCharsets.UTF_8); + redisConnection.watch(key); + + final long prevVersion = oldValue == null ? -1L : oldValue.getVersion(); + + final byte[] currValue = redisConnection.get(key); + final RedisStateMap currStateMap = serDe.deserialize(currValue); + final long currVersion = currStateMap == null ? -1L : currStateMap.getVersion(); + + // the replace API expects that you can't call replace on a non-existing value, so unwatch and return + if (!allowReplaceMissing && currVersion == -1) { + redisConnection.unwatch(); + return false; + } + + // start a transaction + redisConnection.multi(); + + // compare-and-set + if (prevVersion == currVersion) { + // build the new RedisStateMap incrementing the version, using latest encoding, and using the passed in values + final RedisStateMap newStateMap = new RedisStateMap.Builder() + .version(currVersion + 1) + .encodingVersion(ENCODING_VERSION) + .stateValues(newValue) + .build(); + + // if we use set(k, newVal) then the results list will always have size == 0 b/c when convertPipelineAndTxResults is set to true, + // status responses like "OK" are skipped over, so by using getSet we can rely on the results list to know if the transaction succeeded + redisConnection.getSet(key, serDe.serialize(newStateMap)); + } + + // execute the transaction + final List<Object> results = redisConnection.exec(); + + // if we have a result then the replace succeeded + if (results.size() > 0) { + replaced = true; + } + + return replaced; + }); + } + + @Override + public void clear(final String componentId) throws IOException { + int attempted = 0; + boolean updated = false; + + while (!updated && attempted < 20) { + final StateMap currStateMap = getState(componentId); + updated = replace(currStateMap, Collections.emptyMap(), componentId, true); + + final String result = updated ? "successful" : "unsuccessful"; + logger.debug("Attempt # {} to clear state for component {} was {}", new Object[] { attempted + 1, componentId, result}); + + attempted++; + } + + if (!updated) { + throw new IOException("Unable to update state due to concurrent modifications"); + } + } + + @Override + public void onComponentRemoved(final String componentId) throws IOException { + withConnection(redisConnection -> { + final byte[] key = getComponentKey(componentId).getBytes(StandardCharsets.UTF_8); + redisConnection.del(key); + return true; + }); + } + + @Override + public Scope[] getSupportedScopes() { + return new Scope[] {Scope.CLUSTER}; + } + + private String getComponentKey(final String componentId) { + return keyPrefix + componentId; + } + + private void verifyEnabled() throws IOException { + if (!isEnabled()) { + throw new IOException("Cannot update or retrieve cluster state because node is no longer connected to a cluster."); + } + } + + // visible for testing + synchronized RedisConnection getRedis() { + if (connectionFactory == null) { + connectionFactory = RedisUtils.createConnectionFactory(context, logger); + } + + return connectionFactory.getConnection(); + } + + private <T> T withConnection(final RedisAction<T> action) throws IOException { + RedisConnection redisConnection = null; + try { + redisConnection = getRedis(); + return action.execute(redisConnection); + } finally { + if (redisConnection != null) { + try { + redisConnection.close(); + } catch (Exception e) { + logger.warn("Error closing connection: " + e.getMessage(), e); + } + } + } + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisAction.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisAction.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisAction.java new file mode 100644 index 0000000..ada0c6e --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisAction.java @@ -0,0 +1,30 @@ +/* + * 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.redis.util; + +import org.springframework.data.redis.connection.RedisConnection; + +import java.io.IOException; + +/** + * An action to be executed with a RedisConnection. + */ +public interface RedisAction<T> { + + T execute(RedisConnection redisConnection) throws IOException; + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java new file mode 100644 index 0000000..229c438 --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/java/org/apache/nifi/redis/util/RedisUtils.java @@ -0,0 +1,428 @@ +/* + * 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.redis.util; + +import org.apache.nifi.components.AllowableValue; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.context.PropertyContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.redis.RedisType; +import org.apache.nifi.util.StringUtils; +import org.springframework.data.redis.connection.RedisClusterConfiguration; +import org.springframework.data.redis.connection.RedisSentinelConfiguration; +import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import redis.clients.jedis.JedisPoolConfig; +import redis.clients.jedis.JedisShardInfo; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.concurrent.TimeUnit; + +public class RedisUtils { + + // These properties are shared between the connection pool controller service and the state provider, the name + // is purposely set to be more human-readable since that will be referenced in state-management.xml + + public static final AllowableValue REDIS_MODE_STANDALONE = new AllowableValue(RedisType.STANDALONE.getDisplayName(), RedisType.STANDALONE.getDisplayName(), RedisType.STANDALONE.getDescription()); + public static final AllowableValue REDIS_MODE_SENTINEL = new AllowableValue(RedisType.SENTINEL.getDisplayName(), RedisType.SENTINEL.getDisplayName(), RedisType.SENTINEL.getDescription()); + public static final AllowableValue REDIS_MODE_CLUSTER = new AllowableValue(RedisType.CLUSTER.getDisplayName(), RedisType.CLUSTER.getDisplayName(), RedisType.CLUSTER.getDescription()); + + public static final PropertyDescriptor REDIS_MODE = new PropertyDescriptor.Builder() + .name("Redis Mode") + .displayName("Redis Mode") + .description("The type of Redis being communicated with - standalone, sentinel, or clustered.") + .allowableValues(REDIS_MODE_STANDALONE, REDIS_MODE_SENTINEL, REDIS_MODE_CLUSTER) + .defaultValue(REDIS_MODE_STANDALONE.getValue()) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .required(true) + .build(); + + public static final PropertyDescriptor CONNECTION_STRING = new PropertyDescriptor.Builder() + .name("Connection String") + .displayName("Connection String") + .description("The connection string for Redis. In a standalone instance this value will be of the form hostname:port. " + + "In a sentinel instance this value will be the comma-separated list of sentinels, such as host1:port1,host2:port2,host3:port3. " + + "In a clustered instance this value will be the comma-separated list of cluster masters, such as host1:port,host2:port,host3:port.") + .required(true) + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(true) + .build(); + + public static final PropertyDescriptor DATABASE = new PropertyDescriptor.Builder() + .name("Database Index") + .displayName("Database Index") + .description("The database index to be used by connections created from this connection pool. " + + "See the databases property in redis.conf, by default databases 0-15 will be available.") + .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR) + .defaultValue("0") + .expressionLanguageSupported(true) + .required(true) + .build(); + + public static final PropertyDescriptor COMMUNICATION_TIMEOUT = new PropertyDescriptor.Builder() + .name("Communication Timeout") + .displayName("Communication Timeout") + .description("The timeout to use when attempting to communicate with Redis.") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .defaultValue("10 seconds") + .required(true) + .build(); + + public static final PropertyDescriptor CLUSTER_MAX_REDIRECTS = new PropertyDescriptor.Builder() + .name("Cluster Max Redirects") + .displayName("Cluster Max Redirects") + .description("The maximum number of redirects that can be performed when clustered.") + .addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR) + .defaultValue("5") + .required(true) + .build(); + + public static final PropertyDescriptor SENTINEL_MASTER = new PropertyDescriptor.Builder() + .name("Sentinel Master") + .displayName("Sentinel Master") + .description("The name of the sentinel master, require when Mode is set to Sentinel") + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(true) + .build(); + + public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder() + .name("Password") + .displayName("Password") + .description("The password used to authenticate to the Redis server. See the requirepass property in redis.conf.") + .addValidator(StandardValidators.NON_BLANK_VALIDATOR) + .expressionLanguageSupported(true) + .sensitive(true) + .build(); + + public static final PropertyDescriptor POOL_MAX_TOTAL = new PropertyDescriptor.Builder() + .name("Pool - Max Total") + .displayName("Pool - Max Total") + .description("The maximum number of connections that can be allocated by the pool (checked out to clients, or idle awaiting checkout). " + + "A negative value indicates that there is no limit.") + .addValidator(StandardValidators.INTEGER_VALIDATOR) + .defaultValue("8") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_MAX_IDLE = new PropertyDescriptor.Builder() + .name("Pool - Max Idle") + .displayName("Pool - Max Idle") + .description("The maximum number of idle connections that can be held in the pool, or a negative value if there is no limit.") + .addValidator(StandardValidators.INTEGER_VALIDATOR) + .defaultValue("8") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_MIN_IDLE = new PropertyDescriptor.Builder() + .name("Pool - Min Idle") + .displayName("Pool - Min Idle") + .description("The target for the minimum number of idle connections to maintain in the pool. If the configured value of Min Idle is " + + "greater than the configured value for Max Idle, then the value of Max Idle will be used instead.") + .addValidator(StandardValidators.INTEGER_VALIDATOR) + .defaultValue("0") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_BLOCK_WHEN_EXHAUSTED = new PropertyDescriptor.Builder() + .name("Pool - Block When Exhausted") + .displayName("Pool - Block When Exhausted") + .description("Whether or not clients should block and wait when trying to obtain a connection from the pool when the pool has no available connections. " + + "Setting this to false means an error will occur immediately when a client requests a connection and none are available.") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .allowableValues("true", "false") + .defaultValue("true") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_MAX_WAIT_TIME = new PropertyDescriptor.Builder() + .name("Pool - Max Wait Time") + .displayName("Pool - Max Wait Time") + .description("The amount of time to wait for an available connection when Block When Exhausted is set to true.") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .defaultValue("10 seconds") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_MIN_EVICTABLE_IDLE_TIME = new PropertyDescriptor.Builder() + .name("Pool - Min Evictable Idle Time") + .displayName("Pool - Min Evictable Idle Time") + .description("The minimum amount of time an object may sit idle in the pool before it is eligible for eviction.") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .defaultValue("60 seconds") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_TIME_BETWEEN_EVICTION_RUNS = new PropertyDescriptor.Builder() + .name("Pool - Time Between Eviction Runs") + .displayName("Pool - Time Between Eviction Runs") + .description("The amount of time between attempting to evict idle connections from the pool.") + .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR) + .defaultValue("30 seconds") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_NUM_TESTS_PER_EVICTION_RUN = new PropertyDescriptor.Builder() + .name("Pool - Num Tests Per Eviction Run") + .displayName("Pool - Num Tests Per Eviction Run") + .description("The number of connections to tests per eviction attempt. A negative value indicates to test all connections.") + .addValidator(StandardValidators.INTEGER_VALIDATOR) + .defaultValue("-1") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_TEST_ON_CREATE = new PropertyDescriptor.Builder() + .name("Pool - Test On Create") + .displayName("Pool - Test On Create") + .description("Whether or not connections should be tested upon creation.") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .allowableValues("true", "false") + .defaultValue("false") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_TEST_ON_BORROW = new PropertyDescriptor.Builder() + .name("Pool - Test On Borrow") + .displayName("Pool - Test On Borrow") + .description("Whether or not connections should be tested upon borrowing from the pool.") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .allowableValues("true", "false") + .defaultValue("false") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_TEST_ON_RETURN = new PropertyDescriptor.Builder() + .name("Pool - Test On Return") + .displayName("Pool - Test On Return") + .description("Whether or not connections should be tested upon returning to the pool.") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .allowableValues("true", "false") + .defaultValue("false") + .required(true) + .build(); + + public static final PropertyDescriptor POOL_TEST_WHILE_IDLE = new PropertyDescriptor.Builder() + .name("Pool - Test While Idle") + .displayName("Pool - Test While Idle") + .description("Whether or not connections should be tested while idle.") + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .allowableValues("true", "false") + .defaultValue("true") + .required(true) + .build(); + + public static final List<PropertyDescriptor> REDIS_CONNECTION_PROPERTY_DESCRIPTORS; + static { + final List<PropertyDescriptor> props = new ArrayList<>(); + props.add(RedisUtils.REDIS_MODE); + props.add(RedisUtils.CONNECTION_STRING); + props.add(RedisUtils.DATABASE); + props.add(RedisUtils.COMMUNICATION_TIMEOUT); + props.add(RedisUtils.CLUSTER_MAX_REDIRECTS); + props.add(RedisUtils.SENTINEL_MASTER); + props.add(RedisUtils.PASSWORD); + props.add(RedisUtils.POOL_MAX_TOTAL); + props.add(RedisUtils.POOL_MAX_IDLE); + props.add(RedisUtils.POOL_MIN_IDLE); + props.add(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED); + props.add(RedisUtils.POOL_MAX_WAIT_TIME); + props.add(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME); + props.add(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS); + props.add(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN); + props.add(RedisUtils.POOL_TEST_ON_CREATE); + props.add(RedisUtils.POOL_TEST_ON_BORROW); + props.add(RedisUtils.POOL_TEST_ON_RETURN); + props.add(RedisUtils.POOL_TEST_WHILE_IDLE); + REDIS_CONNECTION_PROPERTY_DESCRIPTORS = Collections.unmodifiableList(props); + } + + + public static JedisConnectionFactory createConnectionFactory(final PropertyContext context, final ComponentLog logger) { + final String redisMode = context.getProperty(RedisUtils.REDIS_MODE).getValue(); + final String connectionString = context.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue(); + final Integer dbIndex = context.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger(); + final String password = context.getProperty(RedisUtils.PASSWORD).evaluateAttributeExpressions().getValue(); + final Integer timeout = context.getProperty(RedisUtils.COMMUNICATION_TIMEOUT).asTimePeriod(TimeUnit.MILLISECONDS).intValue(); + final JedisPoolConfig poolConfig = createJedisPoolConfig(context); + + JedisConnectionFactory connectionFactory; + + if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) { + final JedisShardInfo jedisShardInfo = createJedisShardInfo(connectionString, timeout, password); + + logger.info("Connecting to Redis in standalone mode at " + connectionString); + connectionFactory = new JedisConnectionFactory(jedisShardInfo); + + } else if (RedisUtils.REDIS_MODE_SENTINEL.getValue().equals(redisMode)) { + final String[] sentinels = connectionString.split("[,]"); + final String sentinelMaster = context.getProperty(RedisUtils.SENTINEL_MASTER).evaluateAttributeExpressions().getValue(); + final RedisSentinelConfiguration sentinelConfiguration = new RedisSentinelConfiguration(sentinelMaster, new HashSet<>(getTrimmedValues(sentinels))); + final JedisShardInfo jedisShardInfo = createJedisShardInfo(sentinels[0], timeout, password); + + logger.info("Connecting to Redis in sentinel mode..."); + logger.info("Redis master = " + sentinelMaster); + + for (final String sentinel : sentinels) { + logger.info("Redis sentinel at " + sentinel); + } + + connectionFactory = new JedisConnectionFactory(sentinelConfiguration, poolConfig); + connectionFactory.setShardInfo(jedisShardInfo); + + } else { + final String[] clusterNodes = connectionString.split("[,]"); + final Integer maxRedirects = context.getProperty(RedisUtils.CLUSTER_MAX_REDIRECTS).asInteger(); + + final RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration(getTrimmedValues(clusterNodes)); + clusterConfiguration.setMaxRedirects(maxRedirects); + + logger.info("Connecting to Redis in clustered mode..."); + for (final String clusterNode : clusterNodes) { + logger.info("Redis cluster node at " + clusterNode); + } + + connectionFactory = new JedisConnectionFactory(clusterConfiguration, poolConfig); + } + + connectionFactory.setUsePool(true); + connectionFactory.setPoolConfig(poolConfig); + connectionFactory.setDatabase(dbIndex); + connectionFactory.setTimeout(timeout); + + if (!StringUtils.isBlank(password)) { + connectionFactory.setPassword(password); + } + + // need to call this to initialize the pool/connections + connectionFactory.afterPropertiesSet(); + return connectionFactory; + } + + private static List<String> getTrimmedValues(final String[] values) { + final List<String> trimmedValues = new ArrayList<>(); + for (final String value : values) { + trimmedValues.add(value.trim()); + } + return trimmedValues; + } + + private static JedisShardInfo createJedisShardInfo(final String hostAndPort, final Integer timeout, final String password) { + final String[] hostAndPortSplit = hostAndPort.split("[:]"); + final String host = hostAndPortSplit[0].trim(); + final Integer port = Integer.parseInt(hostAndPortSplit[1].trim()); + + final JedisShardInfo jedisShardInfo = new JedisShardInfo(host, port); + jedisShardInfo.setConnectionTimeout(timeout); + jedisShardInfo.setSoTimeout(timeout); + + if (!StringUtils.isEmpty(password)) { + jedisShardInfo.setPassword(password); + } + + return jedisShardInfo; + } + + private static JedisPoolConfig createJedisPoolConfig(final PropertyContext context) { + final JedisPoolConfig poolConfig = new JedisPoolConfig(); + poolConfig.setMaxTotal(context.getProperty(RedisUtils.POOL_MAX_TOTAL).asInteger()); + poolConfig.setMaxIdle(context.getProperty(RedisUtils.POOL_MAX_IDLE).asInteger()); + poolConfig.setMinIdle(context.getProperty(RedisUtils.POOL_MIN_IDLE).asInteger()); + poolConfig.setBlockWhenExhausted(context.getProperty(RedisUtils.POOL_BLOCK_WHEN_EXHAUSTED).asBoolean()); + poolConfig.setMaxWaitMillis(context.getProperty(RedisUtils.POOL_MAX_WAIT_TIME).asTimePeriod(TimeUnit.MILLISECONDS)); + poolConfig.setMinEvictableIdleTimeMillis(context.getProperty(RedisUtils.POOL_MIN_EVICTABLE_IDLE_TIME).asTimePeriod(TimeUnit.MILLISECONDS)); + poolConfig.setTimeBetweenEvictionRunsMillis(context.getProperty(RedisUtils.POOL_TIME_BETWEEN_EVICTION_RUNS).asTimePeriod(TimeUnit.MILLISECONDS)); + poolConfig.setNumTestsPerEvictionRun(context.getProperty(RedisUtils.POOL_NUM_TESTS_PER_EVICTION_RUN).asInteger()); + poolConfig.setTestOnCreate(context.getProperty(RedisUtils.POOL_TEST_ON_CREATE).asBoolean()); + poolConfig.setTestOnBorrow(context.getProperty(RedisUtils.POOL_TEST_ON_BORROW).asBoolean()); + poolConfig.setTestOnReturn(context.getProperty(RedisUtils.POOL_TEST_ON_RETURN).asBoolean()); + poolConfig.setTestWhileIdle(context.getProperty(RedisUtils.POOL_TEST_WHILE_IDLE).asBoolean()); + return poolConfig; + } + + public static List<ValidationResult> validate(ValidationContext validationContext) { + final List<ValidationResult> results = new ArrayList<>(); + + final String redisMode = validationContext.getProperty(RedisUtils.REDIS_MODE).getValue(); + final String connectionString = validationContext.getProperty(RedisUtils.CONNECTION_STRING).evaluateAttributeExpressions().getValue(); + final Integer dbIndex = validationContext.getProperty(RedisUtils.DATABASE).evaluateAttributeExpressions().asInteger(); + + if (StringUtils.isBlank(connectionString)) { + results.add(new ValidationResult.Builder() + .subject(RedisUtils.CONNECTION_STRING.getDisplayName()) + .valid(false) + .explanation("Connection String cannot be blank") + .build()); + } else if (RedisUtils.REDIS_MODE_STANDALONE.getValue().equals(redisMode)) { + final String[] hostAndPort = connectionString.split("[:]"); + if (hostAndPort == null || hostAndPort.length != 2 || StringUtils.isBlank(hostAndPort[0]) || StringUtils.isBlank(hostAndPort[1]) || !isInteger(hostAndPort[1])) { + results.add(new ValidationResult.Builder() + .subject(RedisUtils.CONNECTION_STRING.getDisplayName()) + .input(connectionString) + .valid(false) + .explanation("Standalone Connection String must be in the form host:port") + .build()); + } + } else { + for (final String connection : connectionString.split("[,]")) { + final String[] hostAndPort = connection.split("[:]"); + if (hostAndPort == null || hostAndPort.length != 2 || StringUtils.isBlank(hostAndPort[0]) || StringUtils.isBlank(hostAndPort[1]) || !isInteger(hostAndPort[1])) { + results.add(new ValidationResult.Builder() + .subject(RedisUtils.CONNECTION_STRING.getDisplayName()) + .input(connection) + .valid(false) + .explanation("Connection String must be in the form host:port,host:port,host:port,etc.") + .build()); + } + } + } + + if (RedisUtils.REDIS_MODE_CLUSTER.getValue().equals(redisMode) && dbIndex > 0) { + results.add(new ValidationResult.Builder() + .subject(RedisUtils.DATABASE.getDisplayName()) + .valid(false) + .explanation("Database Index must be 0 when using clustered Redis") + .build()); + } + + if (RedisUtils.REDIS_MODE_SENTINEL.getValue().equals(redisMode)) { + final String sentinelMaster = validationContext.getProperty(RedisUtils.SENTINEL_MASTER).evaluateAttributeExpressions().getValue(); + if (StringUtils.isEmpty(sentinelMaster)) { + results.add(new ValidationResult.Builder() + .subject(RedisUtils.SENTINEL_MASTER.getDisplayName()) + .valid(false) + .explanation("Sentinel Master must be provided when Mode is Sentinel") + .build()); + } + } + + return results; + } + + private static boolean isInteger(final String number) { + try { + Integer.parseInt(number); + return true; + } catch (Exception e) { + return false; + } + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/resources/META-INF/services/org.apache.nifi.components.state.StateProvider ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/resources/META-INF/services/org.apache.nifi.components.state.StateProvider b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/resources/META-INF/services/org.apache.nifi.components.state.StateProvider new file mode 100644 index 0000000..c7445ab --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/resources/META-INF/services/org.apache.nifi.components.state.StateProvider @@ -0,0 +1,15 @@ +# 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. +org.apache.nifi.redis.state.RedisStateProvider \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService new file mode 100644 index 0000000..5d4073f --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/main/resources/META-INF/services/org.apache.nifi.controller.ControllerService @@ -0,0 +1,16 @@ +# 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. +org.apache.nifi.redis.service.RedisConnectionPoolService +org.apache.nifi.redis.service.RedisDistributedMapCacheClientService \ No newline at end of file http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/FakeRedisProcessor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/FakeRedisProcessor.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/FakeRedisProcessor.java new file mode 100644 index 0000000..7057c90 --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/FakeRedisProcessor.java @@ -0,0 +1,53 @@ +/* + * 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.redis.service; + +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.redis.RedisConnectionPool; + +import java.util.Collections; +import java.util.List; + +/** + * Fake processor used for testing RedisConnectionPoolService. + */ +public class FakeRedisProcessor extends AbstractProcessor { + + public static final PropertyDescriptor REDIS_SERVICE = new PropertyDescriptor.Builder() + .name("redis-service") + .displayName("Redis Service") + .identifiesControllerService(RedisConnectionPool.class) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .required(true) + .build(); + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return Collections.singletonList(REDIS_SERVICE); + } + + @Override + public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException { + + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java new file mode 100644 index 0000000..9d43e67 --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/ITRedisDistributedMapCacheClientService.java @@ -0,0 +1,264 @@ +/* + * 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.redis.service; + +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.distributed.cache.client.AtomicCacheEntry; +import org.apache.nifi.distributed.cache.client.AtomicDistributedMapCacheClient; +import org.apache.nifi.distributed.cache.client.Deserializer; +import org.apache.nifi.distributed.cache.client.Serializer; +import org.apache.nifi.distributed.cache.client.exception.DeserializationException; +import org.apache.nifi.distributed.cache.client.exception.SerializationException; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.AbstractProcessor; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.Relationship; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.redis.util.RedisUtils; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.util.TestRunner; +import org.apache.nifi.util.TestRunners; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import redis.embedded.RedisServer; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.net.StandardSocketOptions; +import java.nio.channels.SocketChannel; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** + * This is an integration test that is meant to be run against a real Redis instance. + */ +public class ITRedisDistributedMapCacheClientService { + + private TestRedisProcessor proc; + private TestRunner testRunner; + private RedisServer redisServer; + private RedisConnectionPoolService redisConnectionPool; + private RedisDistributedMapCacheClientService redisMapCacheClientService; + private int redisPort; + + @Before + public void setup() throws IOException { + this.redisPort = getAvailablePort(); + + this.redisServer = new RedisServer(redisPort); + redisServer.start(); + + proc = new TestRedisProcessor(); + testRunner = TestRunners.newTestRunner(proc); + } + + private int getAvailablePort() throws IOException { + try (SocketChannel socket = SocketChannel.open()) { + socket.setOption(StandardSocketOptions.SO_REUSEADDR, true); + socket.bind(new InetSocketAddress("localhost", 0)); + return socket.socket().getLocalPort(); + } + } + + @After + public void teardown() throws IOException { + if (redisServer != null) { + redisServer.stop(); + } + } + + @Test + public void testStandaloneRedis() throws InitializationException { + try { + // create, configure, and enable the RedisConnectionPool service + redisConnectionPool = new RedisConnectionPoolService(); + testRunner.addControllerService("redis-connection-pool", redisConnectionPool); + testRunner.setProperty(redisConnectionPool, RedisUtils.CONNECTION_STRING, "localhost:" + redisPort); + + // uncomment this to test using a different database index than the default 0 + //testRunner.setProperty(redisConnectionPool, RedisUtils.DATABASE, "1"); + + // uncomment this to test using a password to authenticate to redis + //testRunner.setProperty(redisConnectionPool, RedisUtils.PASSWORD, "foobared"); + + testRunner.enableControllerService(redisConnectionPool); + + setupRedisMapCacheClientService(); + executeProcessor(); + } finally { + if (redisConnectionPool != null) { + redisConnectionPool.onDisabled(); + } + } + } + + private void setupRedisMapCacheClientService() throws InitializationException { + // create, configure, and enable the RedisDistributedMapCacheClient service + redisMapCacheClientService = new RedisDistributedMapCacheClientService(); + testRunner.addControllerService("redis-map-cache-client", redisMapCacheClientService); + testRunner.setProperty(redisMapCacheClientService, RedisDistributedMapCacheClientService.REDIS_CONNECTION_POOL, "redis-connection-pool"); + testRunner.enableControllerService(redisMapCacheClientService); + testRunner.setProperty(TestRedisProcessor.REDIS_MAP_CACHE, "redis-map-cache-client"); + } + + private void executeProcessor() { + // queue a flow file to trigger the processor and executeProcessor it + testRunner.enqueue("trigger"); + testRunner.run(); + testRunner.assertAllFlowFilesTransferred(TestRedisProcessor.REL_SUCCESS, 1); + } + + /** + * Test processor that exercises RedisDistributedMapCacheClient. + */ + private static class TestRedisProcessor extends AbstractProcessor { + + public static final PropertyDescriptor REDIS_MAP_CACHE = new PropertyDescriptor.Builder() + .name("redis-map-cache") + .displayName("Redis Map Cache") + .identifiesControllerService(AtomicDistributedMapCacheClient.class) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .required(true) + .build(); + + public static final Relationship REL_SUCCESS = new Relationship.Builder().name("success").build(); + public static final Relationship REL_FAILURE = new Relationship.Builder().name("failure").build(); + + @Override + protected List<PropertyDescriptor> getSupportedPropertyDescriptors() { + return Collections.singletonList(REDIS_MAP_CACHE); + } + + @Override + public Set<Relationship> getRelationships() { + return new HashSet<>(Arrays.asList(REL_SUCCESS, REL_FAILURE)); + } + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + final FlowFile flowFile = session.get(); + if (flowFile == null) { + return; + } + + final ByteArrayOutputStream out = new ByteArrayOutputStream(); + final Serializer<String> stringSerializer = new StringSerializer(); + final Deserializer<String> stringDeserializer = new StringDeserializer(); + + final AtomicDistributedMapCacheClient cacheClient = context.getProperty(REDIS_MAP_CACHE).asControllerService(AtomicDistributedMapCacheClient.class); + + try { + final long timestamp = System.currentTimeMillis(); + final String key = "test-redis-processor-" + timestamp; + final String value = "the time is " + timestamp; + + // verify the key doesn't exists, put the key/value, then verify it exists + Assert.assertFalse(cacheClient.containsKey(key, stringSerializer)); + cacheClient.put(key, value, stringSerializer, stringSerializer); + Assert.assertTrue(cacheClient.containsKey(key, stringSerializer)); + + // verify get returns the expected value we set above + final String retrievedValue = cacheClient.get(key, stringSerializer, stringDeserializer); + Assert.assertEquals(value, retrievedValue); + + // verify remove removes the entry and contains key returns false after + Assert.assertTrue(cacheClient.remove(key, stringSerializer)); + Assert.assertFalse(cacheClient.containsKey(key, stringSerializer)); + + // verify putIfAbsent works the first time and returns false the second time + Assert.assertTrue(cacheClient.putIfAbsent(key, value, stringSerializer, stringSerializer)); + Assert.assertFalse(cacheClient.putIfAbsent(key, "some other value", stringSerializer, stringSerializer)); + Assert.assertEquals(value, cacheClient.get(key, stringSerializer, stringDeserializer)); + + // verify that getAndPutIfAbsent returns the existing value and doesn't modify it in the cache + final String getAndPutIfAbsentResult = cacheClient.getAndPutIfAbsent(key, value, stringSerializer, stringSerializer, stringDeserializer); + Assert.assertEquals(value, getAndPutIfAbsentResult); + Assert.assertEquals(value, cacheClient.get(key, stringSerializer, stringDeserializer)); + + // verify that getAndPutIfAbsent on a key that doesn't exist returns null + final String keyThatDoesntExist = key + "_DOES_NOT_EXIST"; + Assert.assertFalse(cacheClient.containsKey(keyThatDoesntExist, stringSerializer)); + final String getAndPutIfAbsentResultWhenDoesntExist = cacheClient.getAndPutIfAbsent(keyThatDoesntExist, value, stringSerializer, stringSerializer, stringDeserializer); + Assert.assertEquals(null, getAndPutIfAbsentResultWhenDoesntExist); + Assert.assertEquals(value, cacheClient.get(keyThatDoesntExist, stringSerializer, stringDeserializer)); + + // verify atomic fetch returns the correct entry + final AtomicCacheEntry<String,String,byte[]> entry = cacheClient.fetch(key, stringSerializer, stringDeserializer); + Assert.assertEquals(key, entry.getKey()); + Assert.assertEquals(value, entry.getValue()); + Assert.assertTrue(Arrays.equals(value.getBytes(StandardCharsets.UTF_8), entry.getRevision().orElse(null))); + + final AtomicCacheEntry<String,String,byte[]> notLatestEntry = new AtomicCacheEntry<>(entry.getKey(), entry.getValue(), "not previous".getBytes(StandardCharsets.UTF_8)); + + // verify atomic replace does not replace when previous value is not equal + Assert.assertFalse(cacheClient.replace(notLatestEntry, stringSerializer, stringSerializer)); + Assert.assertEquals(value, cacheClient.get(key, stringSerializer, stringDeserializer)); + + // verify atomic replace does replace when previous value is equal + final String replacementValue = "this value has been replaced"; + entry.setValue(replacementValue); + Assert.assertTrue(cacheClient.replace(entry, stringSerializer, stringSerializer)); + Assert.assertEquals(replacementValue, cacheClient.get(key, stringSerializer, stringDeserializer)); + + // verify atomic replace does replace no value previous existed + final String replaceKeyDoesntExist = key + "_REPLACE_DOES_NOT_EXIST"; + final AtomicCacheEntry<String,String,byte[]> entryDoesNotExist = new AtomicCacheEntry<>(replaceKeyDoesntExist, replacementValue, null); + Assert.assertTrue(cacheClient.replace(entryDoesNotExist, stringSerializer, stringSerializer)); + Assert.assertEquals(replacementValue, cacheClient.get(replaceKeyDoesntExist, stringSerializer, stringDeserializer)); + + final int numToDelete = 2000; + for (int i=0; i < numToDelete; i++) { + cacheClient.put(key + "-" + i, value, stringSerializer, stringSerializer); + } + + Assert.assertTrue(cacheClient.removeByPattern("test-redis-processor-*") >= numToDelete); + Assert.assertFalse(cacheClient.containsKey(key, stringSerializer)); + + session.transfer(flowFile, REL_SUCCESS); + } catch (final Exception e) { + getLogger().error("Routing to failure due to: " + e.getMessage(), e); + session.transfer(flowFile, REL_FAILURE); + } + } + } + + private static class StringSerializer implements Serializer<String> { + @Override + public void serialize(String value, OutputStream output) throws SerializationException, IOException { + if (value != null) { + output.write(value.getBytes(StandardCharsets.UTF_8)); + } + } + } + + private static class StringDeserializer implements Deserializer<String> { + @Override + public String deserialize(byte[] input) throws DeserializationException, IOException { + return input == null ? null : new String(input, StandardCharsets.UTF_8); + } + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/TestRedisConnectionPoolService.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/TestRedisConnectionPoolService.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/TestRedisConnectionPoolService.java new file mode 100644 index 0000000..2017c3b --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/service/TestRedisConnectionPoolService.java @@ -0,0 +1,95 @@ +/* + * 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.redis.service; + +import org.apache.nifi.redis.RedisConnectionPool; +import org.apache.nifi.redis.util.RedisUtils; +import org.apache.nifi.reporting.InitializationException; +import org.apache.nifi.util.TestRunner; +import org.apache.nifi.util.TestRunners; +import org.junit.Before; +import org.junit.Test; + +public class TestRedisConnectionPoolService { + + private TestRunner testRunner; + private FakeRedisProcessor proc; + private RedisConnectionPool redisService; + + @Before + public void setup() throws InitializationException { + proc = new FakeRedisProcessor(); + testRunner = TestRunners.newTestRunner(proc); + + redisService = new RedisConnectionPoolService(); + testRunner.addControllerService("redis-service", redisService); + } + + @Test + public void testValidateConnectionString() { + testRunner.assertNotValid(redisService); + + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, " "); + testRunner.assertNotValid(redisService); + + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, "${redis.connection}"); + testRunner.assertNotValid(redisService); + + testRunner.setVariable("redis.connection", "localhost:6379"); + testRunner.assertValid(redisService); + + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, "localhost"); + testRunner.assertNotValid(redisService); + + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, "localhost:a"); + testRunner.assertNotValid(redisService); + + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, "localhost:6379"); + testRunner.assertValid(redisService); + + // standalone can only have one host:port pair + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, "localhost:6379,localhost:6378"); + testRunner.assertNotValid(redisService); + + // cluster can have multiple host:port pairs + testRunner.setProperty(redisService, RedisUtils.REDIS_MODE, RedisUtils.REDIS_MODE_CLUSTER.getValue()); + testRunner.assertValid(redisService); + + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, "localhost:6379,localhost"); + testRunner.assertNotValid(redisService); + + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, "local:host:6379,localhost:6378"); + testRunner.assertNotValid(redisService); + + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, "localhost:a,localhost:b"); + testRunner.assertNotValid(redisService); + + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, "localhost :6379, localhost :6378, localhost:6377"); + testRunner.assertValid(redisService); + } + + @Test + public void testValidateSentinelMasterRequiredInSentinelMode() { + testRunner.setProperty(redisService, RedisUtils.REDIS_MODE, RedisUtils.REDIS_MODE_SENTINEL.getValue()); + testRunner.setProperty(redisService, RedisUtils.CONNECTION_STRING, "localhost:6379,localhost:6378"); + testRunner.assertNotValid(redisService); + + testRunner.setProperty(redisService, RedisUtils.SENTINEL_MASTER, "mymaster"); + testRunner.assertValid(redisService); + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java new file mode 100644 index 0000000..6a5fb82 --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/ITRedisStateProvider.java @@ -0,0 +1,318 @@ +/* + * 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.redis.state; + +import org.apache.nifi.attribute.expression.language.StandardPropertyValue; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.components.state.StateMap; +import org.apache.nifi.components.state.StateProvider; +import org.apache.nifi.components.state.StateProviderInitializationContext; +import org.apache.nifi.logging.ComponentLog; +import org.apache.nifi.redis.util.RedisUtils; +import org.apache.nifi.util.MockComponentLog; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import redis.embedded.RedisServer; + +import javax.net.ssl.SSLContext; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.StandardSocketOptions; +import java.nio.channels.SocketChannel; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.Map; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +/** + * NOTE: These test cases should be kept in-sync with AbstractTestStateProvider which is in the framework + * and couldn't be extended here. + */ +public class ITRedisStateProvider { + + protected final String componentId = "111111111-1111-1111-1111-111111111111"; + + private RedisServer redisServer; + private RedisStateProvider provider; + + @Before + public void setup() throws Exception { + final int redisPort = getAvailablePort(); + + this.redisServer = new RedisServer(redisPort); + redisServer.start(); + + final Map<PropertyDescriptor, String> properties = new HashMap<>(); + properties.put(RedisUtils.CONNECTION_STRING, "localhost:" + redisPort); + this.provider = createProvider(properties); + } + + @After + public void teardown() throws IOException { + if (provider != null) { + try { + provider.clear(componentId); + } catch (IOException e) { + } + provider.disable(); + provider.shutdown(); + } + + if (redisServer != null) { + redisServer.stop(); + } + } + + public StateProvider getProvider() { + return provider; + } + + @Test + public void testSetAndGet() throws IOException { + getProvider().setState(Collections.singletonMap("testSetAndGet", "value"), componentId); + assertEquals("value", getProvider().getState(componentId).get("testSetAndGet")); + } + + @Test + public void testReplaceSuccessful() throws IOException { + final String key = "testReplaceSuccessful"; + final StateProvider provider = getProvider(); + + StateMap map = provider.getState(componentId); + assertNotNull(map); + assertEquals(-1, map.getVersion()); + + assertNotNull(map.toMap()); + assertTrue(map.toMap().isEmpty()); + provider.setState(Collections.singletonMap(key, "value1"), componentId); + + map = provider.getState(componentId); + assertNotNull(map); + assertEquals(0, map.getVersion()); + assertEquals("value1", map.get(key)); + assertEquals("value1", map.toMap().get(key)); + + final Map<String, String> newMap = new HashMap<>(map.toMap()); + newMap.put(key, "value2"); + assertTrue(provider.replace(map, newMap, componentId)); + + map = provider.getState(componentId); + assertEquals("value2", map.get(key)); + assertEquals(1L, map.getVersion()); + } + + @Test + public void testReplaceWithWrongVersion() throws IOException { + final String key = "testReplaceWithWrongVersion"; + final StateProvider provider = getProvider(); + provider.setState(Collections.singletonMap(key, "value1"), componentId); + + StateMap stateMap = provider.getState(componentId); + assertNotNull(stateMap); + assertEquals("value1", stateMap.get(key)); + assertEquals(0, stateMap.getVersion()); + + provider.setState(Collections.singletonMap(key, "intermediate value"), componentId); + + assertFalse(provider.replace(stateMap, Collections.singletonMap(key, "value2"), componentId)); + stateMap = provider.getState(componentId); + assertEquals(key, stateMap.toMap().keySet().iterator().next()); + assertEquals(1, stateMap.toMap().size()); + assertEquals("intermediate value", stateMap.get(key)); + assertEquals(1, stateMap.getVersion()); + } + + + @Test + public void testToMap() throws IOException { + final String key = "testKeySet"; + final StateProvider provider = getProvider(); + Map<String, String> map = provider.getState(componentId).toMap(); + assertNotNull(map); + assertTrue(map.isEmpty()); + + provider.setState(Collections.singletonMap(key, "value"), componentId); + map = provider.getState(componentId).toMap(); + assertNotNull(map); + assertEquals(1, map.size()); + assertEquals("value", map.get(key)); + + provider.setState(Collections.<String, String> emptyMap(), componentId); + + final StateMap stateMap = provider.getState(componentId); + map = stateMap.toMap(); + assertNotNull(map); + assertTrue(map.isEmpty()); + assertEquals(1, stateMap.getVersion()); + } + + @Test + public void testClear() throws IOException { + final StateProvider provider = getProvider(); + StateMap stateMap = provider.getState(componentId); + assertNotNull(stateMap); + assertEquals(-1L, stateMap.getVersion()); + assertTrue(stateMap.toMap().isEmpty()); + + provider.setState(Collections.singletonMap("testClear", "value"), componentId); + + stateMap = provider.getState(componentId); + assertNotNull(stateMap); + assertEquals(0, stateMap.getVersion()); + assertEquals("value", stateMap.get("testClear")); + + provider.clear(componentId); + + stateMap = provider.getState(componentId); + assertNotNull(stateMap); + assertEquals(1L, stateMap.getVersion()); + assertTrue(stateMap.toMap().isEmpty()); + } + + @Test + public void testReplaceWithNonExistingValue() throws Exception { + final StateProvider provider = getProvider(); + StateMap stateMap = provider.getState(componentId); + assertNotNull(stateMap); + + final Map<String, String> newValue = new HashMap<>(); + newValue.put("value", "value"); + + final boolean replaced = provider.replace(stateMap, newValue, componentId); + assertFalse(replaced); + } + + @Test + public void testReplaceWithNonExistingValueAndVersionGreaterThanNegativeOne() throws Exception { + final StateProvider provider = getProvider(); + final StateMap stateMap = new StateMap() { + @Override + public long getVersion() { + return 4; + } + + @Override + public String get(String key) { + return null; + } + + @Override + public Map<String, String> toMap() { + return Collections.emptyMap(); + } + }; + + final Map<String, String> newValue = new HashMap<>(); + newValue.put("value", "value"); + + final boolean replaced = provider.replace(stateMap, newValue, componentId); + assertFalse(replaced); + } + + @Test + public void testOnComponentRemoved() throws IOException, InterruptedException { + final StateProvider provider = getProvider(); + final Map<String, String> newValue = new HashMap<>(); + newValue.put("value", "value"); + + provider.setState(newValue, componentId); + final StateMap stateMap = provider.getState(componentId); + assertEquals(0L, stateMap.getVersion()); + + provider.onComponentRemoved(componentId); + + // wait for the background process to complete + Thread.sleep(1000L); + + final StateMap stateMapAfterRemoval = provider.getState(componentId); + + // version should be -1 because the state has been removed entirely. + assertEquals(-1L, stateMapAfterRemoval.getVersion()); + } + + + private void initializeProvider(final RedisStateProvider provider, final Map<PropertyDescriptor, String> properties) throws IOException { + provider.initialize(new StateProviderInitializationContext() { + @Override + public String getIdentifier() { + return "Unit Test Provider Initialization Context"; + } + + @Override + public Map<PropertyDescriptor, PropertyValue> getProperties() { + final Map<PropertyDescriptor, PropertyValue> propValueMap = new HashMap<>(); + for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) { + propValueMap.put(entry.getKey(), new StandardPropertyValue(entry.getValue(), null)); + } + return propValueMap; + } + + @Override + public Map<String,String> getAllProperties() { + final Map<String,String> propValueMap = new LinkedHashMap<>(); + for (final Map.Entry<PropertyDescriptor, String> entry : properties.entrySet()) { + propValueMap.put(entry.getKey().getName(), entry.getValue()); + } + return propValueMap; + } + + @Override + public PropertyValue getProperty(final PropertyDescriptor property) { + String prop = properties.get(property); + + if (prop == null) { + prop = property.getDefaultValue(); + } + + return new StandardPropertyValue(prop, null); + } + + @Override + public SSLContext getSSLContext() { + return null; + } + + @Override + public ComponentLog getLogger() { + return new MockComponentLog("Unit Test RedisStateProvider", provider); + } + }); + } + + private RedisStateProvider createProvider(final Map<PropertyDescriptor, String> properties) throws Exception { + final RedisStateProvider provider = new RedisStateProvider(); + initializeProvider(provider, properties); + provider.enable(); + return provider; + } + + private int getAvailablePort() throws IOException { + try (SocketChannel socket = SocketChannel.open()) { + socket.setOption(StandardSocketOptions.SO_REUSEADDR, true); + socket.bind(new InetSocketAddress("localhost", 0)); + return socket.socket().getLocalPort(); + } + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/TestRedisStateMapJsonSerDe.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/TestRedisStateMapJsonSerDe.java b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/TestRedisStateMapJsonSerDe.java new file mode 100644 index 0000000..f147045 --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-extensions/src/test/java/org/apache/nifi/redis/state/TestRedisStateMapJsonSerDe.java @@ -0,0 +1,79 @@ +/* + * 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.redis.state; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.io.IOException; + +public class TestRedisStateMapJsonSerDe { + + private RedisStateMapSerDe serDe; + + @Before + public void setup() { + serDe = new RedisStateMapJsonSerDe(); + } + + @Test + public void testSerializeDeserialize() throws IOException { + final RedisStateMap stateMap = new RedisStateMap.Builder() + .version(2L) + .encodingVersion(3) + .stateValue("field1", "value1") + .stateValue("field2", "value2") + .stateValue("field3", "value3") + .build(); + + final byte[] serialized = serDe.serialize(stateMap); + Assert.assertNotNull(serialized); + + final RedisStateMap deserialized = serDe.deserialize(serialized); + Assert.assertNotNull(deserialized); + Assert.assertEquals(stateMap.getVersion(), deserialized.getVersion()); + Assert.assertEquals(stateMap.getEncodingVersion(), deserialized.getEncodingVersion()); + Assert.assertEquals(stateMap.toMap(), deserialized.toMap()); + } + + @Test + public void testSerializeWhenNull() throws IOException { + Assert.assertNull(serDe.serialize(null)); + } + + @Test + public void testDeserializeWhenNull() throws IOException { + Assert.assertNull(serDe.deserialize(null)); + } + + @Test + public void testDefaultSerialization() throws IOException { + final RedisStateMap stateMap = new RedisStateMap.Builder().build(); + + final byte[] serialized = serDe.serialize(stateMap); + Assert.assertNotNull(serialized); + + final RedisStateMap deserialized = serDe.deserialize(serialized); + Assert.assertNotNull(deserialized); + Assert.assertEquals(RedisStateMap.DEFAULT_VERSION.longValue(), stateMap.getVersion()); + Assert.assertEquals(RedisStateMap.DEFAULT_ENCODING, stateMap.getEncodingVersion()); + Assert.assertNotNull(deserialized.toMap()); + Assert.assertEquals(0, deserialized.toMap().size()); + } + +} http://git-wip-us.apache.org/repos/asf/nifi/blob/aabd4a25/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-nar/pom.xml ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-nar/pom.xml b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-nar/pom.xml new file mode 100644 index 0000000..de1f5d7 --- /dev/null +++ b/nifi-nar-bundles/nifi-redis-bundle/nifi-redis-nar/pom.xml @@ -0,0 +1,46 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-redis-bundle</artifactId> + <version>1.4.0-SNAPSHOT</version> + </parent> + + <artifactId>nifi-redis-nar</artifactId> + <version>1.4.0-SNAPSHOT</version> + <packaging>nar</packaging> + <properties> + <maven.javadoc.skip>true</maven.javadoc.skip> + <source.skip>true</source.skip> + </properties> + + <dependencies> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-redis-extensions</artifactId> + <version>1.4.0-SNAPSHOT</version> + </dependency> + <dependency> + <groupId>org.apache.nifi</groupId> + <artifactId>nifi-redis-service-api-nar</artifactId> + <type>nar</type> + </dependency> + </dependencies> + +</project>
