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 2014/11/14 01:29:00 UTC
[7/9] incubator-streams git commit: omni-bus update
omni-bus update
primarily addresses STREAMS-218 (Generic Activity Serializer Processor)
addresses STREAMS-68 (ActivitySerializer should be ActivityConverter) - was touching related classes anyway
some additional refactoring of class and method names is advisable but this is a big chunk
resolves STREAMS-219 (.gitignore files generated by maven plugins)
needed to knock this out to avoid committing these files along with other changes
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/aad0e887
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/aad0e887
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/aad0e887
Branch: refs/heads/STREAMS-68,218
Commit: aad0e88742b9156e6e11d1bf06d23133ec6ec961
Parents: dc7ba80
Author: sblackmon <sb...@apache.org>
Authored: Thu Nov 13 17:08:21 2014 -0600
Committer: sblackmon <sb...@apache.org>
Committed: Thu Nov 13 17:08:21 2014 -0600
----------------------------------------------------------------------
.gitignore | 6 +
streams-components/pom.xml | 1 +
streams-components/streams-converters/pom.xml | 86 +++++
.../converter/ActivityConverterProcessor.java | 113 +++++++
.../BaseActivityConverterResolver.java | 28 ++
.../converter/BaseDocumentClassifier.java | 62 ++++
.../BaseObjectNodeActivityConverter.java | 61 ++++
.../converter/BaseStringActivityConverter.java | 61 ++++
.../CleanAdditionalPropertiesProcessor.java | 62 ++++
.../converter/TypeConverterProcessor.java | 84 +++++
.../streams/converter/TypeConverterUtil.java | 50 +++
.../test/ActivityConverterProcessorTest.java | 102 ++++++
.../test/TypeConverterProcessorTest.java | 93 ++++++
streams-contrib/pom.xml | 1 -
.../streams-processor-jackson/pom.xml | 86 -----
.../CleanAdditionalPropertiesProcessor.java | 62 ----
.../streams/jackson/TypeConverterProcessor.java | 112 -------
.../test/TypeConverterProcessorTest.java | 95 ------
.../streams-provider-datasift/pom.xml | 2 +-
.../DatasiftActivitySerializerProcessor.java | 9 +-
.../DatasiftTypeConverterProcessor.java | 8 +-
.../serializer/DatasiftActivityConverter.java | 77 +++++
.../serializer/DatasiftActivitySerializer.java | 65 ----
.../serializer/DatasiftConverterResolver.java | 55 ++++
.../serializer/DatasiftEventClassifier.java | 34 +-
.../DatasiftInstagramActivityConverter.java | 124 ++++++++
.../DatasiftInstagramActivitySerializer.java | 124 --------
.../DatasiftInteractionActivityConverter.java | 222 +++++++++++++
.../DatasiftInteractionActivitySerializer.java | 222 -------------
.../DatasiftTwitterActivityConverter.java | 272 ++++++++++++++++
.../DatasiftTwitterActivitySerializer.java | 272 ----------------
.../com/datasift/test/DatasiftSerDeTest.java | 2 +-
.../DatasiftActivityConverterTest.java | 85 +++++
.../DatasiftActivitySerializerTest.java | 86 -----
.../serializer/DatasiftEventClassifierTest.java | 23 +-
.../DatasiftInstagramActivityConverterTest.java | 35 +++
...DatasiftInstagramActivitySerializerTest.java | 43 ---
...atasiftInteractionActivityConverterTest.java | 40 +++
...tasiftInteractionActivitySerializerTest.java | 48 ---
.../DatasiftTwitterActivityConverterTest.java | 35 +++
.../DatasiftTwitterActivitySerializerTest.java | 43 ---
.../api/FacebookPageActivityConverter.java | 76 +++++
.../api/FacebookPageActivitySerializer.java | 66 ----
.../api/FacebookPostActivityConverter.java | 313 +++++++++++++++++++
.../api/FacebookPostActivitySerializer.java | 304 ------------------
.../processor/FacebookTypeConverter.java | 22 +-
.../provider/FacebookEventClassifier.java | 57 ----
.../serializer/FacebookConverterResolver.java | 62 ++++
.../serializer/FacebookDocumentClassifier.java | 75 +++++
.../FacebookStreamsPostConverter.java | 60 ++++
.../FacebookStreamsPostSerializer.java | 60 ----
.../test/FacebookActivityActorSerDeTest.java | 6 +-
.../test/FacebookActivitySerDeTest.java | 4 +-
.../test/FacebookDocumentClassifierTest.java | 57 ++++
.../test/FacebookEventClassifierTest.java | 57 ----
.../streams/facebook/test/SimplePageTest.java | 4 +-
.../com/gplus/api/GPlusActivityConverter.java | 93 ++++++
.../com/gplus/api/GPlusActivitySerializer.java | 93 ------
.../com/gplus/api/GPlusEDCAsActivityTest.java | 4 +-
.../com/reddit/api/RedditActivityConverter.java | 107 +++++++
.../reddit/api/RedditActivitySerializer.java | 107 -------
.../reddit/api/RedditEDCAsActivityJSONTest.java | 4 +-
.../ActivityXMLActivityConverter.java | 240 ++++++++++++++
.../ActivityXMLActivitySerializer.java | 240 --------------
.../powertrack/PowerTrackActivityConverter.java | 121 +++++++
.../PowerTrackActivitySerializer.java | 121 -------
.../test/PowerTrackDeserializationTest.java | 6 +-
.../gmail/provider/GMailImapProviderTask.java | 5 +-
.../provider/GMailMessageActivityConverter.java | 202 ++++++++++++
.../GMailMessageActivitySerializer.java | 212 -------------
.../gplus/provider/GPlusActivityConverter.java | 67 ++++
.../gplus/provider/GPlusActivitySerializer.java | 67 ----
.../gplus/provider/GPlusEventProcessor.java | 2 +-
.../processor/InstagramTypeConverter.java | 6 +-
.../InstagramJsonActivityConverter.java | 78 +++++
.../InstagramJsonActivitySerializer.java | 78 -----
.../serializer/InstagramUserInfoConverter.java | 81 +++++
.../serializer/InstagramUserInfoSerializer.java | 83 -----
.../data/MoreoverJsonActivityConverter.java | 93 ++++++
.../data/MoreoverJsonActivitySerializer.java | 93 ------
.../data/MoreoverXmlActivityConverter.java | 105 +++++++
.../data/MoreoverXmlActivitySerializer.java | 105 -------
.../data/MoreoverJsonActivityConverterTest.java | 65 ++++
.../MoreoverJsonActivitySerializerTest.java | 72 -----
.../data/MoreoverXmlActivityConverterTest.java | 61 ++++
.../data/MoreoverXmlActivitySerializerTest.java | 61 ----
.../streams/rss/processor/RssTypeConverter.java | 6 +-
.../streams/rss/provider/RssEventProcessor.java | 5 +-
.../serializer/SyndEntryActivityConverter.java | 232 ++++++++++++++
.../serializer/SyndEntryActivitySerializer.java | 235 --------------
.../test/SyndEntryActivitySerizlizerTest.java | 4 +-
.../streams-provider-twitter/pom.xml | 3 +-
.../FetchAndReplaceTwitterProcessor.java | 4 +-
.../processor/TwitterProfileProcessor.java | 4 +-
.../provider/TwitterEventClassifier.java | 99 ------
.../TwitterUserInformationProvider.java | 2 +-
.../serializer/TwitterConverterResolver.java | 72 +++++
.../serializer/TwitterDocumentClassifier.java | 82 +++++
.../TwitterJsonActivityConverter.java | 79 +++++
.../TwitterJsonActivitySerializer.java | 70 -----
.../TwitterJsonDeleteActivityConverter.java | 84 +++++
.../TwitterJsonDeleteActivitySerializer.java | 94 ------
.../TwitterJsonRetweetActivityConverter.java | 78 +++++
.../TwitterJsonRetweetActivitySerializer.java | 79 -----
.../TwitterJsonTweetActivityConverter.java | 76 +++++
.../TwitterJsonTweetActivitySerializer.java | 77 -----
.../TwitterJsonUserActivityConverter.java | 77 +++++
.../TwitterJsonUserActivitySerializer.java | 78 -----
...terJsonUserstreameventActivityConverter.java | 115 +++++++
...erJsonUserstreameventActivitySerializer.java | 115 -------
.../streams/twitter/test/FullTweetTest.java | 119 +++++++
.../streams/twitter/test/SimpleTweetTest.java | 43 ++-
.../twitter/test/TweetActivitySerDeTest.java | 32 +-
.../streams/twitter/test/TweetSerDeTest.java | 8 +-
.../test/TwitterDocumentClassifierTest.java | 117 +++++++
.../test/TwitterEventClassifierTest.java | 100 ------
.../apache/streams/data/ActivityConverter.java | 60 ++++
.../streams/data/ActivityConverterFactory.java | 29 ++
.../streams/data/ActivityConverterResolver.java | 38 +++
.../apache/streams/data/ActivitySerializer.java | 60 ----
.../apache/streams/data/DocumentClassifier.java | 40 +++
.../apache/streams/data/util/ActivityUtil.java | 3 +
.../streams/jackson/StreamsJacksonMapper.java | 14 +
.../streams/pig/StreamsComponentFactory.java | 13 +-
.../streams/pig/StreamsProcessDocumentExec.java | 11 -
.../streams/pig/StreamsSerializerExec.java | 16 +-
.../streams/pig/test/PigProcessDatumTest.java | 6 -
.../pig/test/PigProcessDocumentTest.java | 3 -
.../streams/pig/test/PigSerializerTest.java | 12 +-
129 files changed, 5167 insertions(+), 4313 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/.gitignore
----------------------------------------------------------------------
diff --git a/.gitignore b/.gitignore
index 23d6cc9..e04fafd 100644
--- a/.gitignore
+++ b/.gitignore
@@ -22,3 +22,9 @@ dependency-reduced-pom.xml
Icon?
ehthumbs.db
Thumbs.db
+
+# Maven plug-in generated files #
+DEPENDENCIES
+DISCLAIMER
+LICENSE
+NOTICE
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-components/pom.xml
----------------------------------------------------------------------
diff --git a/streams-components/pom.xml b/streams-components/pom.xml
index 9942e14..138eeef 100644
--- a/streams-components/pom.xml
+++ b/streams-components/pom.xml
@@ -37,6 +37,7 @@
</properties>
<modules>
+ <module>streams-converters</module>
<module>streams-http</module>
</modules>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-components/streams-converters/pom.xml
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/pom.xml b/streams-components/streams-converters/pom.xml
new file mode 100644
index 0000000..cf990cb
--- /dev/null
+++ b/streams-components/streams-converters/pom.xml
@@ -0,0 +1,86 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one
+ ~ or more contributor license agreements. See the NOTICE file
+ ~ distributed with this work for additional information
+ ~ regarding copyright ownership. The ASF licenses this file
+ ~ to you under the Apache License, Version 2.0 (the
+ ~ "License"); you may not use this file except in compliance
+ ~
+ ~ http://www.apache.org/licenses/LICENSE-2.0
+ ~
+ ~ Unless required by applicable law or agreed to in writing,
+ ~ software distributed under the License is distributed on an
+ ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ ~ KIND, either express or implied. See the License for the
+ ~ specific language governing permissions and limitations
+ ~ under the License.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+ <artifactId>streams-converters</artifactId>
+ <version>0.1-SNAPSHOT</version>
+
+ <parent>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-components</artifactId>
+ <version>0.1-SNAPSHOT</version>
+ </parent>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-config</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streams</groupId>
+ <artifactId>streams-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.jayway.jsonpath</groupId>
+ <artifactId>json-path</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.jayway.jsonpath</groupId>
+ <artifactId>json-path-assert</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <sourceDirectory>src/main/java</sourceDirectory>
+ <testSourceDirectory>src/test/java</testSourceDirectory>
+ <resources>
+ <resource>
+ <directory>src/main/resources</directory>
+ </resource>
+ </resources>
+ <testResources>
+ <testResource>
+ <directory>src/test/resources</directory>
+ </testResource>
+ </testResources>
+ <plugins>
+
+ </plugins>
+ </build>
+
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/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
new file mode 100644
index 0000000..46ba788
--- /dev/null
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/ActivityConverterProcessor.java
@@ -0,0 +1,113 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.streams.converter;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.data.ActivityConverterFactory;
+import org.apache.streams.data.ActivityConverterResolver;
+import org.apache.streams.data.DocumentClassifier;
+import org.apache.streams.pojo.json.Activity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ *
+ */
+public class ActivityConverterProcessor extends TypeConverterProcessor {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(ActivityConverterProcessor.class);
+
+ private List<DocumentClassifier> classifiers;
+ private List<ActivityConverterResolver> resolvers;
+
+ public ActivityConverterProcessor() {
+ super(Activity.class);
+ this.classifiers = Lists.newArrayList((DocumentClassifier)BaseDocumentClassifier.getInstance());
+ this.resolvers = Lists.newArrayList((ActivityConverterResolver) BaseActivityConverterResolver.getInstance());
+ }
+
+ public ActivityConverterProcessor(List<DocumentClassifier> classifiers, List<ActivityConverterResolver> factories) {
+ super(Activity.class);
+ this.classifiers = classifiers;
+ this.resolvers = factories;
+ this.classifiers.add(BaseDocumentClassifier.getInstance());
+ this.resolvers.add(BaseActivityConverterResolver.getInstance());
+ }
+
+ @Override
+ public List<StreamsDatum> process(StreamsDatum entry) {
+
+ Preconditions.checkArgument(classifiers.size() > 0);
+ Preconditions.checkArgument(resolvers.size() > 0);
+
+ List<StreamsDatum> result = Lists.newLinkedList();
+ Object inDoc = entry.getDocument();
+
+ try {
+
+ // This implementation is primitive, greedy, takes first it can resolve
+ Class datumClass = null;
+ for( DocumentClassifier classifier : classifiers ) {
+ datumClass = classifier.detectClass(inDoc);
+ if( classifier != null )
+ break;
+ }
+
+ // This implementation is primitive, greedy, takes first it can resolve
+ Class converterClass = null;
+ for( ActivityConverterResolver resolver : resolvers ) {
+ converterClass = resolver.bestSerializer(datumClass);
+ if( converterClass != null )
+ break;
+ }
+
+ ActivityConverter converter = ActivityConverterFactory.getInstance(converterClass);
+
+ Object typedDoc;
+ if( datumClass.isInstance(inDoc) )
+ typedDoc = inDoc;
+ else
+ typedDoc = TypeConverterUtil.convert(inDoc, datumClass, mapper);
+
+ Activity activity = converter.deserialize(typedDoc);
+
+ entry.setDocument(activity);
+
+ result.add(entry);
+
+ } catch( Throwable e ) {
+ LOGGER.warn("Unable to serialize!", e.getMessage());
+ }
+
+ return result;
+ }
+
+ @Override
+ public void prepare(Object configurationObject) {
+ super.prepare(configurationObject);
+ Preconditions.checkArgument(classifiers.size() > 0);
+ Preconditions.checkArgument(resolvers.size() > 0);
+ }
+
+};
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseActivityConverterResolver.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseActivityConverterResolver.java b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseActivityConverterResolver.java
new file mode 100644
index 0000000..3347d25
--- /dev/null
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseActivityConverterResolver.java
@@ -0,0 +1,28 @@
+package org.apache.streams.converter;
+
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.data.ActivityConverterResolver;
+import org.apache.streams.exceptions.ActivitySerializerException;
+
+/**
+ * Created by sblackmon on 11/12/14.
+ */
+public class BaseActivityConverterResolver implements ActivityConverterResolver {
+
+ private static BaseActivityConverterResolver instance = new BaseActivityConverterResolver();
+
+ public static BaseActivityConverterResolver getInstance() {
+ return instance;
+ }
+
+ @Override
+ public Class bestSerializer(Class documentClass) throws ActivitySerializerException {
+ if( documentClass == String.class) {
+ return BaseStringActivityConverter.class;
+ } else if( documentClass == ObjectNode.class) {
+ return BaseObjectNodeActivityConverter.class;
+ } else return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseDocumentClassifier.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseDocumentClassifier.java b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseDocumentClassifier.java
new file mode 100644
index 0000000..3f92f5f
--- /dev/null
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseDocumentClassifier.java
@@ -0,0 +1,62 @@
+package org.apache.streams.converter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import org.apache.streams.data.DocumentClassifier;
+import org.apache.streams.data.util.ActivityUtil;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+
+import java.io.IOException;
+
+/**
+ * Created by sblackmon on 11/12/14.
+ */
+public class BaseDocumentClassifier implements DocumentClassifier {
+
+ private static BaseDocumentClassifier instance = new BaseDocumentClassifier();
+
+ public static BaseDocumentClassifier getInstance() {
+ return instance;
+ }
+
+ private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
+
+ @Override
+ public Class detectClass(Object document) {
+ Preconditions.checkArgument(
+ document instanceof String
+ || document instanceof ObjectNode);
+
+ Activity activity = null;
+ ObjectNode node = null;
+
+ // Soon javax.validation will available in jackson
+ // That will make this simpler and more powerful
+ if( document instanceof String ) {
+ try {
+ activity = this.mapper.readValue((String)document, Activity.class);
+ if(ActivityUtil.isValid(activity))
+ return Activity.class;
+ else
+ return ObjectNode.class;
+ } catch (IOException e1) {
+ try {
+ node = this.mapper.readValue((String)document, ObjectNode.class);
+ return ObjectNode.class;
+ } catch (IOException e2) {
+ return String.class;
+ }
+ }
+ } else if( document instanceof ObjectNode ){
+ activity = this.mapper.convertValue((ObjectNode)document, Activity.class);
+ if(ActivityUtil.isValid(activity))
+ return Activity.class;
+ else
+ return ObjectNode.class;
+ } else return document.getClass();
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseObjectNodeActivityConverter.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseObjectNodeActivityConverter.java b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseObjectNodeActivityConverter.java
new file mode 100644
index 0000000..a4d0ea4
--- /dev/null
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseObjectNodeActivityConverter.java
@@ -0,0 +1,61 @@
+package org.apache.streams.converter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Lists;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.exceptions.ActivitySerializerException;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+
+import java.util.List;
+
+/**
+ * Created by sblackmon on 11/13/14.
+ */
+public class BaseObjectNodeActivityConverter implements ActivityConverter<ObjectNode> {
+
+ private ObjectMapper mapper = new StreamsJacksonMapper();
+
+ private static BaseObjectNodeActivityConverter instance = new BaseObjectNodeActivityConverter();
+
+ public static BaseObjectNodeActivityConverter getInstance() {
+ return instance;
+ }
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public ObjectNode serialize(Activity deserialized) throws ActivitySerializerException {
+ try {
+ return mapper.convertValue(deserialized, ObjectNode.class);
+ } catch (Exception e) {
+ throw new ActivitySerializerException();
+ }
+ }
+
+ @Override
+ public Activity deserialize(ObjectNode serialized) throws ActivitySerializerException {
+ try {
+ return mapper.convertValue(serialized, Activity.class);
+ } catch (Exception e) {
+ throw new ActivitySerializerException();
+ }
+ }
+
+ @Override
+ public List<Activity> deserializeAll(List<ObjectNode> serializedList) {
+ List<Activity> result = Lists.newArrayList();
+ for( ObjectNode item : serializedList ) {
+ try {
+ Activity activity = deserialize(item);
+ result.add(activity);
+ } catch (ActivitySerializerException e) {}
+ }
+ return result;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseStringActivityConverter.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseStringActivityConverter.java b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseStringActivityConverter.java
new file mode 100644
index 0000000..ea0c108
--- /dev/null
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/BaseStringActivityConverter.java
@@ -0,0 +1,61 @@
+package org.apache.streams.converter;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Lists;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.exceptions.ActivitySerializerException;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+
+import java.util.List;
+
+/**
+ * Created by sblackmon on 11/13/14.
+ */
+public class BaseStringActivityConverter implements ActivityConverter<String> {
+
+ private ObjectMapper mapper = new StreamsJacksonMapper();
+
+ private static BaseStringActivityConverter instance = new BaseStringActivityConverter();
+
+ public static BaseStringActivityConverter getInstance() {
+ return instance;
+ }
+
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public String serialize(Activity deserialized) throws ActivitySerializerException {
+ try {
+ return mapper.writeValueAsString(deserialized);
+ } catch (JsonProcessingException e) {
+ throw new ActivitySerializerException();
+ }
+ }
+
+ @Override
+ public Activity deserialize(String serialized) throws ActivitySerializerException {
+ try {
+ return mapper.readValue(serialized, Activity.class);
+ } catch (Exception e) {
+ throw new ActivitySerializerException();
+ }
+ }
+
+ @Override
+ public List<Activity> deserializeAll(List<String> serializedList) {
+ List<Activity> result = Lists.newArrayList();
+ for( String item : serializedList ) {
+ try {
+ Activity activity = deserialize(item);
+ result.add(activity);
+ } catch (ActivitySerializerException e) {}
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-components/streams-converters/src/main/java/org/apache/streams/converter/CleanAdditionalPropertiesProcessor.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/CleanAdditionalPropertiesProcessor.java b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/CleanAdditionalPropertiesProcessor.java
new file mode 100644
index 0000000..4b4641e
--- /dev/null
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/CleanAdditionalPropertiesProcessor.java
@@ -0,0 +1,62 @@
+package org.apache.streams.converter;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.datatype.jsonorg.JsonOrgModule;
+import com.google.common.collect.Lists;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProcessor;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * This processor walks an input objectnode and corrects any artifacts
+ * that may have occured from improper serialization of jsonschema2pojo beans.
+ *
+ * The logic is also available for inclusion in other module via static import.
+ */
+public class CleanAdditionalPropertiesProcessor implements StreamsProcessor {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(CleanAdditionalPropertiesProcessor.class);
+
+ private ObjectMapper mapper;
+
+ @Override
+ public List<StreamsDatum> process(StreamsDatum datum) {
+ List<StreamsDatum> result = Lists.newLinkedList();
+ ObjectNode activity = this.mapper.convertValue(datum.getDocument(), ObjectNode.class);
+ cleanAdditionalProperties(activity);
+ datum.setDocument(activity);
+ result.add(datum);
+ return result;
+ }
+
+ @Override
+ public void prepare(Object o) {
+ this.mapper = StreamsJacksonMapper.getInstance();
+ this.mapper.registerModule(new JsonOrgModule());
+ }
+
+ @Override
+ public void cleanUp() {
+
+ }
+
+ public static void cleanAdditionalProperties(ObjectNode node) {
+ if( node.get("additionalProperties") != null ) {
+ ObjectNode additionalProperties = (ObjectNode) node.get("additionalProperties");
+ cleanAdditionalProperties(additionalProperties);
+ Iterator<Map.Entry<String, JsonNode>> jsonNodeIterator = additionalProperties.fields();
+ while( jsonNodeIterator.hasNext() ) {
+ Map.Entry<String, JsonNode> entry = jsonNodeIterator.next();
+ node.put(entry.getKey(), entry.getValue());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/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
new file mode 100644
index 0000000..6bfeead
--- /dev/null
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/TypeConverterProcessor.java
@@ -0,0 +1,84 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+package org.apache.streams.converter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Lists;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProcessor;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ *
+ */
+public class TypeConverterProcessor implements StreamsProcessor {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(TypeConverterProcessor.class);
+
+ private List<String> formats = Lists.newArrayList();
+
+ protected ObjectMapper mapper;
+
+ protected Class outClass;
+
+ public TypeConverterProcessor(Class outClass) {
+ this.outClass = outClass;
+ }
+
+ public TypeConverterProcessor(Class outClass, List<String> formats) {
+ this.outClass = outClass;
+ this.formats = formats;
+ }
+
+ @Override
+ public List<StreamsDatum> process(StreamsDatum entry) {
+
+ List<StreamsDatum> result = Lists.newLinkedList();
+ Object inDoc = entry.getDocument();
+
+ Object outDoc = TypeConverterUtil.convert(inDoc, outClass, mapper);
+
+ if( outDoc != null ) {
+ entry.setDocument(outDoc);
+ result.add(entry);
+ }
+
+ return result;
+ }
+
+ @Override
+ public void prepare(Object configurationObject) {
+ if( formats.size() > 0 )
+ this.mapper = StreamsJacksonMapper.getInstance(formats);
+ else
+ this.mapper = StreamsJacksonMapper.getInstance();
+ }
+
+ @Override
+ public void cleanUp() {
+ this.mapper = null;
+ }
+
+};
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-components/streams-converters/src/main/java/org/apache/streams/converter/TypeConverterUtil.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/main/java/org/apache/streams/converter/TypeConverterUtil.java b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/TypeConverterUtil.java
new file mode 100644
index 0000000..8a39ac5
--- /dev/null
+++ b/streams-components/streams-converters/src/main/java/org/apache/streams/converter/TypeConverterUtil.java
@@ -0,0 +1,50 @@
+package org.apache.streams.converter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+
+/**
+ * Created by sblackmon on 11/13/14.
+ */
+public class TypeConverterUtil {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(TypeConverterUtil.class);
+
+ public static Object convert(Object object, Class outClass) {
+ return TypeConverterUtil.convert(object, outClass, StreamsJacksonMapper.getInstance());
+ }
+
+ public static Object convert(Object object, Class outClass, ObjectMapper mapper) {
+ ObjectNode node = null;
+ Object outDoc = null;
+ if( object instanceof String ) {
+ try {
+ node = mapper.readValue((String)object, ObjectNode.class);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ } else {
+ node = mapper.convertValue(object, ObjectNode.class);
+ }
+
+ if(node != null) {
+ try {
+ if( outClass == String.class )
+ outDoc = mapper.writeValueAsString(node);
+ else
+ outDoc = mapper.convertValue(node, outClass);
+
+ } catch (Throwable e) {
+ LOGGER.warn(e.getMessage());
+ LOGGER.warn(node.toString());
+ }
+ }
+
+ return outDoc;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-components/streams-converters/src/test/java/org/apache/streams/jackson/test/ActivityConverterProcessorTest.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/test/java/org/apache/streams/jackson/test/ActivityConverterProcessorTest.java b/streams-components/streams-converters/src/test/java/org/apache/streams/jackson/test/ActivityConverterProcessorTest.java
new file mode 100644
index 0000000..6d626b2
--- /dev/null
+++ b/streams-components/streams-converters/src/test/java/org/apache/streams/jackson/test/ActivityConverterProcessorTest.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.streams.jackson.test;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.streams.converter.ActivityConverterProcessor;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProcessor;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+import static junit.framework.Assert.*;
+
+/**
+ *
+ */
+public class ActivityConverterProcessorTest {
+
+ private static final ObjectMapper mapper = new StreamsJacksonMapper();
+
+ @Test
+ public void testBaseActivitySerializerProcessorInvalid() {
+ String INVALID_DOCUMENT = " 38Xs}";
+ StreamsProcessor processor = new ActivityConverterProcessor();
+ processor.prepare(null);
+ StreamsDatum datum = new StreamsDatum(INVALID_DOCUMENT);
+ List<StreamsDatum> result = processor.process(datum);
+ assertNotNull(result);
+ assertEquals(0, result.size());
+ }
+
+ @Test
+ public void testBaseActivitySerializerProcessorString() {
+ String STRING_DOCUMENT = "{\"verb\":\"post\"}";
+ StreamsProcessor processor = new ActivityConverterProcessor();
+ processor.prepare(null);
+ StreamsDatum datum = new StreamsDatum(STRING_DOCUMENT);
+ List<StreamsDatum> result = processor.process(datum);
+ assertNotNull(result);
+ assertEquals(1, result.size());
+ StreamsDatum resultDatum = result.get(0);
+ assertNotNull(resultDatum);
+ assertNotNull(resultDatum.getDocument());
+ assertTrue(resultDatum.getDocument() instanceof Activity);
+ assertTrue(((Activity)resultDatum.getDocument()).getVerb().equals("post"));
+ }
+
+ @Test
+ public void testBaseActivitySerializerProcessorObject() throws IOException {
+ ObjectNode OBJECT_DOCUMENT = mapper.readValue("{\"verb\":\"share\"}", ObjectNode.class);
+ StreamsProcessor processor = new ActivityConverterProcessor();
+ processor.prepare(null);
+ StreamsDatum datum = new StreamsDatum(OBJECT_DOCUMENT);
+ List<StreamsDatum> result = processor.process(datum);
+ assertNotNull(result);
+ assertEquals(1, result.size());
+ StreamsDatum resultDatum = result.get(0);
+ assertNotNull(resultDatum);
+ assertNotNull(resultDatum.getDocument());
+ assertTrue(resultDatum.getDocument() instanceof Activity);
+ assertTrue(((Activity)resultDatum.getDocument()).getVerb().equals("share"));
+ }
+
+// @Test
+// public void testTypeConverterObjectNodeToString() throws IOException {
+// final String ID = "1";
+// StreamsProcessor processor = new ActivitySerializerProcessor(ObjectNode.class, String.class, Lists.newArrayList(DATASIFT_FORMAT));
+// processor.prepare(null);
+// ObjectMapper mapper = StreamsJacksonMapper.getInstance(Lists.newArrayList(DATASIFT_FORMAT));
+// ObjectNode node = mapper.readValue(DATASIFT_JSON, ObjectNode.class);
+// StreamsDatum datum = new StreamsDatum(node, ID);
+// List<StreamsDatum> result = processor.process(datum);
+// assertNotNull(result);
+// assertEquals(1, result.size());
+// StreamsDatum resultDatum = result.get(0);
+// assertNotNull(resultDatum);
+// assertNotNull(resultDatum.getDocument());
+// assertTrue(resultDatum.getDocument() instanceof String);
+// assertEquals(ID, resultDatum.getId());
+// }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-components/streams-converters/src/test/java/org/apache/streams/jackson/test/TypeConverterProcessorTest.java
----------------------------------------------------------------------
diff --git a/streams-components/streams-converters/src/test/java/org/apache/streams/jackson/test/TypeConverterProcessorTest.java b/streams-components/streams-converters/src/test/java/org/apache/streams/jackson/test/TypeConverterProcessorTest.java
new file mode 100644
index 0000000..8a6c732
--- /dev/null
+++ b/streams-components/streams-converters/src/test/java/org/apache/streams/jackson/test/TypeConverterProcessorTest.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.streams.jackson.test;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Lists;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProcessor;
+import org.apache.streams.converter.TypeConverterProcessor;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+import static junit.framework.Assert.*;
+
+/**
+ *
+ */
+public class TypeConverterProcessorTest {
+
+ private static final String DATASIFT_JSON = "{\"demographic\":{\"gender\":\"female\"},\"interaction\":{\"schema\":{\"version\":3},\"source\":\"Twitter for Android\",\"author\":{\"username\":\"ViiOLeee\",\"name\":\"Violeta Anguita\",\"id\":70931384,\"avatar\":\"http://pbs.twimg.com/profile_images/378800000851401229/bbf480cde2e9923a1d20acd393da0212_normal.jpeg\",\"link\":\"http://twitter.com/ViiOLeee\",\"language\":\"en\"},\"type\":\"twitter\",\"created_at\":\"Tue, 27 May 2014 22:38:15 +0000\",\"received_at\":1.401230295658E9,\"content\":\"RT @AliiAnguita: \\\"@Pharrell: Loved working with @edsheeran on Sing. He's a genius. https://t.co/wB2qKyJMRw\\\" @ViiOLeee look at this!\",\"id\":\"1e3e5ef97532a580e0741841f5746728\",\"link\":\"http://twitter.com/ViiOLeee/status/471420141989666817\",\"mentions\":[\"Pharrell\",\"edsheeran\",\"ViiOLeee\",\"AliiAnguita\"],\"mention_ids\":[338084918,85452649,70931384]},\"klout\":{\"score\":34},\"language\":{\"tag\":\"en\",\"tag_extended\":\"en\",\
"confidence\":98},\"links\":{\"code\":[200],\"created_at\":[\"Tue, 27 May 2014 14:28:06 +0000\"],\"meta\":{\"charset\":[\"UTF-8\"],\"content_type\":[\"text/html\"],\"description\":[\"Official Video for Ed Sheeran's track SING Get this track on iTunes: http://smarturl.it/EdSing Pre-order 'x' on iTunes and get 'One' instantly: http://smartu...\"],\"keywords\":[[\"ed sheeran\",\"ed sheeran sing\",\"ed sheeran new album\",\"Ed Sheeran (Musical Artist)\",\"ed sheeran one\",\"ed sheeran fault in our stars\",\"ed sheeran all of the stars\",\"s...\"]],\"lang\":[\"en\"],\"opengraph\":[{\"site_name\":\"YouTube\",\"url\":\"http://www.youtube.com/watch?v=tlYcUqEPN58\",\"title\":\"Ed Sheeran - SING [Official Video]\",\"image\":\"https://i1.ytimg.com/vi/tlYcUqEPN58/maxresdefault.jpg\",\"description\":\"Official Video for Ed Sheeran's track SING Get this track on iTunes: http://smarturl.it/EdSing Pre-order 'x' on iTunes and get 'One' instantly: http://smartu
...\",\"type\":\"video\"}],\"twitter\":[{\"card\":\"player\",\"site\":\"@youtube\",\"url\":\"http://www.youtube.com/watch?v=tlYcUqEPN58\",\"title\":\"Ed Sheeran - SING [Official Video]\",\"description\":\"Official Video for Ed Sheeran's track SING Get this track on iTunes: http://smarturl.it/EdSing Pre-order 'x' on iTunes and get 'One' instantly: http://smartu...\",\"image\":\"https://i1.ytimg.com/vi/tlYcUqEPN58/maxresdefault.jpg\",\"app\":{\"iphone\":{\"name\":\"YouTube\",\"id\":\"544007664\",\"url\":\"vnd.youtube://watch/tlYcUqEPN58\"},\"ipad\":{\"name\":\"YouTube\",\"id\":\"544007664\",\"url\":\"vnd.youtube://watch/tlYcUqEPN58\"},\"googleplay\":{\"name\":\"YouTube\",\"id\":\"com.google.android.youtube\",\"url\":\"http://www.youtube.com/watch?v=tlYcUqEPN58\"}},\"player\":\"https://www.youtube.com/embed/tlYcUqEPN58\",\"player_width\":\"1280\",\"player_height\":\"720\"}]},\"normalized_url\":[\"https://youtube.com/watch?v=tlYcUqEPN58\"],\"retweet_count\":[0],\"tit
le\":[\"Ed Sheeran - SING [Official Video] - YouTube\"],\"url\":[\"https://www.youtube.com/watch?v=tlYcUqEPN58\"]},\"twitter\":{\"id\":\"471420141989666817\",\"retweet\":{\"text\":\"\\\"@Pharrell: Loved working with @edsheeran on Sing. He's a genius. https://t.co/wB2qKyJMRw\\\" @ViiOLeee look at this!\",\"id\":\"471420141989666817\",\"user\":{\"name\":\"Violeta Anguita\",\"description\":\"La vida no seria la fiesta que todos esperamos, pero mientras estemos aqui debemos BAILAR!!! #ErasmusOnceErasmusForever\",\"location\":\"Espanhaa..Olaa!\",\"statuses_count\":5882,\"followers_count\":249,\"friends_count\":1090,\"screen_name\":\"ViiOLeee\",\"profile_image_url\":\"http://pbs.twimg.com/profile_images/378800000851401229/bbf480cde2e9923a1d20acd393da0212_normal.jpeg\",\"profile_image_url_https\":\"https://pbs.twimg.com/profile_images/378800000851401229/bbf480cde2e9923a1d20acd393da0212_normal.jpeg\",\"lang\":\"en\",\"time_zone\":\"Madrid\",\"utc_offset\":7200,\"listed_count\":1,\"id\":709
31384,\"id_str\":\"70931384\",\"geo_enabled\":false,\"verified\":false,\"favourites_count\":275,\"created_at\":\"Wed, 02 Sep 2009 10:19:59 +0000\"},\"source\":\"<a href=\\\"http://twitter.com/download/android\\\" rel=\\\"nofollow\\\">Twitter for Android</a>\",\"count\":1,\"created_at\":\"Tue, 27 May 2014 22:38:15 +0000\",\"mentions\":[\"Pharrell\",\"edsheeran\",\"ViiOLeee\",\"AliiAnguita\"],\"mention_ids\":[338084918,85452649,70931384],\"links\":[\"https://www.youtube.com/watch?v=tlYcUqEPN58\"],\"display_urls\":[\"youtube.com/watch?v=tlYcUq���\"],\"domains\":[\"www.youtube.com\"],\"lang\":\"en\"},\"retweeted\":{\"id\":\"471419867078209536\",\"user\":{\"name\":\"Alicia Anguita \",\"description\":\"Estudiante de Ingenieria de la Edificaci��n en Granada.\",\"statuses_count\":371,\"followers_count\":185,\"friends_count\":404,\"screen_name\":\"AliiAnguita\",\"profile_image_url\":\"http://pbs.twimg.com/profile_images/424248659677442048/qCPZL8c9_normal.jpeg\",\"profile_image_url_
https\":\"https://pbs.twimg.com/profile_images/424248659677442048/qCPZL8c9_normal.jpeg\",\"lang\":\"es\",\"listed_count\":0,\"id\":561201891,\"id_str\":\"561201891\",\"geo_enabled\":false,\"verified\":false,\"favourites_count\":17,\"created_at\":\"Mon, 23 Apr 2012 13:11:44 +0000\"},\"source\":\"<a href=\\\"http://twitter.com/download/android\\\" rel=\\\"nofollow\\\">Twitter for Android</a>\",\"created_at\":\"Tue, 27 May 2014 22:37:09 +0000\"}}}";
+
+ public static final String DATASIFT_FORMAT = "EEE, dd MMM yyyy HH:mm:ss Z";
+
+ @Test
+ public void testTypeConverterStringToString() {
+ final String ID = "1";
+ StreamsProcessor processor = new TypeConverterProcessor(String.class, Lists.newArrayList(DATASIFT_FORMAT));
+ processor.prepare(null);
+ StreamsDatum datum = new StreamsDatum(DATASIFT_JSON, ID);
+ List<StreamsDatum> result = processor.process(datum);
+ assertNotNull(result);
+ assertEquals(1, result.size());
+ StreamsDatum resultDatum = result.get(0);
+ assertNotNull(resultDatum);
+ assertNotNull(resultDatum.getDocument());
+ assertTrue(resultDatum.getDocument() instanceof String);
+ assertEquals(ID, resultDatum.getId());
+ }
+
+ @Test
+ public void testTypeConverterStringToObjectNode() {
+ final String ID = "1";
+ StreamsProcessor processor = new TypeConverterProcessor(ObjectNode.class, Lists.newArrayList(DATASIFT_FORMAT));
+ processor.prepare(null);
+ StreamsDatum datum = new StreamsDatum(DATASIFT_JSON, ID);
+ List<StreamsDatum> result = processor.process(datum);
+ assertNotNull(result);
+ assertEquals(1, result.size());
+ StreamsDatum resultDatum = result.get(0);
+ assertNotNull(resultDatum);
+ assertNotNull(resultDatum.getDocument());
+ assertTrue(resultDatum.getDocument() instanceof ObjectNode);
+ assertEquals(ID, resultDatum.getId());
+ }
+
+ @Test
+ public void testTypeConverterObjectNodeToString() throws IOException {
+ final String ID = "1";
+ StreamsProcessor processor = new TypeConverterProcessor(String.class, Lists.newArrayList(DATASIFT_FORMAT));
+ processor.prepare(null);
+ ObjectMapper mapper = StreamsJacksonMapper.getInstance(DATASIFT_FORMAT);
+ ObjectNode node = mapper.readValue(DATASIFT_JSON, ObjectNode.class);
+ StreamsDatum datum = new StreamsDatum(node, ID);
+ List<StreamsDatum> result = processor.process(datum);
+ assertNotNull(result);
+ assertEquals(1, result.size());
+ StreamsDatum resultDatum = result.get(0);
+ assertNotNull(resultDatum);
+ assertNotNull(resultDatum.getDocument());
+ assertTrue(resultDatum.getDocument() instanceof String);
+ assertEquals(ID, resultDatum.getId());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-contrib/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/pom.xml b/streams-contrib/pom.xml
index fcec297..3162028 100644
--- a/streams-contrib/pom.xml
+++ b/streams-contrib/pom.xml
@@ -47,7 +47,6 @@
<module>streams-amazon-aws</module>
<!--<module>streams-processor-lucene</module>-->
<!--<module>streams-processor-tika</module>-->
- <module>streams-processor-jackson</module>
<module>streams-processor-json</module>
<module>streams-processor-urls</module>
<module>streams-processor-peoplepattern</module>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-contrib/streams-processor-jackson/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-jackson/pom.xml b/streams-contrib/streams-processor-jackson/pom.xml
deleted file mode 100644
index 8bb44fd..0000000
--- a/streams-contrib/streams-processor-jackson/pom.xml
+++ /dev/null
@@ -1,86 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ Licensed to the Apache Software Foundation (ASF) under one
- ~ or more contributor license agreements. See the NOTICE file
- ~ distributed with this work for additional information
- ~ regarding copyright ownership. The ASF licenses this file
- ~ to you under the Apache License, Version 2.0 (the
- ~ "License"); you may not use this file except in compliance
- ~
- ~ http://www.apache.org/licenses/LICENSE-2.0
- ~
- ~ Unless required by applicable law or agreed to in writing,
- ~ software distributed under the License is distributed on an
- ~ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- ~ KIND, either express or implied. See the License for the
- ~ specific language governing permissions and limitations
- ~ under the License.
- -->
-<project xmlns="http://maven.apache.org/POM/4.0.0"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-
- <modelVersion>4.0.0</modelVersion>
- <artifactId>streams-processor-jackson</artifactId>
- <version>0.1-SNAPSHOT</version>
-
- <parent>
- <groupId>org.apache.streams</groupId>
- <artifactId>streams-contrib</artifactId>
- <version>0.1-SNAPSHOT</version>
- </parent>
-
- <dependencies>
- <dependency>
- <groupId>org.apache.streams</groupId>
- <artifactId>streams-config</artifactId>
- </dependency>
- <dependency>
- <groupId>org.apache.streams</groupId>
- <artifactId>streams-core</artifactId>
- </dependency>
- <dependency>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-core</artifactId>
- </dependency>
- <dependency>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-annotations</artifactId>
- </dependency>
- <dependency>
- <groupId>com.jayway.jsonpath</groupId>
- <artifactId>json-path</artifactId>
- </dependency>
- <dependency>
- <groupId>com.jayway.jsonpath</groupId>
- <artifactId>json-path-assert</artifactId>
- </dependency>
- <dependency>
- <groupId>junit</groupId>
- <artifactId>junit</artifactId>
- </dependency>
- <dependency>
- <groupId>commons-io</groupId>
- <artifactId>commons-io</artifactId>
- </dependency>
- </dependencies>
-
- <build>
- <sourceDirectory>src/main/java</sourceDirectory>
- <testSourceDirectory>src/test/java</testSourceDirectory>
- <resources>
- <resource>
- <directory>src/main/resources</directory>
- </resource>
- </resources>
- <testResources>
- <testResource>
- <directory>src/test/resources</directory>
- </testResource>
- </testResources>
- <plugins>
-
- </plugins>
- </build>
-
-</project>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/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
deleted file mode 100644
index 56b0c5c..0000000
--- a/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/CleanAdditionalPropertiesProcessor.java
+++ /dev/null
@@ -1,62 +0,0 @@
-package org.apache.streams.jackson;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.fasterxml.jackson.datatype.jsonorg.JsonOrgModule;
-import com.google.common.collect.Lists;
-import org.apache.streams.core.StreamsDatum;
-import org.apache.streams.core.StreamsProcessor;
-import org.apache.streams.jackson.StreamsJacksonMapper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-
-/**
- * This processor walks an input objectnode and corrects any artifacts
- * that may have occured from improper serialization of jsonschema2pojo beans.
- *
- * The logic is also available for inclusion in other module via static import.
- */
-public class CleanAdditionalPropertiesProcessor implements StreamsProcessor {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(CleanAdditionalPropertiesProcessor.class);
-
- private ObjectMapper mapper;
-
- @Override
- public List<StreamsDatum> process(StreamsDatum datum) {
- List<StreamsDatum> result = Lists.newLinkedList();
- ObjectNode activity = this.mapper.convertValue(datum.getDocument(), ObjectNode.class);
- cleanAdditionalProperties(activity);
- datum.setDocument(activity);
- result.add(datum);
- return result;
- }
-
- @Override
- public void prepare(Object o) {
- this.mapper = StreamsJacksonMapper.getInstance();
- this.mapper.registerModule(new JsonOrgModule());
- }
-
- @Override
- public void cleanUp() {
-
- }
-
- public static void cleanAdditionalProperties(ObjectNode node) {
- if( node.get("additionalProperties") != null ) {
- ObjectNode additionalProperties = (ObjectNode) node.get("additionalProperties");
- cleanAdditionalProperties(additionalProperties);
- Iterator<Map.Entry<String, JsonNode>> jsonNodeIterator = additionalProperties.fields();
- while( jsonNodeIterator.hasNext() ) {
- Map.Entry<String, JsonNode> entry = jsonNodeIterator.next();
- node.put(entry.getKey(), entry.getValue());
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/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
deleted file mode 100644
index 63d03e9..0000000
--- a/streams-contrib/streams-processor-jackson/src/main/java/org/apache/streams/jackson/TypeConverterProcessor.java
+++ /dev/null
@@ -1,112 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
-*/
-package org.apache.streams.jackson;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.google.common.collect.Lists;
-import org.apache.streams.core.StreamsDatum;
-import org.apache.streams.core.StreamsProcessor;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.List;
-
-/**
- *
- */
-public class TypeConverterProcessor implements StreamsProcessor {
-
- private final static Logger LOGGER = LoggerFactory.getLogger(TypeConverterProcessor.class);
-
- private List<String> formats = Lists.newArrayList();
-
- private ObjectMapper mapper;
-
- private Class inClass;
- private Class outClass;
-
- public TypeConverterProcessor(Class inClass, Class outClass, ObjectMapper mapper) {
- this.inClass = inClass;
- this.outClass = outClass;
- this.mapper = mapper;
- }
-
- public TypeConverterProcessor(Class inClass, Class outClass, List<String> formats) {
- this.inClass = inClass;
- this.outClass = outClass;
- this.formats = formats;
- }
-
- public TypeConverterProcessor(Class inClass, Class outClass) {
- this.inClass = inClass;
- this.outClass = outClass;
- }
-
- @Override
- public List<StreamsDatum> process(StreamsDatum entry) {
- List<StreamsDatum> result = Lists.newLinkedList();
- Object inDoc = entry.getDocument();
- ObjectNode node = null;
- if( inClass == String.class ||
- inDoc instanceof String ) {
- try {
- node = this.mapper.readValue((String)entry.getDocument(), ObjectNode.class);
- } catch (IOException e) {
- e.printStackTrace();
- }
- } else {
- node = this.mapper.convertValue(inDoc, ObjectNode.class);
- }
-
- if(node != null) {
- Object outDoc;
- try {
- if( outClass == String.class )
- outDoc = this.mapper.writeValueAsString(node);
- else
- outDoc = this.mapper.convertValue(node, outClass);
-
- StreamsDatum outDatum = new StreamsDatum(outDoc, entry.getId(), entry.getTimestamp(), entry.getSequenceid());
- outDatum.setMetadata(entry.getMetadata());
- result.add(outDatum);
- } catch (Throwable e) {
- LOGGER.warn(e.getMessage());
- LOGGER.warn(node.toString());
- }
- }
-
- return result;
- }
-
- @Override
- public void prepare(Object configurationObject) {
- if( formats.size() > 0 )
- this.mapper = StreamsJacksonMapper.getInstance(formats);
- else
- this.mapper = StreamsJacksonMapper.getInstance();
- }
-
- @Override
- public void cleanUp() {
-
- }
-};
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-contrib/streams-processor-jackson/src/test/java/org/apache/streams/jackson/test/TypeConverterProcessorTest.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-jackson/src/test/java/org/apache/streams/jackson/test/TypeConverterProcessorTest.java b/streams-contrib/streams-processor-jackson/src/test/java/org/apache/streams/jackson/test/TypeConverterProcessorTest.java
deleted file mode 100644
index 1316d5c..0000000
--- a/streams-contrib/streams-processor-jackson/src/test/java/org/apache/streams/jackson/test/TypeConverterProcessorTest.java
+++ /dev/null
@@ -1,95 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.streams.jackson.test;
-
-import com.fasterxml.jackson.core.JsonParseException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.google.common.collect.Lists;
-import org.apache.streams.core.StreamsDatum;
-import org.apache.streams.core.StreamsProcessor;
-import org.apache.streams.jackson.StreamsJacksonMapper;
-import org.apache.streams.jackson.TypeConverterProcessor;
-import org.apache.streams.pojo.json.Activity;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.List;
-
-import static junit.framework.Assert.*;
-
-/**
- *
- */
-public class TypeConverterProcessorTest {
-
- private static final String DATASIFT_JSON = "{\"demographic\":{\"gender\":\"female\"},\"interaction\":{\"schema\":{\"version\":3},\"source\":\"Twitter for Android\",\"author\":{\"username\":\"ViiOLeee\",\"name\":\"Violeta Anguita\",\"id\":70931384,\"avatar\":\"http://pbs.twimg.com/profile_images/378800000851401229/bbf480cde2e9923a1d20acd393da0212_normal.jpeg\",\"link\":\"http://twitter.com/ViiOLeee\",\"language\":\"en\"},\"type\":\"twitter\",\"created_at\":\"Tue, 27 May 2014 22:38:15 +0000\",\"received_at\":1.401230295658E9,\"content\":\"RT @AliiAnguita: \\\"@Pharrell: Loved working with @edsheeran on Sing. He's a genius. https://t.co/wB2qKyJMRw\\\" @ViiOLeee look at this!\",\"id\":\"1e3e5ef97532a580e0741841f5746728\",\"link\":\"http://twitter.com/ViiOLeee/status/471420141989666817\",\"mentions\":[\"Pharrell\",\"edsheeran\",\"ViiOLeee\",\"AliiAnguita\"],\"mention_ids\":[338084918,85452649,70931384]},\"klout\":{\"score\":34},\"language\":{\"tag\":\"en\",\"tag_extended\":\"en\",\
"confidence\":98},\"links\":{\"code\":[200],\"created_at\":[\"Tue, 27 May 2014 14:28:06 +0000\"],\"meta\":{\"charset\":[\"UTF-8\"],\"content_type\":[\"text/html\"],\"description\":[\"Official Video for Ed Sheeran's track SING Get this track on iTunes: http://smarturl.it/EdSing Pre-order 'x' on iTunes and get 'One' instantly: http://smartu...\"],\"keywords\":[[\"ed sheeran\",\"ed sheeran sing\",\"ed sheeran new album\",\"Ed Sheeran (Musical Artist)\",\"ed sheeran one\",\"ed sheeran fault in our stars\",\"ed sheeran all of the stars\",\"s...\"]],\"lang\":[\"en\"],\"opengraph\":[{\"site_name\":\"YouTube\",\"url\":\"http://www.youtube.com/watch?v=tlYcUqEPN58\",\"title\":\"Ed Sheeran - SING [Official Video]\",\"image\":\"https://i1.ytimg.com/vi/tlYcUqEPN58/maxresdefault.jpg\",\"description\":\"Official Video for Ed Sheeran's track SING Get this track on iTunes: http://smarturl.it/EdSing Pre-order 'x' on iTunes and get 'One' instantly: http://smartu
...\",\"type\":\"video\"}],\"twitter\":[{\"card\":\"player\",\"site\":\"@youtube\",\"url\":\"http://www.youtube.com/watch?v=tlYcUqEPN58\",\"title\":\"Ed Sheeran - SING [Official Video]\",\"description\":\"Official Video for Ed Sheeran's track SING Get this track on iTunes: http://smarturl.it/EdSing Pre-order 'x' on iTunes and get 'One' instantly: http://smartu...\",\"image\":\"https://i1.ytimg.com/vi/tlYcUqEPN58/maxresdefault.jpg\",\"app\":{\"iphone\":{\"name\":\"YouTube\",\"id\":\"544007664\",\"url\":\"vnd.youtube://watch/tlYcUqEPN58\"},\"ipad\":{\"name\":\"YouTube\",\"id\":\"544007664\",\"url\":\"vnd.youtube://watch/tlYcUqEPN58\"},\"googleplay\":{\"name\":\"YouTube\",\"id\":\"com.google.android.youtube\",\"url\":\"http://www.youtube.com/watch?v=tlYcUqEPN58\"}},\"player\":\"https://www.youtube.com/embed/tlYcUqEPN58\",\"player_width\":\"1280\",\"player_height\":\"720\"}]},\"normalized_url\":[\"https://youtube.com/watch?v=tlYcUqEPN58\"],\"retweet_count\":[0],\"tit
le\":[\"Ed Sheeran - SING [Official Video] - YouTube\"],\"url\":[\"https://www.youtube.com/watch?v=tlYcUqEPN58\"]},\"twitter\":{\"id\":\"471420141989666817\",\"retweet\":{\"text\":\"\\\"@Pharrell: Loved working with @edsheeran on Sing. He's a genius. https://t.co/wB2qKyJMRw\\\" @ViiOLeee look at this!\",\"id\":\"471420141989666817\",\"user\":{\"name\":\"Violeta Anguita\",\"description\":\"La vida no seria la fiesta que todos esperamos, pero mientras estemos aqui debemos BAILAR!!! #ErasmusOnceErasmusForever\",\"location\":\"Espanhaa..Olaa!\",\"statuses_count\":5882,\"followers_count\":249,\"friends_count\":1090,\"screen_name\":\"ViiOLeee\",\"profile_image_url\":\"http://pbs.twimg.com/profile_images/378800000851401229/bbf480cde2e9923a1d20acd393da0212_normal.jpeg\",\"profile_image_url_https\":\"https://pbs.twimg.com/profile_images/378800000851401229/bbf480cde2e9923a1d20acd393da0212_normal.jpeg\",\"lang\":\"en\",\"time_zone\":\"Madrid\",\"utc_offset\":7200,\"listed_count\":1,\"id\":709
31384,\"id_str\":\"70931384\",\"geo_enabled\":false,\"verified\":false,\"favourites_count\":275,\"created_at\":\"Wed, 02 Sep 2009 10:19:59 +0000\"},\"source\":\"<a href=\\\"http://twitter.com/download/android\\\" rel=\\\"nofollow\\\">Twitter for Android</a>\",\"count\":1,\"created_at\":\"Tue, 27 May 2014 22:38:15 +0000\",\"mentions\":[\"Pharrell\",\"edsheeran\",\"ViiOLeee\",\"AliiAnguita\"],\"mention_ids\":[338084918,85452649,70931384],\"links\":[\"https://www.youtube.com/watch?v=tlYcUqEPN58\"],\"display_urls\":[\"youtube.com/watch?v=tlYcUq���\"],\"domains\":[\"www.youtube.com\"],\"lang\":\"en\"},\"retweeted\":{\"id\":\"471419867078209536\",\"user\":{\"name\":\"Alicia Anguita \",\"description\":\"Estudiante de Ingenieria de la Edificaci��n en Granada.\",\"statuses_count\":371,\"followers_count\":185,\"friends_count\":404,\"screen_name\":\"AliiAnguita\",\"profile_image_url\":\"http://pbs.twimg.com/profile_images/424248659677442048/qCPZL8c9_normal.jpeg\",\"profile_image_url_
https\":\"https://pbs.twimg.com/profile_images/424248659677442048/qCPZL8c9_normal.jpeg\",\"lang\":\"es\",\"listed_count\":0,\"id\":561201891,\"id_str\":\"561201891\",\"geo_enabled\":false,\"verified\":false,\"favourites_count\":17,\"created_at\":\"Mon, 23 Apr 2012 13:11:44 +0000\"},\"source\":\"<a href=\\\"http://twitter.com/download/android\\\" rel=\\\"nofollow\\\">Twitter for Android</a>\",\"created_at\":\"Tue, 27 May 2014 22:37:09 +0000\"}}}";
-
- public static final String DATASIFT_FORMAT = "EEE, dd MMM yyyy HH:mm:ss Z";
-
- @Test
- public void testTypeConverterStringToString() {
- final String ID = "1";
- StreamsProcessor processor = new TypeConverterProcessor(String.class, String.class, Lists.newArrayList(DATASIFT_FORMAT));
- processor.prepare(null);
- StreamsDatum datum = new StreamsDatum(DATASIFT_JSON, ID);
- List<StreamsDatum> result = processor.process(datum);
- assertNotNull(result);
- assertEquals(1, result.size());
- StreamsDatum resultDatum = result.get(0);
- assertNotNull(resultDatum);
- assertNotNull(resultDatum.getDocument());
- assertTrue(resultDatum.getDocument() instanceof String);
- assertEquals(ID, resultDatum.getId());
- }
-
- @Test
- public void testTypeConverterStringToObjectNode() {
- final String ID = "1";
- StreamsProcessor processor = new TypeConverterProcessor(String.class, ObjectNode.class, Lists.newArrayList(DATASIFT_FORMAT));
- processor.prepare(null);
- StreamsDatum datum = new StreamsDatum(DATASIFT_JSON, ID);
- List<StreamsDatum> result = processor.process(datum);
- assertNotNull(result);
- assertEquals(1, result.size());
- StreamsDatum resultDatum = result.get(0);
- assertNotNull(resultDatum);
- assertNotNull(resultDatum.getDocument());
- assertTrue(resultDatum.getDocument() instanceof ObjectNode);
- assertEquals(ID, resultDatum.getId());
- }
-
- @Test
- public void testTypeConverterObjectNodeToString() throws IOException {
- final String ID = "1";
- StreamsProcessor processor = new TypeConverterProcessor(ObjectNode.class, String.class, Lists.newArrayList(DATASIFT_FORMAT));
- processor.prepare(null);
- ObjectMapper mapper = StreamsJacksonMapper.getInstance(Lists.newArrayList(DATASIFT_FORMAT));
- ObjectNode node = mapper.readValue(DATASIFT_JSON, ObjectNode.class);
- StreamsDatum datum = new StreamsDatum(node, ID);
- List<StreamsDatum> result = processor.process(datum);
- assertNotNull(result);
- assertEquals(1, result.size());
- StreamsDatum resultDatum = result.get(0);
- assertNotNull(resultDatum);
- assertNotNull(resultDatum.getDocument());
- assertTrue(resultDatum.getDocument() instanceof String);
- assertEquals(ID, resultDatum.getId());
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-contrib/streams-provider-datasift/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/pom.xml b/streams-contrib/streams-provider-datasift/pom.xml
index ae96d4e..23165af 100644
--- a/streams-contrib/streams-provider-datasift/pom.xml
+++ b/streams-contrib/streams-provider-datasift/pom.xml
@@ -37,7 +37,7 @@
</dependency>
<dependency>
<groupId>org.apache.streams</groupId>
- <artifactId>streams-processor-jackson</artifactId>
+ <artifactId>streams-converters</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/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 a4a4b5a..bc982e9 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
@@ -23,8 +23,7 @@ import com.google.common.collect.Lists;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.core.StreamsProcessor;
import org.apache.streams.datasift.Datasift;
-import org.apache.streams.datasift.provider.DatasiftConverter;
-import org.apache.streams.datasift.serializer.DatasiftActivitySerializer;
+import org.apache.streams.datasift.serializer.DatasiftActivityConverter;
import org.apache.streams.datasift.util.StreamsDatasiftMapper;
import org.apache.streams.jackson.StreamsJacksonMapper;
import org.apache.streams.pojo.json.Activity;
@@ -42,7 +41,7 @@ public class DatasiftActivitySerializerProcessor implements StreamsProcessor {
private ObjectMapper mapper;
private Class outClass;
- private DatasiftActivitySerializer datasiftActivitySerializer;
+ private DatasiftActivityConverter datasiftActivitySerializer;
public final static String TERMINATE = new String("TERMINATE");
@@ -77,8 +76,8 @@ public class DatasiftActivitySerializerProcessor implements StreamsProcessor {
@Override
public void prepare(Object configurationObject) {
- this.mapper = StreamsJacksonMapper.getInstance(Lists.newArrayList(StreamsDatasiftMapper.DATASIFT_FORMAT));
- this.datasiftActivitySerializer = new DatasiftActivitySerializer();
+ this.mapper = StreamsJacksonMapper.getInstance(StreamsDatasiftMapper.DATASIFT_FORMAT);
+ this.datasiftActivitySerializer = new DatasiftActivityConverter();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/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 1166b2e..a0d2fc3 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
@@ -25,10 +25,10 @@ import org.apache.streams.core.StreamsDatum;
import org.apache.streams.core.StreamsProcessor;
import org.apache.streams.datasift.Datasift;
import org.apache.streams.datasift.provider.DatasiftConverter;
-import org.apache.streams.datasift.serializer.DatasiftActivitySerializer;
+import org.apache.streams.datasift.serializer.DatasiftActivityConverter;
import org.apache.streams.datasift.util.StreamsDatasiftMapper;
import org.apache.streams.pojo.json.Activity;
-import org.apache.streams.jackson.CleanAdditionalPropertiesProcessor;
+import org.apache.streams.converter.CleanAdditionalPropertiesProcessor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,7 +43,7 @@ public class DatasiftTypeConverterProcessor implements StreamsProcessor {
private ObjectMapper mapper;
private Class outClass;
- private DatasiftActivitySerializer datasiftInteractionActivitySerializer;
+ private DatasiftActivityConverter datasiftInteractionActivitySerializer;
private DatasiftConverter converter;
public final static String TERMINATE = new String("TERMINATE");
@@ -75,7 +75,7 @@ public class DatasiftTypeConverterProcessor implements StreamsProcessor {
@Override
public void prepare(Object configurationObject) {
this.mapper = StreamsDatasiftMapper.getInstance();
- this.datasiftInteractionActivitySerializer = new DatasiftActivitySerializer();
+ this.datasiftInteractionActivitySerializer = new DatasiftActivityConverter();
if(this.outClass.equals(Activity.class)) {
this.converter = new ActivityConverter();
} else if (this.outClass.equals(String.class)) {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftActivityConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftActivityConverter.java b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftActivityConverter.java
new file mode 100644
index 0000000..65bebce
--- /dev/null
+++ b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftActivityConverter.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.streams.datasift.serializer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Lists;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.data.ActivityConverterFactory;
+import org.apache.streams.datasift.Datasift;
+import org.apache.streams.datasift.twitter.Twitter;
+import org.apache.streams.datasift.util.StreamsDatasiftMapper;
+import org.apache.streams.exceptions.ActivitySerializerException;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+
+import java.util.List;
+
+/**
+ * We should be able to @Deprecate this soon and adopt ActivityConverterProcessor
+ */
+public class DatasiftActivityConverter implements ActivityConverter<Datasift> {
+
+ private static final ObjectMapper MAPPER = StreamsJacksonMapper.getInstance(StreamsDatasiftMapper.DATASIFT_FORMAT);
+
+ private static DatasiftActivityConverter instance = new DatasiftActivityConverter();
+
+ public static DatasiftActivityConverter getInstance() {
+ return instance;
+ }
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public Datasift serialize(Activity deserialized) throws ActivitySerializerException {
+ return null;
+ }
+
+ @Override
+ public Activity deserialize(Datasift serialized) throws ActivitySerializerException {
+ Class detectedClass = DatasiftEventClassifier.getInstance().detectClass(serialized);
+ Class converterClass = DatasiftConverterResolver.getInstance().bestSerializer(detectedClass);
+ ActivityConverter serializer = ActivityConverterFactory.getInstance(converterClass);
+ return serializer.deserialize(serialized);
+ }
+
+ public Activity deserialize(String json) throws ActivitySerializerException {
+ try {
+ return deserialize(MAPPER.readValue(json, Datasift.class));
+ } catch (Exception e) {
+ throw new ActivitySerializerException(e);
+ }
+ }
+
+ @Override
+ public List<Activity> deserializeAll(List<Datasift> serializedList) {
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftActivitySerializer.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftActivitySerializer.java b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftActivitySerializer.java
deleted file mode 100644
index b587cd6..0000000
--- a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftActivitySerializer.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.streams.datasift.serializer;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.streams.data.ActivitySerializer;
-import org.apache.streams.datasift.Datasift;
-import org.apache.streams.datasift.util.StreamsDatasiftMapper;
-import org.apache.streams.exceptions.ActivitySerializerException;
-import org.apache.streams.pojo.json.Activity;
-
-import java.util.List;
-
-/**
- *
- */
-public class DatasiftActivitySerializer implements ActivitySerializer<Datasift> {
-
- private static final ObjectMapper MAPPER = StreamsDatasiftMapper.getInstance();
-
- @Override
- public String serializationFormat() {
- return null;
- }
-
- @Override
- public Datasift serialize(Activity deserialized) throws ActivitySerializerException {
- return null;
- }
-
- @Override
- public Activity deserialize(Datasift serialized) throws ActivitySerializerException {
- ActivitySerializer serializer = DatasiftEventClassifier.bestSerializer(serialized);
- return serializer.deserialize(serialized);
- }
-
- public Activity deserialize(String json) throws ActivitySerializerException {
- try {
- return deserialize(MAPPER.readValue(json, Datasift.class));
- } catch (Exception e) {
- throw new ActivitySerializerException(e);
- }
- }
-
- @Override
- public List<Activity> deserializeAll(List<Datasift> serializedList) {
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/aad0e887/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftConverterResolver.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftConverterResolver.java b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftConverterResolver.java
new file mode 100644
index 0000000..67cccd8
--- /dev/null
+++ b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftConverterResolver.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.streams.datasift.serializer;
+
+import com.google.common.base.Preconditions;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.data.ActivityConverterResolver;
+import org.apache.streams.data.DocumentClassifier;
+import org.apache.streams.datasift.Datasift;
+import org.apache.streams.datasift.instagram.Instagram;
+import org.apache.streams.datasift.interaction.Interaction;
+import org.apache.streams.datasift.twitter.Twitter;
+
+/**
+ * Created by sblackmon on 11/6/14.
+ */
+public class DatasiftConverterResolver implements ActivityConverterResolver {
+
+ public DatasiftConverterResolver() {
+
+ }
+
+ private static DatasiftConverterResolver instance = new DatasiftConverterResolver();
+
+ public static DatasiftConverterResolver getInstance() {
+ return instance;
+ }
+
+ public Class bestSerializer(Class documentClass) {
+
+ if(documentClass == Twitter.class) {
+ return DatasiftTwitterActivityConverter.class;
+ } else if(documentClass == Instagram.class) {
+ return DatasiftInstagramActivityConverter.class;
+ } else {
+ return DatasiftInteractionActivityConverter.class;
+ }
+ }
+}