You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streams.apache.org by mf...@apache.org on 2014/09/04 21:42:03 UTC
[1/6] git commit: Merge pull request #3 from apache/master
Repository: incubator-streams
Updated Branches:
refs/heads/master fe4cec680 -> d8a3380bf
Merge pull request #3 from apache/master
Apache master merge 2014/08/19
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/dca64751
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/dca64751
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/dca64751
Branch: refs/heads/master
Commit: dca647510984d189904674511b41fa1d4889b788
Parents: 9c6164c ee6b4d5
Author: Ryan Ebanks <ry...@raveldata.com>
Authored: Tue Aug 19 11:11:26 2014 -0500
Committer: Ryan Ebanks <ry...@raveldata.com>
Committed: Tue Aug 19 11:11:26 2014 -0500
----------------------------------------------------------------------
pom.xml | 1 +
.../ElasticsearchPersistWriter.java | 3 +
.../streams-provider-facebook/pom.xml | 2 +
.../api/FacebookPageActivitySerializer.java | 66 ++++++
.../processor/FacebookTypeConverter.java | 41 ++--
.../provider/FacebookEventClassifier.java | 57 ++++++
.../provider/FacebookStreamConfigurator.java | 66 ++++++
.../serializer/FacebookActivityUtil.java | 125 ++++++++++++
.../org/apache/streams/facebook/Page.json | 168 ++++++++++++++++
.../test/FacebookActivityActorSerDeTest.java | 65 ++++++
.../test/FacebookActivitySerDeTest.java | 7 -
.../test/FacebookEventClassifierTest.java | 57 ++++++
.../facebook/test/FacebookPageSerDeTest.java | 75 +++++++
.../facebook/test/FacebookPostSerDeTest.java | 1 -
.../streams/facebook/test/SimplePageTest.java | 100 ++++++++++
.../src/test/resources/testpage.json | 1 +
.../instagram/provider/InstagramOauthToken.java | 43 ++++
.../provider/InstagramRecentMediaCollector.java | 200 ++++++++++---------
.../provider/InstagramRecentMediaProvider.java | 110 ++++++++--
.../com/instagram/InstagramConfiguration.json | 63 +++++-
.../InstagramRecentMediaCollectorTest.java | 130 ++++++------
.../InstagramRecentMediaProviderTest.java | 21 +-
streams-util/pom.xml | 6 +-
.../org/apache/streams/util/ComponentUtils.java | 42 +++-
.../backoff/AbstractBackOffStrategy.java | 78 ++++++++
.../api/requests/backoff/BackOffException.java | 62 ++++++
.../api/requests/backoff/BackOffStrategy.java | 50 +++++
.../impl/ConstantTimeBackOffStrategy.java | 47 +++++
.../impl/ExponentialBackOffStrategy.java | 46 +++++
.../backoff/impl/LinearTimeBackOffStrategy.java | 39 ++++
.../util/oauth/tokens/AbstractOauthToken.java | 33 +++
.../tokens/tokenmanager/SimpleTokenManager.java | 53 +++++
.../tokenmanager/impl/BasicTokenManger.java | 86 ++++++++
.../requests/backoff/BackOffStrategyTest.java | 70 +++++++
.../ConstantTimeBackOffStrategyTest.java | 26 +++
.../backoff/ExponentialBackOffStrategyTest.java | 23 +++
.../backoff/LinearTimeBackOffStartegyTest.java | 22 ++
.../tokenmanager/TestBasicTokenManager.java | 176 ++++++++++++++++
38 files changed, 2034 insertions(+), 227 deletions(-)
----------------------------------------------------------------------
[5/6] git commit: Streams-159 | Added mindmup documentation and
template
Posted by mf...@apache.org.
Streams-159 | Added mindmup documentation and template
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/6d67fd62
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/6d67fd62
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/6d67fd62
Branch: refs/heads/master
Commit: 6d67fd62c118609403b20726ee8bb1c700b23b1c
Parents: 5795335
Author: Ryan Ebanks <re...@Informations-MacBook-Pro-3.local>
Authored: Thu Aug 28 16:01:44 2014 -0500
Committer: Ryan Ebanks <re...@Informations-MacBook-Pro-3.local>
Committed: Thu Aug 28 16:01:44 2014 -0500
----------------------------------------------------------------------
.../metadata/facebook_post.mup | 694 +++++++++++++++++++
.../metadata/facebook_post.png | Bin 0 -> 1038039 bytes
streams-pojo/metadata/Serializer.mup | 371 ++++++++++
streams-pojo/metadata/serializer.png | Bin 0 -> 604337 bytes
4 files changed, 1065 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/6d67fd62/streams-contrib/streams-provider-facebook/metadata/facebook_post.mup
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/metadata/facebook_post.mup b/streams-contrib/streams-provider-facebook/metadata/facebook_post.mup
new file mode 100644
index 0000000..ccedae4
--- /dev/null
+++ b/streams-contrib/streams-provider-facebook/metadata/facebook_post.mup
@@ -0,0 +1,694 @@
+{
+ "title": "Serializer",
+ "id": 1,
+ "formatVersion": 2,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {
+ "1": {
+ "title": "Activity",
+ "id": 2,
+ "ideas": {
+ "1": {
+ "title": "actor",
+ "id": 4,
+ "ideas": {
+ "1": {
+ "title": "summary",
+ "id": 28
+ },
+ "7": {
+ "title": "url",
+ "id": 41
+ },
+ "0.5": {
+ "title": "displayName",
+ "id": 32,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.75": {
+ "title": "extensions",
+ "id": 33,
+ "ideas": {
+ "1": {
+ "title": "followers",
+ "id": 34
+ },
+ "2": {
+ "title": "handle",
+ "id": 35
+ },
+ "3": {
+ "title": "location",
+ "id": 36
+ },
+ "4": {
+ "title": "location",
+ "id": 37
+ },
+ "5": {
+ "title": "screenName",
+ "id": 38
+ },
+ "6": {
+ "title": "posts",
+ "id": 39
+ },
+ "7": {
+ "title": "favorites",
+ "id": 40
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.875": {
+ "title": "handle",
+ "id": 30,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.9375": {
+ "title": "id",
+ "id": 29,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.96875": {
+ "title": "image",
+ "id": 31,
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "2": {
+ "title": "id",
+ "id": 5
+ },
+ "3": {
+ "title": "verb",
+ "id": 6
+ },
+ "1.5": {
+ "title": "content",
+ "id": 16,
+ "attr": {
+ "style": {}
+ }
+ },
+ "2.5": {
+ "title": "icon",
+ "id": 13,
+ "attr": {
+ "style": {}
+ }
+ },
+ "1.75": {
+ "title": "generator",
+ "id": 12,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 62
+ },
+ "0.5": {
+ "title": "displayName",
+ "id": 63,
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "2.75": {
+ "title": "links",
+ "id": 18,
+ "attr": {
+ "style": {}
+ }
+ },
+ "2.875": {
+ "title": "object",
+ "id": 8,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {
+ "1": {
+ "title": "content",
+ "id": 22
+ },
+ "2": {
+ "title": "id",
+ "id": 23
+ },
+ "4": {
+ "title": "objectType",
+ "id": 25
+ },
+ "1.5": {
+ "title": "geo",
+ "id": 24,
+ "ideas": {
+ "1": {
+ "title": "coordinates",
+ "id": 26
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "2.9375": {
+ "title": "published",
+ "id": 10,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {}
+ },
+ "2.90625": {
+ "title": "provider",
+ "id": 14,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 20
+ },
+ "0.5": {
+ "title": "displayName",
+ "id": 21,
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "2.96875": {
+ "title": "target",
+ "id": 9,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 64
+ },
+ "2": {
+ "title": "image",
+ "id": 65
+ },
+ "4": {
+ "title": "summary",
+ "id": 67
+ },
+ "0.5": {
+ "title": "displayName",
+ "id": 66,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.25": {
+ "title": "content",
+ "id": 68,
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "2.984375": {
+ "title": "updated",
+ "id": 11,
+ "attr": {
+ "style": {}
+ }
+ },
+ "2.9765625": {
+ "title": "title",
+ "id": 15,
+ "attr": {
+ "style": {}
+ }
+ },
+ "2.9921875": {
+ "title": "url",
+ "id": 17,
+ "attr": {
+ "style": {}
+ }
+ },
+ "1.625": {
+ "title": "extensions",
+ "id": 42,
+ "ideas": {
+ "1": {
+ "title": "facebook",
+ "id": 43,
+ "ideas": {}
+ },
+ "6": {
+ "title": "user_mentions",
+ "id": 58,
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 59
+ },
+ "2": {
+ "title": "handle",
+ "id": 60
+ },
+ "3": {
+ "title": "displayName",
+ "id": 61
+ }
+ }
+ },
+ "0.5": {
+ "title": "hashtags",
+ "id": 57,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.75": {
+ "title": "location",
+ "id": 48,
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 49
+ },
+ "2": {
+ "title": "coordinates",
+ "id": 50,
+ "ideas": {
+ "1": {
+ "title": "type",
+ "id": 52
+ },
+ "2": {
+ "title": "coordinates",
+ "id": 53
+ }
+ }
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.625": {
+ "title": "likes",
+ "id": 54,
+ "ideas": {
+ "1": {
+ "title": "perspectival",
+ "id": 55
+ },
+ "2": {
+ "title": "count",
+ "id": 56
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.875": {
+ "title": "rebroadcasts",
+ "id": 45,
+ "ideas": {
+ "1": {
+ "title": "perspectival",
+ "id": 46
+ },
+ "2": {
+ "title": "count",
+ "id": 47
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "-1": {
+ "title": "Facebook Post",
+ "id": 3,
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 69
+ },
+ "2": {
+ "title": "likes",
+ "id": 92,
+ "ideas": {
+ "1": {
+ "title": "data",
+ "id": 93,
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 94
+ },
+ "2": {
+ "title": "name",
+ "id": 95
+ }
+ }
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "3": {
+ "title": "message",
+ "id": 74
+ },
+ "4": {
+ "title": "picture",
+ "id": 75
+ },
+ "7": {
+ "title": "shares",
+ "id": 97,
+ "ideas": {
+ "1": {
+ "title": "count",
+ "id": 98
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "10": {
+ "title": "type",
+ "id": 96
+ },
+ "15": {
+ "title": "updated_time",
+ "id": 105
+ },
+ "0.5": {
+ "title": "actions",
+ "id": 78,
+ "ideas": {
+ "1": {
+ "title": "name",
+ "id": 79
+ },
+ "0.5": {
+ "title": "link",
+ "id": 80,
+ "attr": {
+ "style": {}
+ }
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.75": {
+ "title": "application",
+ "id": 100,
+ "ideas": {
+ "1": {
+ "title": "name",
+ "id": 101
+ },
+ "3": {
+ "title": "namespace",
+ "id": 103
+ },
+ "0.5": {
+ "title": "id",
+ "id": 102,
+ "attr": {
+ "style": {}
+ }
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.875": {
+ "title": "comments",
+ "id": 81,
+ "ideas": {
+ "1": {
+ "title": "data",
+ "id": 82,
+ "ideas": {
+ "1": {
+ "title": "user_likes",
+ "id": 83
+ },
+ "0.5": {
+ "title": "can_remove",
+ "id": 87,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.75": {
+ "title": "created_time",
+ "id": 91,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.875": {
+ "title": "id",
+ "id": 85,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {}
+ },
+ "0.8125": {
+ "title": "from",
+ "id": 88,
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 89
+ },
+ "2": {
+ "title": "name",
+ "id": 90
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.9375": {
+ "title": "like_count",
+ "id": 86,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.96875": {
+ "title": "message",
+ "id": 84,
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.9375": {
+ "title": "from",
+ "id": 70,
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 71
+ },
+ "2": {
+ "title": "name",
+ "id": 72
+ },
+ "0.5": {
+ "title": "category",
+ "id": 73,
+ "attr": {
+ "style": {}
+ }
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.96875": {
+ "title": "icon",
+ "id": 77,
+ "attr": {
+ "style": {}
+ }
+ },
+ "2.5": {
+ "title": "link",
+ "id": 76,
+ "attr": {
+ "style": {}
+ }
+ },
+ "3.5": {
+ "title": "object_id",
+ "id": 99,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.8125": {
+ "title": "created_time",
+ "id": 104,
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ }
+ },
+ "links": [
+ {
+ "ideaIdFrom": 104,
+ "ideaIdTo": 10,
+ "attr": {
+ "style": {
+ "color": "#FF0000",
+ "lineStyle": "dashed"
+ }
+ }
+ },
+ {
+ "ideaIdFrom": 21,
+ "ideaIdTo": 101,
+ "attr": {
+ "style": {
+ "color": "#FF0000",
+ "lineStyle": "dashed"
+ }
+ }
+ },
+ {
+ "ideaIdFrom": 102,
+ "ideaIdTo": 20,
+ "attr": {
+ "style": {
+ "color": "#FF0000",
+ "lineStyle": "dashed"
+ }
+ }
+ },
+ {
+ "ideaIdFrom": 43,
+ "ideaIdTo": 3,
+ "attr": {
+ "style": {
+ "color": "#FF0000",
+ "lineStyle": "dashed"
+ }
+ }
+ },
+ {
+ "ideaIdFrom": 72,
+ "ideaIdTo": 32,
+ "attr": {
+ "style": {
+ "color": "#FF0000",
+ "lineStyle": "dashed"
+ }
+ }
+ },
+ {
+ "ideaIdFrom": 13,
+ "ideaIdTo": 77,
+ "attr": {
+ "style": {
+ "color": "#FF0000",
+ "lineStyle": "dashed"
+ }
+ }
+ },
+ {
+ "ideaIdFrom": 18,
+ "ideaIdTo": 76,
+ "attr": {
+ "style": {
+ "color": "#FF0000",
+ "lineStyle": "dashed"
+ }
+ }
+ },
+ {
+ "ideaIdFrom": 74,
+ "ideaIdTo": 16,
+ "attr": {
+ "style": {
+ "color": "#FF0000",
+ "lineStyle": "dashed"
+ }
+ }
+ },
+ {
+ "ideaIdFrom": 98,
+ "ideaIdTo": 47,
+ "attr": {
+ "style": {
+ "color": "#FF0000",
+ "lineStyle": "dashed"
+ }
+ }
+ },
+ {
+ "ideaIdFrom": 105,
+ "ideaIdTo": 11,
+ "attr": {
+ "style": {
+ "color": "#FF0000",
+ "lineStyle": "dashed"
+ }
+ }
+ }
+ ]
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/6d67fd62/streams-contrib/streams-provider-facebook/metadata/facebook_post.png
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/metadata/facebook_post.png b/streams-contrib/streams-provider-facebook/metadata/facebook_post.png
new file mode 100644
index 0000000..2d5554f
Binary files /dev/null and b/streams-contrib/streams-provider-facebook/metadata/facebook_post.png differ
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/6d67fd62/streams-pojo/metadata/Serializer.mup
----------------------------------------------------------------------
diff --git a/streams-pojo/metadata/Serializer.mup b/streams-pojo/metadata/Serializer.mup
new file mode 100644
index 0000000..e8f88ca
--- /dev/null
+++ b/streams-pojo/metadata/Serializer.mup
@@ -0,0 +1,371 @@
+{
+ "title": "Serializer",
+ "id": 1,
+ "formatVersion": 2,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {
+ "1": {
+ "title": "Activity",
+ "id": 2,
+ "ideas": {
+ "1": {
+ "title": "actor",
+ "id": 4,
+ "ideas": {
+ "1": {
+ "title": "summary",
+ "id": 28
+ },
+ "7": {
+ "title": "url",
+ "id": 41
+ },
+ "0.5": {
+ "title": "displayName",
+ "id": 32,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.75": {
+ "title": "extensions",
+ "id": 33,
+ "ideas": {
+ "1": {
+ "title": "followers",
+ "id": 34
+ },
+ "2": {
+ "title": "handle",
+ "id": 35
+ },
+ "3": {
+ "title": "location",
+ "id": 36
+ },
+ "4": {
+ "title": "location",
+ "id": 37
+ },
+ "5": {
+ "title": "screenName",
+ "id": 38
+ },
+ "6": {
+ "title": "posts",
+ "id": 39
+ },
+ "7": {
+ "title": "favorites",
+ "id": 40
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.875": {
+ "title": "handle",
+ "id": 30,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.9375": {
+ "title": "id",
+ "id": 29,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.96875": {
+ "title": "image",
+ "id": 31,
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "2": {
+ "title": "id",
+ "id": 5
+ },
+ "3": {
+ "title": "verb",
+ "id": 6
+ },
+ "1.5": {
+ "title": "content",
+ "id": 16,
+ "attr": {
+ "style": {}
+ }
+ },
+ "2.5": {
+ "title": "icon",
+ "id": 13,
+ "attr": {
+ "style": {}
+ }
+ },
+ "1.75": {
+ "title": "generator",
+ "id": 12,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 62
+ },
+ "0.5": {
+ "title": "displayName",
+ "id": 63,
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "2.75": {
+ "title": "links",
+ "id": 18,
+ "attr": {
+ "style": {}
+ }
+ },
+ "2.875": {
+ "title": "object",
+ "id": 8,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {
+ "1": {
+ "title": "content",
+ "id": 22
+ },
+ "2": {
+ "title": "id",
+ "id": 23
+ },
+ "4": {
+ "title": "objectType",
+ "id": 25
+ },
+ "1.5": {
+ "title": "geo",
+ "id": 24,
+ "ideas": {
+ "1": {
+ "title": "coordinates",
+ "id": 26
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "2.9375": {
+ "title": "published",
+ "id": 10,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {}
+ },
+ "2.90625": {
+ "title": "provider",
+ "id": 14,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 20
+ },
+ "0.5": {
+ "title": "displayName",
+ "id": 21,
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "2.96875": {
+ "title": "target",
+ "id": 9,
+ "attr": {
+ "style": {}
+ },
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 64
+ },
+ "2": {
+ "title": "image",
+ "id": 65
+ },
+ "4": {
+ "title": "summary",
+ "id": 67
+ },
+ "0.5": {
+ "title": "displayName",
+ "id": 66,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.25": {
+ "title": "content",
+ "id": 68,
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "2.984375": {
+ "title": "updated",
+ "id": 11,
+ "attr": {
+ "style": {}
+ }
+ },
+ "2.9765625": {
+ "title": "title",
+ "id": 15,
+ "attr": {
+ "style": {}
+ }
+ },
+ "2.9921875": {
+ "title": "url",
+ "id": 17,
+ "attr": {
+ "style": {}
+ }
+ },
+ "1.625": {
+ "title": "extensions",
+ "id": 42,
+ "ideas": {
+ "1": {
+ "title": "source",
+ "id": 43,
+ "ideas": {
+ "1": {
+ "title": "source json",
+ "id": 44
+ }
+ }
+ },
+ "6": {
+ "title": "user_mentions",
+ "id": 58,
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 59
+ },
+ "2": {
+ "title": "handle",
+ "id": 60
+ },
+ "3": {
+ "title": "displayName",
+ "id": 61
+ }
+ }
+ },
+ "0.5": {
+ "title": "hashtags",
+ "id": 57,
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.75": {
+ "title": "location",
+ "id": 48,
+ "ideas": {
+ "1": {
+ "title": "id",
+ "id": 49
+ },
+ "2": {
+ "title": "coordinates",
+ "id": 50,
+ "ideas": {
+ "1": {
+ "title": "type",
+ "id": 52
+ },
+ "2": {
+ "title": "coordinates",
+ "id": 53
+ }
+ }
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.625": {
+ "title": "likes",
+ "id": 54,
+ "ideas": {
+ "1": {
+ "title": "perspectival",
+ "id": 55
+ },
+ "2": {
+ "title": "count",
+ "id": 56
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ },
+ "0.875": {
+ "title": "rebroadcasts",
+ "id": 45,
+ "ideas": {
+ "1": {
+ "title": "perspectival",
+ "id": 46
+ },
+ "2": {
+ "title": "count",
+ "id": 47
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ }
+ },
+ "attr": {
+ "style": {}
+ }
+ }
+ }
+ },
+ "-1": {
+ "title": "Source Data",
+ "id": 3
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/6d67fd62/streams-pojo/metadata/serializer.png
----------------------------------------------------------------------
diff --git a/streams-pojo/metadata/serializer.png b/streams-pojo/metadata/serializer.png
new file mode 100644
index 0000000..3f99b8e
Binary files /dev/null and b/streams-pojo/metadata/serializer.png differ
[4/6] git commit: Added TODO prefix
Posted by mf...@apache.org.
Added TODO prefix
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/5795335a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/5795335a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/5795335a
Branch: refs/heads/master
Commit: 5795335aeb55d0c9b9379f1e58dde7ff752e2928
Parents: a628676
Author: Ryan Ebanks <re...@Informations-MacBook-Pro-3.local>
Authored: Wed Aug 27 13:10:40 2014 -0500
Committer: Ryan Ebanks <re...@Informations-MacBook-Pro-3.local>
Committed: Wed Aug 27 13:10:40 2014 -0500
----------------------------------------------------------------------
.../facebook/provider/pagefeed/FacebookPageFeedDataCollector.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/5795335a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedDataCollector.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedDataCollector.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedDataCollector.java
index 184d273..2027c37 100644
--- a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedDataCollector.java
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedDataCollector.java
@@ -114,7 +114,7 @@ public class FacebookPageFeedDataCollector extends FacebookDataCollector {
} catch (FacebookException fe) {
LOGGER.error("Facebook returned an exception : {}", fe);
LOGGER.error("Facebook returned an exception while trying to get feed for page, {} : {}", pageId, fe.getMessage());
- // Rate limit exceptions with facebook4j unclear http://facebook4j.org/oldjavadocs/1.1.12-2.0.0/2.0.0/index.html?facebook4j/internal/http/HttpResponseCode.html
+ //TODO Rate limit exceptions with facebook4j unclear http://facebook4j.org/oldjavadocs/1.1.12-2.0.0/2.0.0/index.html?facebook4j/internal/http/HttpResponseCode.html
// back off at all exceptions until figured out.
super.backOff.backOff();
}
[2/6] git commit: Updated facebook provider to have page feed
provider and fixed serializer for posts
Posted by mf...@apache.org.
Updated facebook provider to have page feed provider and fixed serializer for posts
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/e7b90335
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/e7b90335
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/e7b90335
Branch: refs/heads/master
Commit: e7b90335de261f2f2537a3ae94095da06ce4b592
Parents: dca6475
Author: Ryan Ebanks <re...@Informations-MacBook-Pro-3.local>
Authored: Wed Aug 27 06:51:56 2014 -0500
Committer: Ryan Ebanks <re...@Informations-MacBook-Pro-3.local>
Committed: Wed Aug 27 06:51:56 2014 -0500
----------------------------------------------------------------------
.../streams/console/ConsolePersistWriter.java | 5 +-
.../api/FacebookPostActivitySerializer.java | 29 +++-
.../processor/FacebookTypeConverter.java | 5 +
.../pagefeed/FacebookDataCollector.java | 138 ++++++++++++++++++
.../pagefeed/FacebookPageFeedDataCollector.java | 124 +++++++++++++++++
.../pagefeed/FacebookPageFeedProvider.java | 28 ++++
.../provider/pagefeed/FacebookProvider.java | 139 +++++++++++++++++++
.../FacebookStreamsPostSerializer.java | 60 ++++++++
.../streams/facebook/FacebookConfiguration.json | 33 +++++
.../provider/pagefeed/TestFacebookProvider.java | 94 +++++++++++++
10 files changed, 648 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e7b90335/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistWriter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistWriter.java b/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistWriter.java
index 3140479..96d116f 100644
--- a/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistWriter.java
+++ b/streams-contrib/streams-persist-console/src/main/java/org/apache/streams/console/ConsolePersistWriter.java
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
import org.apache.streams.core.StreamsDatum;
import org.apache.streams.core.StreamsPersistWriter;
+import org.apache.streams.jackson.StreamsJacksonMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,7 +36,7 @@ public class ConsolePersistWriter implements StreamsPersistWriter {
protected volatile Queue<StreamsDatum> persistQueue;
- private ObjectMapper mapper = new ObjectMapper();
+ private ObjectMapper mapper = StreamsJacksonMapper.getInstance();
public ConsolePersistWriter() {
this.persistQueue = new ConcurrentLinkedQueue<StreamsDatum>();
@@ -60,7 +61,7 @@ public class ConsolePersistWriter implements StreamsPersistWriter {
String text = mapper.writeValueAsString(entry);
- System.out.println(text);
+ System.out.println("\n"+text+"\n");
// LOGGER.info(text);
} catch (JsonProcessingException e) {
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e7b90335/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/api/FacebookPostActivitySerializer.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/api/FacebookPostActivitySerializer.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/api/FacebookPostActivitySerializer.java
index f1687ea..aa718fb 100644
--- a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/api/FacebookPostActivitySerializer.java
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/api/FacebookPostActivitySerializer.java
@@ -19,9 +19,12 @@
package org.apache.streams.facebook.api;
+import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import org.apache.commons.lang.NotImplementedException;
import org.apache.streams.data.ActivitySerializer;
import org.apache.streams.exceptions.ActivitySerializerException;
@@ -67,6 +70,7 @@ public class FacebookPostActivitySerializer implements ActivitySerializer<org.ap
@Override
public Activity deserialize(Post post) throws ActivitySerializerException {
+
Activity activity = new Activity();
activity.setPublished(post.getCreatedTime());
activity.setUpdated(post.getUpdatedTime());
@@ -76,6 +80,16 @@ public class FacebookPostActivitySerializer implements ActivitySerializer<org.ap
parseObject(activity, mapper.convertValue(post, ObjectNode.class));
fixObjectId(activity);
fixContentFromSummary(activity);
+ activity.setVerb("post");
+ List<String> links = Lists.newLinkedList();
+ links.add(post.getLink());
+ activity.setLinks(links);
+ ensureExtensions(activity).put("facebook", post);
+ if(post.getLikes() != null) {
+ Map<String, Object> likes = Maps.newHashMap();
+ likes.put("count", post.getLikes().size());
+ ensureExtensions(activity).put("likes", likes);
+ }
return activity;
}
@@ -108,7 +122,7 @@ public class FacebookPostActivitySerializer implements ActivitySerializer<org.ap
private void setProvider(Activity activity) {
Provider provider = new Provider();
- provider.setId(getProviderId(PROVIDER_NAME));
+ provider.setId("id:provider:"+PROVIDER_NAME);
provider.setDisplayName(PROVIDER_NAME);
activity.setProvider(provider);
}
@@ -180,7 +194,9 @@ public class FacebookPostActivitySerializer implements ActivitySerializer<org.ap
private void addLikeExtension(Activity activity, JsonNode value) {
Map<String, Object> extensions = ensureExtensions(activity);
- extensions.put(LIKES_EXTENSION, value.asInt());
+ Map<String, Object> likes = Maps.newHashMap();
+ likes.put("count", value.asLong());
+ extensions.put(LIKES_EXTENSION, likes);
}
private void addLocationExtension(Activity activity, JsonNode value) {
@@ -213,7 +229,7 @@ public class FacebookPostActivitySerializer implements ActivitySerializer<org.ap
}
private void addId(Activity activity, JsonNode value) {
- activity.setId(getActivityId(PROVIDER_NAME, value.asText()));
+ activity.setId("id:"+PROVIDER_NAME+":"+value.asText());
}
private void addObjectLink(Activity activity, JsonNode value) {
@@ -223,7 +239,10 @@ public class FacebookPostActivitySerializer implements ActivitySerializer<org.ap
private void addRebroadcastExtension(Activity activity, JsonNode value) {
Map<String, Object> extensions = ensureExtensions(activity);
if(value.has("count")) {
- extensions.put(REBROADCAST_EXTENSION, value.get("count").asInt());
+ Map<String, Object> rebroadCast = Maps.newHashMap();
+ rebroadCast.put("count", value.get("count").asLong());
+ rebroadCast.put("perspectival", true);
+ extensions.put(REBROADCAST_EXTENSION, rebroadCast);
}
}
@@ -245,7 +264,7 @@ public class FacebookPostActivitySerializer implements ActivitySerializer<org.ap
actor.setDisplayName(value.get("name").asText());
}
if(value.has("id")) {
- actor.setId(getPersonId(PROVIDER_NAME, value.get("id").asText()));
+ actor.setId("id:"+PROVIDER_NAME+":"+value.get("id").asText());
}
activity.setActor(actor);
}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e7b90335/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
index 7eb0e65..e160c77 100644
--- a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
@@ -175,6 +175,11 @@ public class FacebookTypeConverter implements StreamsProcessor {
if( out != null && validate(out, outClass))
result = new StreamsDatum(out);
+ } else if(item instanceof Post) {
+ Object out = convert(mapper.convertValue(item, ObjectNode.class), inClass, outClass);
+
+ if( out != null && validate(out, outClass))
+ result = new StreamsDatum(out);
}
} catch (Exception e) {
e.printStackTrace();
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e7b90335/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookDataCollector.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookDataCollector.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookDataCollector.java
new file mode 100644
index 0000000..196eaca
--- /dev/null
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookDataCollector.java
@@ -0,0 +1,138 @@
+/*
+ * 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.facebook.provider.pagefeed;
+
+import com.google.common.annotations.VisibleForTesting;
+import facebook4j.Facebook;
+import facebook4j.FacebookFactory;
+import facebook4j.auth.AccessToken;
+import facebook4j.conf.ConfigurationBuilder;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.facebook.FacebookConfiguration;
+import org.apache.streams.facebook.IdConfig;
+import org.apache.streams.util.api.requests.backoff.BackOffStrategy;
+import org.apache.streams.util.api.requests.backoff.impl.ExponentialBackOffStrategy;
+import org.apache.streams.util.oauth.tokens.tokenmanager.SimpleTokenManager;
+import org.apache.streams.util.oauth.tokens.tokenmanager.impl.BasicTokenManger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Abstract data collector for Facebook. Iterates over ids and queues data to be output
+ * by a {@link org.apache.streams.core.StreamsProvider}
+ */
+public abstract class FacebookDataCollector implements Runnable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FacebookDataCollector.class);
+ private static final String READ_ONLY = "read_streams";
+
+ @VisibleForTesting
+ protected AtomicBoolean isComplete;
+ protected BackOffStrategy backOff;
+
+ private FacebookConfiguration config;
+ private BlockingQueue<StreamsDatum> queue;
+ private SimpleTokenManager<String> authTokens;
+
+
+ public FacebookDataCollector(FacebookConfiguration config, BlockingQueue<StreamsDatum> queue) {
+ this.config = config;
+ this.queue = queue;
+ this.isComplete = new AtomicBoolean(false);
+ this.backOff = new ExponentialBackOffStrategy(5);
+ this.authTokens = new BasicTokenManger<String>();
+ if(config.getUserAccessTokens() != null) {
+ for(String token : config.getUserAccessTokens()) {
+ this.authTokens.addTokenToPool(token);
+ }
+ }
+ }
+
+ /**
+ * Returns true when the collector has finished querying facebook and has queued all data
+ * for the provider
+ * @return
+ */
+ public boolean isComplete(){
+ return this.isComplete.get();
+ }
+
+ /**
+ * Queues facebook data
+ * @param data
+ * @param id
+ */
+ protected void outputData(Object data, String id) {
+ try {
+ this.queue.put(new StreamsDatum(data, id));
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ /**
+ * Gets a Facebook client. If multiple authenticated users for this app are available
+ * it will rotate through the users oauth credentials
+ * @return
+ */
+ protected Facebook getNextFacebookClient() {
+ ConfigurationBuilder cb = new ConfigurationBuilder();
+ cb.setDebugEnabled(true)
+ .setOAuthAppId(this.config.getOauth().getAppId())
+ .setOAuthAppSecret(this.config.getOauth().getAppSecret());
+ if(this.authTokens.numAvailableTokens() > 0)
+ cb.setOAuthAccessToken(this.authTokens.getNextAvailableToken());
+ else {
+ cb.setOAuthAccessToken(this.config.getOauth().getAppAccessToken());
+ LOGGER.debug("appAccessToken : {}", this.config.getOauth().getAppAccessToken());
+ }
+ cb.setOAuthPermissions(READ_ONLY)
+ .setJSONStoreEnabled(true);
+ LOGGER.debug("appId : {}", this.config.getOauth().getAppId());
+ LOGGER.debug("appSecret: {}", this.config.getOauth().getAppSecret());
+ FacebookFactory ff = new FacebookFactory(cb.build());
+ return ff.getInstance();
+ }
+
+ /**
+ * Queries facebook and queues the resulting data
+ * @param id
+ * @throws Exception
+ */
+ protected abstract void getData(IdConfig id) throws Exception;
+
+
+ @Override
+ public void run() {
+ for( IdConfig id : this.config.getIds()) {
+ try {
+ getData(id);
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ } catch (Exception e) {
+ LOGGER.error("Caught Exception while trying to poll data for page : {}", id);
+ LOGGER.error("Exception while getting page feed data: {}", e);
+ }
+ }
+ this.isComplete.set(true);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e7b90335/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedDataCollector.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedDataCollector.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedDataCollector.java
new file mode 100644
index 0000000..184d273
--- /dev/null
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedDataCollector.java
@@ -0,0 +1,124 @@
+/*
+ * 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.facebook.provider.pagefeed;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import facebook4j.*;
+import facebook4j.json.DataObjectFactory;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.facebook.FacebookConfiguration;
+import org.apache.streams.facebook.IdConfig;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.BlockingQueue;
+
+/**
+ * Collects the page feed data from public Facebook pages
+ */
+public class FacebookPageFeedDataCollector extends FacebookDataCollector {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FacebookPageFeedDataCollector.class);
+ private static final int MAX_ATTEMPTS = 5;
+ private static final ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
+ private static final int LIMIT = 100;
+
+ public FacebookPageFeedDataCollector(BlockingQueue<StreamsDatum> queue, FacebookConfiguration configuration) {
+ super(configuration, queue);
+ }
+
+ @Override
+ protected void getData(IdConfig id) throws Exception {
+ boolean exit = false;
+
+ ResponseList<Post> facebookPosts = getPosts(id.getId());
+ LOGGER.debug("Post received : {}", facebookPosts.size());
+ backOff.reset();
+ do {
+ for(Post post : facebookPosts) {
+ if(id.getBeforeDate() != null && id.getAfterDate() != null) {
+ if(id.getBeforeDate().isAfter(post.getCreatedTime().getTime())
+ && id.getAfterDate().isBefore(post.getCreatedTime().getTime())) {
+ super.outputData(MAPPER.readValue(DataObjectFactory.getRawJSON(post), org.apache.streams.facebook.Post.class), post.getId());
+
+ }
+ } else if(id.getBeforeDate() != null && id.getBeforeDate().isAfter(post.getCreatedTime().getTime())) {
+ super.outputData(MAPPER.readValue(DataObjectFactory.getRawJSON(post), org.apache.streams.facebook.Post.class), post.getId());
+ } else if(id.getAfterDate() != null && id.getAfterDate().isBefore(post.getCreatedTime().getTime())) {
+ super.outputData(MAPPER.readValue(DataObjectFactory.getRawJSON(post), org.apache.streams.facebook.Post.class), post.getId());
+ } else if(id.getBeforeDate() == null && id.getAfterDate() == null) {
+ super.outputData(MAPPER.readValue(DataObjectFactory.getRawJSON(post), org.apache.streams.facebook.Post.class), post.getId());
+ } else {
+ exit = true;
+ LOGGER.debug("Breaking on post, {}, with createdAtDate {}", post.getId(), post.getCreatedTime());
+ break;
+ }
+ }
+ if(facebookPosts.getPaging() != null && !exit) {
+ LOGGER.debug("Paging. . .");
+ facebookPosts = getPosts(facebookPosts.getPaging());
+ backOff.reset();
+ LOGGER.debug("Paging received {} posts*", facebookPosts.size());
+ } else {
+ LOGGER.debug("No more paging.");
+ facebookPosts = null;
+ }
+ } while(facebookPosts != null && facebookPosts.size() != 0);
+
+
+ }
+
+ private ResponseList<Post> getPosts(Paging<Post> paging) throws Exception{
+ return getPosts(null, paging);
+ }
+
+ private ResponseList<Post> getPosts(String pageId) throws Exception {
+ return getPosts(pageId, null);
+ }
+
+ /**
+ * Queries facebook. Attempts requests up to 5 times and backs off on each facebook exception.
+ * @param pageId
+ * @param paging
+ * @return
+ * @throws Exception
+ */
+ private ResponseList<Post> getPosts(String pageId, Paging<Post> paging) throws Exception {
+ int attempt = 0;
+ while(attempt < MAX_ATTEMPTS) {
+ ++attempt;
+ try {
+ if (pageId != null) {
+ Reading reading = new Reading();
+ reading.limit(LIMIT);
+ return getNextFacebookClient().getPosts(pageId, reading);
+ }
+ else
+ return getNextFacebookClient().fetchNext(paging);
+ } catch (FacebookException fe) {
+ LOGGER.error("Facebook returned an exception : {}", fe);
+ LOGGER.error("Facebook returned an exception while trying to get feed for page, {} : {}", pageId, fe.getMessage());
+ // Rate limit exceptions with facebook4j unclear http://facebook4j.org/oldjavadocs/1.1.12-2.0.0/2.0.0/index.html?facebook4j/internal/http/HttpResponseCode.html
+ // back off at all exceptions until figured out.
+ super.backOff.backOff();
+ }
+ }
+ throw new Exception("Failed to get data from facebook after "+MAX_ATTEMPTS);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e7b90335/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedProvider.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedProvider.java
new file mode 100644
index 0000000..35507b9
--- /dev/null
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookPageFeedProvider.java
@@ -0,0 +1,28 @@
+/*
+ * 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.facebook.provider.pagefeed;
+
+/**
+ *
+ */
+public class FacebookPageFeedProvider extends FacebookProvider{
+ @Override
+ protected FacebookDataCollector getDataCollector() {
+ return new FacebookPageFeedDataCollector(super.datums, super.configuration);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e7b90335/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookProvider.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookProvider.java
new file mode 100644
index 0000000..47f798b
--- /dev/null
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/provider/pagefeed/FacebookProvider.java
@@ -0,0 +1,139 @@
+/*
+ * 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.facebook.provider.pagefeed;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.Queues;
+import com.google.common.collect.Sets;
+import com.sun.corba.se.spi.orb.DataCollector;
+import com.typesafe.config.ConfigRenderOptions;
+import org.apache.streams.config.StreamsConfigurator;
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.core.StreamsProvider;
+import org.apache.streams.core.StreamsResultSet;
+import org.apache.streams.facebook.FacebookConfiguration;
+import org.apache.streams.facebook.IdConfig;
+import org.apache.streams.jackson.StreamsJacksonMapper;
+import org.apache.streams.util.ComponentUtils;
+import org.apache.streams.util.SerializationUtil;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.math.BigInteger;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Abstract {@link org.apache.streams.core.StreamsProvider} for facebook.
+ */
+public abstract class FacebookProvider implements StreamsProvider {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(FacebookProvider.class);
+ private static final ObjectMapper MAPPER = StreamsJacksonMapper.getInstance();
+ private static final int MAX_BATCH_SIZE = 2000;
+
+ protected FacebookConfiguration configuration;
+ protected BlockingQueue<StreamsDatum> datums;
+
+ private AtomicBoolean isComplete;
+ private ExecutorService executor;
+ private FacebookDataCollector dataCollector;
+
+
+ public FacebookProvider() {
+ try {
+ this.configuration = MAPPER.readValue(StreamsConfigurator.config.getConfig("facebook").root().render(ConfigRenderOptions.concise()), FacebookConfiguration.class);
+ } catch (IOException ioe) {
+ LOGGER.error("Exception trying to read default config : {}", ioe);
+ }
+ }
+
+ public FacebookProvider(FacebookConfiguration configuration) {
+ this.configuration = (FacebookConfiguration) SerializationUtil.cloneBySerialization(configuration);
+ }
+
+
+ @Override
+ public void startStream() {
+ this.dataCollector = getDataCollector();
+ this.executor.submit(dataCollector);
+ }
+
+ protected abstract FacebookDataCollector getDataCollector();
+
+ @Override
+ public StreamsResultSet readCurrent() {
+ int batchSize = 0;
+ BlockingQueue<StreamsDatum> batch = Queues.newLinkedBlockingQueue();
+ while(!this.datums.isEmpty() && batchSize < MAX_BATCH_SIZE) {
+ ComponentUtils.offerUntilSuccess(ComponentUtils.pollWhileNotEmpty(this.datums), batch);
+ ++batchSize;
+ }
+ this.isComplete.set(batch.isEmpty() && this.datums.isEmpty() && this.dataCollector.isComplete());
+ return new StreamsResultSet(batch);
+ }
+
+ @Override
+ public StreamsResultSet readNew(BigInteger sequence) {
+ return null;
+ }
+
+ @Override
+ public StreamsResultSet readRange(DateTime start, DateTime end) {
+ return null;
+ }
+
+ @Override
+ public boolean isRunning() {
+ return !this.isComplete.get();
+ }
+
+ @Override
+ public void prepare(Object configurationObject) {
+ this.datums = Queues.newLinkedBlockingQueue();
+ this.isComplete = new AtomicBoolean(false);
+ this.executor = Executors.newFixedThreadPool(1);
+ }
+
+ @Override
+ public void cleanUp() {
+ ComponentUtils.shutdownExecutor(executor, 5, 5);
+ executor = null;
+ }
+
+ /**
+ * Overrides the ids and addedAfter time in the configuration
+ * @param idsToAfterDate
+ */
+ public void overrideIds(Map<String, DateTime> idsToAfterDate) {
+ Set<IdConfig> ids = Sets.newHashSet();
+ for(String id : idsToAfterDate.keySet()) {
+ IdConfig idConfig = new IdConfig();
+ idConfig.setId(id);
+ idConfig.setAfterDate(idsToAfterDate.get(id));
+ ids.add(idConfig);
+ }
+ this.configuration.setIds(ids);
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e7b90335/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/serializer/FacebookStreamsPostSerializer.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/serializer/FacebookStreamsPostSerializer.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/serializer/FacebookStreamsPostSerializer.java
new file mode 100644
index 0000000..3c5cae6
--- /dev/null
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/serializer/FacebookStreamsPostSerializer.java
@@ -0,0 +1,60 @@
+package org.apache.streams.facebook.serializer;
+
+import org.apache.streams.data.ActivitySerializer;
+import org.apache.streams.exceptions.ActivitySerializerException;
+import org.apache.streams.facebook.Post;
+import org.apache.streams.pojo.json.Activity;
+import org.apache.streams.pojo.json.Actor;
+import org.apache.streams.pojo.json.Provider;
+
+import java.util.List;
+
+/**
+ * Converts {@link org.apache.streams.facebook.Post} to {@link org.apache.streams.pojo.json.Activity}
+ */
+public class FacebookStreamsPostSerializer implements ActivitySerializer<Post> {
+
+ private static final String FACEBOOK_STREAMS_ID = "id:provider:facebook";
+ private static final String ID_PREFIX = "id:facebook:";
+ private static final String PROVIDER_DISPLAY = "Facebook";
+
+ @Override
+ public String serializationFormat() {
+ return null;
+ }
+
+ @Override
+ public Post serialize(Activity deserialized) throws ActivitySerializerException {
+ return null;
+ }
+
+ @Override
+ public Activity deserialize(Post post) throws ActivitySerializerException {
+ Activity activity = new Activity();
+ activity.setActor(createActor(post));
+
+ activity.setId(post.getId());
+ activity.setContent(post.getMessage());
+ return null;
+ }
+
+ @Override
+ public List<Activity> deserializeAll(List<Post> serializedList) {
+ return null;
+ }
+
+ public Actor createActor(Post post) {
+ Actor actor = new Actor();
+ actor.setDisplayName(post.getFrom().getName());
+ actor.setId(ID_PREFIX+post.getFrom().getId());
+ return actor;
+ }
+
+ public Provider createProvider(Post post) {
+ Provider provider = new Provider();
+ provider.setId(FACEBOOK_STREAMS_ID);
+ provider.setDisplayName(PROVIDER_DISPLAY);
+ return provider;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e7b90335/streams-contrib/streams-provider-facebook/src/main/jsonschema/org/apache/streams/facebook/FacebookConfiguration.json
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/jsonschema/org/apache/streams/facebook/FacebookConfiguration.json b/streams-contrib/streams-provider-facebook/src/main/jsonschema/org/apache/streams/facebook/FacebookConfiguration.json
index b4e5afb..3356623 100644
--- a/streams-contrib/streams-provider-facebook/src/main/jsonschema/org/apache/streams/facebook/FacebookConfiguration.json
+++ b/streams-contrib/streams-provider-facebook/src/main/jsonschema/org/apache/streams/facebook/FacebookConfiguration.json
@@ -44,6 +44,39 @@
"type": "string"
}
}
+ },
+ "userAccessTokens": {
+ "type": "array",
+ "uniqueItems": true,
+ "items": {
+ "type": "string",
+ "description": "User access tokens"
+ },
+ "description": "User access tokens that have been authorized for this app"
+ },
+ "ids": {
+ "type": "array",
+ "uniqueItems": true,
+ "items": {
+ "type": "object",
+ "description": "id for the application to use. user id, page id, etc. with date ranges",
+ "javaType": "org.apache.streams.facebook.IdConfig",
+ "javaInterfaces": ["java.io.Serializable"],
+ "properties": {
+ "id": {
+ "type": "string"
+ },
+ "beforeDate": {
+ "type": "string",
+ "format": "date-time"
+ },
+ "afterDate": {
+ "type": "string",
+ "format": "date-time"
+ }
+ }
+ },
+ "description": "ids for the application to use. user ids, page ids, etc."
}
}
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/e7b90335/streams-contrib/streams-provider-facebook/src/test/java/org/apache/streams/facebook/provider/pagefeed/TestFacebookProvider.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/test/java/org/apache/streams/facebook/provider/pagefeed/TestFacebookProvider.java b/streams-contrib/streams-provider-facebook/src/test/java/org/apache/streams/facebook/provider/pagefeed/TestFacebookProvider.java
new file mode 100644
index 0000000..5b4aaa2
--- /dev/null
+++ b/streams-contrib/streams-provider-facebook/src/test/java/org/apache/streams/facebook/provider/pagefeed/TestFacebookProvider.java
@@ -0,0 +1,94 @@
+/*
+ * 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.facebook.provider.pagefeed;
+
+import org.apache.streams.core.StreamsDatum;
+import org.apache.streams.facebook.FacebookConfiguration;
+import org.apache.streams.facebook.IdConfig;
+import org.junit.Test;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CyclicBarrier;
+
+import static org.junit.Assert.*;
+
+/**
+ * Unit Tests For {@link org.apache.streams.facebook.provider.pagefeed.FacebookProvider}
+ */
+public class TestFacebookProvider {
+
+ @Test
+ public void testFacebookProvider() throws Exception {
+ //Test that streams starts and shut downs.
+ final CyclicBarrier barrier = new CyclicBarrier(2);
+ FacebookProvider provider = new FacebookProvider(new FacebookConfiguration()) {
+ @Override
+ protected FacebookDataCollector getDataCollector() {
+ return new TestFacebookDataCollector(barrier, super.configuration, super.datums);
+ }
+ };
+ provider.prepare(null);
+ provider.startStream();
+ assertTrue(provider.isRunning());
+ barrier.await();
+ assertTrue(provider.isRunning());
+ assertEquals(5, provider.readCurrent().size());
+ barrier.await();
+ assertEquals(0, provider.readCurrent().size());
+ assertFalse(provider.isRunning());
+ provider.cleanUp();
+ }
+
+ private class TestFacebookDataCollector extends FacebookDataCollector {
+
+ private CyclicBarrier barrier;
+ private BlockingQueue<StreamsDatum> queue;
+
+ public TestFacebookDataCollector(CyclicBarrier barrier, FacebookConfiguration config, BlockingQueue<StreamsDatum> queue) {
+ super(config, queue);
+ this.barrier = barrier;
+ this.queue = queue;
+
+ }
+
+ @Override
+ protected void getData(IdConfig id) throws Exception {
+
+ }
+
+ @Override
+ public void run() {
+ try {
+ for(int i=0; i < 5; ++i) {
+ super.outputData(new Integer(i), ""+i);
+ }
+ this.barrier.await();
+ super.isComplete.set(true);
+ this.barrier.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ } catch (BrokenBarrierException bbe) {
+ fail();
+ }
+ }
+ }
+
+
+
+}
[6/6] git commit: Merge PR#76 'rbnks/STREAMS-159'
Posted by mf...@apache.org.
Merge PR#76 'rbnks/STREAMS-159'
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/d8a3380b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/d8a3380b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/d8a3380b
Branch: refs/heads/master
Commit: d8a3380bfd1b5bfcf54c9b29efbe3795da883527
Parents: fe4cec6 6d67fd6
Author: mfranklin <mf...@apache.org>
Authored: Thu Sep 4 13:35:58 2014 -0400
Committer: mfranklin <mf...@apache.org>
Committed: Thu Sep 4 13:35:58 2014 -0400
----------------------------------------------------------------------
.../streams/console/ConsolePersistWriter.java | 5 +-
.../metadata/facebook_post.mup | 694 +++++++++++++++++++
.../metadata/facebook_post.png | Bin 0 -> 1038039 bytes
.../api/FacebookPostActivitySerializer.java | 29 +-
.../processor/FacebookTypeConverter.java | 12 +-
.../pagefeed/FacebookDataCollector.java | 138 ++++
.../pagefeed/FacebookPageFeedDataCollector.java | 124 ++++
.../pagefeed/FacebookPageFeedProvider.java | 28 +
.../provider/pagefeed/FacebookProvider.java | 139 ++++
.../FacebookStreamsPostSerializer.java | 60 ++
.../streams/facebook/FacebookConfiguration.json | 33 +
.../provider/pagefeed/TestFacebookProvider.java | 94 +++
streams-pojo/metadata/Serializer.mup | 371 ++++++++++
streams-pojo/metadata/serializer.png | Bin 0 -> 604337 bytes
14 files changed, 1718 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
[3/6] git commit: Added statement to log exception
Posted by mf...@apache.org.
Added statement to log exception
Project: http://git-wip-us.apache.org/repos/asf/incubator-streams/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-streams/commit/a6286765
Tree: http://git-wip-us.apache.org/repos/asf/incubator-streams/tree/a6286765
Diff: http://git-wip-us.apache.org/repos/asf/incubator-streams/diff/a6286765
Branch: refs/heads/master
Commit: a6286765303267dca25e6302f99b673e78f12832
Parents: e7b9033
Author: Ryan Ebanks <re...@Informations-MacBook-Pro-3.local>
Authored: Wed Aug 27 13:09:27 2014 -0500
Committer: Ryan Ebanks <re...@Informations-MacBook-Pro-3.local>
Committed: Wed Aug 27 13:09:27 2014 -0500
----------------------------------------------------------------------
.../streams/facebook/processor/FacebookTypeConverter.java | 7 +++++--
1 file changed, 5 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-streams/blob/a6286765/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
----------------------------------------------------------------------
diff --git a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
index e160c77..d247e51 100644
--- a/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
+++ b/streams-contrib/streams-provider-facebook/src/main/java/org/apache/streams/facebook/processor/FacebookTypeConverter.java
@@ -181,8 +181,11 @@ public class FacebookTypeConverter implements StreamsProcessor {
if( out != null && validate(out, outClass))
result = new StreamsDatum(out);
}
- } catch (Exception e) {
- e.printStackTrace();
+ } catch (Exception e) {
+ LOGGER.error("Exception switching types : {}", e);
+ if(e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
}
if( result != null )