plugins/TwitterBridge/daemons/twitterdaemon.php

   1 #!/usr/bin/env php
   2 <?php
   3 /*
   4  * StatusNet - the distributed open-source microblogging tool
   5  * Copyright (C) 2008-2010, StatusNet, Inc.
   6  *
   7  * This program is free software: you can redistribute it and/or modify
   8  * it under the terms of the GNU Affero General Public License as published by
   9  * the Free Software Foundation, either version 3 of the License, or
  10  * (at your option) any later version.
  11  *
  12  * This program is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15  * GNU Affero General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Affero General Public License
  18  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  19  */
  20
  21 define('INSTALLDIR', realpath(dirname(__FILE__) . '/../../..'));
  22
  23 $shortoptions = 'fi::a';
  24 $longoptions = array('id::', 'foreground', 'all');
  25
  26 $helptext = <<<END_OF_XMPP_HELP
  27 Daemon script for receiving new notices from Twitter users.
  28
  29     -i --id           Identity (default none)
  30     -a --all          Handle Twitter for all local sites
  31                       (requires Stomp queue handler, status_network setup)
  32     -f --foreground   Stay in the foreground (default background)
  33
  34 END_OF_XMPP_HELP;
  35
  36 require_once INSTALLDIR.'/scripts/commandline.inc';
  37
  38 require_once INSTALLDIR . '/lib/jabber.php';
  39
  40 class TwitterDaemon extends SpawningDaemon
  41 {
  42     protected $allsites = false;
  43
  44     function __construct($id=null, $daemonize=true, $threads=1, $allsites=false)
  45     {
  46         if ($threads != 1) {
  47             // This should never happen. :)
  48             throw new Exception("TwitterDaemon must run single-threaded");
  49         }
  50         parent::__construct($id, $daemonize, $threads);
  51         $this->allsites = $allsites;
  52     }
  53
  54     function runThread()
  55     {
  56         common_log(LOG_INFO, 'Waiting to listen to Twitter and queues');
  57
  58         $master = new TwitterMaster($this->get_id(), $this->processManager());
  59         $master->init($this->allsites);
  60         $master->service();
  61
  62         common_log(LOG_INFO, 'terminating normally');
  63
  64         return $master->respawn ? self::EXIT_RESTART : self::EXIT_SHUTDOWN;
  65     }
  66
  67 }
  68
  69 class TwitterMaster extends IoMaster
  70 {
  71     protected $processManager;
  72
  73     function __construct($id, $processManager)
  74     {
  75         parent::__construct($id);
  76         $this->processManager = $processManager;
  77     }
  78
  79     /**
  80      * Initialize IoManagers for the currently configured site
  81      * which are appropriate to this instance.
  82      */
  83     function initManagers()
  84     {
  85         $qm = QueueManager::get();
  86         $qm->setActiveGroup('twitter');
  87         $this->instantiate($qm);
  88         $this->instantiate(new TwitterManager());
  89         $this->instantiate($this->processManager);
  90     }
  91 }
  92
  93
  94 class TwitterManager extends IoManager
  95 {
  96     // Recommended resource limits from http://dev.twitter.com/pages/site_streams
  97     const MAX_STREAMS = 1000;
  98     const USERS_PER_STREAM = 100;
  99     const STREAMS_PER_SECOND = 20;
 100
 101     protected $streams;
 102     protected $users;
 103
 104     /**
 105      * Pull the site's active Twitter-importing users and start spawning
 106      * some data streams for them!
 107      *
 108      * @fixme check their last-id and check whether we'll need to do a manual pull.
 109      * @fixme abstract out the fetching so we can work over multiple sites.
 110      */
 111     protected function initStreams()
 112     {
 113         common_log(LOG_INFO, 'init...');
 114         // Pull Twitter user IDs for all users we want to pull data for
 115         $flink = new Foreign_link();
 116         $flink->service = TWITTER_SERVICE;
 117         // @fixme probably should do the bitfield check in a whereAdd but it's ugly :D
 118         $flink->find();
 119
 120         $userIds = array();
 121         while ($flink->fetch()) {
 122             if (($flink->noticesync & FOREIGN_NOTICE_RECV) ==
 123                 FOREIGN_NOTICE_RECV) {
 124                 $userIds[] = $flink->foreign_id;
 125
 126                 if (count($userIds) >= self::USERS_PER_STREAM) {
 127                     $this->spawnStream($userIds);
 128                     $userIds = array();
 129                 }
 130             }
 131         }
 132
 133         if (count($userIds)) {
 134             $this->spawnStream($userIds);
 135         }
 136     }
 137
 138     /**
 139      * Prepare a Site Stream connection for the given chunk of users.
 140      * The actual connection will be opened later.
 141      *
 142      * @param $userIds array of Twitter-side user IDs
 143      */
 144     protected function spawnStream($userIds)
 145     {
 146         $stream = $this->initSiteStream();
 147         $stream->followUsers($userIds);
 148
 149         // Slip the stream reader into our list of active streams.
 150         // We'll manage its actual connection on the next go-around.
 151         $this->streams[] = $stream;
 152
 153         // Record the user->stream mappings; this makes it easier for us to know
 154         // later if we need to kill something.
 155         foreach ($userIds as $id) {
 156             $this->users[$id] = $stream;
 157         }
 158     }
 159
 160     /**
 161      * Initialize a generic site streams connection object.
 162      * All our connections will look like this, then we'll add users to them.
 163      *
 164      * @return TwitterStreamReader
 165      */
 166     protected function initSiteStream()
 167     {
 168         $auth = $this->siteStreamAuth();
 169         $stream = new TwitterSiteStream($auth);
 170
 171         // Add our event handler callbacks. Whee!
 172         $this->setupEvents($stream);
 173         return $stream;
 174     }
 175
 176     /**
 177      * Fetch the Twitter OAuth credentials to use to connect to the Site Streams API.
 178      *
 179      * This will use the locally-stored credentials for the applictation's owner account
 180      * from the site configuration. These should be configured through the administration
 181      * panels or manually in the config file.
 182      *
 183      * Will throw an exception if no credentials can be found -- but beware that invalid
 184      * credentials won't cause breakage until later.
 185      *
 186      * @return TwitterOAuthClient
 187      */
 188     protected function siteStreamAuth()
 189     {
 190         $token = common_config('twitter', 'stream_token');
 191         $secret = common_config('twitter', 'stream_secret');
 192         if (empty($token) || empty($secret)) {
 193             throw new ServerException('Twitter site streams have not been correctly configured. Configure the app owner account via the admin panel.');
 194         }
 195         return new TwitterOAuthClient($token, $secret);
 196     }
 197
 198     /**
 199      * Collect the sockets for all active connections for i/o monitoring.
 200      *
 201      * @return array of resources
 202      */
 203     public function getSockets()
 204     {
 205         $sockets = array();
 206         foreach ($this->streams as $stream) {
 207             foreach ($stream->getSockets() as $socket) {
 208                 $sockets[] = $socket;
 209             }
 210         }
 211         return $sockets;
 212     }
 213
 214     /**
 215      * We're ready to process input from one of our data sources! Woooooo!
 216      * @fixme is there an easier way to map from socket back to owning module? :(
 217      *
 218      * @param resource $socket
 219      * @return boolean success
 220      */
 221     public function handleInput($socket)
 222     {
 223         foreach ($this->streams as $stream) {
 224             foreach ($stream->getSockets() as $aSocket) {
 225                 if ($socket === $aSocket) {
 226                     $stream->handleInput($socket);
 227                 }
 228             }
 229         }
 230         return true;
 231     }
 232
 233     /**
 234      * Start the i/o system up! Prepare our connections and start opening them.
 235      *
 236      * @fixme do some rate-limiting on the stream setup
 237      * @fixme do some sensible backoff on failure etc
 238      */
 239     public function start()
 240     {
 241         $this->initStreams();
 242         foreach ($this->streams as $stream) {
 243             $stream->connect();
 244         }
 245         return true;
 246     }
 247
 248     /**
 249      * Close down our connections when the daemon wraps up for business.
 250      */
 251     public function finish()
 252     {
 253         foreach ($this->streams as $index => $stream) {
 254             $stream->close();
 255             unset($this->streams[$index]);
 256         }
 257         return true;
 258     }
 259
 260     public static function get()
 261     {
 262         throw new Exception('not a singleton');
 263     }
 264
 265     /**
 266      * Set up event handlers on the streaming interface.
 267      *
 268      * @fixme add more event types as we add handling for them
 269      */
 270     protected function setupEvents(TwitterStreamReader $stream)
 271     {
 272         $handlers = array(
 273             'status',
 274         );
 275         foreach ($handlers as $event) {
 276             $stream->hookEvent($event, array($this, 'onTwitter' . ucfirst($event)));
 277         }
 278     }
 279
 280     /**
 281      * Event callback notifying that a user has a new message in their home timeline.
 282      * We store the incoming message into the queues for processing, keeping our own
 283      * daemon running as shiny-fast as possible.
 284      *
 285      * @param object $status JSON data: Twitter status update
 286      * @fixme in all-sites mode we may need to route queue items into another site's
 287      *        destination queues, or multiple sites.
 288      */
 289     protected function onTwitterStatus($status, $context)
 290     {
 291         $data = array(
 292             'status' => $status,
 293             'for_user' => $context->for_user,
 294         );
 295         $qm = QueueManager::get();
 296         $qm->enqueue($data, 'tweetin');
 297     }
 298 }
 299
 300
 301 if (have_option('i', 'id')) {
 302     $id = get_option_value('i', 'id');
 303 } else if (count($args) > 0) {
 304     $id = $args[0];
 305 } else {
 306     $id = null;
 307 }
 308
 309 $foreground = have_option('f', 'foreground');
 310 $all = have_option('a') || have_option('--all');
 311
 312 $daemon = new TwitterDaemon($id, !$foreground, 1, $all);
 313
 314 $daemon->runOnce();