You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by xi...@apache.org on 2017/07/16 11:12:30 UTC
[1/2] storm git commit: STORM-2633: Apply new code style to
storm-sql-redis
Repository: storm
Updated Branches:
refs/heads/master 73c33bf9a -> 2fc414d00
STORM-2633: Apply new code style to storm-sql-redis
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/2dbe80a5
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/2dbe80a5
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/2dbe80a5
Branch: refs/heads/master
Commit: 2dbe80a508c9b17ecf60fdc9256412841f4c0819
Parents: d7c7818
Author: Xin Wang <be...@163.com>
Authored: Sat Jul 15 11:59:55 2017 +0800
Committer: Xin Wang <be...@163.com>
Committed: Sat Jul 15 11:59:55 2017 +0800
----------------------------------------------------------------------
sql/storm-sql-external/storm-sql-redis/pom.xml | 3 -
.../sql/redis/RedisDataSourcesProvider.java | 292 ++++++++++---------
2 files changed, 149 insertions(+), 146 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/2dbe80a5/sql/storm-sql-external/storm-sql-redis/pom.xml
----------------------------------------------------------------------
diff --git a/sql/storm-sql-external/storm-sql-redis/pom.xml b/sql/storm-sql-external/storm-sql-redis/pom.xml
index c2fb1d1..82aa13b 100644
--- a/sql/storm-sql-external/storm-sql-redis/pom.xml
+++ b/sql/storm-sql-external/storm-sql-redis/pom.xml
@@ -77,9 +77,6 @@
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-checkstyle-plugin</artifactId>
<!--Note - the version would be inherited-->
- <configuration>
- <maxAllowedViolations>129</maxAllowedViolations>
- </configuration>
</plugin>
</plugins>
</build>
http://git-wip-us.apache.org/repos/asf/storm/blob/2dbe80a5/sql/storm-sql-external/storm-sql-redis/src/jvm/org/apache/storm/sql/redis/RedisDataSourcesProvider.java
----------------------------------------------------------------------
diff --git a/sql/storm-sql-external/storm-sql-redis/src/jvm/org/apache/storm/sql/redis/RedisDataSourcesProvider.java b/sql/storm-sql-external/storm-sql-redis/src/jvm/org/apache/storm/sql/redis/RedisDataSourcesProvider.java
index 68933b2..b020a15 100644
--- a/sql/storm-sql-external/storm-sql-redis/src/jvm/org/apache/storm/sql/redis/RedisDataSourcesProvider.java
+++ b/sql/storm-sql-external/storm-sql-redis/src/jvm/org/apache/storm/sql/redis/RedisDataSourcesProvider.java
@@ -15,9 +15,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
package org.apache.storm.sql.redis;
import com.google.common.base.Preconditions;
+
+import java.io.Serializable;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+
import org.apache.storm.redis.common.config.JedisClusterConfig;
import org.apache.storm.redis.common.config.JedisPoolConfig;
import org.apache.storm.redis.common.mapper.RedisDataTypeDescription;
@@ -32,197 +41,194 @@ import org.apache.storm.sql.runtime.FieldInfo;
import org.apache.storm.sql.runtime.IOutputSerializer;
import org.apache.storm.sql.runtime.ISqlTridentDataSource;
import org.apache.storm.sql.runtime.SimpleSqlTridentConsumer;
-import org.apache.storm.sql.runtime.utils.SerdeUtils;
import org.apache.storm.sql.runtime.utils.FieldInfoUtils;
+import org.apache.storm.sql.runtime.utils.SerdeUtils;
import org.apache.storm.trident.spout.ITridentDataSource;
import org.apache.storm.trident.state.StateFactory;
import org.apache.storm.trident.state.StateUpdater;
import org.apache.storm.tuple.ITuple;
import redis.clients.util.JedisURIHelper;
-import java.io.Serializable;
-import java.net.InetSocketAddress;
-import java.net.URI;
-import java.util.Collections;
-import java.util.List;
-import java.util.Properties;
-
/**
* Create a Redis sink based on the URI and properties. The URI has the format of
* redis://:[password]@[host]:[port]/[dbIdx]. Only host is mandatory and others can be set to default.
- *
+ * <p/>
* The properties are in JSON format which specifies the config of the Redis data type and etc.
* Please note that when "use.redis.cluster" is "true", cluster discovery is only done from given URI.
*/
public class RedisDataSourcesProvider implements DataSourcesProvider {
- private static final int DEFAULT_REDIS_PORT = 6379;
- private static final int DEFAULT_TIMEOUT = 2000;
+ private static final int DEFAULT_REDIS_PORT = 6379;
+ private static final int DEFAULT_TIMEOUT = 2000;
- private abstract static class AbstractRedisTridentDataSource implements ISqlTridentDataSource, Serializable {
- protected abstract StateFactory newStateFactory();
- protected abstract StateUpdater newStateUpdater(RedisStoreMapper storeMapper);
+ private abstract static class AbstractRedisTridentDataSource implements ISqlTridentDataSource, Serializable {
+ protected abstract StateFactory newStateFactory();
- private final Properties props;
- private final List<FieldInfo> fields;
- private final IOutputSerializer serializer;
+ protected abstract StateUpdater newStateUpdater(RedisStoreMapper storeMapper);
- AbstractRedisTridentDataSource(Properties props, List<FieldInfo> fields, IOutputSerializer serializer) {
- this.props = props;
- this.fields = fields;
- this.serializer = serializer;
- }
+ private final Properties props;
+ private final List<FieldInfo> fields;
+ private final IOutputSerializer serializer;
- @Override
- public ITridentDataSource getProducer() {
- throw new UnsupportedOperationException(this.getClass().getName() + " doesn't provide Producer");
- }
+ AbstractRedisTridentDataSource(Properties props, List<FieldInfo> fields, IOutputSerializer serializer) {
+ this.props = props;
+ this.fields = fields;
+ this.serializer = serializer;
+ }
- @Override
- public SqlTridentConsumer getConsumer() {
- RedisDataTypeDescription dataTypeDescription = getDataTypeDesc(props);
+ @Override
+ public ITridentDataSource getProducer() {
+ throw new UnsupportedOperationException(this.getClass().getName() + " doesn't provide Producer");
+ }
- RedisStoreMapper storeMapper = new TridentRedisStoreMapper(dataTypeDescription, fields, serializer);
+ @Override
+ public SqlTridentConsumer getConsumer() {
+ RedisDataTypeDescription dataTypeDescription = getDataTypeDesc(props);
- StateFactory stateFactory = newStateFactory();
- StateUpdater stateUpdater = newStateUpdater(storeMapper);
+ RedisStoreMapper storeMapper = new TridentRedisStoreMapper(dataTypeDescription, fields, serializer);
- return new SimpleSqlTridentConsumer(stateFactory, stateUpdater);
- }
+ StateFactory stateFactory = newStateFactory();
+ StateUpdater stateUpdater = newStateUpdater(storeMapper);
- private RedisDataTypeDescription getDataTypeDesc(Properties props) {
- Preconditions.checkArgument(props.containsKey("data.type"),
- "Redis data source must contain \"data.type\" config");
+ return new SimpleSqlTridentConsumer(stateFactory, stateUpdater);
+ }
+
+ private RedisDataTypeDescription getDataTypeDesc(Properties props) {
+ Preconditions.checkArgument(props.containsKey("data.type"),
+ "Redis data source must contain \"data.type\" config");
- RedisDataTypeDescription.RedisDataType dataType = RedisDataTypeDescription.RedisDataType.valueOf(props.getProperty("data.type").toUpperCase());
- String additionalKey = props.getProperty("data.additional.key");
+ RedisDataTypeDescription.RedisDataType dataType = RedisDataTypeDescription.RedisDataType.valueOf(
+ props.getProperty("data.type").toUpperCase());
+ String additionalKey = props.getProperty("data.additional.key");
- return new RedisDataTypeDescription(dataType, additionalKey);
+ return new RedisDataTypeDescription(dataType, additionalKey);
+ }
}
- }
- private static class RedisClusterTridentDataSource extends AbstractRedisTridentDataSource {
- private final JedisClusterConfig config;
+ private static class RedisClusterTridentDataSource extends AbstractRedisTridentDataSource {
+ private final JedisClusterConfig config;
- RedisClusterTridentDataSource(JedisClusterConfig config, Properties props, List<FieldInfo> fields, IOutputSerializer serializer) {
- super(props, fields, serializer);
- this.config = config;
- }
+ RedisClusterTridentDataSource(JedisClusterConfig config, Properties props, List<FieldInfo> fields, IOutputSerializer serializer) {
+ super(props, fields, serializer);
+ this.config = config;
+ }
- @Override
- protected StateFactory newStateFactory() {
- return new RedisClusterState.Factory(config);
- }
+ @Override
+ protected StateFactory newStateFactory() {
+ return new RedisClusterState.Factory(config);
+ }
- @Override
- protected StateUpdater newStateUpdater(RedisStoreMapper storeMapper) {
- return new RedisClusterStateUpdater(storeMapper);
+ @Override
+ protected StateUpdater newStateUpdater(RedisStoreMapper storeMapper) {
+ return new RedisClusterStateUpdater(storeMapper);
+ }
}
- }
- private static class RedisTridentDataSource extends AbstractRedisTridentDataSource {
- private final JedisPoolConfig config;
+ private static class RedisTridentDataSource extends AbstractRedisTridentDataSource {
+ private final JedisPoolConfig config;
- RedisTridentDataSource(JedisPoolConfig config, Properties props, List<FieldInfo> fields, IOutputSerializer serializer) {
- super(props, fields, serializer);
- this.config = config;
+ RedisTridentDataSource(JedisPoolConfig config, Properties props, List<FieldInfo> fields, IOutputSerializer serializer) {
+ super(props, fields, serializer);
+ this.config = config;
+ }
+
+ @Override
+ protected StateFactory newStateFactory() {
+ return new RedisState.Factory(config);
+ }
+
+ @Override
+ protected StateUpdater newStateUpdater(RedisStoreMapper storeMapper) {
+ return new RedisStateUpdater(storeMapper);
+ }
}
@Override
- protected StateFactory newStateFactory() {
- return new RedisState.Factory(config);
+ public String scheme() {
+ return "redis";
}
@Override
- protected StateUpdater newStateUpdater(RedisStoreMapper storeMapper) {
- return new RedisStateUpdater(storeMapper);
+ public DataSource construct(URI uri, String inputFormatClass, String outputFormatClass, List<FieldInfo> fields) {
+ throw new UnsupportedOperationException();
}
- }
-
- @Override
- public String scheme() {
- return "redis";
- }
-
- @Override
- public DataSource construct(URI uri, String inputFormatClass, String outputFormatClass, List<FieldInfo> fields) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public ISqlTridentDataSource constructTrident(URI uri, String inputFormatClass, String outputFormatClass, Properties props, List<FieldInfo> fields) {
- Preconditions.checkArgument(JedisURIHelper.isValid(uri), "URI is not valid for Redis: " + uri);
-
- String host = uri.getHost();
- int port = uri.getPort() != -1 ? uri.getPort() : DEFAULT_REDIS_PORT;
- int dbIdx = JedisURIHelper.getDBIndex(uri);
- String password = JedisURIHelper.getPassword(uri);
-
- int timeout = Integer.parseInt(props.getProperty("redis.timeout", String.valueOf(DEFAULT_TIMEOUT)));
-
- boolean clusterMode = Boolean.valueOf(props.getProperty("use.redis.cluster", "false"));
-
- List<String> fieldNames = FieldInfoUtils.getFieldNames(fields);
- IOutputSerializer serializer = SerdeUtils.getSerializer(outputFormatClass, props, fieldNames);
- if (clusterMode) {
- JedisClusterConfig config = new JedisClusterConfig.Builder()
- .setNodes(Collections.singleton(new InetSocketAddress(host, port)))
- .setTimeout(timeout)
- .build();
- return new RedisClusterTridentDataSource(config, props, fields, serializer);
- } else {
- JedisPoolConfig config = new JedisPoolConfig(host, port, timeout, password, dbIdx);
- return new RedisTridentDataSource(config, props, fields, serializer);
+
+ @Override
+ public ISqlTridentDataSource constructTrident(
+ URI uri, String inputFormatClass, String outputFormatClass, Properties props, List<FieldInfo> fields) {
+ Preconditions.checkArgument(JedisURIHelper.isValid(uri), "URI is not valid for Redis: " + uri);
+
+ String host = uri.getHost();
+ int port = uri.getPort() != -1 ? uri.getPort() : DEFAULT_REDIS_PORT;
+ int dbIdx = JedisURIHelper.getDBIndex(uri);
+ String password = JedisURIHelper.getPassword(uri);
+
+ int timeout = Integer.parseInt(props.getProperty("redis.timeout", String.valueOf(DEFAULT_TIMEOUT)));
+
+ boolean clusterMode = Boolean.valueOf(props.getProperty("use.redis.cluster", "false"));
+
+ List<String> fieldNames = FieldInfoUtils.getFieldNames(fields);
+ IOutputSerializer serializer = SerdeUtils.getSerializer(outputFormatClass, props, fieldNames);
+ if (clusterMode) {
+ JedisClusterConfig config = new JedisClusterConfig.Builder()
+ .setNodes(Collections.singleton(new InetSocketAddress(host, port)))
+ .setTimeout(timeout)
+ .build();
+ return new RedisClusterTridentDataSource(config, props, fields, serializer);
+ } else {
+ JedisPoolConfig config = new JedisPoolConfig(host, port, timeout, password, dbIdx);
+ return new RedisTridentDataSource(config, props, fields, serializer);
+ }
}
- }
- private static class TridentRedisStoreMapper implements RedisStoreMapper {
- private final RedisDataTypeDescription dataTypeDescription;
- private final FieldInfo primaryKeyField;
- private final IOutputSerializer outputSerializer;
+ private static class TridentRedisStoreMapper implements RedisStoreMapper {
+ private final RedisDataTypeDescription dataTypeDescription;
+ private final FieldInfo primaryKeyField;
+ private final IOutputSerializer outputSerializer;
- private TridentRedisStoreMapper(RedisDataTypeDescription dataTypeDescription, List<FieldInfo> fields, IOutputSerializer outputSerializer) {
- this.dataTypeDescription = dataTypeDescription;
- this.outputSerializer = outputSerializer;
+ private TridentRedisStoreMapper(RedisDataTypeDescription dataTypeDescription, List<FieldInfo> fields,
+ IOutputSerializer outputSerializer) {
+ this.dataTypeDescription = dataTypeDescription;
+ this.outputSerializer = outputSerializer;
- // find primary key from constructor
- FieldInfo pkField = findPrimaryKeyField(fields);
- Preconditions.checkArgument(pkField != null, "Primary key must be presented to field list");
+ // find primary key from constructor
+ FieldInfo pkField = findPrimaryKeyField(fields);
+ Preconditions.checkArgument(pkField != null, "Primary key must be presented to field list");
- this.primaryKeyField = pkField;
- }
+ this.primaryKeyField = pkField;
+ }
- private FieldInfo findPrimaryKeyField(List<FieldInfo> fields) {
- FieldInfo pkField = null;
- for (FieldInfo field : fields) {
- if (field.isPrimary()) {
- // TODO: this assumes key is only from the one field
- // if not we need to have order of fields in PK
- pkField = field;
- break;
- }
- }
- return pkField;
- }
+ private FieldInfo findPrimaryKeyField(List<FieldInfo> fields) {
+ FieldInfo pkField = null;
+ for (FieldInfo field : fields) {
+ if (field.isPrimary()) {
+ // TODO: this assumes key is only from the one field
+ // if not we need to have order of fields in PK
+ pkField = field;
+ break;
+ }
+ }
+ return pkField;
+ }
- @Override
- public RedisDataTypeDescription getDataTypeDescription() {
- return dataTypeDescription;
- }
+ @Override
+ public RedisDataTypeDescription getDataTypeDescription() {
+ return dataTypeDescription;
+ }
- @Override
- public String getKeyFromTuple(ITuple tuple) {
- String keyFieldName = primaryKeyField.name();
- Object key = tuple.getValueByField(keyFieldName);
- if (key == null) {
- throw new NullPointerException("key field " + keyFieldName + " is null");
- }
- return String.valueOf(key);
- }
+ @Override
+ public String getKeyFromTuple(ITuple tuple) {
+ String keyFieldName = primaryKeyField.name();
+ Object key = tuple.getValueByField(keyFieldName);
+ if (key == null) {
+ throw new NullPointerException("key field " + keyFieldName + " is null");
+ }
+ return String.valueOf(key);
+ }
- @Override
- public String getValueFromTuple(ITuple tuple) {
- byte[] array = outputSerializer.write(tuple.getValues(), null).array();
- return new String(array);
+ @Override
+ public String getValueFromTuple(ITuple tuple) {
+ byte[] array = outputSerializer.write(tuple.getValues(), null).array();
+ return new String(array);
+ }
}
- }
}
[2/2] storm git commit: Merge branch 'STORM-2633' of
https://github.com/vesense/storm
Posted by xi...@apache.org.
Merge branch 'STORM-2633' of https://github.com/vesense/storm
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/2fc414d0
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/2fc414d0
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/2fc414d0
Branch: refs/heads/master
Commit: 2fc414d00317d5d50ba5e648ccfdb4a4b36495f9
Parents: 73c33bf 2dbe80a
Author: Xin Wang <be...@163.com>
Authored: Sun Jul 16 19:12:24 2017 +0800
Committer: Xin Wang <be...@163.com>
Committed: Sun Jul 16 19:12:24 2017 +0800
----------------------------------------------------------------------
sql/storm-sql-external/storm-sql-redis/pom.xml | 3 -
.../sql/redis/RedisDataSourcesProvider.java | 292 ++++++++++---------
2 files changed, 149 insertions(+), 146 deletions(-)
----------------------------------------------------------------------