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         if (common_config('twitter', 'enabled')) {
  86             $qm = QueueManager::get();
  87             $qm->setActiveGroup('twitter');
  88             $this->instantiate($qm);
  89             $this->instantiate(TwitterManager::get());
  90             $this->instantiate($this->processManager);
  91         }
  92     }
  93 }
  94
  95
  96 class TwitterManager extends IoManager
  97 {
  98     // Recommended resource limits from http://dev.twitter.com/pages/site_streams
  99     const MAX_STREAMS = 1000;
 100     const USERS_PER_STREAM = 100;
 101     const STREAMS_PER_SECOND = 20;
 102
 103     protected $twitterStreams;
 104     protected $twitterUsers;
 105
 106     function __construct()
 107     {
 108     }
 109
 110     /**
 111      * Pull the site's active Twitter-importing users and start spawning
 112      * some data streams for them!
 113      *
 114      * @fixme check their last-id and check whether we'll need to do a manual pull.
 115      * @fixme abstract out the fetching so we can work over multiple sites.
 116      */
 117     protected function initStreams()
 118     {
 119         // Pull Twitter user IDs for all users we want to pull data for
 120         $flink = new Foreign_link();
 121         $flink->service = TWITTER_SERVICE;
 122         // @fixme probably should do the bitfield check in a whereAdd but it's ugly :D
 123         $flink->find();
 124
 125         $userIds = array();
 126         while ($flink->fetch()) {
 127             if (($flink->noticesync & FOREIGN_NOTICE_RECV) ==
 128                 FOREIGN_NOTICE_RECV) {
 129                 $userIds[] = $flink->foreign_id;
 130
 131                 if (count($userIds) >= self::USERS_PER_STREAM) {
 132                     $this->spawnStream($userIds);
 133                     $userIds = array();
 134                 }
 135             }
 136         }
 137
 138         if (count($userIds)) {
 139             $this->spawnStream($userIds);
 140         }
 141     }
 142
 143     /**
 144      * Prepare a Site Stream connection for the given chunk of users.
 145      * The actual connection will be opened later.
 146      *
 147      * @param $users array of Twitter-side user IDs
 148      */
 149     protected function spawnStream($users)
 150     {
 151         $stream = $this->initSiteStream();
 152         $stream->followUsers($userIds);
 153
 154         // Slip the stream reader into our list of active streams.
 155         // We'll manage its actual connection on the next go-around.
 156         $this->streams[] = $stream;
 157
 158         // Record the user->stream mappings; this makes it easier for us to know
 159         // later if we need to kill something.
 160         foreach ($userIds as $id) {
 161             $this->users[$id] = $stream;
 162         }
 163     }
 164
 165     /**
 166      * Initialize a generic site streams connection object.
 167      * All our connections will look like this, then we'll add users to them.
 168      *
 169      * @return TwitterStreamReader
 170      */
 171     protected function initSiteStream()
 172     {
 173         $auth = $this->siteStreamAuth();
 174         $stream = new TwitterSiteStream($auth);
 175
 176         // Add our event handler callbacks. Whee!
 177         $this->setupEvents($stream);
 178         return $stream;
 179     }
 180
 181     /**
 182      * Fetch the Twitter OAuth credentials to use to connect to the Site Streams API.
 183      *
 184      * This will use the locally-stored credentials for the applictation's owner account
 185      * from the site configuration. These should be configured through the administration
 186      * panels or manually in the config file.
 187      *
 188      * Will throw an exception if no credentials can be found -- but beware that invalid
 189      * credentials won't cause breakage until later.
 190      *
 191      * @return TwitterOAuthClient
 192      */
 193     protected function siteStreamAuth()
 194     {
 195         $token = common_config('twitter', 'stream_token');
 196         $secret = common_config('twitter', 'stream_secret');
 197         if (empty($token) || empty($secret)) {
 198             throw new ServerException('Twitter site streams have not been correctly configured. Configure the app owner account via the admin panel.');
 199         }
 200         return new TwitterOAuthClient($token, $secret);
 201     }
 202
 203     /**
 204      * Collect the sockets for all active connections for i/o monitoring.
 205      *
 206      * @return array of resources
 207      */
 208     public function getSockets()
 209     {
 210         $sockets = array();
 211         foreach ($this->streams as $stream) {
 212             foreach ($stream->getSockets() as $socket) {
 213                 $sockets[] = $socket;
 214             }
 215         }
 216         return $streams;
 217     }
 218
 219     /**
 220      * We're ready to process input from one of our data sources! Woooooo!
 221      * @fixme is there an easier way to map from socket back to owning module? :(
 222      *
 223      * @param resource $socket
 224      * @return boolean success
 225      */
 226     public function handleInput($socket)
 227     {
 228         foreach ($this->streams as $stream) {
 229             foreach ($stream->getSockets() as $aSocket) {
 230                 if ($socket === $aSocket) {
 231                     $stream->handleInput($socket);
 232                 }
 233             }
 234         }
 235         return true;
 236     }
 237
 238     /**
 239      * Start the i/o system up! Prepare our connections and start opening them.
 240      *
 241      * @fixme do some rate-limiting on the stream setup
 242      * @fixme do some sensible backoff on failure etc
 243      */
 244     public function start()
 245     {
 246         $this->initStreams();
 247         foreach ($this->streams as $stream) {
 248             $stream->connect();
 249         }
 250         return true;
 251     }
 252
 253     /**
 254      * Close down our connections when the daemon wraps up for business.
 255      */
 256     public function finish()
 257     {
 258         foreach ($this->streams as $index => $stream) {
 259             $stream->close();
 260             unset($this->streams[$index]);
 261         }
 262         return true;
 263     }
 264
 265     public static function get()
 266     {
 267         throw new Exception('not a singleton');
 268     }
 269
 270     /**
 271      * Set up event handlers on the streaming interface.
 272      *
 273      * @fixme add more event types as we add handling for them
 274      */
 275     protected function setupEvents(TwitterStream $stream)
 276     {
 277         $handlers = array(
 278             'status',
 279         );
 280         foreach ($handlers as $event) {
 281             $stream->hookEvent($event, array($this, 'onTwitter' . ucfirst($event)));
 282         }
 283     }
 284
 285     /**
 286      * Event callback notifying that a user has a new message in their home timeline.
 287      * We store the incoming message into the queues for processing, keeping our own
 288      * daemon running as shiny-fast as possible.
 289      *
 290      * @param object $status JSON data: Twitter status update
 291      * @fixme in all-sites mode we may need to route queue items into another site's
 292      *        destination queues, or multiple sites.
 293      */
 294     protected function onTwitterStatus($status, $context)
 295     {
 296         $data = array(
 297             'status' => $status,
 298             'for_user' => $context->for_user,
 299         );
 300         $qm = QueueManager::get();
 301         $qm->enqueue($data, 'tweetin');
 302     }
 303 }
 304
 305
 306 if (have_option('i', 'id')) {
 307     $id = get_option_value('i', 'id');
 308 } else if (count($args) > 0) {
 309     $id = $args[0];
 310 } else {
 311     $id = null;
 312 }
 313
 314 $foreground = have_option('f', 'foreground');
 315 $all = have_option('a') || have_option('--all');
 316
 317 $daemon = new TwitterDaemon($id, !$foreground, 1, $all);
 318
 319 $daemon->runOnce();