You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by sb...@apache.org on 2015/11/03 23:55:20 UTC
[2/3] incubator-streams git commit: resolves STREAMS-368
resolves STREAMS-368
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/86f920d2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/86f920d2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/86f920d2
Branch: refs/heads/master
Commit: 86f920d215571c0dcd3ffcc34c0bf69fd7b2041e
Parents: 1470b92
Author: Steve Blackmon (@steveblackmon) <sb...@apache.org>
Authored: Sun Sep 13 16:59:11 2015 -0500
Committer: Steve Blackmon (@steveblackmon) <sb...@apache.org>
Committed: Sun Sep 13 16:59:11 2015 -0500
----------------------------------------------------------------------
.../streams/converter/ActivityConverterProcessor.java | 7 +++++++
.../streams/converter/HoconConverterProcessor.java | 7 +++++++
.../streams/converter/PromoteExtensionsProcessor.java | 7 +++++++
.../apache/streams/converter/TypeConverterProcessor.java | 7 +++++++
.../apache/streams/filters/VerbDefinitionDropFilter.java | 5 +++++
.../apache/streams/filters/VerbDefinitionKeepFilter.java | 5 +++++
.../http/persist/SimpleHTTPPostPersistWriter.java | 4 ++++
.../http/processor/SimpleHTTPGetProcessor.java | 4 ++++
.../http/processor/SimpleHTTPPostProcessor.java | 4 ++++
.../components/http/provider/SimpleHttpProvider.java | 5 +++++
.../streams/amazon/kinesis/KinesisPersistReader.java | 5 +++++
.../streams/amazon/kinesis/KinesisPersistWriter.java | 5 +++++
.../main/java/org/apache/streams/s3/S3PersistReader.java | 5 +++++
.../main/java/org/apache/streams/s3/S3PersistWriter.java | 5 +++++
.../org/apache/streams/console/ConsolePersistReader.java | 6 ++++++
.../org/apache/streams/console/ConsolePersistWriter.java | 7 +++++++
.../elasticsearch/ElasticsearchPersistDeleter.java | 7 +++++++
.../elasticsearch/ElasticsearchPersistReader.java | 6 ++++++
.../elasticsearch/ElasticsearchPersistUpdater.java | 7 +++++++
.../elasticsearch/ElasticsearchPersistWriter.java | 5 +++++
.../processor/DatumFromMetadataAsDocumentProcessor.java | 5 +++++
.../processor/DatumFromMetadataProcessor.java | 5 +++++
.../processor/DocumentToMetadataProcessor.java | 5 +++++
.../processor/MetadataFromDocumentProcessor.java | 5 +++++
.../elasticsearch/processor/PercolateTagProcessor.java | 5 +++++
.../streams/filebuffer/FileBufferPersistReader.java | 7 +++++++
.../streams/filebuffer/FileBufferPersistWriter.java | 7 +++++++
.../java/org/apache/streams/graph/GraphVertexReader.java | 5 +++++
.../org/apache/streams/hbase/HbasePersistWriter.java | 9 ++++++++-
.../org/apache/streams/hdfs/WebHdfsPersistReader.java | 5 +++++
.../org/apache/streams/hdfs/WebHdfsPersistWriter.java | 5 +++++
.../org/apache/streams/kafka/KafkaPersistReader.java | 7 +++++++
.../org/apache/streams/kafka/KafkaPersistWriter.java | 7 +++++++
.../org/apache/streams/mongo/MongoPersistReader.java | 5 +++++
.../org/apache/streams/mongo/MongoPersistWriter.java | 8 ++++++++
.../jackson/CleanAdditionalPropertiesProcessor.java | 7 +++++++
.../apache/streams/jackson/TypeConverterProcessor.java | 7 +++++++
.../java/org/apache/streams/json/JsonPathExtractor.java | 5 +++++
.../java/org/apache/streams/json/JsonPathFilter.java | 5 +++++
.../org/apache/streams/regex/RegexHashtagExtractor.java | 10 ++++++++++
.../org/apache/streams/regex/RegexMentionsExtractor.java | 8 ++++++++
.../java/org/apache/streams/regex/RegexUrlExtractor.java | 7 +++++++
.../org/apache/streams/urls/LinkResolverProcessor.java | 5 +++++
.../processor/DatasiftActivitySerializerProcessor.java | 7 +++++++
.../processor/DatasiftTypeConverterProcessor.java | 7 +++++++
.../streams/datasift/provider/DatasiftPushProvider.java | 7 +++++++
.../datasift/provider/DatasiftStreamProvider.java | 7 +++++++
.../facebook/processor/FacebookTypeConverter.java | 5 +++++
.../facebook/provider/FacebookFriendFeedProvider.java | 6 ++++++
.../facebook/provider/FacebookFriendUpdatesProvider.java | 6 ++++++
.../streams/facebook/provider/FacebookProvider.java | 7 ++++++-
.../provider/FacebookUserInformationProvider.java | 6 ++++++
.../facebook/provider/FacebookUserstreamProvider.java | 6 ++++++
.../java/com/google/gmail/provider/GMailProvider.java | 7 +++++++
.../gplus/processor/GooglePlusCommentProcessor.java | 5 +++++
.../google/gplus/processor/GooglePlusTypeConverter.java | 5 +++++
.../com/google/gplus/provider/AbstractGPlusProvider.java | 9 +++++++--
.../google/gplus/provider/GPlusUserActivityProvider.java | 6 ++++++
.../com/google/gplus/provider/GPlusUserDataProvider.java | 6 ++++++
.../instagram/processor/InstagramTypeConverter.java | 5 +++++
.../instagram/provider/InstagramAbstractProvider.java | 11 +++++++++++
.../apache/streams/data/moreover/MoreoverProvider.java | 5 +++++
.../apache/streams/rss/processor/RssTypeConverter.java | 7 +++++++
.../apache/streams/rss/provider/RssStreamProvider.java | 8 ++++++++
.../streams/sysomos/processor/SysomosTypeConverter.java | 7 +++++++
.../apache/streams/sysomos/provider/SysomosProvider.java | 7 ++++++-
.../processor/FetchAndReplaceTwitterProcessor.java | 5 +++++
.../streams/twitter/processor/TwitterEventProcessor.java | 5 +++++
.../twitter/processor/TwitterProfileProcessor.java | 7 +++++++
.../twitter/processor/TwitterUrlApiProcessor.java | 7 +++++++
.../streams/twitter/provider/TwitterStreamProvider.java | 6 ++++++
.../twitter/provider/TwitterTimelineProvider.java | 6 ++++++
.../twitter/provider/TwitterUserInformationProvider.java | 6 ++++++
.../java/com/youtube/processor/YoutubeTypeConverter.java | 6 ++++++
.../main/java/com/youtube/provider/YoutubeProvider.java | 8 ++++++++
.../java/org/apache/streams/core/StreamsOperation.java | 5 +++++
.../streams/dropwizard/GenericWebhookResource.java | 7 +++++++
.../local/test/processors/DoNothingProcessor.java | 7 +++++++
.../processors/PassthroughDatumCounterProcessor.java | 7 +++++++
.../streams/local/test/processors/SlowProcessor.java | 8 ++++++++
.../local/test/providers/EmptyResultSetProvider.java | 6 ++++++
.../local/test/providers/NumericMessageProvider.java | 5 +++++
.../streams/local/test/writer/DatumCounterWriter.java | 5 +++++
.../streams/local/test/writer/DoNothingWriter.java | 6 ++++++
.../streams/local/test/writer/SystemOutWriter.java | 6 ++++++
.../test/component/ExpectedDatumsPersistWriter.java | 5 +++++
.../streams/test/component/FileReaderProvider.java | 5 +++++
.../apache/streams/pig/test/AppendStringProcessor.java | 7 +++++++
.../org/apache/streams/pig/test/CopyThriceProcessor.java | 7 +++++++
.../org/apache/streams/pig/test/DoNothingProcessor.java | 7 +++++++
90 files changed, 553 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-components/streams-converters/src/main/java/org/apache/streams/converter/ActivityConverterProcessor.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/ActivityConverterProcessor.java b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/ActivityConverterProcessor.java
index be19e8e..0e3e92d 100644
--- a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/ActivityConverterProcessor.java
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/ActivityConverterProcessor.java
@@ -41,6 +41,8 @@ import java.util.List;
*/
public class ActivityConverterProcessor implements StreamsProcessor {
+ public static final String STREAMS_ID = "ActivityConverterProcessor";
+
private final static Logger LOGGER = LoggerFactory.getLogger(ActivityConverterProcessor.class);
private ActivityConverterUtil converterUtil;
@@ -55,6 +57,11 @@ public class ActivityConverterProcessor implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newLinkedList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-components/streams-converters/src/main/java/org/apache/streams/converter/HoconConverterProcessor.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/HoconConverterProcessor.java b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/HoconConverterProcessor.java
index 6ec00b8..b3ee72f 100644
--- a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/HoconConverterProcessor.java
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/HoconConverterProcessor.java
@@ -37,6 +37,8 @@ import java.util.List;
*/
public class HoconConverterProcessor implements StreamsProcessor {
+ public static final String STREAMS_ID = "HoconConverterProcessor";
+
private final static Logger LOGGER = LoggerFactory.getLogger(HoconConverterProcessor.class);
protected Class outClass;
@@ -52,6 +54,11 @@ public class HoconConverterProcessor implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newLinkedList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-components/streams-converters/src/main/java/org/apache/streams/converter/PromoteExtensionsProcessor.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/PromoteExtensionsProcessor.java b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/PromoteExtensionsProcessor.java
index c5be883..2c4fa03 100644
--- a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/PromoteExtensionsProcessor.java
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/PromoteExtensionsProcessor.java
@@ -43,6 +43,8 @@ import java.util.List;
*/
public class PromoteExtensionsProcessor implements StreamsProcessor, Serializable {
+ public static final String STREAMS_ID = "PromoteExtensionsProcessor";
+
private final static Logger LOGGER = LoggerFactory.getLogger(PromoteExtensionsProcessor.class);
private ExtensionUtil extensionUtil;
@@ -52,6 +54,11 @@ public class PromoteExtensionsProcessor implements StreamsProcessor, Serializabl
protected Class outClass;
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newLinkedList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-components/streams-converters/src/main/java/org/apache/streams/converter/TypeConverterProcessor.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/TypeConverterProcessor.java b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/TypeConverterProcessor.java
index edd1907..edd70f4 100644
--- a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/TypeConverterProcessor.java
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/TypeConverterProcessor.java
@@ -40,6 +40,8 @@ import java.util.List;
*/
public class TypeConverterProcessor implements StreamsProcessor, Serializable {
+ public static final String STREAMS_ID = "TypeConverterProcessor";
+
private final static Logger LOGGER = LoggerFactory.getLogger(TypeConverterProcessor.class);
private List<String> formats = Lists.newArrayList();
@@ -58,6 +60,11 @@ public class TypeConverterProcessor implements StreamsProcessor, Serializable {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newLinkedList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionDropFilter.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionDropFilter.java b/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionDropFilter.java
index 93aab59..e0c7a68 100644
--- a/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionDropFilter.java
+++ b/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionDropFilter.java
@@ -56,6 +56,11 @@ public class VerbDefinitionDropFilter implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionKeepFilter.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionKeepFilter.java b/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionKeepFilter.java
index acd89ba..82e8c99 100644
--- a/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionKeepFilter.java
+++ b/streams-components/streams-filters/src/main/java/org/apache/streams/filters/VerbDefinitionKeepFilter.java
@@ -55,6 +55,11 @@ public class VerbDefinitionKeepFilter implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-components/streams-http/src/main/java/org/apache/streams/components/http/persist/SimpleHTTPPostPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-http/src/main/java/org/apache/streams/components/http/persist/SimpleHTTPPostPersistWriter.java b/streams-components/streams-http/src/main/java/org/apache/streams/components/http/persist/SimpleHTTPPostPersistWriter.java
index f9d2103..7e20009 100644
--- a/streams-components/streams-http/src/main/java/org/apache/streams/components/http/persist/SimpleHTTPPostPersistWriter.java
+++ b/streams-components/streams-http/src/main/java/org/apache/streams/components/http/persist/SimpleHTTPPostPersistWriter.java
@@ -72,6 +72,10 @@ public class SimpleHTTPPostPersistWriter implements StreamsPersistWriter {
this.configuration = configuration;
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
@Override
public void write(StreamsDatum entry) {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-components/streams-http/src/main/java/org/apache/streams/components/http/processor/SimpleHTTPGetProcessor.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-http/src/main/java/org/apache/streams/components/http/processor/SimpleHTTPGetProcessor.java b/streams-components/streams-http/src/main/java/org/apache/streams/components/http/processor/SimpleHTTPGetProcessor.java
index 3f5a5c4..0d6747e 100644
--- a/streams-components/streams-http/src/main/java/org/apache/streams/components/http/processor/SimpleHTTPGetProcessor.java
+++ b/streams-components/streams-http/src/main/java/org/apache/streams/components/http/processor/SimpleHTTPGetProcessor.java
@@ -80,6 +80,10 @@ public class SimpleHTTPGetProcessor implements StreamsProcessor {
this.configuration = processorConfiguration;
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
/**
Override this to store a result other than exact json representation of response
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-components/streams-http/src/main/java/org/apache/streams/components/http/processor/SimpleHTTPPostProcessor.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-http/src/main/java/org/apache/streams/components/http/processor/SimpleHTTPPostProcessor.java b/streams-components/streams-http/src/main/java/org/apache/streams/components/http/processor/SimpleHTTPPostProcessor.java
index f0ba2fd..528e50c 100644
--- a/streams-components/streams-http/src/main/java/org/apache/streams/components/http/processor/SimpleHTTPPostProcessor.java
+++ b/streams-components/streams-http/src/main/java/org/apache/streams/components/http/processor/SimpleHTTPPostProcessor.java
@@ -81,6 +81,10 @@ public class SimpleHTTPPostProcessor implements StreamsProcessor {
this.configuration = processorConfiguration;
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
/**
Override this to store a result other than exact json representation of response
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-components/streams-http/src/main/java/org/apache/streams/components/http/provider/SimpleHttpProvider.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-http/src/main/java/org/apache/streams/components/http/provider/SimpleHttpProvider.java b/streams-components/streams-http/src/main/java/org/apache/streams/components/http/provider/SimpleHttpProvider.java
index 1a4a557..46d1cca 100644
--- a/streams-components/streams-http/src/main/java/org/apache/streams/components/http/provider/SimpleHttpProvider.java
+++ b/streams-components/streams-http/src/main/java/org/apache/streams/components/http/provider/SimpleHttpProvider.java
@@ -107,6 +107,11 @@ public class SimpleHttpProvider implements StreamsProvider {
this.configuration = providerConfiguration;
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
/**
Override this to add parameters to the request
*/
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-amazon-aws/streams-persist-kinesis/src/main/java/org/apache/streams/amazon/kinesis/KinesisPersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-amazon-aws/streams-persist-kinesis/src/main/java/org/apache/streams/amazon/kinesis/KinesisPersistReader.java b/streams-contrib/streams-amazon-aws/streams-persist-kinesis/src/main/java/org/apache/streams/amazon/kinesis/KinesisPersistReader.java
index b755579..fc00321 100644
--- a/streams-contrib/streams-amazon-aws/streams-persist-kinesis/src/main/java/org/apache/streams/amazon/kinesis/KinesisPersistReader.java
+++ b/streams-contrib/streams-amazon-aws/streams-persist-kinesis/src/main/java/org/apache/streams/amazon/kinesis/KinesisPersistReader.java
@@ -90,6 +90,11 @@ public class KinesisPersistReader implements StreamsPersistReader, Serializable
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void startStream() {
this.streamNames = this.config.getStreams();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-amazon-aws/streams-persist-kinesis/src/main/java/org/apache/streams/amazon/kinesis/KinesisPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-amazon-aws/streams-persist-kinesis/src/main/java/org/apache/streams/amazon/kinesis/KinesisPersistWriter.java b/streams-contrib/streams-amazon-aws/streams-persist-kinesis/src/main/java/org/apache/streams/amazon/kinesis/KinesisPersistWriter.java
index 2caa028..d262cd4 100644
--- a/streams-contrib/streams-amazon-aws/streams-persist-kinesis/src/main/java/org/apache/streams/amazon/kinesis/KinesisPersistWriter.java
+++ b/streams-contrib/streams-amazon-aws/streams-persist-kinesis/src/main/java/org/apache/streams/amazon/kinesis/KinesisPersistWriter.java
@@ -66,6 +66,11 @@ public class KinesisPersistWriter implements StreamsPersistWriter {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void write(StreamsDatum entry) {
String document = (String) TypeConverterUtil.getInstance().convert(entry.getDocument(), String.class);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-amazon-aws/streams-persist-s3/src/main/java/org/apache/streams/s3/S3PersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-amazon-aws/streams-persist-s3/src/main/java/org/apache/streams/s3/S3PersistReader.java b/streams-contrib/streams-amazon-aws/streams-persist-s3/src/main/java/org/apache/streams/s3/S3PersistReader.java
index 5709f22..5041dd8 100644
--- a/streams-contrib/streams-amazon-aws/streams-persist-s3/src/main/java/org/apache/streams/s3/S3PersistReader.java
+++ b/streams-contrib/streams-amazon-aws/streams-persist-s3/src/main/java/org/apache/streams/s3/S3PersistReader.java
@@ -99,6 +99,11 @@ public class S3PersistReader implements StreamsPersistReader, DatumStatusCountab
this.s3ReaderConfiguration = s3ReaderConfiguration;
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
public void prepare(Object configurationObject) {
// Connect to S3
synchronized (this)
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-amazon-aws/streams-persist-s3/src/main/java/org/apache/streams/s3/S3PersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-amazon-aws/streams-persist-s3/src/main/java/org/apache/streams/s3/S3PersistWriter.java b/streams-contrib/streams-amazon-aws/streams-persist-s3/src/main/java/org/apache/streams/s3/S3PersistWriter.java
index 09f16ff..93cdb65 100644
--- a/streams-contrib/streams-amazon-aws/streams-persist-s3/src/main/java/org/apache/streams/s3/S3PersistWriter.java
+++ b/streams-contrib/streams-amazon-aws/streams-persist-s3/src/main/java/org/apache/streams/s3/S3PersistWriter.java
@@ -112,6 +112,11 @@ public class S3PersistWriter implements StreamsPersistWriter, DatumStatusCountab
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void write(StreamsDatum streamsDatum) {
synchronized (this) {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistReader.java b/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistReader.java
index 8afba85..8793333 100644
--- a/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistReader.java
+++ b/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistReader.java
@@ -55,6 +55,12 @@ public class ConsolePersistReader implements StreamsPersistReader {
this();
this.inputStream = inputStream;
}
+
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
public void prepare(Object o) {
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistWriter.java b/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistWriter.java
index 53bb8d7..6d284ba 100644
--- a/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistWriter.java
+++ b/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistWriter.java
@@ -33,6 +33,8 @@ import java.util.concurrent.ConcurrentLinkedQueue;
public class ConsolePersistWriter implements StreamsPersistWriter {
+ private final static String STREAMS_ID = "ConsolePersistWriter";
+
private static final Logger LOGGER = LoggerFactory.getLogger(ConsolePersistWriter.class);
protected PrintStream printStream = System.out;
@@ -50,6 +52,11 @@ public class ConsolePersistWriter implements StreamsPersistWriter {
this.printStream = printStream;
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
public void prepare(Object o) {
Preconditions.checkNotNull(persistQueue);
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistDeleter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistDeleter.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistDeleter.java
index 9604ccc..af754ad 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistDeleter.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistDeleter.java
@@ -29,6 +29,8 @@ import java.util.Map;
public class ElasticsearchPersistDeleter extends ElasticsearchPersistWriter implements StreamsPersistWriter {
+ public static final String STREAMS_ID = ElasticsearchPersistDeleter.class.getCanonicalName();
+
private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchPersistDeleter.class);
public ElasticsearchPersistDeleter() {
@@ -40,6 +42,11 @@ public class ElasticsearchPersistDeleter extends ElasticsearchPersistWriter impl
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void write(StreamsDatum streamsDatum) {
if(streamsDatum == null || streamsDatum.getDocument() == null)
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistReader.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistReader.java
index 3a2f8fb..9103614 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistReader.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistReader.java
@@ -45,6 +45,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
*/
public class ElasticsearchPersistReader implements StreamsPersistReader, Serializable {
+
public static final String STREAMS_ID = "ElasticsearchPersistReader";
private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchPersistReader.class);
@@ -65,6 +66,11 @@ public class ElasticsearchPersistReader implements StreamsPersistReader, Seriali
this.config = config;
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
//PersistReader methods
@Override
public void startStream() {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistUpdater.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistUpdater.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistUpdater.java
index 8ab55ca..f712248 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistUpdater.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistUpdater.java
@@ -30,6 +30,8 @@ import java.util.Map;
public class ElasticsearchPersistUpdater extends ElasticsearchPersistWriter implements StreamsPersistWriter {
+ public static final String STREAMS_ID = ElasticsearchPersistUpdater.class.getCanonicalName();
+
private final static Logger LOGGER = LoggerFactory.getLogger(ElasticsearchPersistUpdater.class);
public ElasticsearchPersistUpdater() {
@@ -41,6 +43,11 @@ public class ElasticsearchPersistUpdater extends ElasticsearchPersistWriter impl
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void write(StreamsDatum streamsDatum) {
if(streamsDatum == null || streamsDatum.getDocument() == null)
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
index 549920b..faa4d1f 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/ElasticsearchPersistWriter.java
@@ -135,6 +135,11 @@ public class ElasticsearchPersistWriter implements StreamsPersistWriter, DatumSt
public boolean isConnected() { return (this.manager.getClient() != null); }
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void write(StreamsDatum streamsDatum) {
if(streamsDatum == null || streamsDatum.getDocument() == null)
return;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataAsDocumentProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataAsDocumentProcessor.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataAsDocumentProcessor.java
index 79d0f4a..355c471 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataAsDocumentProcessor.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataAsDocumentProcessor.java
@@ -69,6 +69,11 @@ public class DatumFromMetadataAsDocumentProcessor implements StreamsProcessor, S
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataProcessor.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataProcessor.java
index 2faf80b..512e191 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataProcessor.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DatumFromMetadataProcessor.java
@@ -59,6 +59,11 @@ public class DatumFromMetadataProcessor implements StreamsProcessor, Serializabl
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DocumentToMetadataProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DocumentToMetadataProcessor.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DocumentToMetadataProcessor.java
index bac5ba6..41cc12d 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DocumentToMetadataProcessor.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/DocumentToMetadataProcessor.java
@@ -61,6 +61,11 @@ public class DocumentToMetadataProcessor implements StreamsProcessor, Serializab
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/MetadataFromDocumentProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/MetadataFromDocumentProcessor.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/MetadataFromDocumentProcessor.java
index 7a9135d..aba9000 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/MetadataFromDocumentProcessor.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/MetadataFromDocumentProcessor.java
@@ -54,6 +54,11 @@ public class MetadataFromDocumentProcessor implements StreamsProcessor, Serializ
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/PercolateTagProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/PercolateTagProcessor.java b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/PercolateTagProcessor.java
index d1de6f9..850fb15 100644
--- a/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/PercolateTagProcessor.java
+++ b/streams-contrib/streams-persist-elasticsearch/src/main/java/org/apache/streams/elasticsearch/processor/PercolateTagProcessor.java
@@ -112,6 +112,11 @@ public class PercolateTagProcessor implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-filebuffer/src/main/java/org/apache/streams/filebuffer/FileBufferPersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-filebuffer/src/main/java/org/apache/streams/filebuffer/FileBufferPersistReader.java b/streams-contrib/streams-persist-filebuffer/src/main/java/org/apache/streams/filebuffer/FileBufferPersistReader.java
index 8aee1cb..bdf950c 100644
--- a/streams-contrib/streams-persist-filebuffer/src/main/java/org/apache/streams/filebuffer/FileBufferPersistReader.java
+++ b/streams-contrib/streams-persist-filebuffer/src/main/java/org/apache/streams/filebuffer/FileBufferPersistReader.java
@@ -48,6 +48,8 @@ import java.util.concurrent.Executors;
*/
public class FileBufferPersistReader implements StreamsPersistReader, Serializable {
+ public final static String STREAMS_ID = "FileBufferPersistReader";
+
private static final Logger LOGGER = LoggerFactory.getLogger(FileBufferPersistReader.class);
protected volatile Queue<StreamsDatum> persistQueue;
@@ -72,6 +74,11 @@ public class FileBufferPersistReader implements StreamsPersistReader, Serializab
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public StreamsResultSet readAll() {
return readCurrent();
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-filebuffer/src/main/java/org/apache/streams/filebuffer/FileBufferPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-filebuffer/src/main/java/org/apache/streams/filebuffer/FileBufferPersistWriter.java b/streams-contrib/streams-persist-filebuffer/src/main/java/org/apache/streams/filebuffer/FileBufferPersistWriter.java
index a6db4e0..89c45c8 100644
--- a/streams-contrib/streams-persist-filebuffer/src/main/java/org/apache/streams/filebuffer/FileBufferPersistWriter.java
+++ b/streams-contrib/streams-persist-filebuffer/src/main/java/org/apache/streams/filebuffer/FileBufferPersistWriter.java
@@ -41,6 +41,8 @@ import java.util.concurrent.ConcurrentLinkedQueue;
*/
public class FileBufferPersistWriter implements StreamsPersistWriter, Serializable {
+ public final static String STREAMS_ID = "FileBufferPersistWriter";
+
private static final Logger LOGGER = LoggerFactory.getLogger(FileBufferPersistWriter.class);
protected volatile Queue<StreamsDatum> persistQueue;
@@ -60,6 +62,11 @@ public class FileBufferPersistWriter implements StreamsPersistWriter, Serializab
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void write(StreamsDatum entry) {
String key = entry.getId() != null ? entry.getId() : GuidUtils.generateGuid("filewriter");
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-graph/src/main/java/org/apache/streams/graph/GraphVertexReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-graph/src/main/java/org/apache/streams/graph/GraphVertexReader.java b/streams-contrib/streams-persist-graph/src/main/java/org/apache/streams/graph/GraphVertexReader.java
index 93041d8..731159f 100644
--- a/streams-contrib/streams-persist-graph/src/main/java/org/apache/streams/graph/GraphVertexReader.java
+++ b/streams-contrib/streams-persist-graph/src/main/java/org/apache/streams/graph/GraphVertexReader.java
@@ -97,6 +97,11 @@ public class GraphVertexReader extends SimpleHttpProvider implements StreamsPers
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void prepare(Object configurationObject) {
super.prepare(configurationObject);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-hbase/src/main/java/org/apache/streams/hbase/HbasePersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hbase/src/main/java/org/apache/streams/hbase/HbasePersistWriter.java b/streams-contrib/streams-persist-hbase/src/main/java/org/apache/streams/hbase/HbasePersistWriter.java
index e99582d..4a51fab 100644
--- a/streams-contrib/streams-persist-hbase/src/main/java/org/apache/streams/hbase/HbasePersistWriter.java
+++ b/streams-contrib/streams-persist-hbase/src/main/java/org/apache/streams/hbase/HbasePersistWriter.java
@@ -37,6 +37,8 @@ import java.util.concurrent.ConcurrentLinkedQueue;
public class HbasePersistWriter implements StreamsPersistWriter, Flushable, Closeable
{
+ public final static String STREAMS_ID = "HbasePersistWriter";
+
private final static Logger LOGGER = LoggerFactory.getLogger(HbasePersistWriter.class);
protected HConnection connection;
@@ -111,7 +113,12 @@ public class HbasePersistWriter implements StreamsPersistWriter, Flushable, Clos
return;
}
}
-
+
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
@Override
public void write(StreamsDatum streamsDatum) {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReader.java b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReader.java
index 72f3157..8f7331e 100644
--- a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReader.java
+++ b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistReader.java
@@ -161,6 +161,11 @@ public class WebHdfsPersistReader implements StreamsPersistReader, DatumStatusCo
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void prepare(Object configurationObject) {
LOGGER.debug("Prepare");
connectToWebHDFS();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
index 33598c2..acf492f 100644
--- a/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
+++ b/streams-contrib/streams-persist-hdfs/src/main/java/org/apache/streams/hdfs/WebHdfsPersistWriter.java
@@ -153,6 +153,11 @@ public class WebHdfsPersistWriter implements StreamsPersistWriter, Flushable, Cl
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void write(StreamsDatum streamsDatum) {
synchronized (this) {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java b/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java
index a7810b1..7d31d41 100644
--- a/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java
+++ b/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistReader.java
@@ -49,6 +49,8 @@ import java.util.concurrent.TimeUnit;
public class KafkaPersistReader implements StreamsPersistReader, Serializable {
+ public final static String STREAMS_ID = "KafkaPersistReader";
+
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPersistReader.class);
protected volatile Queue<StreamsDatum> persistQueue;
@@ -81,6 +83,11 @@ public class KafkaPersistReader implements StreamsPersistReader, Serializable {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void startStream() {
Properties props = new Properties();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistWriter.java b/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistWriter.java
index c5f029a..0f685c5 100644
--- a/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistWriter.java
+++ b/streams-contrib/streams-persist-kafka/src/main/java/org/apache/streams/kafka/KafkaPersistWriter.java
@@ -38,6 +38,8 @@ import java.util.concurrent.ConcurrentLinkedQueue;
public class KafkaPersistWriter implements StreamsPersistWriter, Serializable, Runnable {
+ public final static String STREAMS_ID = "KafkaPersistWriter";
+
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaPersistWriter.class);
protected volatile Queue<StreamsDatum> persistQueue;
@@ -92,6 +94,11 @@ public class KafkaPersistWriter implements StreamsPersistWriter, Serializable, R
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void write(StreamsDatum entry) {
try {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java
index fb1fe94..ab03f22 100644
--- a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java
+++ b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistReader.java
@@ -105,6 +105,11 @@ public class MongoPersistReader implements StreamsPersistReader {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void prepare(Object configurationObject) {
connectToMongo();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
index 16bf02a..edd8ce5 100644
--- a/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
+++ b/streams-contrib/streams-persist-mongo/src/main/java/org/apache/streams/mongo/MongoPersistWriter.java
@@ -49,7 +49,10 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
public class MongoPersistWriter implements StreamsPersistWriter, Runnable {
+ public final static String STREAMS_ID = "MongoPersistWriter";
+
private final static Logger LOGGER = LoggerFactory.getLogger(MongoPersistWriter.class);
+
private final static long MAX_WRITE_LATENCY = 1000;
protected volatile Queue<StreamsDatum> persistQueue;
@@ -85,6 +88,11 @@ public class MongoPersistWriter implements StreamsPersistWriter, Runnable {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void write(StreamsDatum streamsDatum) {
DBObject dbObject = prepareObject(streamsDatum);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/CleanAdditionalPropertiesProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/CleanAdditionalPropertiesProcessor.java b/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/CleanAdditionalPropertiesProcessor.java
index c5e9a4b..6c08eb1 100644
--- a/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/CleanAdditionalPropertiesProcessor.java
+++ b/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/CleanAdditionalPropertiesProcessor.java
@@ -42,11 +42,18 @@ import java.util.Map;
*/
public class CleanAdditionalPropertiesProcessor implements StreamsProcessor {
+ public static final String STREAMS_ID = "CleanAdditionalPropertiesProcessor";
+
private static final Logger LOGGER = LoggerFactory.getLogger(CleanAdditionalPropertiesProcessor.class);
private ObjectMapper mapper;
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum datum) {
List<StreamsDatum> result = Lists.newLinkedList();
ObjectNode activity = this.mapper.convertValue(datum.getDocument(), ObjectNode.class);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/TypeConverterProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/TypeConverterProcessor.java b/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/TypeConverterProcessor.java
index 4466117..454f99e 100644
--- a/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/TypeConverterProcessor.java
+++ b/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/TypeConverterProcessor.java
@@ -36,6 +36,8 @@ import java.util.List;
*/
public class TypeConverterProcessor implements StreamsProcessor {
+ public static final String STREAMS_ID = "TypeConverterProcessor";
+
private final static Logger LOGGER = LoggerFactory.getLogger(TypeConverterProcessor.class);
private List<String> formats = Lists.newArrayList();
@@ -63,6 +65,11 @@ public class TypeConverterProcessor implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newLinkedList();
Object inDoc = entry.getDocument();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java b/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
index 3aa5216..24288f1 100644
--- a/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
+++ b/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
@@ -61,6 +61,11 @@ public class JsonPathExtractor implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathFilter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathFilter.java b/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathFilter.java
index 55dbda4..fcf34d7 100644
--- a/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathFilter.java
+++ b/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathFilter.java
@@ -67,6 +67,11 @@ public class JsonPathFilter implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexHashtagExtractor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexHashtagExtractor.java b/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexHashtagExtractor.java
index 1e565c8..a59193b 100644
--- a/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexHashtagExtractor.java
+++ b/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexHashtagExtractor.java
@@ -20,6 +20,8 @@
package org.apache.streams.regex;
import org.apache.streams.core.StreamsProcessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Processes the content of an {@link org.apache.streams.pojo.json.Activity} object to extract the Hashtags and add
@@ -27,6 +29,10 @@ import org.apache.streams.core.StreamsProcessor;
*/
public class RegexHashtagExtractor extends AbstractRegexExtensionExtractor<String> implements StreamsProcessor{
+ private final static String STREAMS_ID = "RegexHashtagExtractor";
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(RegexHashtagExtractor.class);
+
public final static String DEFAULT_PATTERN = "#\\w+";
public final static String PATTERN_CONFIG_KEY = "HashtagPattern";
public final static String EXTENSION_KEY = "hashtags";
@@ -35,6 +41,10 @@ public class RegexHashtagExtractor extends AbstractRegexExtensionExtractor<Strin
super(PATTERN_CONFIG_KEY, EXTENSION_KEY, DEFAULT_PATTERN);
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
@Override
protected String prepareObject(String extracted) {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexMentionsExtractor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexMentionsExtractor.java b/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexMentionsExtractor.java
index c1b662d..eabdb04 100644
--- a/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexMentionsExtractor.java
+++ b/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexMentionsExtractor.java
@@ -29,6 +29,9 @@ import java.util.Map;
* them to the appropriate extensions object
*/
public class RegexMentionsExtractor extends AbstractRegexExtensionExtractor<Map<String, Object>> implements StreamsProcessor {
+
+ private final static String STREAMS_ID = "RegexMentionsExtractor";
+
public static final String DEFAULT_PATTERN = "@\\w+";
public static final String PATTERN_CONFIG_KEY = "MentionPattern";
public static final String EXTENSION_KEY = "user_mentions";
@@ -39,6 +42,11 @@ public class RegexMentionsExtractor extends AbstractRegexExtensionExtractor<Map<
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
protected Map<String, Object> prepareObject(String extracted) {
HashMap<String, Object> mention = Maps.newHashMap();
mention.put(DISPLAY_KEY, extracted.substring(1));
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexUrlExtractor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexUrlExtractor.java b/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexUrlExtractor.java
index 5d37b3a..84d3257 100644
--- a/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexUrlExtractor.java
+++ b/streams-contrib/streams-processor-regex/src/main/java/org/apache/streams/regex/RegexUrlExtractor.java
@@ -30,6 +30,13 @@ import java.util.Collection;
*/
public class RegexUrlExtractor extends AbstractRegexExtensionExtractor<String> implements StreamsProcessor {
+ private final static String STREAMS_ID = "RegexUrlExtractor";
+
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
//Temporarily copied from streams-processor-urls so as not to force a dependency on that provider. This should
//be moved to a common utility package
public final static String DEFAULT_PATTERN =
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkResolverProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkResolverProcessor.java b/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkResolverProcessor.java
index 0b4cd7c..203b414 100644
--- a/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkResolverProcessor.java
+++ b/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkResolverProcessor.java
@@ -36,6 +36,11 @@ public class LinkResolverProcessor implements StreamsProcessor {
private static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newArrayList();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/processor/DatasiftActivitySerializerProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/processor/DatasiftActivitySerializerProcessor.java b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/processor/DatasiftActivitySerializerProcessor.java
index 1f5a83a..43b16b2 100644
--- a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/processor/DatasiftActivitySerializerProcessor.java
+++ b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/processor/DatasiftActivitySerializerProcessor.java
@@ -39,6 +39,8 @@ import java.util.List;
*/
public class DatasiftActivitySerializerProcessor implements StreamsProcessor {
+ private final static String STREAMS_ID = "DatasiftActivitySerializerProcessor";
+
private final static Logger LOGGER = LoggerFactory.getLogger(DatasiftActivitySerializerProcessor.class);
private ObjectMapper mapper;
@@ -52,6 +54,11 @@ public class DatasiftActivitySerializerProcessor implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newLinkedList();
Activity activity;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/processor/DatasiftTypeConverterProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/processor/DatasiftTypeConverterProcessor.java b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/processor/DatasiftTypeConverterProcessor.java
index 018c07b..dcffd1a 100644
--- a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/processor/DatasiftTypeConverterProcessor.java
+++ b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/processor/DatasiftTypeConverterProcessor.java
@@ -40,6 +40,8 @@ import java.util.List;
*/
public class DatasiftTypeConverterProcessor implements StreamsProcessor {
+ private final static String STREAMS_ID = "RegexUrlExtractor";
+
private final static Logger LOGGER = LoggerFactory.getLogger(DatasiftTypeConverterProcessor.class);
private ObjectMapper mapper;
@@ -54,6 +56,11 @@ public class DatasiftTypeConverterProcessor implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
List<StreamsDatum> result = Lists.newLinkedList();
Object doc;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftPushProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftPushProvider.java b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftPushProvider.java
index db4a5a7..bdd2c97 100644
--- a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftPushProvider.java
+++ b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftPushProvider.java
@@ -71,6 +71,8 @@ import java.util.regex.Pattern;
@Consumes(MediaType.APPLICATION_JSON)
public class DatasiftPushProvider implements StreamsProvider {
+ private final static String STREAMS_ID = "DatasiftPushProvider";
+
private final static Logger LOGGER = LoggerFactory.getLogger(DatasiftPushProvider.class);
private static ObjectMapper mapper = StreamsJacksonMapper.getInstance();
@@ -81,6 +83,11 @@ public class DatasiftPushProvider implements StreamsProvider {
private static Pattern newLinePattern = Pattern.compile("(\\r\\n?|\\n)", Pattern.MULTILINE);
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
@POST
@Path("json")
public Response json(@Context HttpHeaders headers,
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftStreamProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftStreamProvider.java b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftStreamProvider.java
index 558ae8a..fbd5e52 100644
--- a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftStreamProvider.java
+++ b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/provider/DatasiftStreamProvider.java
@@ -53,6 +53,8 @@ import java.util.concurrent.ConcurrentLinkedQueue;
*/
public class DatasiftStreamProvider implements StreamsProvider {
+ private final static String STREAMS_ID = "DatasiftStreamProvider";
+
private final static Logger LOGGER = LoggerFactory.getLogger(DatasiftStreamProvider.class);
private DatasiftConfiguration config;
@@ -90,6 +92,11 @@ public class DatasiftStreamProvider implements StreamsProvider {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void startStream() {
Preconditions.checkNotNull(this.config);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
index 98e7a91..762b6c0 100644
--- a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
@@ -134,6 +134,11 @@ public class FacebookTypeConverter implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
StreamsDatum result = null;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookFriendFeedProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookFriendFeedProvider.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookFriendFeedProvider.java
index 2e577e4..231ee4f 100644
--- a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookFriendFeedProvider.java
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookFriendFeedProvider.java
@@ -55,6 +55,7 @@ public class FacebookFriendFeedProvider implements StreamsProvider, Serializable
{
public static final String STREAMS_ID = "FacebookFriendFeedProvider";
+
private static final Logger LOGGER = LoggerFactory.getLogger(FacebookFriendFeedProvider.class);
private static final ObjectMapper mapper = StreamsJacksonMapper.getInstance();
@@ -126,6 +127,11 @@ public class FacebookFriendFeedProvider implements StreamsProvider, Serializable
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void startStream() {
shutdownAndAwaitTermination(executor);
running.set(true);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookFriendUpdatesProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookFriendUpdatesProvider.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookFriendUpdatesProvider.java
index bc10f9b..cda868e 100644
--- a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookFriendUpdatesProvider.java
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookFriendUpdatesProvider.java
@@ -56,6 +56,7 @@ public class FacebookFriendUpdatesProvider implements StreamsProvider, Serializa
{
public static final String STREAMS_ID = "FacebookFriendPostsProvider";
+
private static final Logger LOGGER = LoggerFactory.getLogger(FacebookFriendUpdatesProvider.class);
private static final ObjectMapper mapper = StreamsJacksonMapper.getInstance();
@@ -127,6 +128,11 @@ public class FacebookFriendUpdatesProvider implements StreamsProvider, Serializa
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void startStream() {
running.set(true);
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookProvider.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookProvider.java
index 2d60b3d..563e0e9 100644
--- a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookProvider.java
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookProvider.java
@@ -46,6 +46,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
*/
public abstract class FacebookProvider implements StreamsProvider {
+ private final static String STREAMS_ID = "FacebookProvider";
+
private static final Logger LOGGER = LoggerFactory.getLogger(FacebookProvider.class);
private static final ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
private static final int MAX_BATCH_SIZE = 2000;
@@ -57,7 +59,6 @@ public abstract class FacebookProvider implements StreamsProvider {
private ExecutorService executor;
private FacebookDataCollector dataCollector;
-
public FacebookProvider() {
try {
this.configuration = MAPPER.readValue(StreamsConfigurator.config.getConfig("facebook").root().render(ConfigRenderOptions.concise()), FacebookConfiguration.class);
@@ -70,6 +71,10 @@ public abstract class FacebookProvider implements StreamsProvider {
this.configuration = (FacebookConfiguration) SerializationUtil.cloneBySerialization(configuration);
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
@Override
public void startStream() {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookUserInformationProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookUserInformationProvider.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookUserInformationProvider.java
index 9649627..1262106 100644
--- a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookUserInformationProvider.java
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookUserInformationProvider.java
@@ -51,6 +51,7 @@ public class FacebookUserInformationProvider implements StreamsProvider, Seriali
{
public static final String STREAMS_ID = "FacebookUserInformationProvider";
+
private static final Logger LOGGER = LoggerFactory.getLogger(FacebookUserInformationProvider.class);
private static final ObjectMapper mapper = StreamsJacksonMapper.getInstance();
@@ -117,6 +118,11 @@ public class FacebookUserInformationProvider implements StreamsProvider, Seriali
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void startStream() {
running.set(true);
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookUserstreamProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookUserstreamProvider.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookUserstreamProvider.java
index fc324a1..0f2121a 100644
--- a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookUserstreamProvider.java
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/FacebookUserstreamProvider.java
@@ -59,6 +59,7 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
public class FacebookUserstreamProvider implements StreamsProvider, Serializable {
public static final String STREAMS_ID = "FacebookUserstreamProvider";
+
private static final Logger LOGGER = LoggerFactory.getLogger(FacebookUserstreamProvider.class);
private static final ObjectMapper mapper = StreamsJacksonMapper.getInstance();
@@ -134,6 +135,11 @@ public class FacebookUserstreamProvider implements StreamsProvider, Serializable
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void startStream() {
client = getFacebookClient();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-google/google-gmail/src/main/java/com/google/gmail/provider/GMailProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-google/google-gmail/src/main/java/com/google/gmail/provider/GMailProvider.java b/streams-contrib/streams-provider-google/google-gmail/src/main/java/com/google/gmail/provider/GMailProvider.java
index 19a6407..5872e27 100644
--- a/streams-contrib/streams-provider-google/google-gmail/src/main/java/com/google/gmail/provider/GMailProvider.java
+++ b/streams-contrib/streams-provider-google/google-gmail/src/main/java/com/google/gmail/provider/GMailProvider.java
@@ -46,6 +46,8 @@ import java.util.concurrent.*;
*/
public class GMailProvider implements StreamsProvider, DatumStatusCountable {
+ public final static String STREAMS_ID = "GMailProvider";
+
private final static Logger LOGGER = LoggerFactory.getLogger(GMailProvider.class);
private GMailConfiguration config;
@@ -104,6 +106,11 @@ public class GMailProvider implements StreamsProvider, DatumStatusCountable {
protected DatumStatusCounter countersCurrent = new DatumStatusCounter();
@Override
+ public String getId() {
+ return "GMailProvider";
+ }
+
+ @Override
public void startStream() {
task = executor.submit(new GMailImapProviderTask(this));
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/processor/GooglePlusCommentProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/processor/GooglePlusCommentProcessor.java b/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/processor/GooglePlusCommentProcessor.java
index 583c741..d095b32 100644
--- a/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/processor/GooglePlusCommentProcessor.java
+++ b/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/processor/GooglePlusCommentProcessor.java
@@ -39,6 +39,11 @@ public class GooglePlusCommentProcessor implements StreamsProcessor {
private int count;
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
StreamsDatum result = null;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/processor/GooglePlusTypeConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/processor/GooglePlusTypeConverter.java b/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/processor/GooglePlusTypeConverter.java
index 0ec1eb8..d44a487 100644
--- a/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/processor/GooglePlusTypeConverter.java
+++ b/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/processor/GooglePlusTypeConverter.java
@@ -59,6 +59,11 @@ public class GooglePlusTypeConverter implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
StreamsDatum result = null;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/AbstractGPlusProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/AbstractGPlusProvider.java b/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/AbstractGPlusProvider.java
index 791afe4..44e1b03 100644
--- a/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/AbstractGPlusProvider.java
+++ b/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/AbstractGPlusProvider.java
@@ -63,6 +63,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
*/
public abstract class AbstractGPlusProvider implements StreamsProvider {
+ public final static String STREAMS_ID = "AbstractGPlusProvider";
+
private final static Logger LOGGER = LoggerFactory.getLogger(AbstractGPlusProvider.class);
private final static Set<String> SCOPE = new HashSet<String>() {{ add("https://www.googleapis.com/auth/plus.login");}};
private final static int MAX_BATCH_SIZE = 1000;
@@ -82,8 +84,6 @@ public abstract class AbstractGPlusProvider implements StreamsProvider {
protected GoogleCredential credential;
protected Plus plus;
-
-
public AbstractGPlusProvider() {
Config config = StreamsConfigurator.config.getConfig("gplus");
this.config = GPlusConfigurator.detectConfiguration(config);
@@ -112,6 +112,11 @@ public abstract class AbstractGPlusProvider implements StreamsProvider {
protected abstract Runnable getDataCollector(BackOffStrategy strategy, BlockingQueue<StreamsDatum> queue, Plus plus, UserInfo userInfo);
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public StreamsResultSet readCurrent() {
BlockingQueue<StreamsDatum> batch = new LinkedBlockingQueue<>();
int batchCount = 0;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/GPlusUserActivityProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/GPlusUserActivityProvider.java b/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/GPlusUserActivityProvider.java
index 933feab..0ab75e6 100644
--- a/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/GPlusUserActivityProvider.java
+++ b/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/GPlusUserActivityProvider.java
@@ -31,6 +31,7 @@ import java.util.concurrent.BlockingQueue;
*/
public class GPlusUserActivityProvider extends AbstractGPlusProvider{
+ public final static String STREAMS_ID = "GPlusUserActivityProvider";
public GPlusUserActivityProvider() {
super();
@@ -41,6 +42,11 @@ public class GPlusUserActivityProvider extends AbstractGPlusProvider{
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
protected Runnable getDataCollector(BackOffStrategy strategy, BlockingQueue<StreamsDatum> queue, Plus plus, UserInfo userInfo) {
return new GPlusUserActivityCollector(plus, queue, strategy, userInfo);
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/GPlusUserDataProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/GPlusUserDataProvider.java b/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/GPlusUserDataProvider.java
index 1833d41..2effdea 100644
--- a/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/GPlusUserDataProvider.java
+++ b/streams-contrib/streams-provider-google/google-gplus/src/main/java/com/google/gplus/provider/GPlusUserDataProvider.java
@@ -31,6 +31,8 @@ import java.util.concurrent.BlockingQueue;
*/
public class GPlusUserDataProvider extends AbstractGPlusProvider{
+ public final static String STREAMS_ID = "GPlusUserDataProvider";
+
public GPlusUserDataProvider() {
super();
}
@@ -39,6 +41,10 @@ public class GPlusUserDataProvider extends AbstractGPlusProvider{
super(config);
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
@Override
protected Runnable getDataCollector(BackOffStrategy strategy, BlockingQueue<StreamsDatum> queue, Plus plus, UserInfo userInfo) {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-instagram/src/main/java/org/apache/streams/instagram/processor/InstagramTypeConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-instagram/src/main/java/org/apache/streams/instagram/processor/InstagramTypeConverter.java b/streams-contrib/streams-provider-instagram/src/main/java/org/apache/streams/instagram/processor/InstagramTypeConverter.java
index f0101fd..6595e8e 100644
--- a/streams-contrib/streams-provider-instagram/src/main/java/org/apache/streams/instagram/processor/InstagramTypeConverter.java
+++ b/streams-contrib/streams-provider-instagram/src/main/java/org/apache/streams/instagram/processor/InstagramTypeConverter.java
@@ -60,6 +60,11 @@ public class InstagramTypeConverter implements StreamsProcessor {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public List<StreamsDatum> process(StreamsDatum entry) {
StreamsDatum result = null;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-instagram/src/main/java/org/apache/streams/instagram/provider/InstagramAbstractProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-instagram/src/main/java/org/apache/streams/instagram/provider/InstagramAbstractProvider.java b/streams-contrib/streams-provider-instagram/src/main/java/org/apache/streams/instagram/provider/InstagramAbstractProvider.java
index 3f12493..8bbd900 100644
--- a/streams-contrib/streams-provider-instagram/src/main/java/org/apache/streams/instagram/provider/InstagramAbstractProvider.java
+++ b/streams-contrib/streams-provider-instagram/src/main/java/org/apache/streams/instagram/provider/InstagramAbstractProvider.java
@@ -27,6 +27,8 @@ import org.apache.streams.instagram.UsersInfo;
import org.apache.streams.util.ComponentUtils;
import org.apache.streams.util.SerializationUtil;
import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.math.BigInteger;
import java.util.Collection;
@@ -43,6 +45,10 @@ import java.util.concurrent.atomic.AtomicBoolean;
*/
public abstract class InstagramAbstractProvider implements StreamsProvider {
+ public static final String STREAMS_ID = "InstagramAbstractProvider";
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(InstagramAbstractProvider.class);
+
private static final int MAX_BATCH_SIZE = 2000;
protected InstagramConfiguration config;
@@ -60,6 +66,11 @@ public abstract class InstagramAbstractProvider implements StreamsProvider {
}
@Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
+ @Override
public void startStream() {
this.dataCollector = getInstagramDataCollector();
this.executorService = Executors.newSingleThreadExecutor();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/86f920d2/streams-contrib/streams-provider-moreover/src/main/java/org/apache/streams/data/moreover/MoreoverProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-moreover/src/main/java/org/apache/streams/data/moreover/MoreoverProvider.java b/streams-contrib/streams-provider-moreover/src/main/java/org/apache/streams/data/moreover/MoreoverProvider.java
index de5c88a..1add4d2 100644
--- a/streams-contrib/streams-provider-moreover/src/main/java/org/apache/streams/data/moreover/MoreoverProvider.java
+++ b/streams-contrib/streams-provider-moreover/src/main/java/org/apache/streams/data/moreover/MoreoverProvider.java
@@ -56,6 +56,11 @@ public class MoreoverProvider implements StreamsProvider {
}
}
+ @Override
+ public String getId() {
+ return STREAMS_ID;
+ }
+
public void startStream() {
for(MoreoverKeyData key : keys) {