diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/AbstractRedisBolt.java b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/AbstractRedisBolt.java deleted file mode 100644 index 0c64f43005f..00000000000 --- a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/AbstractRedisBolt.java +++ /dev/null @@ -1,109 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.storm.redis.bolt; - -import org.apache.storm.task.OutputCollector; -import org.apache.storm.task.TopologyContext; -import org.apache.storm.topology.base.BaseRichBolt; -import org.apache.storm.redis.common.config.JedisClusterConfig; -import org.apache.storm.redis.common.config.JedisPoolConfig; -import org.apache.storm.redis.common.container.JedisCommandsContainerBuilder; -import org.apache.storm.redis.common.container.JedisCommandsInstanceContainer; -import redis.clients.jedis.JedisCommands; - -import java.util.Map; - -/** - * AbstractRedisBolt class is for users to implement custom bolts which makes interaction with Redis. - *

- * Due to environment abstraction, AbstractRedisBolt provides JedisCommands which contains only single key operations. - *

- * Custom Bolts may want to follow this pattern: - * 

- * JedisCommands jedisCommands = null;
- * try {
- *     jedisCommand = getInstance();
- *     // do some works
- * } finally {
- *     if (jedisCommand != null) {
- *         returnInstance(jedisCommand);
- *     }
- * }
- *
- * - */ -// TODO: Separate Jedis / JedisCluster to provide full operations for each environment to users -public abstract class AbstractRedisBolt extends BaseRichBolt { - protected OutputCollector collector; - - private transient JedisCommandsInstanceContainer container; - - private JedisPoolConfig jedisPoolConfig; - private JedisClusterConfig jedisClusterConfig; - - /** - * Constructor for single Redis environment (JedisPool) - * @param config configuration for initializing JedisPool - */ - public AbstractRedisBolt(JedisPoolConfig config) { - this.jedisPoolConfig = config; - } - - /** - * Constructor for Redis Cluster environment (JedisCluster) - * @param config configuration for initializing JedisCluster - */ - public AbstractRedisBolt(JedisClusterConfig config) { - this.jedisClusterConfig = config; - } - - /** - * {@inheritDoc} - */ - @Override - public void prepare(Map map, TopologyContext topologyContext, OutputCollector collector) { - // FIXME: stores map (stormConf), topologyContext and expose these to derived classes - this.collector = collector; - - if (jedisPoolConfig != null) { - this.container = JedisCommandsContainerBuilder.build(jedisPoolConfig); - } else if (jedisClusterConfig != null) { - this.container = JedisCommandsContainerBuilder.build(jedisClusterConfig); - } else { - throw new IllegalArgumentException("Jedis configuration not found"); - } - } - - /** - * Borrow JedisCommands instance from container.

- * JedisCommands is an interface which contains single key operations. - * @return implementation of JedisCommands - * @see JedisCommandsInstanceContainer#getInstance() - */ - protected JedisCommands getInstance() { - return this.container.getInstance(); - } - - /** - * Return borrowed instance to container. - * @param instance borrowed object - */ - protected void returnInstance(JedisCommands instance) { - this.container.returnInstance(instance); - } -} diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/BaseLookupBolt.java b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/BaseLookupBolt.java new file mode 100644 index 00000000000..e1fe3d26b89 --- /dev/null +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/BaseLookupBolt.java @@ -0,0 +1,99 @@ +/** + * 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.storm.redis.bolt; + +import com.google.common.base.Preconditions; +import org.apache.storm.redis.common.mapper.RedisDataTypeDescription; +import org.apache.storm.redis.common.mapper.RedisLookupMapper; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.tuple.Tuple; +import org.apache.storm.tuple.Values; +import redis.clients.jedis.JedisCommands; + +import java.util.List; + +public abstract class BaseLookupBolt extends BaseRedisBolt { + protected OutputCollector collector; + + protected final RedisLookupMapper mapper; + protected final RedisDataTypeDescription.RedisDataType dataType; + protected final String additionalKey; + + public BaseLookupBolt(RedisLookupMapper lookupMapper) { + this.mapper = lookupMapper; + Preconditions.checkNotNull(this.mapper, "Mapper is Null"); + + RedisDataTypeDescription dataTypeDescription = lookupMapper.getDataTypeDescription(); + this.dataType = dataTypeDescription.getDataType(); + this.additionalKey = dataTypeDescription.getAdditionalKey(); + } + + @Override + public void execute(Tuple input) { + String key = mapper.getKeyFromTuple(input); + Object lookupValue; + + JedisCommands jedisCommand = getInstance(); + try { + switch (dataType) { + case STRING: + lookupValue = jedisCommand.get(key); + break; + + case LIST: + lookupValue = jedisCommand.lpop(key); + break; + + case HASH: + lookupValue = jedisCommand.hget(additionalKey, key); + break; + + case SET: + lookupValue = jedisCommand.scard(key); + break; + + case SORTED_SET: + lookupValue = jedisCommand.zscore(additionalKey, key); + break; + + case HYPER_LOG_LOG: + lookupValue = jedisCommand.pfcount(key); + break; + + case GEO: + lookupValue = jedisCommand.geopos(additionalKey, key); + break; + + default: + throw new IllegalArgumentException("Cannot process such data type: " + dataType); + } + + List values = mapper.toTuple(input, lookupValue); + for (Values value : values) { + collector.emit(input, value); + } + + collector.ack(input); + } catch (Exception e) { + this.collector.reportError(e); + this.collector.fail(input); + } finally { + returnInstance(jedisCommand); + } + } +} diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/BaseRedisBolt.java b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/BaseRedisBolt.java new file mode 100644 index 00000000000..e6f6a56e64c --- /dev/null +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/BaseRedisBolt.java @@ -0,0 +1,34 @@ +/** + * 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.storm.redis.bolt; + +import org.apache.storm.redis.common.container.JedisCommandsInstanceContainer; +import org.apache.storm.topology.base.BaseRichBolt; +import redis.clients.jedis.JedisCommands; + +public abstract class BaseRedisBolt extends BaseRichBolt { + protected transient JedisCommandsInstanceContainer container; + + protected JedisCommands getInstance() { + return this.container.getInstance(); + } + + protected void returnInstance(JedisCommands instance) { + this.container.returnInstance(instance); + } +} diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/BaseStoreBolt.java b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/BaseStoreBolt.java new file mode 100644 index 00000000000..f30c66c72cd --- /dev/null +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/BaseStoreBolt.java @@ -0,0 +1,100 @@ +/** + * 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.storm.redis.bolt; + +import com.google.common.base.Preconditions; +import org.apache.storm.redis.common.mapper.RedisDataTypeDescription; +import org.apache.storm.redis.common.mapper.RedisStoreMapper; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.tuple.Tuple; +import redis.clients.jedis.JedisCommands; + +public abstract class BaseStoreBolt extends BaseRedisBolt { + protected OutputCollector collector; + + protected final RedisStoreMapper mapper; + protected final RedisDataTypeDescription.RedisDataType dataType; + protected final String additionalKey; + + public BaseStoreBolt(RedisStoreMapper mapper) { + this.mapper = mapper; + Preconditions.checkNotNull(this.mapper, "Mapper is Null"); + + RedisDataTypeDescription dataTypeDescription = mapper.getDataTypeDescription(); + this.dataType = dataTypeDescription.getDataType(); + this.additionalKey = dataTypeDescription.getAdditionalKey(); + } + + @Override + public void execute(Tuple input) { + String key = mapper.getKeyFromTuple(input); + String value = mapper.getValueFromTuple(input); + + JedisCommands jedisCommand = null; + try { + jedisCommand = getInstance(); + + switch (dataType) { + case STRING: + jedisCommand.set(key, value); + break; + + case LIST: + jedisCommand.rpush(key, value); + break; + + case HASH: + jedisCommand.hset(additionalKey, key, value); + break; + + case SET: + jedisCommand.sadd(key, value); + break; + + case SORTED_SET: + jedisCommand.zadd(additionalKey, Double.valueOf(value), key); + break; + + case HYPER_LOG_LOG: + jedisCommand.pfadd(key, value); + break; + + case GEO: + String[] array = value.split(":"); + if (array.length != 2) { + throw new IllegalArgumentException("value structure should be longitude:latitude"); + } + + double longitude = Double.valueOf(array[0]); + double latitude = Double.valueOf(array[1]); + jedisCommand.geoadd(additionalKey, longitude, latitude, key); + break; + + default: + throw new IllegalArgumentException("Cannot process such data type: " + dataType); + } + + collector.ack(input); + } catch (Exception e) { + this.collector.reportError(e); + this.collector.fail(input); + } finally { + returnInstance(jedisCommand); + } + } +} diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisClusterLookupBolt.java b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisClusterLookupBolt.java new file mode 100644 index 00000000000..50d72de8ca5 --- /dev/null +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisClusterLookupBolt.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.storm.redis.bolt; + +import org.apache.storm.redis.common.config.JedisClusterConfig; +import org.apache.storm.redis.common.container.JedisCommandsContainerBuilder; +import org.apache.storm.redis.common.mapper.RedisLookupMapper; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.OutputFieldsDeclarer; + +import java.util.Map; + +public class RedisClusterLookupBolt extends BaseLookupBolt { + + private final JedisClusterConfig config; + + public RedisClusterLookupBolt(JedisClusterConfig config, RedisLookupMapper lookupMapper) { + super(lookupMapper); + this.config = config; + } + + @Override + public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { + this.collector = collector; + + if (this.config != null) { + this.container = JedisCommandsContainerBuilder.buildClusterContainer(config); + } else { + throw new IllegalArgumentException("Jedis configuration not found"); + } + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + mapper.declareOutputFields(declarer); + } +} diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisClusterStoreBolt.java b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisClusterStoreBolt.java new file mode 100644 index 00000000000..6d74fd45ec5 --- /dev/null +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisClusterStoreBolt.java @@ -0,0 +1,52 @@ +/** + * 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.storm.redis.bolt; + +import org.apache.storm.redis.common.config.JedisClusterConfig; +import org.apache.storm.redis.common.container.JedisCommandsContainerBuilder; +import org.apache.storm.redis.common.mapper.RedisStoreMapper; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.OutputFieldsDeclarer; + +import java.util.Map; + +public class RedisClusterStoreBolt extends BaseStoreBolt { + private final JedisClusterConfig config; + + public RedisClusterStoreBolt(JedisClusterConfig config, RedisStoreMapper mapper) { + super(mapper); + this.config = config; + } + + @Override + public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { + this.collector = collector; + + if (this.config != null) { + this.container = JedisCommandsContainerBuilder.buildClusterContainer(config); + } else { + throw new IllegalArgumentException("Jedis configuration not found"); + } + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + + } +} diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisLookupBolt.java b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisLookupBolt.java index 968ade0daf4..e4f4906f38c 100644 --- a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisLookupBolt.java +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisLookupBolt.java @@ -17,121 +17,42 @@ */ package org.apache.storm.redis.bolt; -import org.apache.storm.topology.OutputFieldsDeclarer; -import org.apache.storm.tuple.Tuple; -import org.apache.storm.tuple.Values; -import org.apache.storm.redis.common.mapper.RedisDataTypeDescription; -import org.apache.storm.redis.common.mapper.RedisLookupMapper; -import org.apache.storm.redis.common.config.JedisClusterConfig; import org.apache.storm.redis.common.config.JedisPoolConfig; -import redis.clients.jedis.JedisCommands; +import org.apache.storm.redis.common.container.JedisCommandsContainerBuilder; +import org.apache.storm.redis.common.mapper.RedisLookupMapper; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.OutputFieldsDeclarer; -import java.util.List; +import java.util.Map; /** * Basic bolt for querying from Redis and emits response as tuple. - *

+ *

* Various data types are supported: STRING, LIST, HASH, SET, SORTED_SET, HYPER_LOG_LOG */ -public class RedisLookupBolt extends AbstractRedisBolt { - private final RedisLookupMapper lookupMapper; - private final RedisDataTypeDescription.RedisDataType dataType; - private final String additionalKey; - - /** - * Constructor for single Redis environment (JedisPool) - * @param config configuration for initializing JedisPool - * @param lookupMapper mapper containing which datatype, query key, output key that Bolt uses - */ - public RedisLookupBolt(JedisPoolConfig config, RedisLookupMapper lookupMapper) { - super(config); +public class RedisLookupBolt extends BaseLookupBolt { - this.lookupMapper = lookupMapper; + private final JedisPoolConfig config; - RedisDataTypeDescription dataTypeDescription = lookupMapper.getDataTypeDescription(); - this.dataType = dataTypeDescription.getDataType(); - this.additionalKey = dataTypeDescription.getAdditionalKey(); - } - - /** - * Constructor for Redis Cluster environment (JedisCluster) - * @param config configuration for initializing JedisCluster - * @param lookupMapper mapper containing which datatype, query key, output key that Bolt uses - */ - public RedisLookupBolt(JedisClusterConfig config, RedisLookupMapper lookupMapper) { - super(config); - - this.lookupMapper = lookupMapper; - - RedisDataTypeDescription dataTypeDescription = lookupMapper.getDataTypeDescription(); - this.dataType = dataTypeDescription.getDataType(); - this.additionalKey = dataTypeDescription.getAdditionalKey(); + public RedisLookupBolt(JedisPoolConfig config, RedisLookupMapper lookupMapper) { + super(lookupMapper); + this.config = config; } - /** - * {@inheritDoc} - */ @Override - public void execute(Tuple input) { - String key = lookupMapper.getKeyFromTuple(input); - Object lookupValue; - - JedisCommands jedisCommand = null; - try { - jedisCommand = getInstance(); - - switch (dataType) { - case STRING: - lookupValue = jedisCommand.get(key); - break; - - case LIST: - lookupValue = jedisCommand.lpop(key); - break; - - case HASH: - lookupValue = jedisCommand.hget(additionalKey, key); - break; - - case SET: - lookupValue = jedisCommand.scard(key); - break; - - case SORTED_SET: - lookupValue = jedisCommand.zscore(additionalKey, key); - break; - - case HYPER_LOG_LOG: - lookupValue = jedisCommand.pfcount(key); - break; - - case GEO: - lookupValue = jedisCommand.geopos(additionalKey, key); - break; - - default: - throw new IllegalArgumentException("Cannot process such data type: " + dataType); - } - - List values = lookupMapper.toTuple(input, lookupValue); - for (Values value : values) { - collector.emit(input, value); - } + public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { + this.collector = collector; - collector.ack(input); - } catch (Exception e) { - this.collector.reportError(e); - this.collector.fail(input); - } finally { - returnInstance(jedisCommand); + if (this.config != null) { + this.container = JedisCommandsContainerBuilder.buildContainer(config); + } else { + throw new IllegalArgumentException("Jedis configuration not found"); } } - /** - * {@inheritDoc} - */ @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { - lookupMapper.declareOutputFields(declarer); + mapper.declareOutputFields(declarer); } } diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisStoreBolt.java b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisStoreBolt.java index 00ff2186dad..71faf5169ee 100644 --- a/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisStoreBolt.java +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/bolt/RedisStoreBolt.java @@ -17,117 +17,41 @@ */ package org.apache.storm.redis.bolt; -import org.apache.storm.redis.common.config.JedisClusterConfig; import org.apache.storm.redis.common.config.JedisPoolConfig; -import org.apache.storm.redis.common.mapper.RedisDataTypeDescription; +import org.apache.storm.redis.common.container.JedisCommandsContainerBuilder; import org.apache.storm.redis.common.mapper.RedisStoreMapper; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; -import org.apache.storm.tuple.Tuple; -import redis.clients.jedis.JedisCommands; + +import java.util.Map; /** * Basic bolt for writing to Redis - *

+ *

* Various data types are supported: STRING, LIST, HASH, SET, SORTED_SET, HYPER_LOG_LOG */ -public class RedisStoreBolt extends AbstractRedisBolt { - private final RedisStoreMapper storeMapper; - private final RedisDataTypeDescription.RedisDataType dataType; - private final String additionalKey; - - /** - * Constructor for single Redis environment (JedisPool) - * @param config configuration for initializing JedisPool - * @param storeMapper mapper containing which datatype, storing value's key that Bolt uses - */ - public RedisStoreBolt(JedisPoolConfig config, RedisStoreMapper storeMapper) { - super(config); - this.storeMapper = storeMapper; - - RedisDataTypeDescription dataTypeDescription = storeMapper.getDataTypeDescription(); - this.dataType = dataTypeDescription.getDataType(); - this.additionalKey = dataTypeDescription.getAdditionalKey(); - } +public class RedisStoreBolt extends BaseStoreBolt { + private final JedisPoolConfig config; - /** - * Constructor for Redis Cluster environment (JedisCluster) - * @param config configuration for initializing JedisCluster - * @param storeMapper mapper containing which datatype, storing value's key that Bolt uses - */ - public RedisStoreBolt(JedisClusterConfig config, RedisStoreMapper storeMapper) { - super(config); - this.storeMapper = storeMapper; - - RedisDataTypeDescription dataTypeDescription = storeMapper.getDataTypeDescription(); - this.dataType = dataTypeDescription.getDataType(); - this.additionalKey = dataTypeDescription.getAdditionalKey(); + public RedisStoreBolt(JedisPoolConfig config, RedisStoreMapper mapper) { + super(mapper); + this.config = config; } - /** - * {@inheritDoc} - */ @Override - public void execute(Tuple input) { - String key = storeMapper.getKeyFromTuple(input); - String value = storeMapper.getValueFromTuple(input); - - JedisCommands jedisCommand = null; - try { - jedisCommand = getInstance(); - - switch (dataType) { - case STRING: - jedisCommand.set(key, value); - break; - - case LIST: - jedisCommand.rpush(key, value); - break; + public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { + this.collector = collector; - case HASH: - jedisCommand.hset(additionalKey, key, value); - break; - - case SET: - jedisCommand.sadd(key, value); - break; - - case SORTED_SET: - jedisCommand.zadd(additionalKey, Double.valueOf(value), key); - break; - - case HYPER_LOG_LOG: - jedisCommand.pfadd(key, value); - break; - - case GEO: - String[] array = value.split(":"); - if (array.length != 2) { - throw new IllegalArgumentException("value structure should be longitude:latitude"); - } - - double longitude = Double.valueOf(array[0]); - double latitude = Double.valueOf(array[1]); - jedisCommand.geoadd(additionalKey, longitude, latitude, key); - break; - - default: - throw new IllegalArgumentException("Cannot process such data type: " + dataType); - } - - collector.ack(input); - } catch (Exception e) { - this.collector.reportError(e); - this.collector.fail(input); - } finally { - returnInstance(jedisCommand); + if (this.config != null) { + this.container = JedisCommandsContainerBuilder.buildContainer(config); + } else { + throw new IllegalArgumentException("Jedis configuration not found"); } } - /** - * {@inheritDoc} - */ @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { + } } diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisClusterConfig.java b/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisClusterConfig.java index d8696aaca7a..e9838a597b4 100644 --- a/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisClusterConfig.java +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisClusterConfig.java @@ -21,7 +21,6 @@ import redis.clients.jedis.HostAndPort; import redis.clients.jedis.Protocol; -import java.io.Serializable; import java.net.InetSocketAddress; import java.util.HashSet; import java.util.Set; @@ -29,7 +28,7 @@ /** * Configuration for JedisCluster. */ -public class JedisClusterConfig implements Serializable { +public class JedisClusterConfig implements JedisConfig { private Set nodes; private int timeout; private int maxRedirections; diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisConfig.java b/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisConfig.java new file mode 100644 index 00000000000..0b3aa9bb6a9 --- /dev/null +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisConfig.java @@ -0,0 +1,24 @@ +/** + * 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.storm.redis.common.config; + +import java.io.Serializable; + +public interface JedisConfig extends Serializable { + +} diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisPoolConfig.java b/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisPoolConfig.java index 149bdad078f..c377999d9ab 100644 --- a/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisPoolConfig.java +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/common/config/JedisPoolConfig.java @@ -19,12 +19,10 @@ import redis.clients.jedis.Protocol; -import java.io.Serializable; - /** * Configuration for JedisPool. */ -public class JedisPoolConfig implements Serializable { +public class JedisPoolConfig implements JedisConfig { private String host; private int port; diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/common/container/JedisCommandsContainerBuilder.java b/external/storm-redis/src/main/java/org/apache/storm/redis/common/container/JedisCommandsContainerBuilder.java index 36a74d6c8d6..3d36a8efcb8 100644 --- a/external/storm-redis/src/main/java/org/apache/storm/redis/common/container/JedisCommandsContainerBuilder.java +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/common/container/JedisCommandsContainerBuilder.java @@ -36,7 +36,7 @@ public class JedisCommandsContainerBuilder { * @param config configuration for JedisPool * @return container for single Redis environment */ - public static JedisCommandsInstanceContainer build(JedisPoolConfig config) { + public static JedisContainer buildContainer(JedisPoolConfig config) { JedisPool jedisPool = new JedisPool(DEFAULT_POOL_CONFIG, config.getHost(), config.getPort(), config.getTimeout(), config.getPassword(), config.getDatabase()); return new JedisContainer(jedisPool); } @@ -46,7 +46,7 @@ public static JedisCommandsInstanceContainer build(JedisPoolConfig config) { * @param config configuration for JedisCluster * @return container for Redis Cluster environment */ - public static JedisCommandsInstanceContainer build(JedisClusterConfig config) { + public static JedisClusterContainer buildClusterContainer(JedisClusterConfig config) { JedisCluster jedisCluster = new JedisCluster(config.getNodes(), config.getTimeout(), config.getMaxRedirections(), DEFAULT_POOL_CONFIG); return new JedisClusterContainer(jedisCluster); } diff --git a/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java b/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java index 6d690d4f5b2..30ad24799fc 100644 --- a/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java +++ b/external/storm-redis/src/main/java/org/apache/storm/redis/state/RedisKeyValueState.java @@ -64,7 +64,7 @@ public RedisKeyValueState(String namespace, JedisPoolConfig poolConfig) { } public RedisKeyValueState(String namespace, JedisPoolConfig poolConfig, Serializer keySerializer, Serializer valueSerializer) { - this(namespace, JedisCommandsContainerBuilder.build(poolConfig), keySerializer, valueSerializer); + this(namespace, JedisCommandsContainerBuilder.buildContainer(poolConfig), keySerializer, valueSerializer); } public RedisKeyValueState(String namespace, JedisCommandsInstanceContainer jedisContainer, diff --git a/external/storm-redis/src/test/java/org/apache/storm/redis/state/DefaultStateSerializerTest.java b/external/storm-redis/src/test/java/org/apache/storm/redis/state/DefaultStateSerializerTest.java index 734698974e2..b9d24635f70 100644 --- a/external/storm-redis/src/test/java/org/apache/storm/redis/state/DefaultStateSerializerTest.java +++ b/external/storm-redis/src/test/java/org/apache/storm/redis/state/DefaultStateSerializerTest.java @@ -23,7 +23,6 @@ import org.junit.Test; import java.util.ArrayList; -import java.util.Collections; import java.util.List; import static org.junit.Assert.*; diff --git a/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateProviderTest.java b/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateProviderTest.java index ec6507a13b1..f3f3ed3265f 100644 --- a/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateProviderTest.java +++ b/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateProviderTest.java @@ -18,11 +18,8 @@ package org.apache.storm.redis.state; import org.apache.storm.Config; -import org.apache.storm.state.State; -import org.junit.Assert; import org.junit.Test; -import java.util.Collections; import java.util.HashMap; import java.util.Map; diff --git a/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java b/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java index ea8cc157bd3..8bb63ea81f8 100644 --- a/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java +++ b/external/storm-redis/src/test/java/org/apache/storm/redis/state/RedisKeyValueStateTest.java @@ -25,16 +25,10 @@ import org.mockito.Mockito; import org.mockito.invocation.InvocationOnMock; import org.mockito.stubbing.Answer; -import redis.clients.jedis.BinaryClient; import redis.clients.jedis.JedisCommands; -import redis.clients.jedis.ScanResult; -import redis.clients.jedis.SortingParams; -import redis.clients.jedis.Tuple; import java.util.HashMap; -import java.util.List; import java.util.Map; -import java.util.Set; import static org.junit.Assert.*; diff --git a/external/storm-redis/src/test/java/org/apache/storm/redis/topology/LookupWordCount.java b/external/storm-redis/src/test/java/org/apache/storm/redis/topology/LookupWordCount.java index f62b7b0b11c..b96816069bc 100644 --- a/external/storm-redis/src/test/java/org/apache/storm/redis/topology/LookupWordCount.java +++ b/external/storm-redis/src/test/java/org/apache/storm/redis/topology/LookupWordCount.java @@ -37,7 +37,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Random; diff --git a/external/storm-redis/src/test/java/org/apache/storm/redis/topology/PersistentWordCount.java b/external/storm-redis/src/test/java/org/apache/storm/redis/topology/PersistentWordCount.java index d46bab6f318..908ae7ef5cb 100644 --- a/external/storm-redis/src/test/java/org/apache/storm/redis/topology/PersistentWordCount.java +++ b/external/storm-redis/src/test/java/org/apache/storm/redis/topology/PersistentWordCount.java @@ -20,22 +20,13 @@ import org.apache.storm.Config; import org.apache.storm.LocalCluster; import org.apache.storm.StormSubmitter; -import org.apache.storm.topology.OutputFieldsDeclarer; import org.apache.storm.topology.TopologyBuilder; import org.apache.storm.tuple.Fields; import org.apache.storm.tuple.ITuple; -import org.apache.storm.tuple.Tuple; -import org.apache.storm.redis.bolt.AbstractRedisBolt; import org.apache.storm.redis.bolt.RedisStoreBolt; -import org.apache.storm.redis.common.config.JedisClusterConfig; import org.apache.storm.redis.common.config.JedisPoolConfig; import org.apache.storm.redis.common.mapper.RedisDataTypeDescription; import org.apache.storm.redis.common.mapper.RedisStoreMapper; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import redis.clients.jedis.JedisCommands; -import redis.clients.jedis.exceptions.JedisConnectionException; -import redis.clients.jedis.exceptions.JedisException; public class PersistentWordCount { private static final String WORD_SPOUT = "WORD_SPOUT"; diff --git a/external/storm-redis/src/test/java/org/apache/storm/redis/topology/WordCounter.java b/external/storm-redis/src/test/java/org/apache/storm/redis/topology/WordCounter.java index 6fa930cdfde..2dd95ff4afa 100644 --- a/external/storm-redis/src/test/java/org/apache/storm/redis/topology/WordCounter.java +++ b/external/storm-redis/src/test/java/org/apache/storm/redis/topology/WordCounter.java @@ -28,8 +28,6 @@ import java.util.Map; -import static org.apache.storm.utils.Utils.tuple; - public class WordCounter implements IBasicBolt { private Map wordCounter = Maps.newHashMap();