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     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     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     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     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     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     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 system up!
 240      * @fixme do some rate-limiting on the stream setup
 241      * @fixme do some sensible backoff on failure etc
 242      */
 243     function start()
 244     {
 245         $this->initStreams();
 246         foreach ($this->streams as $stream) {
 247             $stream->connect();
 248         }
 249         return true;
 250     }
 251
 252     function finish()
 253     {
 254         foreach ($this->streams as $index => $stream) {
 255             $stream->close();
 256             unset($this->streams[$index]);
 257         }
 258         return true;
 259     }
 260
 261     public static function get()
 262     {
 263         throw new Exception('not a singleton');
 264     }
 265
 266     /**
 267      * Set up event handlers on the streaming interface.
 268      *
 269      * @fixme add more event types as we add handling for them
 270      */
 271     protected function setupEvents(TwitterStream $stream)
 272     {
 273         $handlers = array(
 274             'status',
 275         );
 276         foreach ($handlers as $event) {
 277             $stream->hookEvent($event, array($this, 'onTwitter' . ucfirst($event)));
 278         }
 279     }
 280
 281     /**
 282      * Event callback notifying that a user has a new message in their home timeline.
 283      *
 284      * @param object $data JSON data: Twitter status update
 285      */
 286     protected function onTwitterStatus($data, $context)
 287     {
 288         $importer = new TwitterImport();
 289         $notice = $importer->importStatus($data);
 290         if ($notice) {
 291             $user = $this->getTwitterUser($context);
 292             Inbox::insertNotice($user->id, $notice->id);
 293         }
 294     }
 295
 296     /**
 297      * @fixme what about handling multiple sites?
 298      */
 299     function getTwitterUser($context)
 300     {
 301         if ($context->source != 'sitestream') {
 302             throw new ServerException("Unexpected stream source");
 303         }
 304         $flink = Foreign_link::getByForeignID(TWITTER_SERVICE, $context->for_user);
 305         if ($flink) {
 306             return $flink->getUser();
 307         } else {
 308             throw new ServerException("No local user for this Twitter ID");
 309         }
 310     }
 311 }
 312
 313
 314 if (have_option('i', 'id')) {
 315     $id = get_option_value('i', 'id');
 316 } else if (count($args) > 0) {
 317     $id = $args[0];
 318 } else {
 319     $id = null;
 320 }
 321
 322 $foreground = have_option('f', 'foreground');
 323 $all = have_option('a') || have_option('--all');
 324
 325 $daemon = new TwitterDaemon($id, !$foreground, 1, $all);
 326
 327 $daemon->runOnce();