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/05/05 20:54:30 UTC
[43/52] [abbrv] git commit: initial processor-json implementation
breakdown of responsibility in processor-urls and processor-tika additional
message types in provider-twitter
initial processor-json implementation
breakdown of responsibility in processor-urls and processor-tika
additional message types in provider-twitter
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/a79e1150
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/a79e1150
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/a79e1150
Branch: refs/heads/sblackmon
Commit: a79e1150308d9cb749122b7ff1b043ddff5919a9
Parents: b0a7169
Author: sblackmon <sb...@w2odigital.com>
Authored: Tue Apr 15 10:53:52 2014 -0500
Committer: sblackmon <sb...@w2odigital.com>
Committed: Tue Apr 15 10:53:52 2014 -0500
----------------------------------------------------------------------
streams-contrib/pom.xml | 2 +-
streams-contrib/streams-processor-json/pom.xml | 69 ++++++
.../apache/streams/json/JsonPathExtractor.java | 107 ++++++++++
.../apache/streams/json/test/JsonPathTest.java | 92 ++++++++
.../src/test/resources/books.json | 21 ++
.../test/resources/gnip_twitter_extended.json | 144 +++++++++++++
streams-contrib/streams-processor-tika/pom.xml | 3 +-
.../org/apache/streams/tika/TikaProcessor.java | 211 +++++++++++++++----
.../apache/streams/tika/BoilerPipeArticle.json | 27 +--
.../apache/streams/tika/LanguageDetails.json | 13 ++
.../streams/url/TestLinkExpanderProcessor.java | 41 ++++
.../org/apache/streams/urls/LinkCrawler.java | 205 ++++++++++++++++++
.../streams/urls/LinkCrawlerProcessor.java | 115 ++++++++++
.../org/apache/streams/urls/LinkResolver.java | 18 +-
.../org/apache/streams/urls/LinkDetails.json | 6 +
.../streams-provider-twitter/pom.xml | 7 +-
.../provider/TwitterEventClassifier.java | 1 -
...erJsonUserstreameventActivitySerializer.java | 88 ++++++++
.../main/jsonschema/com/twitter/FriendList.json | 20 ++
.../jsonschema/com/twitter/UserstreamEvent.json | 33 +++
.../org/apache/streams/pojo/json/activity.json | 4 +
21 files changed, 1151 insertions(+), 76 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/pom.xml b/streams-contrib/pom.xml
index d6873bb..ec2dbc0 100644
--- a/streams-contrib/pom.xml
+++ b/streams-contrib/pom.xml
@@ -45,7 +45,7 @@
<module>streams-persist-kafka</module>
<module>streams-persist-mongo</module>
<!--<module>streams-processor-lucene</module>-->
- <!--<module>streams-processor-tika</module>-->
+ <module>streams-processor-tika</module>
<module>streams-processor-json</module>
<module>streams-processor-urls</module>
<module>streams-provider-datasift</module>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-json/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-json/pom.xml b/streams-contrib/streams-processor-json/pom.xml
new file mode 100644
index 0000000..27695bc
--- /dev/null
+++ b/streams-contrib/streams-processor-json/pom.xml
@@ -0,0 +1,69 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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-json</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/a79e1150/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java b/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
new file mode 100644
index 0000000..9e5f10e
--- /dev/null
+++ b/streams-contrib/streams-processor-json/src/main/java/org/apache/streams/json/JsonPathExtractor.java
@@ -0,0 +1,107 @@
+package org.apache.streams.json;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+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 com.jayway.jsonpath.JsonPath;
+import net.minidev.json.JSONArray;
+import net.minidev.json.JSONObject;
+import org.apache.commons.lang3.StringUtils;
+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;
+
+/**
+ * Created by sblackmon on 12/10/13.
+ */
+public class JsonPathExtractor implements StreamsProcessor {
+
+ public JsonPathExtractor(String pathExpression) {
+ this.pathExpression = pathExpression;
+ }
+
+ private final static String STREAMS_ID = "JsonPathExtractor";
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(JsonPathExtractor.class);
+
+ private ObjectMapper mapper = new StreamsJacksonMapper();
+
+ private String pathExpression;
+ private JsonPath jsonPath;
+
+ @Override
+ public List<StreamsDatum> process(StreamsDatum entry) {
+
+ List<StreamsDatum> result = Lists.newArrayList();
+
+ String json = null;
+
+ LOGGER.debug("{} processing {}", STREAMS_ID);
+
+ if( entry.getDocument() instanceof ObjectNode ) {
+ ObjectNode node = (ObjectNode) entry.getDocument();
+ try {
+ json = mapper.writeValueAsString(node);
+ } catch (JsonProcessingException e) {
+ e.printStackTrace();
+ }
+ } else if( entry.getDocument() instanceof String ) {
+ json = (String) entry.getDocument();
+ }
+
+ if( StringUtils.isNotEmpty(json)) {
+
+ try {
+ Object readResult = JsonPath.read(json, pathExpression);
+
+ if (readResult instanceof String) {
+ String match = (String) readResult;
+ StreamsDatum matchDatum = new StreamsDatum(match);
+ result.add(matchDatum);
+ } else if (readResult instanceof JSONArray) {
+ JSONArray array = (JSONArray) readResult;
+ Iterator iterator = array.iterator();
+ while (iterator.hasNext()) {
+ Object item = iterator.next();
+ if( item instanceof String ) {
+ String match = (String) item;
+ StreamsDatum matchDatum = new StreamsDatum(match);
+ result.add(matchDatum);
+ } else if ( item instanceof JSONObject ) {
+ ObjectNode objectNode = mapper.readValue(mapper.writeValueAsString(item), ObjectNode.class);
+ StreamsDatum matchDatum = new StreamsDatum(objectNode);
+ result.add(matchDatum);
+ }
+ }
+ } else {
+
+ }
+
+ } catch( Exception e ) {
+ LOGGER.warn(e.getMessage());
+ }
+
+ }
+
+ return result;
+
+ }
+
+ @Override
+ public void prepare(Object configurationObject) {
+ jsonPath = JsonPath.compile(pathExpression);
+ mapper.registerModule(new JsonOrgModule());
+ }
+
+ @Override
+ public void cleanUp() {
+
+ }
+};
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-json/src/test/java/org/apache/streams/json/test/JsonPathTest.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-json/src/test/java/org/apache/streams/json/test/JsonPathTest.java b/streams-contrib/streams-processor-json/src/test/java/org/apache/streams/json/test/JsonPathTest.java
new file mode 100644
index 0000000..26fec94
--- /dev/null
+++ b/streams-contrib/streams-processor-json/src/test/java/org/apache/streams/json/test/JsonPathTest.java
@@ -0,0 +1,92 @@
+package org.apache.streams.json.test;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.exceptions.ActivitySerializerException;
+import org.apache.streams.json.JsonPathExtractor;
+import org.apache.streams.pojo.json.Activity;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.util.List;
+import java.util.Scanner;
+
+import static org.hamcrest.CoreMatchers.*;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+/**
+* Created with IntelliJ IDEA.
+* User: sblackmon
+* Date: 8/20/13
+* Time: 5:57 PM
+* To change this template use File | Settings | File Templates.
+*/
+public class JsonPathTest {
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(JsonPathTest.class);
+
+ private String testJson;
+
+ @Before
+ public void initialize() {
+ try {
+ testJson = FileUtils.readFileToString(new File("src/test/resources/books.json"));
+ } catch (IOException e) {
+ e.printStackTrace();
+ Assert.fail();
+ }
+ }
+
+ @Test
+ public void test1()
+ {
+ JsonPathExtractor extractor = new JsonPathExtractor("$.store.book[*].author");
+ extractor.prepare(null);
+ List<StreamsDatum> result = extractor.process(new StreamsDatum(testJson));
+ assertThat(result.size(), is(2));
+ assertTrue(result.get(0).getDocument() instanceof String);
+ assertTrue(result.get(1).getDocument() instanceof String);
+ }
+
+ @Test
+ public void test2()
+ {
+ JsonPathExtractor extractor = new JsonPathExtractor("$.store.book[?(@.category == 'reference')]");
+ extractor.prepare(null);
+ List<StreamsDatum> result = extractor.process(new StreamsDatum(testJson));
+ assertThat(result.size(), is(1));
+ assertTrue(result.get(0).getDocument() instanceof ObjectNode);
+ }
+
+ @Test
+ public void test3()
+ {
+ JsonPathExtractor extractor = new JsonPathExtractor("$.store.book[?(@.price > 10)]");
+ extractor.prepare(null);
+ List<StreamsDatum> result = extractor.process(new StreamsDatum(testJson));
+ assertThat(result.size(), is(1));
+ assertTrue(result.get(0).getDocument() instanceof ObjectNode);
+ }
+
+ @Test
+ public void test4()
+ {
+ JsonPathExtractor extractor = new JsonPathExtractor("$.store.book[?(@.isbn)]");
+ extractor.prepare(null);
+ List<StreamsDatum> result = extractor.process(new StreamsDatum(testJson));
+ assertThat(result.size(), is(1));
+ assertTrue(result.get(0).getDocument() instanceof ObjectNode);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-json/src/test/resources/books.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-json/src/test/resources/books.json b/streams-contrib/streams-processor-json/src/test/resources/books.json
new file mode 100644
index 0000000..30cf954
--- /dev/null
+++ b/streams-contrib/streams-processor-json/src/test/resources/books.json
@@ -0,0 +1,21 @@
+{
+ "store": {
+ "book": [
+ { "category": "reference",
+ "author": "Nigel Rees",
+ "title": "Sayings of the Century",
+ "price": 8.95
+ },
+ { "category": "fiction",
+ "author": "Evelyn Waugh",
+ "title": "Sword of Honour",
+ "price": 12.99,
+ "isbn": "0-553-21311-3"
+ }
+ ],
+ "bicycle": {
+ "color": "red",
+ "price": 19.95
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-lucene/src/test/resources/gnip_twitter_extended.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-lucene/src/test/resources/gnip_twitter_extended.json b/streams-contrib/streams-processor-lucene/src/test/resources/gnip_twitter_extended.json
new file mode 100644
index 0000000..5ba26be
--- /dev/null
+++ b/streams-contrib/streams-processor-lucene/src/test/resources/gnip_twitter_extended.json
@@ -0,0 +1,144 @@
+{
+ "link": "http://twitter.com/spoffff/statuses/372802927385403392",
+ "postedTime": "2013-08-28T19:28:38.000Z",
+ "provider": {
+ "id": "{link}",
+ "upstreamDuplicates": [],
+ "link": "http://www.twitter.com",
+ "attachments": [],
+ "displayName": "Twitter",
+ "objectType": "service",
+ "downstreamDuplicates": []
+ },
+ "links": [],
+ "actor": {
+ "twitterTimeZone": "Brussels",
+ "friendsCount": 6,
+ "favoritesCount": 0,
+ "link": "http://www.twitter.com/spoffff",
+ "image": {
+ "url": "https://si0.twimg.com/sticky/default_profile_images/default_profile_1_normal.png"
+ },
+ "postedTime": "2012-05-18T15:14:35.000Z",
+ "links": [
+ {
+ "rel": "me",
+ "href": null
+ }
+ ],
+ "listedCount": 0,
+ "downstreamDuplicates": [],
+ "id": "id:twitter.com:583891967",
+ "upstreamDuplicates": [],
+ "languages": [
+ "fr"
+ ],
+ "verified": false,
+ "utcOffset": "7200",
+ "followersCount": 0,
+ "attachments": [],
+ "displayName": "Sabine Chappuis",
+ "preferredUsername": "spoffff",
+ "statusesCount": 87,
+ "objectType": "person"
+ },
+ "object": {
+ "id": "object:search.twitter.com,2005:372802927385403392",
+ "summary": "Le Grand Schtroumpf confirme que la cascade magique n'est \"Plus tr?�s loin.\" http://t.co/Ytn45Pbttk #SmurfsVillage @BeelineGames",
+ "upstreamDuplicates": [],
+ "link": "http://twitter.com/spoffff/statuses/372802927385403392",
+ "postedTime": "2013-08-28T19:28:38.000Z",
+ "attachments": [],
+ "objectType": "note",
+ "downstreamDuplicates": []
+ },
+ "extensions": {
+ "w2o": {
+ "tags": [
+ "brand-cascade",
+ "language-en",
+ "country-ca"
+ ],
+ "provider": "twitter",
+ "analyzer": "romance_analyzer",
+ "lang": {
+ "primaryLanguage": "en"
+ }
+ },
+ "gnip": {
+ "matching_rules": [
+ {
+ "tag": "cascade_CA_CA_en"
+ }
+ ],
+ "klout_score": 10,
+ "urls": [
+ {
+ "expanded_url": "https://itunes.apple.com/us/app/smurfs-village/id399648212?mt=8",
+ "url": "http://t.co/Ytn45Pbttk"
+ }
+ ],
+ "klout_profile": {
+ "topics": [],
+ "klout_user_id": "257268143479895040",
+ "link": "http://klout.com/user/id/257268143479895040"
+ },
+ "language": {
+ "value": "fr"
+ }
+ },
+ "twitter": {
+ "retweetCount": 0,
+ "entities": {
+ "symbols": [],
+ "urls": [
+ {
+ "expanded_url": "http://bit.ly/hUmoRz",
+ "indices": [
+ 77,
+ 99
+ ],
+ "display_url": "bit.ly/hUmoRz",
+ "url": "http://t.co/Ytn45Pbttk"
+ }
+ ],
+ "hashtags": [
+ {
+ "text": "SmurfsVillage",
+ "indices": [
+ 100,
+ 114
+ ]
+ }
+ ],
+ "user_mentions": [
+ {
+ "id": 188075479,
+ "name": "Beeline Interactive",
+ "indices": [
+ 115,
+ 128
+ ],
+ "screen_name": "BeelineGames",
+ "id_str": "188075479"
+ }
+ ]
+ }
+ }
+ },
+ "twitter_filter_level": "medium",
+ "guid": "A8fccSz7rpKfDJY078VLyw==_201308",
+ "content": "Le Grand Schtroumpf confirme que la cascade magique n'est \"Plus tr?�s loin.\" http://t.co/Ytn45Pbttk #SmurfsVillage @BeelineGames",
+ "id": "tag:search.twitter.com,2005:372802927385403392",
+ "verb": "post",
+ "generator": {
+ "id": "{link}",
+ "upstreamDuplicates": [],
+ "link": "https://itunes.apple.com/us/app/smurfs-village/id399648212?mt=8&uo=4",
+ "attachments": [],
+ "displayName": "Smurfs' Village on iOS",
+ "downstreamDuplicates": []
+ },
+ "published": "2013-08-28T19:28:38Z",
+ "objectType": "activity"
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-tika/pom.xml
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-tika/pom.xml b/streams-contrib/streams-processor-tika/pom.xml
index 81026a3..cc1cf21 100644
--- a/streams-contrib/streams-processor-tika/pom.xml
+++ b/streams-contrib/streams-processor-tika/pom.xml
@@ -118,7 +118,8 @@
<addCompileSourceRoot>true</addCompileSourceRoot>
<generateBuilders>true</generateBuilders>
<sourcePaths>
- <sourcePath>src/main/jsonschema/org/apache/streams/tika/LinkDetails.json</sourcePath>
+ <sourcePath>src/main/jsonschema/org/apache/streams/tika/BoilerPipeArticle.json</sourcePath>
+ <sourcePath>src/main/jsonschema/org/apache/streams/tika/LanguageDetails.json</sourcePath>
</sourcePaths>
<outputDirectory>target/generated-sources/jsonschema2pojo</outputDirectory>
<targetPackage>org.apache.streams.tika</targetPackage>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-tika/src/main/java/org/apache/streams/tika/TikaProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-tika/src/main/java/org/apache/streams/tika/TikaProcessor.java b/streams-contrib/streams-processor-tika/src/main/java/org/apache/streams/tika/TikaProcessor.java
index 1e694c5..c9d9bbc 100644
--- a/streams-contrib/streams-processor-tika/src/main/java/org/apache/streams/tika/TikaProcessor.java
+++ b/streams-contrib/streams-processor-tika/src/main/java/org/apache/streams/tika/TikaProcessor.java
@@ -1,18 +1,35 @@
package org.apache.streams.tika;
+import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsonorg.JsonOrgModule;
import com.google.common.collect.Lists;
+import de.l3s.boilerpipe.document.TextBlock;
import org.apache.commons.lang.NotImplementedException;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.core.StreamsProcessor;
import org.apache.streams.jackson.StreamsJacksonMapper;
import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.urls.LinkCrawler;
+import org.apache.streams.urls.LinkCrawlerProcessor;
+import org.apache.streams.urls.LinkDetails;
+import org.apache.streams.util.DateUtil;
+import org.apache.tika.exception.TikaException;
+import org.apache.tika.language.LanguageIdentifier;
+import org.apache.tika.metadata.Metadata;
+import org.apache.tika.parser.AutoDetectParser;
+import org.apache.tika.parser.ParseContext;
+import org.apache.tika.parser.html.BoilerpipeContentHandler;
+import org.apache.tika.parser.html.HtmlParser;
+import org.apache.tika.sax.BodyContentHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.xml.sax.SAXException;
-import java.util.List;
+import java.io.*;
+import java.text.ParseException;
+import java.util.*;
/**
* References:
@@ -26,12 +43,82 @@ import java.util.List;
public class TikaProcessor implements StreamsProcessor
{
- private final static String STREAMS_ID = "LinkCrawlerProcessor";
+ private final static String STREAMS_ID = "TikaProcessor";
private final static Logger LOGGER = LoggerFactory.getLogger(TikaProcessor.class);
+ private AutoDetectParser autoDetectParser;
+
private ObjectMapper mapper;
+ protected LanguageDetails language;
+ protected BoilerPipeArticle article;
+
+ private static final Collection<String> AUTHOR_SEARCH = new ArrayList<String>() {{
+ add("og:author");
+ add("dc:author");
+ add("author");
+ }};
+
+ private static final Collection<String> DESCRIPTION_SEARCH = new ArrayList<String>() {{
+ add("og:description");
+ add("dc:description");
+ add("description");
+ }};
+
+ private static final Collection<String> MEDIUM_SEARCH = new ArrayList<String>() {{
+ add("og:medium");
+ add("dc:medium");
+ add("medium");
+ }};
+
+ private static final Collection<String> IMAGE_SEARCH = new ArrayList<String>() {{
+ add("og:image");
+ add("twitter:image");
+ add("image");
+ }};
+
+ private static final Collection<String> KEYWORDS_SEARCH = new ArrayList<String>() {{
+ add("keywords");
+ add("news_keywords");
+ }};
+
+ private static final Collection<String> PUB_DATE_SEARCH = new ArrayList<String>() {{
+ add("pubdate");
+ add("os:pubdate");
+ add("dc:pubdate");
+ }};
+
+ private static final Collection<String> MODIFIED_DATE_SEARCH = new ArrayList<String>() {{
+ add("lastmod");
+ add("last-modified");
+ }};
+
+ private static final Collection<String> LOCALE_SEARCH = new ArrayList<String>() {{
+ add("locale");
+ add("os:locale");
+ add("dc:local");
+ }};
+
+ // Social Searchers
+ private static final Collection<String> FACEBOOK_PAGE_SEARCH = new ArrayList<String>() {{
+ add("fb:page_id");
+ }};
+
+ private static final Collection<String> FACEBOOK_APP_SEARCH = new ArrayList<String>() {{
+ add("fb:app_id");
+ }};
+
+ private static final Collection<String> TWITTER_SITE_SEARCH = new ArrayList<String>() {{
+ add("twitter:site:id");
+ add("twitter:site");
+ }};
+
+ private static final Collection<String> TWITTER_CREATOR_SEARCH = new ArrayList<String>() {{
+ add("twitter:creator:id");
+ add("twitter:creator");
+ }};
+
@Override
public List<StreamsDatum> process(StreamsDatum entry) {
@@ -61,57 +148,27 @@ public class TikaProcessor implements StreamsProcessor
}
else throw new NotImplementedException();
- List<String> outputLinks = activity.getLinks();
- // for each
- for( String link : outputLinks ) {
-
- System.out.println( "pulling " + link);
+ Object crawlerResult = activity.getExtensions().getAdditionalProperties().get(LinkCrawlerProcessor.STREAMS_ID);
- try {
- StreamsDatum outputDatum = expandLink(link, entry);
- if( outputDatum != null )
- result.add(outputDatum);
- } catch (Exception e) {
- //drop unexpandable links
- LOGGER.debug("Failed to expand link : {}", link);
- LOGGER.debug("Excpetion expanding link : {}", e);
- }
+ LinkDetails linkDetails = mapper.convertValue(crawlerResult, LinkDetails.class);
+ try {
+ parsePlainText(linkDetails.getContent());
+ detectLanguage(article.getPlainText());
+ parseMainContent(article.getPlainText());
+ } catch( Exception e ) {
+ LOGGER.warn( e.getMessage() );
}
return result;
}
- private StreamsDatum expandLink(String link, StreamsDatum input) {
-
- LinkCrawler expander = new LinkCrawler((String)link);
- expander.run();
- StreamsDatum datum = null;
- if(input.getId() == null)
- try {
- datum = new StreamsDatum(this.mapper.writeValueAsString(expander.getArticle()), expander.getFinalURL());
- } catch (JsonProcessingException e) {
- e.printStackTrace();
- return null;
- }
- else
- try {
- datum = new StreamsDatum(this.mapper.writeValueAsString(expander.getArticle()), input.getId());
- } catch (JsonProcessingException e) {
- e.printStackTrace();
- return null;
- }
- datum.setSequenceid(input.getSequenceid());
- datum.setMetadata(input.getMetadata());
- datum.setTimestamp(input.getTimestamp());
- return datum;
-
- }
@Override
public void prepare(Object o) {
this.mapper = StreamsJacksonMapper.getInstance();
this.mapper.registerModule(new JsonOrgModule());
+ this.autoDetectParser = new AutoDetectParser();
}
@Override
@@ -119,4 +176,76 @@ public class TikaProcessor implements StreamsProcessor
}
+ private void parsePlainText(String content) throws Exception {
+ BodyContentHandler handler = new BodyContentHandler();
+ Metadata metadata = new Metadata();
+ InputStream is = new ByteArrayInputStream(content.getBytes());
+ new HtmlParser().parse(is, handler, metadata, new ParseContext());
+ language = new LanguageDetails();
+
+ article.setPlainText(handler.toString());
+
+ }
+
+ private void detectLanguage(String plainText) throws Exception {
+
+ LanguageIdentifier languageIdentifier = new LanguageIdentifier(plainText);
+ LanguageDetails languageDetails = new LanguageDetails();
+ languageDetails.setLanguageCode(languageIdentifier.getLanguage());
+ languageDetails.setIsLanguageReasonablyCertain(languageIdentifier.isReasonablyCertain());
+ language = languageDetails;
+
+ }
+
+ private void parseMainContent(String plainText) throws IOException, SAXException, TikaException, ParseException
+ {
+ Metadata rawMetaData = new Metadata();
+ StringWriter stringWriter = new StringWriter();
+
+ BoilerpipeContentHandler boilerpipeContentHandler = new BoilerpipeContentHandler(stringWriter);
+
+ InputStream is = new ByteArrayInputStream(plainText.getBytes());
+
+ autoDetectParser.parse(is,
+ boilerpipeContentHandler,
+ rawMetaData);
+
+ article.setBody(boilerpipeContentHandler.getTextDocument().getContent());
+ article.setTitle(boilerpipeContentHandler.getTextDocument().getTitle());
+
+ // this map is for ourselves so we convert it to lower-case to make it easier to search.
+ // the meta data that is going to be returned will be unmodified meta data.
+ for(String name : rawMetaData.names())
+ if(rawMetaData.get(name) != null)
+ article.getAdditionalProperties().put(name.toLowerCase(), rawMetaData.get(name));
+
+ article.setAuthor(metaDataSearcher(AUTHOR_SEARCH));
+ article.setDescription(metaDataSearcher(DESCRIPTION_SEARCH));
+ article.setMedium(metaDataSearcher(MEDIUM_SEARCH));
+ article.setImageURL(metaDataSearcher(IMAGE_SEARCH));
+ article.setLocale(metaDataSearcher(LOCALE_SEARCH));
+
+ mergeSet(KEYWORDS_SEARCH, article.getKeywords());
+
+ article.setPublishedDate(DateUtil.determineDate(metaDataSearcher(PUB_DATE_SEARCH)));
+ article.setLastModifiedDate(DateUtil.determineDate(metaDataSearcher(MODIFIED_DATE_SEARCH)));
+
+ if(article.getBody().length() > 50)
+ article.setSiteStatus(BoilerPipeArticle.SiteStatus.SUCCESS);
+ }
+
+ private String metaDataSearcher(Collection<String> itemsToSearch) {
+ for(String s : itemsToSearch)
+ if(article.getAdditionalProperties().containsKey(s))
+ return (String) article.getAdditionalProperties().get(s);
+
+ // the meta searcher returned nothing.
+ return null;
+ }
+
+ private void mergeSet(Collection<String> itemsToSearch, Set<String> set) {
+ for(String s : itemsToSearch)
+ Collections.addAll(set, s == null || s.equals("") ? new String[]{} : s.split(","));
+ }
+
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-tika/src/main/jsonschema/org/apache/streams/tika/BoilerPipeArticle.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-tika/src/main/jsonschema/org/apache/streams/tika/BoilerPipeArticle.json b/streams-contrib/streams-processor-tika/src/main/jsonschema/org/apache/streams/tika/BoilerPipeArticle.json
index 137a4be..438b984 100644
--- a/streams-contrib/streams-processor-tika/src/main/jsonschema/org/apache/streams/tika/BoilerPipeArticle.json
+++ b/streams-contrib/streams-processor-tika/src/main/jsonschema/org/apache/streams/tika/BoilerPipeArticle.json
@@ -39,17 +39,6 @@
"imageURL": {
"type": "string"
},
- "languageDetected": {
- "type": "object",
- "properties": {
- "languageCode": {
- "type": "string"
- },
- "isLanguageReasonablyCertain": {
- "type": "boolean"
- }
- }
- },
"keywords": {
"type": "array",
"uniqueItems": true,
@@ -57,17 +46,11 @@
"type": "string"
}
},
- "twitterCreator": {
- "type": "string"
- },
- "twitterSite": {
- "type": "string"
- },
- "facebookPage": {
- "type": "string"
- },
- "facebookApp": {
- "type": "string"
+ "textBlocks": {
+ "type": "array",
+ "items": {
+ "type": "string"
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-tika/src/main/jsonschema/org/apache/streams/tika/LanguageDetails.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-tika/src/main/jsonschema/org/apache/streams/tika/LanguageDetails.json b/streams-contrib/streams-processor-tika/src/main/jsonschema/org/apache/streams/tika/LanguageDetails.json
new file mode 100644
index 0000000..3b9eaaf
--- /dev/null
+++ b/streams-contrib/streams-processor-tika/src/main/jsonschema/org/apache/streams/tika/LanguageDetails.json
@@ -0,0 +1,13 @@
+{
+ "type": "object",
+ "$schema": "http://json-schema.org/draft-03/schema",
+ "id": "#",
+ "properties": {
+ "languageCode": {
+ "type": "string"
+ },
+ "isLanguageReasonablyCertain": {
+ "type": "boolean"
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-tika/src/test/java/org/apache/streams/url/TestLinkExpanderProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-tika/src/test/java/org/apache/streams/url/TestLinkExpanderProcessor.java b/streams-contrib/streams-processor-tika/src/test/java/org/apache/streams/url/TestLinkExpanderProcessor.java
new file mode 100644
index 0000000..c5dc81f
--- /dev/null
+++ b/streams-contrib/streams-processor-tika/src/test/java/org/apache/streams/url/TestLinkExpanderProcessor.java
@@ -0,0 +1,41 @@
+package org.apache.streams.url;
+
+import com.google.common.collect.Lists;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.urls.LinkCrawlerProcessor;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+import java.util.List;
+
+/**
+ * Created by rebanks on 2/27/14.
+ */
+public class TestLinkExpanderProcessor {
+
+ private static String activityString;
+
+ @Test
+ public void testLinkExpanderProcessorNoLinks() {
+ testHelper();
+ }
+
+ @Test
+ public void testLinkExpanderProcessorOneLink() {
+ testHelper("http://www.wcgworld.com");
+ }
+
+ @Test
+ public void testLinkExpanderProcessorMultiLinks() {
+ testHelper("http://www.wcgworld.com", "http://www.espn.com");
+ }
+
+ private void testHelper(String... urls) {
+ LinkCrawlerProcessor processor = new LinkCrawlerProcessor();
+ processor.prepare(null);
+ for( String url : urls ) {
+ assertEquals(1, processor.process(new StreamsDatum(url)).size());
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkCrawler.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkCrawler.java b/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkCrawler.java
new file mode 100644
index 0000000..fe5e567
--- /dev/null
+++ b/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkCrawler.java
@@ -0,0 +1,205 @@
+package org.apache.streams.urls;
+
+import com.google.common.base.Preconditions;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.net.HttpURLConnection;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.net.URLDecoder;
+import java.util.*;
+
+/**
+ * References:
+ * Some helpful references to help
+ * Purpose URL
+ * ------------- ----------------------------------------------------------------
+ * [Status Codes] http://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html
+ * [Test Cases] http://greenbytes.de/tech/tc/httpredirects/
+ * [t.co behavior] https://dev.twitter.com/docs/tco-redirection-behavior
+ */
+
+public class LinkCrawler extends LinkResolver
+{
+ private final static Logger LOGGER = LoggerFactory.getLogger(LinkCrawler.class);
+
+ public LinkCrawler(String originalURL) {
+ super(originalURL);
+ }
+
+ public void run() {
+
+ Preconditions.checkNotNull(linkDetails);
+
+ Preconditions.checkNotNull(linkDetails.getOriginalURL());
+
+ super.run();
+
+ Preconditions.checkNotNull(linkDetails.getFinalURL());
+
+ crawlLink(linkDetails.getFinalURL());
+
+ }
+
+
+ public void crawlLink(String url)
+ {
+
+ HttpURLConnection connection = null;
+
+ try
+ {
+ URL thisURL = new URL(url);
+ connection = (HttpURLConnection)new URL(url).openConnection();
+
+ // now we are going to pretend that we are a browser...
+ // This is the way my mac works.
+ if(!BOTS_ARE_OK.contains(thisURL.getHost()))
+ {
+ connection.addRequestProperty("Host", thisURL.getHost());
+ connection.addRequestProperty("Connection", "Keep-Alive");
+ connection.addRequestProperty("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/31.0.1650.48 Safari/537.36");
+ connection.addRequestProperty("Accept-Language", "en-US,en;q=0.8,zh;q=0.6");
+
+ // the test to seattlemamadoc.com prompted this change.
+ // they auto detect bots by checking the referrer chain and the 'user-agent'
+ // this broke the t.co test. t.co URLs are EXPLICITLY ok with bots
+ // there is a list for URLS that behave this way at the top in BOTS_ARE_OK
+ // smashew 2013-13-2013
+
+ if(linkDetails.getRedirectCount() > 0 && BOTS_ARE_OK.contains(thisURL.getHost()))
+ connection.addRequestProperty("Referrer", linkDetails.getOriginalURL());
+ }
+
+ connection.setReadTimeout(DEFAULT_HTTP_TIMEOUT);
+ connection.setConnectTimeout(DEFAULT_HTTP_TIMEOUT);
+
+ connection.setInstanceFollowRedirects(false);
+
+ if(linkDetails.getCookies() != null)
+ for (String cookie : linkDetails.getCookies())
+ connection.addRequestProperty("Cookie", cookie.split(";", 1)[0]);
+
+ connection.connect();
+
+ linkDetails.setFinalResponseCode((long)connection.getResponseCode());
+
+ InputStreamReader in = new InputStreamReader((InputStream) connection.getContent());
+ BufferedReader buff = new BufferedReader(in);
+ StringBuffer contentBuffer = new StringBuffer();
+ String line;
+ do {
+ line = buff.readLine();
+ contentBuffer.append(line + "\n");
+ } while (line != null);
+
+ linkDetails.setContentType(connection.getContentType());
+ linkDetails.setContent(contentBuffer.toString());
+
+ /**************
+ *
+ */
+ Map<String,List<String>> headers = createCaseInsensitiveMap(connection.getHeaderFields());
+ /******************************************************************
+ * If they want us to set cookies, well, then we will set cookies
+ * Example URL:
+ * http://nyti.ms/1bCpesx
+ *****************************************************************/
+ if(headers.containsKey(SET_COOKIE_IDENTIFIER))
+ linkDetails.getCookies().add(headers.get(SET_COOKIE_IDENTIFIER).get(0));
+
+ switch (linkDetails.getFinalResponseCode().intValue())
+ {
+ case 200: // HTTP OK
+ linkDetails.setFinalURL(connection.getURL().toString());
+ linkDetails.setDomain(new URL(linkDetails.getFinalURL()).getHost());
+ linkDetails.setLinkStatus(LinkDetails.LinkStatus.SUCCESS);
+ break;
+ case 300: // Multiple choices
+ case 301: // URI has been moved permanently
+ case 302: // Found
+ case 303: // Primarily for a HTTP Post
+ case 304: // Not Modified
+ case 306: // This status code is unused but in the redirect block.
+ case 307: // Temporary re-direct
+ /*******************************************************************
+ * Author:
+ * Smashew
+ *
+ * Date: 2013-11-15
+ *
+ * Note:
+ * It is possible that we have already found our final URL. In
+ * the event that we have found our final URL, we are going to
+ * save this URL as long as it isn't the original URL.
+ * We are still going to ask the browser to re-direct, but in the
+ * case of yet another redirect, seen with the redbull test
+ * this can be followed by a 304, a browser, by W3C standards would
+ * still render the page with it's content, but for us to assert
+ * a success, we are really hoping for a 304 message.
+ *******************************************************************/
+ if(!linkDetails.getOriginalURL().toLowerCase().equals(connection.getURL().toString().toLowerCase()))
+ linkDetails.setFinalURL(connection.getURL().toString());
+ if(!headers.containsKey(LOCATION_IDENTIFIER))
+ {
+ LOGGER.info("Headers: {}", headers);
+ linkDetails.setLinkStatus(LinkDetails.LinkStatus.REDIRECT_ERROR);
+ }
+ else
+ {
+ linkDetails.setRedirected(Boolean.TRUE);
+ linkDetails.setRedirectCount(linkDetails.getRedirectCount().longValue()+1);
+ unwindLink(connection.getHeaderField(LOCATION_IDENTIFIER));
+ }
+ break;
+ case 305: // User must use the specified proxy (deprecated by W3C)
+ break;
+ case 401: // Unauthorized (nothing we can do here)
+ linkDetails.setLinkStatus(LinkDetails.LinkStatus.UNAUTHORIZED);
+ break;
+ case 403: // HTTP Forbidden (Nothing we can do here)
+ linkDetails.setLinkStatus(LinkDetails.LinkStatus.FORBIDDEN);
+ break;
+ case 404: // Not Found (Page is not found, nothing we can do with a 404)
+ linkDetails.setLinkStatus(LinkDetails.LinkStatus.NOT_FOUND);
+ break;
+ case 500: // Internal Server Error
+ case 501: // Not Implemented
+ case 502: // Bad Gateway
+ case 503: // Service Unavailable
+ case 504: // Gateway Timeout
+ case 505: // Version not supported
+ linkDetails.setLinkStatus(LinkDetails.LinkStatus.HTTP_ERROR_STATUS);
+ break;
+ default:
+ LOGGER.info("Unrecognized HTTP Response Code: {}", linkDetails.getFinalResponseCode());
+ linkDetails.setLinkStatus(LinkDetails.LinkStatus.NOT_FOUND);
+ break;
+ }
+ }
+ catch (MalformedURLException e)
+ {
+ // the URL is trash, so, it can't load it.
+ linkDetails.setLinkStatus(LinkDetails.LinkStatus.MALFORMED_URL);
+ }
+ catch (IOException ex)
+ {
+ // there was an issue we are going to set to error.
+ linkDetails.setLinkStatus(LinkDetails.LinkStatus.ERROR);
+ }
+ catch (Exception ex)
+ {
+ // there was an unknown issue we are going to set to exception.
+ linkDetails.setLinkStatus(LinkDetails.LinkStatus.EXCEPTION);
+ }
+ finally
+ {
+ if (connection != null)
+ connection.disconnect();
+ }
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkCrawlerProcessor.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkCrawlerProcessor.java b/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkCrawlerProcessor.java
new file mode 100644
index 0000000..a3948d7
--- /dev/null
+++ b/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkCrawlerProcessor.java
@@ -0,0 +1,115 @@
+package org.apache.streams.urls;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsonorg.JsonOrgModule;
+import com.google.common.collect.Lists;
+import org.apache.commons.lang.NotImplementedException;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProcessor;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.pojo.json.Activity;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+
+/**
+ * References:
+ * Some helpful references to help
+ * Purpose URL
+ * ------------- ----------------------------------------------------------------
+ * [Status Codes] http://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html
+ * [Test Cases] http://greenbytes.de/tech/tc/httpredirects/
+ * [t.co behavior] https://dev.twitter.com/docs/tco-redirection-behavior
+ */
+
+public class LinkCrawlerProcessor implements StreamsProcessor
+{
+ public final static String STREAMS_ID = "LinkCrawlerProcessor";
+
+ private final static Logger LOGGER = LoggerFactory.getLogger(LinkCrawlerProcessor.class);
+
+ private ObjectMapper mapper;
+
+ @Override
+ public List<StreamsDatum> process(StreamsDatum entry) {
+
+ List<StreamsDatum> result = Lists.newArrayList();
+
+ LOGGER.debug("{} processing {}", STREAMS_ID, entry.getDocument().getClass());
+
+ Activity activity;
+
+ System.out.println( STREAMS_ID + " processing " + entry.getDocument().getClass());
+ // get list of shared urls
+ if( entry.getDocument() instanceof Activity) {
+
+ activity = (Activity) entry.getDocument();
+
+ }
+ else if(entry.getDocument() instanceof String) {
+
+ try {
+ activity = mapper.readValue((String) entry.getDocument(), Activity.class);
+ } catch (Exception e) {
+ e.printStackTrace();
+ LOGGER.warn(e.getMessage());
+ return(Lists.newArrayList(entry));
+ }
+
+ }
+ else throw new NotImplementedException();
+
+ List<String> outputLinks = activity.getLinks();
+ // for each
+ for( String link : outputLinks ) {
+
+ System.out.println( "pulling " + link);
+
+ try {
+ StreamsDatum outputDatum = crawlLink(link, entry);
+ if( outputDatum != null )
+ result.add(outputDatum);
+ } catch (Exception e) {
+ //drop unexpandable links
+ LOGGER.debug("Failed to expand link : {}", link);
+ LOGGER.debug("Excpetion expanding link : {}", e);
+ }
+
+ }
+
+ return result;
+ }
+
+ private StreamsDatum crawlLink(String link, StreamsDatum input) {
+
+ LinkCrawler crawler = new LinkCrawler((String)link);
+ crawler.run();
+ StreamsDatum datum = null;
+ if(input.getId() == null)
+ try {
+ datum = new StreamsDatum(this.mapper.writeValueAsString(crawler.getLinkDetails()));
+ } catch (JsonProcessingException e) {
+ e.printStackTrace();
+ return null;
+ }
+ datum.setSequenceid(input.getSequenceid());
+ datum.setMetadata(input.getMetadata());
+ datum.setTimestamp(input.getTimestamp());
+ return datum;
+
+ }
+
+ @Override
+ public void prepare(Object o) {
+ this.mapper = StreamsJacksonMapper.getInstance();
+ this.mapper.registerModule(new JsonOrgModule());
+ }
+
+ @Override
+ public void cleanUp() {
+
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkResolver.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkResolver.java b/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkResolver.java
index 7b5b012..ab72613 100644
--- a/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkResolver.java
+++ b/streams-contrib/streams-processor-urls/src/main/java/org/apache/streams/urls/LinkResolver.java
@@ -27,18 +27,18 @@ public class LinkResolver
{
private final static Logger LOGGER = LoggerFactory.getLogger(LinkResolver.class);
- private static final int MAX_ALLOWED_REDIRECTS = 30;
- private static final int DEFAULT_HTTP_TIMEOUT = 5000; //originally 30000
- private static final String LOCATION_IDENTIFIER = "location";
- private static final String SET_COOKIE_IDENTIFIER = "set-cookie";
+ protected static final int MAX_ALLOWED_REDIRECTS = 30;
+ protected static final int DEFAULT_HTTP_TIMEOUT = 5000; //originally 30000
+ protected static final String LOCATION_IDENTIFIER = "location";
+ protected static final String SET_COOKIE_IDENTIFIER = "set-cookie";
- private LinkDetails linkDetails = new LinkDetails();
+ protected LinkDetails linkDetails = new LinkDetails();
- private static final Collection<String> BOTS_ARE_OK = new ArrayList<String>() {{
+ protected static final Collection<String> BOTS_ARE_OK = new ArrayList<String>() {{
add("t.co");
}};
- private static final Collection<String> URL_TRACKING_TO_REMOVE = new ArrayList<String>() {{
+ protected static final Collection<String> URL_TRACKING_TO_REMOVE = new ArrayList<String>() {{
/******************************************************************
* Google uses parameters in the URL string to track referrers
* on their Google Analytics and promotions. These are the
@@ -250,7 +250,7 @@ public class LinkResolver
}
}
- private Map<String,List<String>> createCaseInsensitiveMap(Map<String,List<String>> input) {
+ protected Map<String,List<String>> createCaseInsensitiveMap(Map<String,List<String>> input) {
Map<String,List<String>> toReturn = new HashMap<String, List<String>>();
for(String k : input.keySet())
if(k != null && input.get(k) != null)
@@ -258,7 +258,7 @@ public class LinkResolver
return toReturn;
}
- private String cleanURL(String url)
+ protected String cleanURL(String url)
{
// If they pass us a null URL then we are going to pass that right back to them.
if(url == null)
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-processor-urls/src/main/jsonschema/org/apache/streams/urls/LinkDetails.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-processor-urls/src/main/jsonschema/org/apache/streams/urls/LinkDetails.json b/streams-contrib/streams-processor-urls/src/main/jsonschema/org/apache/streams/urls/LinkDetails.json
index 83f7692..112645c 100644
--- a/streams-contrib/streams-processor-urls/src/main/jsonschema/org/apache/streams/urls/LinkDetails.json
+++ b/streams-contrib/streams-processor-urls/src/main/jsonschema/org/apache/streams/urls/LinkDetails.json
@@ -68,6 +68,12 @@
},
"tookInMills": {
"type": "integer"
+ },
+ "content": {
+ "type": "string"
+ },
+ "contentType": {
+ "type": "string"
}
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/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 bf25631..24b8d0f 100644
--- a/streams-contrib/streams-provider-twitter/pom.xml
+++ b/streams-contrib/streams-provider-twitter/pom.xml
@@ -67,7 +67,12 @@
<artifactId>twitter4j-core</artifactId>
<version>3.0.5</version>
</dependency>
-
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-all</artifactId>
+ <version>1.3</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/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 2ed73bb..3a8bf6d 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
@@ -2,7 +2,6 @@ package org.apache.streams.twitter.provider;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Preconditions;
-import com.jayway.jsonassert.JsonAssert;
import org.apache.commons.lang.StringUtils;
import org.apache.streams.twitter.pojo.*;
import org.apache.streams.twitter.serializer.StreamsTwitterMapper;
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonUserstreameventActivitySerializer.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonUserstreameventActivitySerializer.java b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonUserstreameventActivitySerializer.java
new file mode 100644
index 0000000..3f43477
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/serializer/TwitterJsonUserstreameventActivitySerializer.java
@@ -0,0 +1,88 @@
+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.util.List;
+
+import static org.apache.streams.twitter.serializer.TwitterJsonActivitySerializer.*;
+
+/**
+* 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 TwitterJsonUserstreameventActivitySerializer implements ActivitySerializer<String> {
+
+ @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();
+ activity.setActor(buildActor(delete));
+ activity.setVerb("delete");
+ activity.setObject(buildActivityObject(delete));
+ activity.setId(TwitterJsonActivitySerializer.formatId(activity.getVerb(), delete.getDelete().getStatus().getIdStr()));
+ if(Strings.isNullOrEmpty(activity.getId()))
+ throw new ActivitySerializerException("Unable to determine activity id");
+ activity.setProvider(getProvider());
+ addTwitterExtension(activity, event);
+ return activity;
+ }
+
+ public Actor buildActor(Delete delete) {
+ Actor actor = new Actor();
+ actor.setId(formatId(delete.getDelete().getStatus().getUserIdStr()));
+ return actor;
+ }
+
+ public ActivityObject buildActivityObject(Delete delete) {
+ ActivityObject actObj = new ActivityObject();
+ actObj.setId(formatId(delete.getDelete().getStatus().getIdStr()));
+ actObj.setObjectType("tweet");
+ return actObj;
+ }
+
+ public ActivityObject buildTarget(Tweet tweet) {
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/FriendList.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/FriendList.json b/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/FriendList.json
new file mode 100644
index 0000000..5dd7687
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/FriendList.json
@@ -0,0 +1,20 @@
+{
+ "type": "object",
+ "$schema": "http://json-schema.org/draft-03/schema",
+ "id": "#",
+ "javaType" : "org.apache.streams.twitter.pojo.FriendList",
+ "properties": {
+ "friends": {
+ "type": "array",
+ "items": {
+ "type": "integer"
+ }
+ },
+ "friends_str": {
+ "type": "array",
+ "items": {
+ "type": "string"
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/UserstreamEvent.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/UserstreamEvent.json b/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/UserstreamEvent.json
new file mode 100644
index 0000000..652567e
--- /dev/null
+++ b/streams-contrib/streams-provider-twitter/src/main/jsonschema/com/twitter/UserstreamEvent.json
@@ -0,0 +1,33 @@
+{
+ "type": "object",
+ "$schema": "http://json-schema.org/draft-03/schema",
+ "id": "#",
+ "javaType" : "org.apache.streams.twitter.pojo.UserstreamEvent",
+ "properties": {
+ "created_at": {
+ "type": "string",
+ "format" : "date-time"
+ },
+ "event_type": {
+ "type": "string",
+ "items": {
+ "type": "integer"
+ }
+ },
+ "source": {
+ "type": "string",
+ "items": {
+ "type": "integer"
+ }
+ },
+ "target": {
+ "type": "string",
+ "items": {
+ "type": "integer"
+ }
+ },
+ "target_object": {
+ "type": "object"
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a79e1150/streams-pojo/src/main/jsonschema/org/apache/streams/pojo/json/activity.json
----------------------------------------------------------------------
diff --git a/streams-pojo/src/main/jsonschema/org/apache/streams/pojo/json/activity.json b/streams-pojo/src/main/jsonschema/org/apache/streams/pojo/json/activity.json
index fa74486..593a1ab 100644
--- a/streams-pojo/src/main/jsonschema/org/apache/streams/pojo/json/activity.json
+++ b/streams-pojo/src/main/jsonschema/org/apache/streams/pojo/json/activity.json
@@ -82,6 +82,10 @@
"items": {
"type" : "string"
}
+ },
+ "extensions" : {
+ "type": "object",
+ "description": "Collection of annotations by processing modules"
}
},
"links": [