svn commit: r1566213 - in /incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider: TwitterEventProcessor.java TwitterStreamConfigurator.java

[sb...@apache.org]

Author: sblackmon
Date: Sun Feb  9 04:40:59 2014
New Revision: 1566213

URL: http://svn.apache.org/r1566213
Log:
implemented TwitterTimelineProvider

Modified:
    incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventProcessor.java
    incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamConfigurator.java

Modified: incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventProcessor.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventProcessor.java?rev=1566213&r1=1566212&r2=1566213&view=diff
==============================================================================
--- incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventProcessor.java (original)
+++ incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterEventProcessor.java Sun Feb  9 04:40:59 2014
@@ -16,7 +16,6 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.io.Serializable;
 import java.util.Queue;
 import java.util.Random;
 import java.util.concurrent.BlockingQueue;
@@ -42,15 +41,15 @@ public class TwitterEventProcessor imple
 
     public final static String TERMINATE = new String("TERMINATE");
 
-    public TwitterEventProcessor(BlockingQueue<String> inQueue, Queue<StreamsDatum> outQueue, Class inClass, Class outClass) {
-        this.inQueue = inQueue;
+    public TwitterEventProcessor(Queue<String> inQueue, Queue<StreamsDatum> outQueue, Class inClass, Class outClass) {
+        this.inQueue = (BlockingQueue<String>)inQueue;
         this.outQueue = outQueue;
         this.inClass = inClass;
         this.outClass = outClass;
     }
 
-    public TwitterEventProcessor(BlockingQueue<String> inQueue, Queue<StreamsDatum> outQueue, Class outClass) {
-        this.inQueue = inQueue;
+    public TwitterEventProcessor(Queue<String> inQueue, Queue<StreamsDatum> outQueue, Class outClass) {
+        this.inQueue = (BlockingQueue<String>)inQueue;
         this.outQueue = outQueue;
         this.outClass = outClass;
     }

Modified: incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamConfigurator.java
URL: http://svn.apache.org/viewvc/incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamConfigurator.java?rev=1566213&r1=1566212&r2=1566213&view=diff
==============================================================================
--- incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamConfigurator.java (original)
+++ incubator/streams/branches/sblackmon/streams-contrib/streams-provider-twitter/src/main/java/org/apache/streams/twitter/provider/TwitterStreamConfigurator.java Sun Feb  9 04:40:59 2014
@@ -1,5 +1,6 @@
 package org.apache.streams.twitter.provider;
 
+import com.google.common.collect.Lists;
 import com.typesafe.config.Config;
 import com.typesafe.config.ConfigException;
 import org.apache.streams.config.StreamsConfigurator;
@@ -8,6 +9,8 @@ import org.apache.streams.twitter.Twitte
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.List;
+
 /**
  * Created by sblackmon on 12/10/13.
  */
@@ -19,6 +22,10 @@ public class TwitterStreamConfigurator {
         Config oauth = StreamsConfigurator.config.getConfig("twitter.oauth");
 
         TwitterStreamConfiguration twitterStreamConfiguration = new TwitterStreamConfiguration();
+        twitterStreamConfiguration.setProtocol(twitter.getString("protocol"));
+        twitterStreamConfiguration.setHost(twitter.getString("host"));
+        twitterStreamConfiguration.setPort(twitter.getLong("port"));
+        twitterStreamConfiguration.setVersion(twitter.getString("version"));
         TwitterOAuthConfiguration twitterOAuthConfiguration = new TwitterOAuthConfiguration();
         twitterOAuthConfiguration.setConsumerKey(oauth.getString("consumerKey"));
         twitterOAuthConfiguration.setConsumerSecret(oauth.getString("consumerSecret"));
@@ -30,7 +37,10 @@ public class TwitterStreamConfigurator {
             twitterStreamConfiguration.setTrack(twitter.getStringList("track"));
         } catch( ConfigException ce ) {}
         try {
-            twitterStreamConfiguration.setFollow(twitter.getLongList("follow"));
+            List<Long> follows = Lists.newArrayList();
+            for( Integer id : twitter.getIntList("follow"))
+                follows.add(new Long(id));
+            twitterStreamConfiguration.setFollow(follows);
         } catch( ConfigException ce ) {}
 
         twitterStreamConfiguration.setFilterLevel(twitter.getString("filter-level"));