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/03/26 18:46:46 UTC
[5/8] incubator-streams git commit: updates to make this module use
DocumentClassifier and ActivityConverter interfaces and reflection-based
DateTime handling and Activity conversion. tweaks and renames to the tests to
match the new model. this change im
updates to make this module use DocumentClassifier and ActivityConverter interfaces and reflection-based DateTime handling and Activity conversion.
tweaks and renames to the tests to match the new model.
this change impacts testing in streams-runtime-pig, so PR submitted for STREAMS-278 must merge first.
resolves STREAMS-276
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/2cc6118f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/2cc6118f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/2cc6118f
Branch: refs/heads/master
Commit: 2cc6118f17aaaaa0ced4ea35f3910f90b05b0889
Parents: 8050869
Author: sblackmon <sb...@w2odigital.com>
Authored: Tue Feb 10 15:35:34 2015 -0600
Committer: Steve Blackmon (@steveblackmon) <sb...@apache.org>
Committed: Tue Mar 24 18:24:53 2015 -0500
----------------------------------------------------------------------
.../DatasiftTwitterActivitySerializer.java | 4 +-
.../streams-provider-twitter/pom.xml | 36 +-
.../twitter/converter/StreamsTwitterMapper.java | 87 +++++
.../converter/TwitterDateTimeFormat.java | 34 ++
.../converter/TwitterDocumentClassifier.java | 77 ++++
.../TwitterJsonDeleteActivityConverter.java | 89 +++++
.../TwitterJsonRetweetActivityConverter.java | 83 +++++
.../TwitterJsonTweetActivityConverter.java | 84 +++++
.../TwitterJsonUserActivityConverter.java | 82 +++++
...terJsonUserstreameventActivityConverter.java | 120 +++++++
.../converter/util/TwitterActivityUtil.java | 349 +++++++++++++++++++
.../FetchAndReplaceTwitterProcessor.java | 8 +-
.../processor/TwitterEventProcessor.java | 118 +------
.../processor/TwitterProfileProcessor.java | 2 +-
.../twitter/processor/TwitterTypeConverter.java | 183 +---------
.../provider/TwitterEventClassifier.java | 38 +-
.../provider/TwitterStreamProcessor.java | 2 +-
.../serializer/StreamsTwitterMapper.java | 94 -----
.../TwitterJsonActivitySerializer.java | 70 ----
.../TwitterJsonDeleteActivitySerializer.java | 94 -----
.../TwitterJsonRetweetActivitySerializer.java | 79 -----
.../TwitterJsonTweetActivitySerializer.java | 77 ----
.../TwitterJsonUserActivitySerializer.java | 78 -----
...erJsonUserstreameventActivitySerializer.java | 115 ------
.../serializer/util/TwitterActivityUtil.java | 346 ------------------
.../src/main/jsonschema/com/twitter/User.json | 123 +++++++
.../src/main/jsonschema/com/twitter/tweet.json | 126 +------
.../streams/twitter/test/SimpleTweetTest.java | 121 -------
.../twitter/test/TweetActivitySerDeTest.java | 139 --------
.../streams/twitter/test/TweetSerDeTest.java | 140 --------
.../TwitterActivityConverterProcessorTest.java | 115 ++++++
.../test/TwitterActivityConvertersTest.java | 149 ++++++++
.../test/TwitterDocumentClassifierTest.java | 83 +++++
.../test/TwitterEventClassifierTest.java | 100 ------
.../twitter/test/TwitterObjectMapperTest.java | 135 +++++++
.../twitter/test/TwitterStreamProviderTest.java | 58 ---
36 files changed, 1641 insertions(+), 1997 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftTwitterActivitySerializer.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftTwitterActivitySerializer.java b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftTwitterActivitySerializer.java
index 05e1b35..8242da9 100644
--- a/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftTwitterActivitySerializer.java
+++ b/streams-contrib/streams-provider-datasift/src/main/java/org/apache/streams/datasift/serializer/DatasiftTwitterActivitySerializer.java
@@ -33,7 +33,7 @@ import org.apache.streams.pojo.extensions.ExtensionUtil;
import org.apache.streams.pojo.json.Activity;
import org.apache.streams.pojo.json.Actor;
import org.apache.streams.pojo.json.Image;
-import org.apache.streams.twitter.serializer.util.TwitterActivityUtil;
+import org.apache.streams.twitter.converter.util.TwitterActivityUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,8 +42,6 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import static org.apache.streams.data.util.ActivityUtil.ensureExtensions;
-
/**
*
*/
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/pom.xml b/streams-contrib/streams-provider-twitter/pom.xml
index 4ce1204..0f9d361 100644
--- a/streams-contrib/streams-provider-twitter/pom.xml
+++ b/streams-contrib/streams-provider-twitter/pom.xml
@@ -1,33 +1,15 @@
<?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>
+<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">
<parent>
<groupId>org.apache.streams</groupId>
<artifactId>streams-contrib</artifactId>
<version>0.2-incubating-SNAPSHOT</version>
</parent>
+ <modelVersion>4.0.0</modelVersion>
<artifactId>streams-provider-twitter</artifactId>
- <name>${project.artifactId}</name>
-
- <description>Twitter Provider</description>
<dependencies>
<dependency>
@@ -68,9 +50,8 @@
</dependency>
<dependency>
<groupId>org.apache.streams</groupId>
- <artifactId>streams-processor-jackson</artifactId>
+ <artifactId>streams-converters</artifactId>
<version>${project.version}</version>
- <scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.streams</groupId>
@@ -99,7 +80,11 @@
<artifactId>hbc-core</artifactId>
<version>2.1.0</version>
</dependency>
-
+ <dependency>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ <version>1.2</version>
+ </dependency>
<dependency>
<groupId>org.twitter4j</groupId>
<artifactId>twitter4j-core</artifactId>
@@ -160,6 +145,7 @@
<sourcePath>src/main/jsonschema/com/twitter/FriendList.json</sourcePath>
<sourcePath>src/main/jsonschema/com/twitter/Retweet.json</sourcePath>
<sourcePath>src/main/jsonschema/com/twitter/tweet.json</sourcePath>
+ <sourcePath>src/main/jsonschema/com/twitter/User.json</sourcePath>
</sourcePaths>
<outputDirectory>target/generated-sources/jsonschema2pojo</outputDirectory>
<targetPackage>org.apache.streams.twitter.pojo</targetPackage>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/StreamsTwitterMapper.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/StreamsTwitterMapper.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/StreamsTwitterMapper.java
new file mode 100644
index 0000000..2662b69
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/StreamsTwitterMapper.java
@@ -0,0 +1,87 @@
+/*
+ * 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.twitter.converter;
+
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationContext;
+import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import org.apache.streams.data.util.RFC3339Utils;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.joda.time.DateTime;
+import org.joda.time.format.DateTimeFormat;
+import org.joda.time.format.DateTimeFormatter;
+
+import java.io.IOException;
+
+/**
+ * This class assist with handling twitter's date-time format during conversion
+ *
+ * Deprecated: use StreamsJacksonMapper.getInstance() with TwitterDateTimeFormat on the classpath instead
+ */
+@Deprecated
+public class StreamsTwitterMapper extends StreamsJacksonMapper {
+
+ public static final String TWITTER_FORMAT = "EEE MMM dd HH:mm:ss Z yyyy";
+
+ public static final DateTimeFormatter TWITTER_FORMATTER = DateTimeFormat.forPattern(TWITTER_FORMAT);
+
+ public static final Long getMillis(String dateTime) {
+
+ // this function is for pig which doesn't handle exceptions well
+ try {
+ Long result = TWITTER_FORMATTER.parseMillis(dateTime);
+ return result;
+ } catch( Exception e ) {
+ return null;
+ }
+
+ }
+
+ private static final StreamsTwitterMapper INSTANCE = new StreamsTwitterMapper();
+
+ public static StreamsTwitterMapper getInstance(){
+ return INSTANCE;
+ }
+
+ public StreamsTwitterMapper() {
+ super();
+ registerModule(new SimpleModule()
+ {
+ {
+ addDeserializer(DateTime.class, new StdDeserializer<DateTime>(DateTime.class) {
+ @Override
+ public DateTime deserialize(JsonParser jpar, DeserializationContext context) throws IOException, JsonProcessingException {
+ DateTime result = null;
+ try {
+ result = TWITTER_FORMATTER.parseDateTime(jpar.getValueAsString());
+ } catch( Exception e ) { }
+ try {
+ result = RFC3339Utils.getInstance().parseToUTC(jpar.getValueAsString());
+ } catch( Exception e ) { }
+ return result;
+ }
+ });
+ }
+ });
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterDateTimeFormat.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterDateTimeFormat.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterDateTimeFormat.java
new file mode 100644
index 0000000..133db32
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterDateTimeFormat.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ *
+ * 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.twitter.converter;
+
+import org.apache.streams.jackson.StreamsDateTimeFormat;
+
+/**
+ * Created by sblackmon on 2/10/15.
+ */
+public class TwitterDateTimeFormat implements StreamsDateTimeFormat {
+
+ public static final String TWITTER_FORMAT = "EEE MMM dd HH:mm:ss Z yyyy";
+
+ @Override
+ public String getFormat() {
+ return TWITTER_FORMAT;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterDocumentClassifier.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterDocumentClassifier.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterDocumentClassifier.java
new file mode 100644
index 0000000..ee5fa1d
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterDocumentClassifier.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.twitter.converter;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import org.apache.streams.data.DocumentClassifier;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Follow;
+import org.apache.streams.twitter.pojo.*;
+
+import java.io.IOException;
+import java.util.List;
+
+/**
+ * Ensures twitter documents can be converted to Activity
+ */
+public class TwitterDocumentClassifier implements DocumentClassifier {
+
+ private static ObjectMapper mapper;
+
+ public List<Class> detectClasses(Object document) {
+
+ Preconditions.checkNotNull(document);
+ Preconditions.checkArgument(document instanceof String || document instanceof ObjectNode);
+
+ mapper = new StreamsJacksonMapper(Lists.newArrayList(StreamsTwitterMapper.TWITTER_FORMAT));
+
+ ObjectNode objectNode;
+ try {
+ if( document instanceof String )
+ objectNode = mapper.readValue((String)document, ObjectNode.class);
+ else
+ objectNode = (ObjectNode) document;
+ } catch (IOException e) {
+ e.printStackTrace();
+ return null;
+ }
+
+ List<Class> classList = Lists.newArrayList();
+
+ if( objectNode.findValue("retweeted_status") != null && objectNode.get("retweeted_status") != null)
+ classList.add(Retweet.class);
+ else if( objectNode.findValue("delete") != null )
+ classList.add(Delete.class);
+ else if( objectNode.findValue("friends") != null ||
+ objectNode.findValue("friends_str") != null )
+ classList.add(FriendList.class);
+ else if( objectNode.findValue("target_object") != null )
+ classList.add(UserstreamEvent.class);
+ else if ( objectNode.findValue("location") != null && objectNode.findValue("user") == null)
+ classList.add(User.class);
+ else
+ classList.add(Tweet.class);
+
+ return classList;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonDeleteActivityConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonDeleteActivityConverter.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonDeleteActivityConverter.java
new file mode 100644
index 0000000..eb35b71
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonDeleteActivityConverter.java
@@ -0,0 +1,89 @@
+/*
+ * 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.twitter.converter;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.exceptions.ActivityConversionException;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+import org.apache.streams.twitter.pojo.Delete;
+import org.apache.streams.twitter.pojo.Tweet;
+
+import java.io.Serializable;
+import java.util.List;
+
+import static org.apache.streams.twitter.converter.util.TwitterActivityUtil.*;
+
+
+/**
+* Created with IntelliJ IDEA.
+* User: mdelaet
+* Date: 9/30/13
+* Time: 9:24 AM
+* To change this template use File | Settings | File Templates.
+*/
+public class TwitterJsonDeleteActivityConverter implements ActivityConverter<Delete>, Serializable {
+
+ public static Class requiredClass = Delete.class;
+
+ @Override
+ public Class requiredClass() {
+ return requiredClass;
+ }
+
+ private static TwitterJsonDeleteActivityConverter instance = new TwitterJsonDeleteActivityConverter();
+
+ public static TwitterJsonDeleteActivityConverter getInstance() {
+ return instance;
+ }
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public Delete fromActivity(Activity deserialized) throws ActivityConversionException {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public List<Activity> toActivityList(List<Delete> serializedList) {
+ throw new NotImplementedException();
+ }
+
+ public List<Activity> toActivityList(Delete delete) throws ActivityConversionException {
+
+ Activity activity = new Activity();
+ updateActivity(delete, activity);
+ return Lists.newArrayList(activity);
+ }
+
+ @Override
+ public List<Delete> fromActivityList(List<Activity> list) {
+ throw new NotImplementedException();
+ }
+
+ public ActivityObject buildTarget(Tweet tweet) {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonRetweetActivityConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonRetweetActivityConverter.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonRetweetActivityConverter.java
new file mode 100644
index 0000000..559dfd6
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonRetweetActivityConverter.java
@@ -0,0 +1,83 @@
+/*
+ * 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.twitter.converter;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.exceptions.ActivityConversionException;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.twitter.pojo.Retweet;
+
+import java.io.Serializable;
+import java.util.List;
+
+import static org.apache.streams.twitter.converter.util.TwitterActivityUtil.*;
+
+public class TwitterJsonRetweetActivityConverter implements ActivityConverter<Retweet>, Serializable {
+
+ public static Class requiredClass = Retweet.class;
+
+ @Override
+ public Class requiredClass() {
+ return requiredClass;
+ }
+
+ private static TwitterJsonRetweetActivityConverter instance = new TwitterJsonRetweetActivityConverter();
+
+ public static TwitterJsonRetweetActivityConverter getInstance() {
+ return instance;
+ }
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public Retweet fromActivity(Activity deserialized) throws ActivityConversionException {
+ return null;
+ }
+
+ @Override
+ public List<Activity> toActivityList(Retweet retweet) throws ActivityConversionException {
+
+ Activity activity = new Activity();
+ updateActivity(retweet, activity);
+
+ return Lists.newArrayList(activity);
+ }
+
+ @Override
+ public List<Retweet> fromActivityList(List<Activity> list) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public List<Activity> toActivityList(List<Retweet> serializedList) {
+ List<Activity> result = Lists.newArrayList();
+ for( Retweet item : serializedList ) {
+ try {
+ List<Activity> activities = toActivityList(item);
+ result.addAll(activities);
+ } catch (ActivityConversionException e) {}
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonTweetActivityConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonTweetActivityConverter.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonTweetActivityConverter.java
new file mode 100644
index 0000000..0b9f0ec
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonTweetActivityConverter.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
+ *
+ * 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.twitter.converter;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.exceptions.ActivityConversionException;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.twitter.pojo.Tweet;
+
+import java.io.Serializable;
+import java.util.List;
+
+import static org.apache.streams.twitter.converter.util.TwitterActivityUtil.*;
+
+public class TwitterJsonTweetActivityConverter implements ActivityConverter<Tweet>, Serializable {
+
+ public static Class requiredClass = Tweet.class;
+
+ @Override
+ public Class requiredClass() {
+ return requiredClass;
+ }
+
+ private static TwitterJsonTweetActivityConverter instance = new TwitterJsonTweetActivityConverter();
+
+ public static TwitterJsonTweetActivityConverter getInstance() {
+ return instance;
+ }
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public Tweet fromActivity(Activity deserialized) throws ActivityConversionException {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public List<Activity> toActivityList(Tweet tweet) throws ActivityConversionException {
+
+ Activity activity = new Activity();
+
+ updateActivity(tweet, activity);
+
+ return Lists.newArrayList(activity);
+ }
+
+ @Override
+ public List<Tweet> fromActivityList(List<Activity> list) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public List<Activity> toActivityList(List<Tweet> serializedList) {
+ List<Activity> result = Lists.newArrayList();
+ for( Tweet item : serializedList ) {
+ try {
+ List<Activity> activities = toActivityList(item);
+ result.addAll(activities);
+ } catch (ActivityConversionException e) {}
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonUserActivityConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonUserActivityConverter.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonUserActivityConverter.java
new file mode 100644
index 0000000..1b8ef8f
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonUserActivityConverter.java
@@ -0,0 +1,82 @@
+/*
+ * 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.twitter.converter;
+
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.exceptions.ActivityConversionException;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.twitter.pojo.User;
+
+import java.util.List;
+
+import static org.apache.streams.twitter.converter.util.TwitterActivityUtil.updateActivity;
+
+public class TwitterJsonUserActivityConverter implements ActivityConverter<User> {
+
+ public static Class requiredClass = User.class;
+
+ @Override
+ public Class requiredClass() {
+ return requiredClass;
+ }
+
+ private static TwitterJsonUserActivityConverter instance = new TwitterJsonUserActivityConverter();
+
+ public static TwitterJsonUserActivityConverter getInstance() {
+ return instance;
+ }
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public User fromActivity(Activity deserialized) throws ActivityConversionException {
+ return null;
+ }
+
+ @Override
+ public List<Activity> toActivityList(User user) throws ActivityConversionException {
+
+ Activity activity = new Activity();
+ updateActivity(user, activity);
+
+ return Lists.newArrayList(activity);
+ }
+
+ @Override
+ public List<User> fromActivityList(List<Activity> list) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public List<Activity> toActivityList(List<User> serializedList) {
+ List<Activity> result = Lists.newArrayList();
+ for( User item : serializedList ) {
+ try {
+ List<Activity> activities = toActivityList(item);
+ result.addAll(activities);
+ } catch (ActivityConversionException e) {}
+ }
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonUserstreameventActivityConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonUserstreameventActivityConverter.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonUserstreameventActivityConverter.java
new file mode 100644
index 0000000..357c41c
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/TwitterJsonUserstreameventActivityConverter.java
@@ -0,0 +1,120 @@
+/*
+ * 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.twitter.converter;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.streams.data.ActivityConverter;
+import org.apache.streams.exceptions.ActivityConversionException;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+import org.apache.streams.pojo.json.Actor;
+import org.apache.streams.twitter.pojo.UserstreamEvent;
+
+import java.util.List;
+
+import static org.apache.streams.twitter.converter.util.TwitterActivityUtil.*;
+
+
+/**
+* Created with IntelliJ IDEA.
+* User: mdelaet
+* Date: 9/30/13
+* Time: 9:24 AM
+* To change this template use File | Settings | File Templates.
+*/
+public class TwitterJsonUserstreameventActivityConverter implements ActivityConverter<UserstreamEvent> {
+
+ public static Class requiredClass = UserstreamEvent.class;
+
+ @Override
+ public Class requiredClass() {
+ return requiredClass;
+ }
+
+ private static TwitterJsonUserstreameventActivityConverter instance = new TwitterJsonUserstreameventActivityConverter();
+
+ public static TwitterJsonUserstreameventActivityConverter getInstance() {
+ return instance;
+ }
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public UserstreamEvent fromActivity(Activity deserialized) throws ActivityConversionException {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public List<Activity> toActivityList(UserstreamEvent userstreamEvent) throws ActivityConversionException {
+
+ Activity activity = convert(userstreamEvent);
+ return Lists.newArrayList(activity);
+
+ }
+
+ @Override
+ public List<UserstreamEvent> fromActivityList(List<Activity> list) {
+ throw new NotImplementedException();
+ }
+
+ @Override
+ public List<Activity> toActivityList(List<UserstreamEvent> serializedList) {
+ return null;
+ }
+
+ public Activity convert(UserstreamEvent event) throws ActivityConversionException {
+
+ Activity activity = new Activity();
+ activity.setActor(buildActor(event));
+ activity.setVerb(detectVerb(event));
+ activity.setObject(buildActivityObject(event));
+ activity.setId(formatId(activity.getVerb()));
+ if(Strings.isNullOrEmpty(activity.getId()))
+ throw new ActivityConversionException("Unable to determine activity id");
+ activity.setProvider(getProvider());
+ return activity;
+ }
+
+ public Actor buildActor(UserstreamEvent event) {
+ Actor actor = new Actor();
+ //actor.setId(formatId(delete.getDelete().getStatus().getUserIdStr()));
+ return actor;
+ }
+
+ public ActivityObject buildActivityObject(UserstreamEvent event) {
+ ActivityObject actObj = new ActivityObject();
+ //actObj.setId(formatId(delete.getDelete().getStatus().getIdStr()));
+ //actObj.setObjectType("tweet");
+ return actObj;
+ }
+
+ public String detectVerb(UserstreamEvent event) {
+ return null;
+ }
+
+ public ActivityObject buildTarget(UserstreamEvent event) {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/util/TwitterActivityUtil.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/util/TwitterActivityUtil.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/util/TwitterActivityUtil.java
new file mode 100644
index 0000000..dc0ec88
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/converter/util/TwitterActivityUtil.java
@@ -0,0 +1,349 @@
+/*
+ * 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.twitter.converter.util;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.Joiner;
+import com.google.common.base.Optional;
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+import org.apache.streams.exceptions.ActivityConversionException;
+import org.apache.streams.pojo.extensions.ExtensionUtil;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.ActivityObject;
+import org.apache.streams.pojo.json.Actor;
+import org.apache.streams.pojo.json.Image;
+import org.apache.streams.pojo.json.Provider;
+import org.apache.streams.twitter.Url;
+import org.apache.streams.twitter.pojo.Delete;
+import org.apache.streams.twitter.pojo.Entities;
+import org.apache.streams.twitter.pojo.Hashtag;
+import org.apache.streams.twitter.pojo.Retweet;
+import org.apache.streams.twitter.pojo.Tweet;
+import org.apache.streams.twitter.pojo.User;
+import org.apache.streams.twitter.pojo.UserMentions;
+import org.apache.streams.twitter.converter.StreamsTwitterMapper;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.streams.data.util.ActivityUtil.ensureExtensions;
+
+/**
+ * Provides utilities for working with Activity objects within the context of Twitter
+ */
+public class TwitterActivityUtil {
+
+ /**
+ * Updates the given Activity object with the values from the Tweet
+ * @param tweet the object to use as the source
+ * @param activity the target of the updates. Will receive all values from the tweet.
+ * @throws org.apache.streams.exceptions.ActivityConversionException
+ */
+ public static void updateActivity(Tweet tweet, Activity activity) throws ActivityConversionException {
+ ObjectMapper mapper = StreamsTwitterMapper.getInstance();
+ activity.setActor(buildActor(tweet));
+ activity.setId(formatId(activity.getVerb(),
+ Optional.fromNullable(
+ tweet.getIdStr())
+ .or(Optional.of(tweet.getId().toString()))
+ .orNull()));
+
+ if(tweet instanceof Retweet) {
+ updateActivityContent(activity, ((Retweet) tweet).getRetweetedStatus(), "share");
+ } else {
+ updateActivityContent(activity, tweet, "post");
+ }
+
+ if(Strings.isNullOrEmpty(activity.getId()))
+ throw new ActivityConversionException("Unable to determine activity id");
+ try {
+ activity.setPublished(tweet.getCreatedAt());
+ } catch( Exception e ) {
+ throw new ActivityConversionException("Unable to determine publishedDate", e);
+ }
+ activity.setTarget(buildTarget(tweet));
+ activity.setProvider(getProvider());
+ activity.setUrl(String.format("http://twitter.com/%s/%s/%s", tweet.getUser().getScreenName(),"/status/",tweet.getIdStr()));
+
+ addTwitterExtension(activity, mapper.convertValue(tweet, ObjectNode.class));
+ }
+
+ /**
+ * Updates the given Activity object with the values from the User
+ * @param user the object to use as the source
+ * @param activity the target of the updates. Will receive all values from the tweet.
+ * @throws org.apache.streams.exceptions.ActivityConversionException
+ */
+ public static void updateActivity(User user, Activity activity) {
+ activity.setActor(buildActor(user));
+ activity.setId(null);
+ activity.setVerb(null);
+ }
+
+ /**
+ * Updates the activity for a delete event
+ * @param delete the delete event
+ * @param activity the Activity object to update
+ * @throws org.apache.streams.exceptions.ActivityConversionException
+ */
+ public static void updateActivity(Delete delete, Activity activity) throws ActivityConversionException {
+ activity.setActor(buildActor(delete));
+ activity.setVerb("delete");
+ activity.setObject(buildActivityObject(delete));
+ activity.setId(formatId(activity.getVerb(), delete.getDelete().getStatus().getIdStr()));
+ if(Strings.isNullOrEmpty(activity.getId()))
+ throw new ActivityConversionException("Unable to determine activity id");
+ activity.setProvider(getProvider());
+ addTwitterExtension(activity, StreamsTwitterMapper.getInstance().convertValue(delete, ObjectNode.class));
+ }
+
+ /**
+ * Builds the actor for a delete event
+ * @param delete the delete event
+ * @return a valid Actor
+ */
+ public static Actor buildActor(Delete delete) {
+ Actor actor = new Actor();
+ actor.setId(formatId(delete.getDelete().getStatus().getUserIdStr()));
+ actor.setObjectType("page");
+ return actor;
+ }
+
+ /**
+ * Builds the ActivityObject for the delete event
+ * @param delete the delete event
+ * @return a valid Activity Object
+ */
+ public static ActivityObject buildActivityObject(Delete delete) {
+ ActivityObject actObj = new ActivityObject();
+ actObj.setId(formatId(delete.getDelete().getStatus().getIdStr()));
+ actObj.setObjectType("tweet");
+ return actObj;
+ }
+
+
+ /**
+ * Updates the content, and associated fields, with those from the given tweet
+ * @param activity the target of the updates. Will receive all values from the tweet.
+ * @param tweet the object to use as the source
+ * @param verb the verb for the given activity's type
+ */
+ public static void updateActivityContent(Activity activity, Tweet tweet, String verb) {
+ activity.setVerb(verb);
+ activity.setTitle("");
+ if( tweet != null ) {
+ activity.setObject(buildActivityObject(tweet));
+ activity.setLinks(getLinks(tweet));
+ activity.setContent(tweet.getText());
+ addLocationExtension(activity, tweet);
+ addTwitterExtensions(activity, tweet);
+ }
+ }
+
+ /**
+ * Creates an {@link org.apache.streams.pojo.json.ActivityObject} for the tweet
+ * @param tweet the object to use as the source
+ * @return a valid ActivityObject
+ */
+ public static ActivityObject buildActivityObject(Tweet tweet) {
+ ActivityObject actObj = new ActivityObject();
+ String id = Optional.fromNullable(
+ tweet.getIdStr())
+ .or(Optional.of(tweet.getId().toString()))
+ .orNull();
+ if( id != null )
+ actObj.setId(id);
+ actObj.setObjectType("post");
+ actObj.setContent(tweet.getText());
+ return actObj;
+ }
+
+ /**
+ * Builds the activity {@link org.apache.streams.pojo.json.Actor} object from the tweet
+ * @param tweet the object to use as the source
+ * @return a valid Actor populated from the Tweet
+ */
+ public static Actor buildActor(Tweet tweet) {
+ Actor actor = new Actor();
+ User user = tweet.getUser();
+
+ return buildActor(user);
+ }
+
+ /**
+ * Builds the activity {@link org.apache.streams.pojo.json.Actor} object from the User
+ * @param user the object to use as the source
+ * @return a valid Actor populated from the Tweet
+ */
+ public static Actor buildActor(User user) {
+ Actor actor = new Actor();
+ actor.setId(formatId(
+ Optional.fromNullable(
+ user.getIdStr())
+ .or(Optional.of(user.getId().toString()))
+ .orNull()
+ ));
+ actor.setObjectType("page");
+ actor.setDisplayName(user.getName());
+ actor.setAdditionalProperty("handle", user.getScreenName());
+ actor.setSummary(user.getDescription());
+
+ if (user.getUrl()!=null){
+ actor.setUrl(user.getUrl());
+ }
+
+ Map<String, Object> extensions = new HashMap<String, Object>();
+ extensions.put("location", user.getLocation());
+ extensions.put("posts", user.getStatusesCount());
+ extensions.put("favorites", user.getFavouritesCount());
+ extensions.put("followers", user.getFollowersCount());
+
+ Image profileImage = new Image();
+ profileImage.setUrl(user.getProfileImageUrlHttps());
+ actor.setImage(profileImage);
+
+ extensions.put("screenName", user.getScreenName());
+
+ actor.setAdditionalProperty("extensions", extensions);
+ return actor;
+ }
+
+ /**
+ * Gets the links from the Twitter event
+ * @param tweet the object to use as the source
+ * @return a list of links corresponding to the expanded URL (no t.co)
+ */
+ public static List<String> getLinks(Tweet tweet) {
+ List<String> links = Lists.newArrayList();
+ if( tweet.getEntities().getUrls() != null ) {
+ for (Url url : tweet.getEntities().getUrls()) {
+ links.add(url.getExpandedUrl());
+ }
+ }
+ else
+ System.out.println(" 0 links");
+ return links;
+ }
+
+ /**
+ * Builds the {@link org.apache.streams.twitter.pojo.TargetObject} from the tweet
+ * @param tweet the object to use as the source
+ * @return currently returns null for all activities
+ */
+ public static ActivityObject buildTarget(Tweet tweet) {
+ return null;
+ }
+
+ /**
+ * Adds the location extension and populates with teh twitter data
+ * @param activity the Activity object to update
+ * @param tweet the object to use as the source
+ */
+ public static void addLocationExtension(Activity activity, Tweet tweet) {
+ Map<String, Object> extensions = ensureExtensions(activity);
+ Map<String, Object> location = new HashMap<String, Object>();
+ location.put("id", formatId(
+ Optional.fromNullable(
+ tweet.getIdStr())
+ .or(Optional.of(tweet.getId().toString()))
+ .orNull()
+ ));
+ location.put("coordinates", tweet.getCoordinates());
+ extensions.put("location", location);
+ }
+
+ /**
+ * Gets the common twitter {@link org.apache.streams.pojo.json.Provider} object
+ * @return a provider object representing Twitter
+ */
+ public static Provider getProvider() {
+ Provider provider = new Provider();
+ provider.setId("id:providers:twitter");
+ provider.setObjectType("application");
+ provider.setDisplayName("Twitter");
+ return provider;
+ }
+ /**
+ * Adds the given Twitter event to the activity as an extension
+ * @param activity the Activity object to update
+ * @param event the Twitter event to add as the extension
+ */
+ public static void addTwitterExtension(Activity activity, ObjectNode event) {
+ Map<String, Object> extensions = org.apache.streams.data.util.ActivityUtil.ensureExtensions(activity);
+ extensions.put("twitter", event);
+ }
+ /**
+ * Formats the ID to conform with the Apache Streams activity ID convention
+ * @param idparts the parts of the ID to join
+ * @return a valid Activity ID in format "id:twitter:part1:part2:...partN"
+ */
+ public static String formatId(String... idparts) {
+ return Joiner.on(":").join(Lists.asList("id:twitter", idparts));
+ }
+
+ /**
+ * Takes various parameters from the twitter object that are currently not part of teh
+ * activity schema and stores them in a generic extensions attribute
+ * @param activity
+ * @param tweet
+ */
+ public static void addTwitterExtensions(Activity activity, Tweet tweet) {
+ Map<String, Object> extensions = ensureExtensions(activity);
+
+ List<String> hashtags = new ArrayList<String>();
+ for(Hashtag hashtag : tweet.getEntities().getHashtags()) {
+ hashtags.add(hashtag.getText());
+ }
+ extensions.put("hashtags", hashtags);
+
+ Map<String, Object> likes = new HashMap<String, Object>();
+ likes.put("perspectival", tweet.getFavorited());
+ likes.put("count", tweet.getAdditionalProperties().get("favorite_count"));
+
+ extensions.put("likes", likes);
+
+ Map<String, Object> rebroadcasts = new HashMap<String, Object>();
+ rebroadcasts.put("perspectival", tweet.getRetweeted());
+ rebroadcasts.put("count", tweet.getRetweetCount());
+
+ extensions.put("rebroadcasts", rebroadcasts);
+
+ List<Map<String, Object>> userMentions = new ArrayList<Map<String, Object>>();
+ Entities entities = tweet.getEntities();
+
+ for(UserMentions user : entities.getUserMentions()) {
+ //Map the twitter user object into an actor
+ Map<String, Object> actor = new HashMap<String, Object>();
+ actor.put("id", "id:twitter:" + user.getIdStr());
+ actor.put("displayName", user.getName());
+ actor.put("handle", user.getScreenName());
+
+ userMentions.add(actor);
+ }
+
+ extensions.put("user_mentions", userMentions);
+
+ extensions.put("keywords", tweet.getText());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/FetchAndReplaceTwitterProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/FetchAndReplaceTwitterProcessor.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/FetchAndReplaceTwitterProcessor.java
index 4ca73df..cd6f6e5 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/FetchAndReplaceTwitterProcessor.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/FetchAndReplaceTwitterProcessor.java
@@ -24,7 +24,7 @@ import com.google.common.collect.Lists;
import org.apache.streams.config.StreamsConfigurator;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.core.StreamsProcessor;
-import org.apache.streams.exceptions.ActivitySerializerException;
+import org.apache.streams.exceptions.ActivityConversionException;
import org.apache.streams.pojo.json.Activity;
import org.apache.streams.twitter.TwitterStreamConfiguration;
import org.apache.streams.twitter.pojo.Delete;
@@ -32,7 +32,7 @@ import org.apache.streams.twitter.pojo.Retweet;
import org.apache.streams.twitter.pojo.Tweet;
import org.apache.streams.twitter.provider.TwitterConfigurator;
import org.apache.streams.twitter.provider.TwitterEventClassifier;
-import org.apache.streams.twitter.serializer.StreamsTwitterMapper;
+import org.apache.streams.twitter.converter.StreamsTwitterMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import twitter4j.*;
@@ -40,7 +40,7 @@ import twitter4j.conf.ConfigurationBuilder;
import java.util.List;
-import static org.apache.streams.twitter.serializer.util.TwitterActivityUtil.*;
+import static org.apache.streams.twitter.converter.util.TwitterActivityUtil.*;
/**
* Given an Activity, fetches the tweet by the activity object id and replaces the existing activity with the converted activity
@@ -110,7 +110,7 @@ public class FetchAndReplaceTwitterProcessor implements StreamsProcessor {
}
}
- protected void replace(Activity doc, String json) throws java.io.IOException, ActivitySerializerException {
+ protected void replace(Activity doc, String json) throws java.io.IOException, ActivityConversionException {
Class documentSubType = TwitterEventClassifier.detectClass(json);
Object object = mapper.readValue(json, documentSubType);
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterEventProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterEventProcessor.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterEventProcessor.java
index c570da8..e80d7a8 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterEventProcessor.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterEventProcessor.java
@@ -28,15 +28,9 @@ import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.core.StreamsProcessor;
-import org.apache.streams.exceptions.ActivitySerializerException;
+import org.apache.streams.exceptions.ActivityConversionException;
import org.apache.streams.jackson.StreamsJacksonMapper;
-import org.apache.streams.pojo.json.Activity;
-import org.apache.streams.twitter.pojo.Delete;
-import org.apache.streams.twitter.pojo.Retweet;
-import org.apache.streams.twitter.pojo.Tweet;
-import org.apache.streams.twitter.provider.TwitterEventClassifier;
-import org.apache.streams.twitter.serializer.*;
-import org.apache.streams.util.ComponentUtils;
+import org.apache.streams.twitter.converter.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -63,8 +57,6 @@ public class TwitterEventProcessor implements StreamsProcessor {
private Class inClass;
private Class outClass;
- private TwitterJsonActivitySerializer twitterJsonActivitySerializer;
-
public TwitterEventProcessor(Class inClass, Class outClass) {
this.inClass = inClass;
this.outClass = outClass;
@@ -74,111 +66,12 @@ public class TwitterEventProcessor implements StreamsProcessor {
this(null, outClass);
}
- public Object convert(ObjectNode event, Class inClass, Class outClass) throws ActivitySerializerException, JsonProcessingException {
-
- Object result = null;
-
- Preconditions.checkNotNull(event);
- Preconditions.checkNotNull(mapper);
- Preconditions.checkNotNull(twitterJsonActivitySerializer);
-
- if( outClass.equals( Activity.class )) {
- LOGGER.debug("ACTIVITY");
- result = twitterJsonActivitySerializer.deserialize(
- mapper.writeValueAsString(event));
- } else if( outClass.equals( Tweet.class )) {
- if ( inClass.equals( Tweet.class )) {
- LOGGER.debug("TWEET");
- result = mapper.convertValue(event, Tweet.class);
- }
- } else if( outClass.equals( Retweet.class )) {
- if ( inClass.equals( Retweet.class )) {
- LOGGER.debug("RETWEET");
- result = mapper.convertValue(event, Retweet.class);
- }
- } else if( outClass.equals( Delete.class )) {
- if ( inClass.equals( Delete.class )) {
- LOGGER.debug("DELETE");
- result = mapper.convertValue(event, Delete.class);
- }
- } else if( outClass.equals( ObjectNode.class )) {
- LOGGER.debug("OBJECTNODE");
- result = mapper.convertValue(event, ObjectNode.class);
- }
-
- // no supported conversion were applied
- if( result != null )
- return result;
-
- LOGGER.debug("CONVERT FAILED");
-
- return null;
-
- }
-
- public boolean validate(Object document, Class klass) {
-
- // TODO
- return true;
- }
-
- public boolean isValidJSON(final String json) {
- boolean valid = false;
- try {
- final JsonParser parser = new ObjectMapper().getJsonFactory()
- .createJsonParser(json);
- while (parser.nextToken() != null) {
- }
- valid = true;
- } catch (JsonParseException jpe) {
- LOGGER.warn("validate: {}", jpe);
- } catch (IOException ioe) {
- LOGGER.warn("validate: {}", ioe);
- }
-
- return valid;
- }
-
@Override
public List<StreamsDatum> process(StreamsDatum entry) {
- // first check for valid json
- ObjectNode node = (ObjectNode) entry.getDocument();
-
- LOGGER.debug("{} processing {}", STREAMS_ID, node.getClass());
-
- String json = null;
- try {
- json = mapper.writeValueAsString(node);
- } catch (JsonProcessingException e) {
- e.printStackTrace();
- }
-
- if( StringUtils.isNotEmpty(json)) {
-
- // since data is coming from outside provider, we don't know what type the events are
- Class inClass = TwitterEventClassifier.detectClass(json);
-
- // if the target is string, just pass-through
- if (java.lang.String.class.equals(outClass))
- return Lists.newArrayList(new StreamsDatum(json));
- else {
- // convert to desired format
- Object out = null;
- try {
- out = convert(node, inClass, outClass);
- } catch (ActivitySerializerException e) {
- LOGGER.warn("Failed deserializing", e);
- return Lists.newArrayList();
- } catch (JsonProcessingException e) {
- LOGGER.warn("Failed parsing JSON", e);
- return Lists.newArrayList();
- }
-
- if (out != null && validate(out, outClass))
- return Lists.newArrayList(new StreamsDatum(out));
- }
- }
+ LOGGER.error("You are calling a deprecated / defunct class. Modify your stream to use ActivityConverterProcessor.");
+
+ LOGGER.debug("CONVERT FAILED");
return Lists.newArrayList();
@@ -187,7 +80,6 @@ public class TwitterEventProcessor implements StreamsProcessor {
@Override
public void prepare(Object configurationObject) {
mapper = StreamsJacksonMapper.getInstance();
- twitterJsonActivitySerializer = new TwitterJsonActivitySerializer();
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterProfileProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterProfileProcessor.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterProfileProcessor.java
index 674eef1..b9661f5 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterProfileProcessor.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterProfileProcessor.java
@@ -27,7 +27,7 @@ import org.apache.streams.twitter.pojo.Retweet;
import org.apache.streams.twitter.pojo.Tweet;
import org.apache.streams.twitter.pojo.User;
import org.apache.streams.twitter.provider.TwitterEventClassifier;
-import org.apache.streams.twitter.serializer.StreamsTwitterMapper;
+import org.apache.streams.twitter.converter.StreamsTwitterMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterTypeConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterTypeConverter.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterTypeConverter.java
index 4c52eba..cc1ecb1 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterTypeConverter.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/processor/TwitterTypeConverter.java
@@ -18,27 +18,7 @@
package org.apache.streams.twitter.processor;
-import com.fasterxml.jackson.core.JsonParseException;
-import com.fasterxml.jackson.core.JsonParser;
-import com.fasterxml.jackson.core.JsonProcessingException;
-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.exceptions.ActivitySerializerException;
-import org.apache.streams.pojo.json.Activity;
-import org.apache.streams.twitter.pojo.Delete;
-import org.apache.streams.twitter.pojo.Retweet;
-import org.apache.streams.twitter.pojo.Tweet;
-import org.apache.streams.twitter.provider.TwitterEventClassifier;
-import org.apache.streams.twitter.serializer.*;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.util.List;
-import java.util.Queue;
+import org.apache.streams.converter.ActivityConverterProcessor;
/**
* This class performs conversion of a twitter event to a specified outClass
@@ -46,167 +26,8 @@ import java.util.Queue;
* Deprecated: use TypeConverterProcessor and ActivityConverterProcessor instead
*/
@Deprecated
-public class TwitterTypeConverter implements StreamsProcessor {
+public class TwitterTypeConverter extends ActivityConverterProcessor {
public final static String STREAMS_ID = "TwitterTypeConverter";
- private final static Logger LOGGER = LoggerFactory.getLogger(TwitterTypeConverter.class);
-
- private ObjectMapper mapper;
-
- private Queue<StreamsDatum> inQueue;
- private Queue<StreamsDatum> outQueue;
-
- private Class inClass;
- private Class outClass;
-
- private TwitterJsonActivitySerializer twitterJsonActivitySerializer;
-
- private int count = 0;
-
- public final static String TERMINATE = new String("TERMINATE");
-
- public TwitterTypeConverter(Class inClass, Class outClass) {
- this.inClass = inClass;
- this.outClass = outClass;
- }
-
- public Queue<StreamsDatum> getProcessorOutputQueue() {
- return outQueue;
- }
-
- public void setProcessorInputQueue(Queue<StreamsDatum> inputQueue) {
- inQueue = inputQueue;
- }
-
- public Object convert(ObjectNode event, Class inClass, Class outClass) throws ActivitySerializerException, JsonProcessingException {
-
- Object result = null;
-
- if( outClass.equals( Activity.class )) {
- LOGGER.debug("ACTIVITY");
- result = twitterJsonActivitySerializer.deserialize(
- mapper.writeValueAsString(event));
- } else if( outClass.equals( Tweet.class )) {
- if ( inClass.equals( Tweet.class )) {
- LOGGER.debug("TWEET");
- result = mapper.convertValue(event, Tweet.class);
- }
- } else if( outClass.equals( Retweet.class )) {
- if ( inClass.equals( Retweet.class )) {
- LOGGER.debug("RETWEET");
- result = mapper.convertValue(event, Retweet.class);
- }
- } else if( outClass.equals( Delete.class )) {
- if ( inClass.equals( Delete.class )) {
- LOGGER.debug("DELETE");
- result = mapper.convertValue(event, Delete.class);
- }
- } else if( outClass.equals( ObjectNode.class )) {
- LOGGER.debug("OBJECTNODE");
- result = mapper.convertValue(event, ObjectNode.class);
- }
-
- // no supported conversion were applied
- if( result != null ) {
- count ++;
- return result;
- }
-
- LOGGER.debug("CONVERT FAILED");
-
- return null;
-
- }
-
- public boolean validate(Object document, Class klass) {
-
- // TODO
- return true;
- }
-
- public boolean isValidJSON(final String json) {
- boolean valid = false;
- try {
- final JsonParser parser = new ObjectMapper().getJsonFactory()
- .createJsonParser(json);
- while (parser.nextToken() != null) {
- }
- valid = true;
- } catch (JsonParseException jpe) {
- LOGGER.warn("validate: {}", jpe);
- } catch (IOException ioe) {
- LOGGER.warn("validate: {}", ioe);
- }
-
- return valid;
- }
-
- @Override
- public List<StreamsDatum> process(StreamsDatum entry) {
-
- StreamsDatum result = null;
-
- try {
-
- Object item = entry.getDocument();
- ObjectNode node;
-
- LOGGER.debug("{} processing {}", STREAMS_ID, item.getClass());
-
- if( item instanceof String ) {
-
- // if the target is string, just pass-through
- if( String.class.equals(outClass)) {
- result = entry;
- }
- else {
- // first check for valid json
- node = (ObjectNode)mapper.readTree((String)item);
-
- // since data is coming from outside provider, we don't know what type the events are
- Class inClass = TwitterEventClassifier.detectClass((String) item);
-
- Object out = convert(node, inClass, outClass);
-
- if( out != null && validate(out, outClass))
- result = new StreamsDatum(out);
- }
-
- } else if( item instanceof ObjectNode ) {
-
- // first check for valid json
- node = (ObjectNode)mapper.valueToTree(item);
-
- // since data is coming from outside provider, we don't know what type the events are
- Class inClass = TwitterEventClassifier.detectClass(mapper.writeValueAsString(item));
-
- Object out = convert(node, inClass, outClass);
-
- if( out != null && validate(out, outClass))
- result = new StreamsDatum(out);
-
- }
-
- } catch (Exception e) {
- e.printStackTrace();
- }
-
- if( result != null )
- return Lists.newArrayList(result);
- else
- return Lists.newArrayList();
- }
-
- @Override
- public void prepare(Object o) {
- mapper = new StreamsTwitterMapper();
- twitterJsonActivitySerializer = new TwitterJsonActivitySerializer();
- }
-
- @Override
- public void cleanUp() {
-
- }
-
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventClassifier.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventClassifier.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventClassifier.java
index 99f110a..f54dd0a 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventClassifier.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventClassifier.java
@@ -23,25 +23,21 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import org.apache.commons.lang.StringUtils;
-import org.apache.streams.data.ActivitySerializer;
import org.apache.streams.jackson.StreamsJacksonMapper;
import org.apache.streams.twitter.pojo.*;
-import org.apache.streams.twitter.serializer.StreamsTwitterMapper;
-import org.apache.streams.twitter.serializer.TwitterJsonDeleteActivitySerializer;
-import org.apache.streams.twitter.serializer.TwitterJsonRetweetActivitySerializer;
-import org.apache.streams.twitter.serializer.TwitterJsonTweetActivitySerializer;
-import org.apache.streams.twitter.serializer.TwitterJsonUserActivitySerializer;
-import org.apache.streams.twitter.serializer.TwitterJsonUserstreameventActivitySerializer;
+import org.apache.streams.twitter.converter.TwitterDateTimeFormat;
import java.io.IOException;
import java.io.Serializable;
/**
* TwitterEventClassifier classifies twitter events
+ *
+ * @Deprecated - replaced by TwitterDocumentClassifier - use ActivityConverterProcessor
*/
public class TwitterEventClassifier implements Serializable {
- private static ObjectMapper mapper = new StreamsJacksonMapper(Lists.newArrayList(StreamsTwitterMapper.TWITTER_FORMAT));
+ private static ObjectMapper mapper = new StreamsJacksonMapper(Lists.newArrayList(TwitterDateTimeFormat.TWITTER_FORMAT));
public static Class detectClass( String json ) {
Preconditions.checkNotNull(json);
@@ -69,31 +65,5 @@ public class TwitterEventClassifier implements Serializable {
else
return Tweet.class;
}
- public static ActivitySerializer bestSerializer( String json ) {
-
- Preconditions.checkNotNull(json);
- Preconditions.checkArgument(StringUtils.isNotEmpty(json));
-
- ObjectNode objectNode;
- try {
- objectNode = (ObjectNode) mapper.readTree(json);
- } catch (IOException e) {
- e.printStackTrace();
- return null;
- }
- if( objectNode.findValue("retweeted_status") != null && objectNode.get("retweeted_status") != null)
- return TwitterJsonRetweetActivitySerializer.getInstance();
- else if( objectNode.findValue("delete") != null )
- return TwitterJsonDeleteActivitySerializer.getInstance();
-// else if( objectNode.findValue("friends") != null ||
-// objectNode.findValue("friends_str") != null )
-// return FriendList.class;
- else if( objectNode.findValue("target_object") != null )
- return TwitterJsonUserstreameventActivitySerializer.getInstance();
- else if ( objectNode.findValue("location") != null && objectNode.findValue("user") == null)
- return TwitterJsonUserActivitySerializer.getInstance();
- else
- return TwitterJsonTweetActivitySerializer.getInstance();
- }
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamProcessor.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamProcessor.java
index 83ea5ae..f0690f8 100644
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamProcessor.java
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamProcessor.java
@@ -23,7 +23,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.collect.Lists;
import com.twitter.hbc.core.processor.StringDelimitedProcessor;
import org.apache.streams.core.StreamsDatum;
-import org.apache.streams.twitter.serializer.StreamsTwitterMapper;
+import org.apache.streams.twitter.converter.StreamsTwitterMapper;
import org.apache.streams.util.ComponentUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/StreamsTwitterMapper.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/StreamsTwitterMapper.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/StreamsTwitterMapper.java
deleted file mode 100644
index 4b0a7da..0000000
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/StreamsTwitterMapper.java
+++ /dev/null
@@ -1,94 +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.twitter.serializer;
-
-import com.fasterxml.jackson.annotation.JsonAutoDetect;
-import com.fasterxml.jackson.annotation.PropertyAccessor;
-import com.fasterxml.jackson.core.JsonParser;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.DeserializationContext;
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.deser.std.StdDeserializer;
-import com.fasterxml.jackson.databind.module.SimpleModule;
-import org.apache.streams.data.util.RFC3339Utils;
-import org.apache.streams.jackson.StreamsDateTimeDeserializer;
-import org.apache.streams.jackson.StreamsDateTimeSerializer;
-import org.apache.streams.jackson.StreamsJacksonMapper;
-import org.apache.streams.jackson.StreamsJacksonModule;
-import org.joda.time.DateTime;
-import org.joda.time.format.DateTimeFormat;
-import org.joda.time.format.DateTimeFormatter;
-
-import java.io.IOException;
-
-/**
- * This class assist with handling twitter's date-time format during conversion
- *
- * Deprecated: use StreamsJacksonMapper.getInstance(StreamsTwitterMapper.TWITTER_FORMAT) instead
- */
-@Deprecated
-public class StreamsTwitterMapper extends StreamsJacksonMapper {
-
- public static final String TWITTER_FORMAT = "EEE MMM dd HH:mm:ss Z yyyy";
-
- public static final DateTimeFormatter TWITTER_FORMATTER = DateTimeFormat.forPattern(TWITTER_FORMAT);
-
- public static final Long getMillis(String dateTime) {
-
- // this function is for pig which doesn't handle exceptions well
- try {
- Long result = TWITTER_FORMATTER.parseMillis(dateTime);
- return result;
- } catch( Exception e ) {
- return null;
- }
-
- }
-
- private static final StreamsTwitterMapper INSTANCE = new StreamsTwitterMapper();
-
- public static StreamsTwitterMapper getInstance(){
- return INSTANCE;
- }
-
- public StreamsTwitterMapper() {
- super();
- registerModule(new SimpleModule()
- {
- {
- addDeserializer(DateTime.class, new StdDeserializer<DateTime>(DateTime.class) {
- @Override
- public DateTime deserialize(JsonParser jpar, DeserializationContext context) throws IOException, JsonProcessingException {
- DateTime result = null;
- try {
- result = TWITTER_FORMATTER.parseDateTime(jpar.getValueAsString());
- } catch( Exception e ) { }
- try {
- result = RFC3339Utils.getInstance().parseToUTC(jpar.getValueAsString());
- } catch( Exception e ) { }
- return result;
- }
- });
- }
- });
-
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonActivitySerializer.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonActivitySerializer.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonActivitySerializer.java
deleted file mode 100644
index d1f0de9..0000000
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonActivitySerializer.java
+++ /dev/null
@@ -1,70 +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.twitter.serializer;
-
-import org.apache.commons.lang.NotImplementedException;
-import org.apache.streams.data.ActivitySerializer;
-import org.apache.streams.exceptions.ActivitySerializerException;
-import org.apache.streams.pojo.json.Activity;
-import org.apache.streams.twitter.pojo.*;
-import org.apache.streams.twitter.provider.TwitterEventClassifier;
-
-import java.util.List;
-import java.io.Serializable;
-
-public class TwitterJsonActivitySerializer implements ActivitySerializer<String>, Serializable
-{
-
- public TwitterJsonActivitySerializer() {
-
- }
-
- private static TwitterJsonActivitySerializer instance = new TwitterJsonActivitySerializer();
-
- public static TwitterJsonActivitySerializer getInstance() {
- return instance;
- }
-
- @Override
- public String serializationFormat() {
- return null;
- }
-
- @Override
- public String serialize(Activity deserialized) throws ActivitySerializerException {
- throw new NotImplementedException();
- }
-
- @Override
- public Activity deserialize(String serialized) throws ActivitySerializerException {
-
- ActivitySerializer serializer = TwitterEventClassifier.bestSerializer(serialized);
- Activity activity = serializer.deserialize(serialized);
-
- if( activity == null )
- throw new ActivitySerializerException("unrecognized type");
-
- return activity;
- }
-
- @Override
- public List<Activity> deserializeAll(List<String> serializedList) {
- throw new NotImplementedException();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonDeleteActivitySerializer.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonDeleteActivitySerializer.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonDeleteActivitySerializer.java
deleted file mode 100644
index b368f71..0000000
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonDeleteActivitySerializer.java
+++ /dev/null
@@ -1,94 +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.twitter.serializer;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.google.common.base.Strings;
-import org.apache.commons.lang.NotImplementedException;
-import org.apache.streams.data.ActivitySerializer;
-import org.apache.streams.exceptions.ActivitySerializerException;
-import org.apache.streams.pojo.json.Activity;
-import org.apache.streams.pojo.json.ActivityObject;
-import org.apache.streams.pojo.json.Actor;
-import org.apache.streams.twitter.pojo.Delete;
-import org.apache.streams.twitter.pojo.Tweet;
-
-import java.io.Serializable;
-import java.util.List;
-
-import static org.apache.streams.twitter.serializer.util.TwitterActivityUtil.*;
-
-
-/**
-* Created with IntelliJ IDEA.
-* User: mdelaet
-* Date: 9/30/13
-* Time: 9:24 AM
-* To change this template use File | Settings | File Templates.
-*/
-public class TwitterJsonDeleteActivitySerializer implements ActivitySerializer<String>, Serializable {
-
- private static TwitterJsonDeleteActivitySerializer instance = new TwitterJsonDeleteActivitySerializer();
-
- public static TwitterJsonDeleteActivitySerializer getInstance() {
- return instance;
- }
-
- @Override
- public String serializationFormat() {
- return null;
- }
-
- @Override
- public String serialize(Activity deserialized) throws ActivitySerializerException {
- throw new NotImplementedException();
- }
-
- @Override
- public Activity deserialize(String serialized) throws ActivitySerializerException {
- return null;
- }
-
- @Override
- public List<Activity> deserializeAll(List<String> serializedList) {
- return null;
- }
-
- public Activity convert(ObjectNode event) throws ActivitySerializerException {
-
- ObjectMapper mapper = StreamsTwitterMapper.getInstance();
- Delete delete = null;
- try {
- delete = mapper.treeToValue(event, Delete.class);
- } catch (JsonProcessingException e) {
- e.printStackTrace();
- }
-
- Activity activity = new Activity();
- updateActivity(delete, activity);
- return activity;
- }
-
- public ActivityObject buildTarget(Tweet tweet) {
- return null;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/2cc6118f/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonRetweetActivitySerializer.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonRetweetActivitySerializer.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonRetweetActivitySerializer.java
deleted file mode 100644
index 58cb769..0000000
--- a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonRetweetActivitySerializer.java
+++ /dev/null
@@ -1,79 +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.twitter.serializer;
-
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.apache.streams.data.ActivitySerializer;
-import org.apache.streams.exceptions.ActivitySerializerException;
-import org.apache.streams.pojo.json.Activity;
-import org.apache.streams.twitter.pojo.Retweet;
-
-import java.io.IOException;
-import java.io.Serializable;
-import java.util.List;
-
-import static org.apache.streams.twitter.serializer.util.TwitterActivityUtil.*;
-
-public class TwitterJsonRetweetActivitySerializer implements ActivitySerializer<String>, Serializable {
-
- public TwitterJsonRetweetActivitySerializer() {
-
- }
-
- private static TwitterJsonRetweetActivitySerializer instance = new TwitterJsonRetweetActivitySerializer();
-
- public static TwitterJsonRetweetActivitySerializer getInstance() {
- return instance;
- }
-
- @Override
- public String serializationFormat() {
- return null;
- }
-
- @Override
- public String serialize(Activity deserialized) throws ActivitySerializerException {
- return null;
- }
-
- @Override
- public Activity deserialize(String event) throws ActivitySerializerException {
-
- ObjectMapper mapper = StreamsTwitterMapper.getInstance();
- Retweet retweet = null;
- try {
- retweet = mapper.readValue(event, Retweet.class);
- } catch (JsonProcessingException e) {
- e.printStackTrace();
- } catch (IOException e) {
- e.printStackTrace();
- }
-
- Activity activity = new Activity();
- updateActivity(retweet, activity);
-
- return activity;
- }
-
- @Override
- public List<Activity> deserializeAll(List<String> serializedList) {
- return null;
- }
-}