Twitter streaming API reader: Cleanup input handling & split from HTTP headers to body

2 files changed, 49 insertions, 19 deletions

diff --git a/plugins/TwitterBridge/jsonstreamreader.php b/plugins/TwitterBridge/jsonstreamreader.php
index ad8e2626a..898766357 100644
--- a/plugins/TwitterBridge/jsonstreamreader.php
+++ b/plugins/TwitterBridge/jsonstreamreader.php
@@ -93,18 +93,24 @@ abstract class JsonStreamReader
         $this->state = 'waiting';
     }
 
+    /**
+     * Send some fun data off to the server.
+     *
+     * @param string $buffer
+     */
     function send($buffer)
     {
-        echo "Writing...\n";
-        var_dump($buffer);
         fwrite($this->socket, $buffer);
     }
 
+    /**
+     * Read next packet of data from the socket.
+     *
+     * @return string
+     */
     function read()
     {
-        echo "Reading...\n";
         $buffer = fread($this->socket, 65536);
-        var_dump($buffer);
         return $buffer;
     }
 
@@ -195,12 +201,16 @@ abstract class JsonStreamReader
     {
         $lines = explode(self::CRLF, $buffer);
         foreach ($lines as $line) {
-            if ($line == '') {
-                $this->state = 'active';
-                common_log(LOG_DEBUG, "$this->id connection is active!");
-            } else {
-                common_log(LOG_DEBUG, "$this->id read HTTP header: $line");
-                $this->responseHeaders[] = $line;
+            if ($this->state == 'headers') {
+                if ($line == '') {
+                    $this->state = 'active';
+                    common_log(LOG_DEBUG, "$this->id connection is active!");
+                } else {
+                    common_log(LOG_DEBUG, "$this->id read HTTP header: $line");
+                    $this->responseHeaders[] = $line;
+                }
+            } else if ($this->state == 'active') {
+                $this->handleLineActive($line);
             }
         }
     }
@@ -211,12 +221,21 @@ abstract class JsonStreamReader
         // Will we always deliver on packet boundaries?
         $lines = explode("\n", $buffer);
         foreach ($lines as $line) {
-            $data = json_decode($line, true);
-            if ($data) {
-                $this->handleJson($data);
-            } else {
-                common_log(LOG_ERR, "$this->id received bogus JSON data: " . $line);
-            }
+            $this->handleLineActive($line);
+        }
+    }
+
+    function handleLineActive($line)
+    {
+        if ($line == '') {
+            // Server sends empty lines as keepalive.
+            return;
+        }
+        $data = json_decode($line, true);
+        if ($data) {
+            $this->handleJson($data);
+        } else {
+            common_log(LOG_ERR, "$this->id received bogus JSON data: " . $line);
         }
     }
 
diff --git a/plugins/TwitterBridge/twitterstreamreader.php b/plugins/TwitterBridge/twitterstreamreader.php
index e746228a3..d440bdd4e 100644
--- a/plugins/TwitterBridge/twitterstreamreader.php
+++ b/plugins/TwitterBridge/twitterstreamreader.php
@@ -94,11 +94,22 @@ abstract class TwitterStreamReader extends JsonStreamReader
 
     abstract function routeMessage($data);
 
-    function handleMessage($data, $forUserId=null)
+    /**
+     * Send the decoded JSON object out to any event listeners.
+     *
+     * @param array $data
+     * @param int $forUserId
+     */
+    function handleMessage(array $data, $forUserId=null)
     {
         $this->fireEvent('raw', $data, $forUserId);
-        $known = array('friends');
-        foreach ($known as $key) {
+
+        if (isset($data['id']) && isset($data['text']) && isset($data['user'])) {
+            $this->fireEvent('status', $data);
+        }
+
+        $knownMeta = array('friends', 'delete', 'scrubgeo', 'limit', 'event', 'direct_message');
+        foreach ($knownMeta as $key) {
             if (isset($data[$key])) {
                 $this->fireEvent($key, $data[$key], $forUserId);
             }