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