code:
<?php
/**
 * StatusNet, the distributed open-source microblogging tool
 *
 * Settings for Twitter integration
 *
 * PHP version 5
 *
 * LICENCE: This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * @category  Settings
 * @package   StatusNet
 * @author    Evan Prodromou <evan@status.net>
 * @copyright 2008-2009 StatusNet, Inc.
 * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
 * @link      http://status.net/
 */

if (!defined('STATUSNET') && !defined('LACONICA')) {
    exit(1);
}

require_once dirname(__DIR__) . '/twitter.php';

/**
 * Settings for Twitter integration
 *
 * @category Settings
 * @package  StatusNet
 * @author   Evan Prodromou <evan@status.net>
 * @license  http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
 * @link     http://status.net/
 *
 * @see      SettingsAction
 */
class TwittersettingsAction extends ProfileSettingsAction
{
    /**
     * Title of the page
     *
     * @return string Title of the page
     */

    function title()
    {
        // TRANS: Title for page with Twitter integration settings.
        return _m('Twitter settings');
    }

    /**
     * Instructions for use
     *
     * @return instructions for use
     */

    function getInstructions()
    {
        // TRANS: Instructions for page with Twitter integration settings.
        return _m('Connect your Twitter account to share your updates ' .
                  'with your Twitter friends and vice-versa.');
    }

    /**
     * Content area of the page
     *
     * Shows a form for associating a Twitter account with this
     * StatusNet account. Also lets the user set preferences.
     *
     * @return void
     */
    function showContent()
    {

        $user = common_current_user();

        $profile = $user->getProfile();

        $fuser = null;

        $flink = Foreign_link::getByUserID($user->id, TWITTER_SERVICE);

        if (!empty($flink)) {
            $fuser = $flink->getForeignUser();
        }

        $this->elementStart('form', array('method' => 'post',
                                          'id' => 'form_settings_twitter',
                                          'class' => 'form_settings',
                                          'action' =>
                                          common_local_url('twittersettings')));

        $this->hidden('token', common_session_token());

        $this->elementStart('fieldset', array('id' => 'settings_twitter_account'));

        if (empty($fuser)) {
            $this->elementStart('ul', 'form_data');
            $this->elementStart('li', array('id' => 'settings_twitter_login_button'));
            $this->element('a', array('href' => common_local_url('twitterauthorization')),
                           // TRANS: Link description to connect to a Twitter account.
                           'Connect my Twitter account');
            $this->elementEnd('li');
            $this->elementEnd('ul');

            $this->elementEnd('fieldset');
        } else {
            // TRANS: Fieldset legend.
            $this->element('legend', null, _m('Twitter account'));
            $this->elementStart('p', array('id' => 'form_confirmed'));
            $this->element('a', array('href' => $fuser->uri), $fuser->nickname);
            $this->elementEnd('p');
            $this->element('p', 'form_note',
                           // TRANS: Form note when a Twitter account has been connected.
                           _m('Connected Twitter account'));
            $this->elementEnd('fieldset');

            $this->elementStart('fieldset');

            // TRANS: Fieldset legend.
            $this->element('legend', null, _m('Disconnect my account from Twitter'));

            if (!$user->password) {
                $this->elementStart('p', array('class' => 'form_guide'));
                // TRANS: Form guide. %s is a URL to the password settings.
                // TRANS: This message contains a Markdown link in the form [description](link).
                $message = sprintf(_m('Disconnecting your Twitter account ' .
                                      'could make it impossible to log in! Please ' .
                                      '[set a password](%s) first.'),
                                   common_local_url('passwordsettings'));
                $message = common_markup_to_html($message);
                $this->text($message);
                $this->elementEnd('p');
            } else {
                // TRANS: Form instructions. %1$s is the StatusNet sitename.
                $note = _m('Keep your %1$s account but disconnect from Twitter. ' .
                    'You can use your %1$s password to log in.');
                $site = common_config('site', 'name');

                $this->element('p', 'instructions',
                    sprintf($note, $site));

                // TRANS: Button text for disconnecting a Twitter account.
                $this->submit('disconnect', _m('BUTTON','Disconnect'));
            }

            $this->elementEnd('fieldset');

            $this->elementStart('fieldset', array('id' => 'settings_twitter_preferences'));

            // TRANS: Fieldset legend.
            $this->element('legend', null, _m('Preferences'));
            $this->elementStart('ul', 'form_data');
            $this->elementStart('li');
            $this->checkbox('noticesend',
                            // TRANS: Checkbox label.
                            _m('Automatically send my notices to Twitter.'),
                            ($flink) ?
                            ($flink->noticesync & FOREIGN_NOTICE_SEND) :
                            true);
            $this->elementEnd('li');
            $this->elementStart('li');
            $this->checkbox('replysync',
                            // TRANS: Checkbox label.
                            _m('Send local "@" replies to Twitter.'),
                            ($flink) ?
                            ($flink->noticesync & FOREIGN_NOTICE_SEND_REPLY) :
                            true);
            $this->elementEnd('li');
            $this->elementStart('li');
            $this->checkbox('friendsync',
                            // TRANS: Checkbox label.
                            _m('Subscribe to my Twitter friends here.'),
                            ($flink) ?
                            ($flink->friendsync & FOREIGN_FRIEND_RECV) :
                            false);
            $this->elementEnd('li');

            if (common_config('twitterimport','enabled')) {
                $this->elementStart('li');
                $this->checkbox('noticerecv',
                                // TRANS: Checkbox label.
                                _m('Import my friends timeline.'),
                                ($flink) ?
                                ($flink->noticesync & FOREIGN_NOTICE_RECV) :
                                false);
                $this->elementEnd('li');
            } else {
                // preserve setting even if bidrection bridge toggled off

                if ($flink && ($flink->noticesync & FOREIGN_NOTICE_RECV)) {
                    $this->hidden('noticerecv', true, 'noticerecv');
                }
            }

            $this->elementEnd('ul');

            if ($flink) {
                // TRANS: Button text for saving Twitter integration settings.
                $this->submit('save', _m('BUTTON','Save'));
            } else {
                // TRANS: Button text for adding Twitter integration.
                $this->submit('add', _m('BUTTON','Add'));
            }

            $this->elementEnd('fieldset');
        }

        $this->elementEnd('form');
    }

    /**
     * Handle posts to this form
     *
     * Based on the button that was pressed, muxes out to other functions
     * to do the actual task requested.
     *
     * All sub-functions reload the form with a message -- success or failure.
     *
     * @return void
     */
    function handlePost()
    {
        // CSRF protection
        $token = $this->trimmed('token');
        if (!$token || $token != common_session_token()) {
            // TRANS: Client error displayed when the session token does not match or is not given.
            $this->showForm(_m('There was a problem with your session token. '.
                               'Try again, please.'));
            return;
        }

        if ($this->arg('save')) {
            $this->savePreferences();
        } else if ($this->arg('disconnect')) {
            $this->removeTwitterAccount();
        } else {
            // TRANS: Client error displayed when the submitted form contains unexpected data.
            $this->showForm(_m('Unexpected form submission.'));
        }
    }

    /**
     * Disassociate an existing Twitter account from this account
     *
     * @return void
     */
    function removeTwitterAccount()
    {
        $user = common_current_user();
        $flink = Foreign_link::getByUserID($user->id, TWITTER_SERVICE);

        if (empty($flink)) {
            // TRANS: Client error displayed when trying to remove a connected Twitter account when there isn't one connected.
            $this->clientError(_m('No Twitter connection to remove.'));
        }

        $result = $flink->safeDelete();

        if (empty($result)) {
            common_log_db_error($flink, 'DELETE', __FILE__);
            // TRANS: Server error displayed when trying to remove a connected Twitter account fails.
            $this->serverError(_m('Could not remove Twitter user.'));
        }

        // TRANS: Success message displayed after disconnecting a Twitter account.
        $this->showForm(_m('Twitter account disconnected.'), true);
    }

    /**
     * Save user's Twitter-bridging preferences
     *
     * @return void
     */
    function savePreferences()
    {
        $noticesend = $this->boolean('noticesend');
        $noticerecv = $this->boolean('noticerecv');
        $friendsync = $this->boolean('friendsync');
        $replysync  = $this->boolean('replysync');

        $user = common_current_user();
        $flink = Foreign_link::getByUserID($user->id, TWITTER_SERVICE);

        if (empty($flink)) {
            common_log_db_error($flink, 'SELECT', __FILE__);
            // @todo FIXME: Shouldn't this be a serverError()?
            // TRANS: Server error displayed when saving Twitter integration preferences fails.
            $this->showForm(_m('Could not save Twitter preferences.'));
            return;
        }

        $original = clone($flink);
        $wasReceiving = (bool)($original->noticesync & FOREIGN_NOTICE_RECV);
        $flink->set_flags($noticesend, $noticerecv, $replysync, $friendsync);
        $result = $flink->update($original);

        if ($result === false) {
            common_log_db_error($flink, 'UPDATE', __FILE__);
            // @todo FIXME: Shouldn't this be a serverError()?
            // TRANS: Server error displayed when saving Twitter integration preferences fails.
            $this->showForm(_m('Could not save Twitter preferences.'));
            return;
        }

        if ($wasReceiving xor $noticerecv) {
            $this->notifyDaemon($flink->foreign_id, $noticerecv);
        }

        // TRANS: Success message after saving Twitter integration preferences.
        $this->showForm(_m('Twitter preferences saved.'), true);
    }

    /**
     * Tell the import daemon that we've updated a user's receive status.
     */
    function notifyDaemon($twitterUserId, $receiving)
    {
        // @todo... should use control signals rather than queues
    }
}


code:
<?php
/**
 * Data class for remembering notice-to-status mappings
 *
 * PHP version 5
 *
 * @category Data
 * @package  StatusNet
 * @author   Evan Prodromou <evan@status.net>
 * @license  http://www.fsf.org/licensing/licenses/agpl.html AGPLv3
 * @link     http://status.net/
 *
 * StatusNet - the distributed open-source microblogging tool
 * Copyright (C) 2010, StatusNet, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.     See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

if (!defined('STATUSNET')) {
    exit(1);
}

require_once INSTALLDIR . '/classes/Memcached_DataObject.php';

/**
 * Data class for mapping notices to statuses
 *
 * Notices flow back and forth between Twitter and StatusNet. We use this
 * table to remember which StatusNet notice corresponds to which Twitter
 * status.
 *
 * Note that notice_id is unique only within a single database; if you
 * want to share this data for some reason, get the notice's URI and use
 * that instead, since it's universally unique.
 *
 * @category Action
 * @package  StatusNet
 * @author   Evan Prodromou <evan@status.net>
 * @license  http://www.fsf.org/licensing/licenses/agpl.html AGPLv3
 * @link     http://status.net/
 *
 * @see      DB_DataObject
 */

class Notice_to_status extends Managed_DataObject
{
    public $__table = 'notice_to_status'; // table name
    public $notice_id;                    // int(4)  primary_key not_null
    public $status_id;                    // bigint not_null
    public $created;                      // datetime()   not_null
    public $modified;                     // datetime   not_null default_0000-00-00%2000%3A00%3A00

    public static function schemaDef()
    {
        return array(
            'fields' => array(
                'notice_id' => array('type' => 'int', 'not null' => true, 'description' => 'local notice id'),
                'status_id' => array('type' => 'int', 'size' => 'big', 'not null' => true, 'description' => 'twitter status id'),
                'created' => array('type' => 'datetime', 'not null' => true, 'description' => 'date this record was created'),
                'modified' => array('type' => 'timestamp', 'not null' => true, 'description' => 'date this record was modified'),
            ),
            'primary key' => array('notice_id'),
            'unique keys' => array(
                'status_id_key' => array('status_id'),
            ),
            'foreign keys' => array(
                'notice_to_status_notice_id_fkey' => array('notice', array('notice_id' => 'id')),
            ),
        );
    }

    /**
     * Save a mapping between a notice and a status
     * Warning: status_id values may not fit in 32-bit integers.
     *
     * @param integer $notice_id ID of the notice in StatusNet
     * @param integer $status_id ID of the status in Twitter
     *
     * @return Notice_to_status new object for this value
     */
    static function saveNew($notice_id, $status_id)
    {
        if (empty($notice_id)) {
            throw new Exception("Invalid notice_id $notice_id");
        }
        $n2s = Notice_to_status::getKV('notice_id', $notice_id);

        if (!empty($n2s)) {
            return $n2s;
        }

        if (empty($status_id)) {
            throw new Exception("Invalid status_id $status_id");
        }
        $n2s = Notice_to_status::getKV('status_id', $status_id);

        if (!empty($n2s)) {
            return $n2s;
        }

        common_debug("Mapping notice {$notice_id} to Twitter status {$status_id}");

        $n2s = new Notice_to_status();

        $n2s->notice_id = $notice_id;
        $n2s->status_id = $status_id;
        $n2s->created   = common_sql_now();

        $n2s->insert();

        return $n2s;
    }
}

license: mit


code:
<?php
/**
 * Store last-touched ID for various timelines
 *
 * PHP version 5
 *
 * @category Data
 * @package  StatusNet
 * @author   Evan Prodromou <evan@status.net>
 * @license  http://www.fsf.org/licensing/licenses/agpl.html AGPLv3
 * @link     http://status.net/
 *
 * StatusNet - the distributed open-source microblogging tool
 * Copyright (C) 2010, StatusNet, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.     See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

if (!defined('STATUSNET')) {
    exit(1);
}

require_once INSTALLDIR . '/classes/Memcached_DataObject.php';

/**
 * Store various timeline data
 *
 * We don't want to keep re-fetching the same statuses and direct messages from Twitter.
 * So, we store the last ID we see from a timeline, and store it. Next time
 * around, we use that ID in the since_id parameter.
 *
 * @category Action
 * @package  StatusNet
 * @author   Evan Prodromou <evan@status.net>
 * @license  http://www.fsf.org/licensing/licenses/agpl.html AGPLv3
 * @link     http://status.net/
 *
 * @see      DB_DataObject
 */
class Twitter_synch_status extends Managed_DataObject
{
    public $__table = 'twitter_synch_status'; // table name
    public $foreign_id;                      // bigint primary_key not_null
    public $timeline;                        // varchar(255)  primary_key not_null
    public $last_id;                         // bigint not_null
    public $created;                         // datetime()   not_null
    public $modified;                        // timestamp()   not_null default_CURRENT_TIMESTAMP

    public static function schemaDef()
    {
        return array(
            'fields' => array(
                'foreign_id' => array('type' => 'int', 'size' => 'big', 'not null' => true, 'description' => 'Foreign message ID'),
                'timeline' => array('type' => 'varchar', 'length' => 255, 'description' => 'timeline name'),
                'last_id' => array('type' => 'int', 'size' => 'big', 'not null' => true, 'description' => 'last id fetched'),
                'created' => array('type' => 'datetime', 'not null' => true, 'description' => 'date this record was created'),
                'modified' => array('type' => 'timestamp', 'not null' => true, 'description' => 'date this record was modified'),
            ),
            'primary key' => array('foreign_id', 'timeline'),
        );
    }

    static function getLastId($foreign_id, $timeline)
    {
        $tss = self::pkeyGet(array('foreign_id' => $foreign_id,
                                   'timeline' => $timeline));

        if (empty($tss)) {
            return null;
        } else {
            return $tss->last_id;
        }
    }

    static function setLastId($foreign_id, $timeline, $last_id)
    {
        $tss = self::pkeyGet(array('foreign_id' => $foreign_id,
                                   'timeline' => $timeline));

        if (empty($tss)) {
            $tss = new Twitter_synch_status();

            $tss->foreign_id = $foreign_id;
            $tss->timeline   = $timeline;
            $tss->last_id    = $last_id;
            $tss->created    = common_sql_now();
            $tss->modified   = $tss->created;

            $tss->insert();

            return true;
        } else {
            $orig = clone($tss);

            $tss->last_id  = $last_id;
            $tss->modified = common_sql_now();

            $tss->update();

            return true;
        }
    }
}


code:
#!/usr/bin/env php
<?php
/*
 * StatusNet - the distributed open-source microblogging tool
 * Copyright (C) 2008, 2009, StatusNet, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.     See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.     If not, see <http://www.gnu.org/licenses/>.
 */

define('INSTALLDIR', realpath(dirname(__FILE__) . '/../../..'));

$shortoptions = 'di::';
$longoptions = array('id::', 'debug');

$helptext = <<<END_OF_TRIM_HELP
Batch script for synching local friends with Twitter friends.
  -i --id              Identity (default 'generic')
  -d --debug           Debug (lots of log output)

END_OF_TRIM_HELP;

require_once INSTALLDIR . '/scripts/commandline.inc';
require_once dirname(__DIR__) . '/twitter.php';

/**
 * Daemon to sync local friends with Twitter friends
 *
 * @category Twitter
 * @package  StatusNet
 * @author   Zach Copley <zach@status.net>
 * @author   Evan Prodromou <evan@status.net>
 * @license  http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
 * @link     http://status.net/
 */
class SyncTwitterFriendsDaemon extends ParallelizingDaemon
{
    /**
     *  Constructor
     *
     * @param string  $id           the name/id of this daemon
     * @param int     $interval     sleep this long before doing everything again
     * @param int     $max_children maximum number of child processes at a time
     * @param boolean $debug        debug output flag
     *
     * @return void
     *
     **/
    function __construct($id = null, $interval = 60,
                         $max_children = 2, $debug = null)
    {
        parent::__construct($id, $interval, $max_children, $debug);
    }

    /**
     * Name of this daemon
     *
     * @return string Name of the daemon.
     */
    function name()
    {
        return ('synctwitterfriends.' . $this->_id);
    }

    /**
     * Find all the Twitter foreign links for users who have requested
     * automatically subscribing to their Twitter friends locally.
     *
     * @return array flinks an array of Foreign_link objects
     */
    function getObjects()
    {
        $flinks = array();
        $flink = new Foreign_link();

        $conn = &$flink->getDatabaseConnection();

        $flink->service = TWITTER_SERVICE;
        $flink->orderBy('last_friendsync');
        $flink->limit(25);  // sync this many users during this run
        $flink->find();

        while ($flink->fetch()) {
            if (($flink->friendsync & FOREIGN_FRIEND_RECV) == FOREIGN_FRIEND_RECV) {
                $flinks[] = clone($flink);
            }
        }

        $conn->disconnect();

        global $_DB_DATAOBJECT;
        unset($_DB_DATAOBJECT['CONNECTIONS']);

        return $flinks;
    }

    function childTask($flink) {
        // Each child ps needs its own DB connection

        // Note: DataObject::getDatabaseConnection() creates
        // a new connection if there isn't one already
        $conn = &$flink->getDatabaseConnection();

        $this->subscribeTwitterFriends($flink);

        $flink->last_friendsync = common_sql_now();
        $flink->update();

        $conn->disconnect();

        // XXX: Couldn't find a less brutal way to blow
        // away a cached connection
        global $_DB_DATAOBJECT;
        unset($_DB_DATAOBJECT['CONNECTIONS']);
    }

    function fetchTwitterFriends($flink)
    {
        $friends = array();

        $client = null;

        if (TwitterOAuthClient::isPackedToken($flink->credentials)) {
            $token = TwitterOAuthClient::unpackToken($flink->credentials);
            $client = new TwitterOAuthClient($token->key, $token->secret);
            common_debug($this->name() . '- Grabbing friends IDs with OAuth.');
        } else {
            common_debug("Skipping Twitter friends for {$flink->user_id} since not OAuth.");
            return $friends;
        }

        try {
            $friends_ids = $client->friendsIds();
        } catch (Exception $e) {
            common_log(LOG_WARNING, $this->name() .
                       ' - error getting friend ids: ' .
                       $e->getMessage());
            return $friends;
        }

        if (empty($friends_ids)) {
            common_debug($this->name() .
                         " - Twitter user $flink->foreign_id " .
                         'doesn\'t have any friends!');
            return $friends;
        }

        common_debug($this->name() . ' - Twitter\'s API says Twitter user id ' .
                     "$flink->foreign_id has " .
                     count($friends_ids) . ' friends.');

        // Calculate how many pages to get...
        $pages = ceil(count($friends_ids) / 100);

        if ($pages == 0) {
            common_debug($this->name() . " - $user seems to have no friends.");
        }

        for ($i = 1; $i <= $pages; $i++) {

        try {
            $more_friends = $client->statusesFriends(null, null, null, $i);
        } catch (Exception $e) {
            common_log(LOG_WARNING, $this->name() .
                       ' - cURL error getting Twitter statuses/friends ' .
                       "page $i - " . $e->getCode() . ' - ' .
                       $e->getMessage());
        }

            if (empty($more_friends)) {
                common_log(LOG_WARNING, $this->name() .
                           " - Couldn't retrieve page $i " .
                           "of Twitter user $flink->foreign_id friends.");
                continue;
            } else {
                if (is_array($more_friends)) {
                    $friends = array_merge($friends, $more_friends);
                }
            }
        }

        return $friends;
    }

    function subscribeTwitterFriends($flink)
    {
        $friends = $this->fetchTwitterFriends($flink);

        if (empty($friends)) {
            common_debug($this->name() .
                         ' - Couldn\'t get friends from Twitter for ' .
                         "Twitter user $flink->foreign_id.");
            return false;
        }

        $profile = $flink->getProfile();

        foreach ($friends as $friend) {

            $friend_name = $friend->screen_name;
            $friend_id = (int) $friend->id;

            // Update or create the Foreign_user record for each
            // Twitter friend

            if (!save_twitter_user($friend_id, $friend_name)) {
                common_log(LOG_WARNING, $this->name() .
                           " - Couldn't save $screen_name's friend, $friend_name.");
                continue;
            }

            // Check to see if there's a related local user

            $friend_flink = Foreign_link::getByForeignID($friend_id,
                                                         TWITTER_SERVICE);

            if (!empty($friend_flink)) {

                // Get associated user and subscribe her

                $friend_profile = Profile::getKV('id', $friend_flink->user_id);

                if ($friend_profile instanceof Profile) {
                    try {
                        $other = Profile::getKV('id', $invites->user_id);
                        Subscription::start($profile, $friend_profile);
                        common_log(LOG_INFO,
                                   $this->name() . ' - Subscribed ' .
                                   "{$friend_profile->nickname} to {$profile->nickname}.");
                    } catch (Exception $e) {
                        common_debug($this->name() .
                                     ' - Tried and failed subscribing ' .
                                     "{$friend_profile->nickname} to {$profile->nickname} - " .
                                     $e->getMessage());
                    }
                }
            }
        }

        return true;
    }

}

$id    = null;
$debug = null;

if (have_option('i')) {
    $id = get_option_value('i');
} else if (have_option('--id')) {
    $id = get_option_value('--id');
} else if (count($args) > 0) {
    $id = $args[0];
} else {
    $id = null;
}

if (have_option('d') || have_option('debug')) {
    $debug = true;
}

$syncer = new SyncTwitterFriendsDaemon($id, 60, 2, $debug);
$syncer->runOnce();


code:
#!/usr/bin/env php
<?php
/*
 * StatusNet - the distributed open-source microblogging tool
 * Copyright (C) 2008-2010, StatusNet, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

define('INSTALLDIR', realpath(dirname(__FILE__) . '/../../..'));

$shortoptions = 'fi::a';
$longoptions = array('id::', 'foreground', 'all');

$helptext = <<<END_OF_TWITTERDAEMON_HELP
Daemon script for receiving new notices from Twitter users.

    -i --id           Identity (default none)
    -a --all          Handle Twitter for all local sites
                      (requires Stomp queue handler, status_network setup)
    -f --foreground   Stay in the foreground (default background)

END_OF_TWITTERDAEMON_HELP;

require_once INSTALLDIR.'/scripts/commandline.inc';

class TwitterDaemon extends SpawningDaemon
{
    protected $allsites = false;

    function __construct($id=null, $daemonize=true, $threads=1, $allsites=false)
    {
        if ($threads != 1) {
            // This should never happen. :)
            throw new Exception("TwitterDaemon must run single-threaded");
        }
        parent::__construct($id, $daemonize, $threads);
        $this->allsites = $allsites;
    }

    function runThread()
    {
        common_log(LOG_INFO, 'Waiting to listen to Twitter and queues');

        $master = new TwitterMaster($this->get_id(), $this->processManager());
        $master->init($this->allsites);
        $master->service();

        common_log(LOG_INFO, 'terminating normally');

        return $master->respawn ? self::EXIT_RESTART : self::EXIT_SHUTDOWN;
    }

}

class TwitterMaster extends IoMaster
{
    protected $processManager;

    function __construct($id, $processManager)
    {
        parent::__construct($id);
        $this->processManager = $processManager;
    }

    /**
     * Initialize IoManagers for the currently configured site
     * which are appropriate to this instance.
     */
    function initManagers()
    {
        $qm = QueueManager::get();
        $qm->setActiveGroup('twitter');
        $this->instantiate($qm);
        $this->instantiate(new TwitterManager());
        $this->instantiate($this->processManager);
    }
}

class TwitterManager extends IoManager
{
    // Recommended resource limits from http://dev.twitter.com/pages/site_streams
    const MAX_STREAMS = 1000;
    const USERS_PER_STREAM = 100;
    const STREAMS_PER_SECOND = 20;

    protected $streams;
    protected $users;

    /**
     * Pull the site's active Twitter-importing users and start spawning
     * some data streams for them!
     *
     * @fixme check their last-id and check whether we'll need to do a manual pull.
     * @fixme abstract out the fetching so we can work over multiple sites.
     */
    protected function initStreams()
    {
        common_log(LOG_INFO, 'init...');
        // Pull Twitter user IDs for all users we want to pull data for
        $flink = new Foreign_link();
        $flink->service = TWITTER_SERVICE;
        // @fixme probably should do the bitfield check in a whereAdd but it's ugly :D
        $flink->find();

        $userIds = array();
        while ($flink->fetch()) {
            if (($flink->noticesync & FOREIGN_NOTICE_RECV) ==
                FOREIGN_NOTICE_RECV) {
                $userIds[] = $flink->foreign_id;

                if (count($userIds) >= self::USERS_PER_STREAM) {
                    $this->spawnStream($userIds);
                    $userIds = array();
                }
            }
        }

        if (count($userIds)) {
            $this->spawnStream($userIds);
        }
    }

    /**
     * Prepare a Site Stream connection for the given chunk of users.
     * The actual connection will be opened later.
     *
     * @param $userIds array of Twitter-side user IDs
     */
    protected function spawnStream($userIds)
    {
        $stream = $this->initSiteStream();
        $stream->followUsers($userIds);

        // Slip the stream reader into our list of active streams.
        // We'll manage its actual connection on the next go-around.
        $this->streams[] = $stream;

        // Record the user->stream mappings; this makes it easier for us to know
        // later if we need to kill something.
        foreach ($userIds as $id) {
            $this->users[$id] = $stream;
        }
    }

    /**
     * Initialize a generic site streams connection object.
     * All our connections will look like this, then we'll add users to them.
     *
     * @return TwitterStreamReader
     */
    protected function initSiteStream()
    {
        $auth = $this->siteStreamAuth();
        $stream = new TwitterSiteStream($auth);

        // Add our event handler callbacks. Whee!
        $this->setupEvents($stream);
        return $stream;
    }

    /**
     * Fetch the Twitter OAuth credentials to use to connect to the Site Streams API.
     *
     * This will use the locally-stored credentials for the applictation's owner account
     * from the site configuration. These should be configured through the administration
     * panels or manually in the config file.
     *
     * Will throw an exception if no credentials can be found -- but beware that invalid
     * credentials won't cause breakage until later.
     *
     * @return TwitterOAuthClient
     */
    protected function siteStreamAuth()
    {
        $token = common_config('twitter', 'stream_token');
        $secret = common_config('twitter', 'stream_secret');
        if (empty($token) || empty($secret)) {
            throw new ServerException('Twitter site streams have not been correctly configured. Configure the app owner account via the admin panel.');
        }
        return new TwitterOAuthClient($token, $secret);
    }

    /**
     * Collect the sockets for all active connections for i/o monitoring.
     *
     * @return array of resources
     */
    public function getSockets()
    {
        $sockets = array();
        foreach ($this->streams as $stream) {
            foreach ($stream->getSockets() as $socket) {
                $sockets[] = $socket;
            }
        }
        return $sockets;
    }

    /**
     * We're ready to process input from one of our data sources! Woooooo!
     * @fixme is there an easier way to map from socket back to owning module? :(
     *
     * @param resource $socket
     * @return boolean success
     */
    public function handleInput($socket)
    {
        foreach ($this->streams as $stream) {
            foreach ($stream->getSockets() as $aSocket) {
                if ($socket === $aSocket) {
                    $stream->handleInput($socket);
                }
            }
        }
        return true;
    }

    /**
     * Start the i/o system up! Prepare our connections and start opening them.
     *
     * @fixme do some rate-limiting on the stream setup
     * @fixme do some sensible backoff on failure etc
     */
    public function start()
    {
        $this->initStreams();
        foreach ($this->streams as $stream) {
            $stream->connect();
        }
        return true;
    }

    /**
     * Close down our connections when the daemon wraps up for business.
     */
    public function finish()
    {
        foreach ($this->streams as $index => $stream) {
            $stream->close();
            unset($this->streams[$index]);
        }
        return true;
    }

    public static function get()
    {
        throw new Exception('not a singleton');
    }

    /**
     * Set up event handlers on the streaming interface.
     *
     * @fixme add more event types as we add handling for them
     */
    protected function setupEvents(TwitterStreamReader $stream)
    {
        $handlers = array(
            'status',
        );
        foreach ($handlers as $event) {
            $stream->hookEvent($event, array($this, 'onTwitter' . ucfirst($event)));
        }
    }

    /**
     * Event callback notifying that a user has a new message in their home timeline.
     * We store the incoming message into the queues for processing, keeping our own
     * daemon running as shiny-fast as possible.
     *
     * @param object $status JSON data: Twitter status update
     * @fixme in all-sites mode we may need to route queue items into another site's
     *        destination queues, or multiple sites.
     */
    protected function onTwitterStatus($status, $context)
    {
        $data = array(
            'status' => $status,
            'for_user' => $context->for_user,
        );
        $qm = QueueManager::get();
        $qm->enqueue($data, 'tweetin');
    }
}

if (have_option('i', 'id')) {
    $id = get_option_value('i', 'id');
} else if (count($args) > 0) {
    $id = $args[0];
} else {
    $id = null;
}

$foreground = have_option('f', 'foreground');
$all = have_option('a') || have_option('--all');

$daemon = new TwitterDaemon($id, !$foreground, 1, $all);

$daemon->runOnce();


code:
#!/usr/bin/env php
<?php
/**
 * StatusNet - the distributed open-source microblogging tool
 * Copyright (C) 2008-2010, StatusNet, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.     See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.     If not, see <http://www.gnu.org/licenses/>.
 */

define('INSTALLDIR', realpath(dirname(__FILE__) . '/../../..'));

// Tune number of processes and how often to poll Twitter
// XXX: Should these things be in config.php?
define('MAXCHILDREN', 2);
define('POLL_INTERVAL', 70); // in seconds, Twitter API v1.1 says 15 calls every 15 mins

$shortoptions = 'di::';
$longoptions = array('id::', 'debug');

$helptext = <<<END_OF_TRIM_HELP
Batch script for retrieving Twitter messages from foreign service.

  -i --id              Identity (default 'generic')
  -d --debug           Debug (lots of log output)

END_OF_TRIM_HELP;

require_once INSTALLDIR . '/scripts/commandline.inc';
require_once INSTALLDIR . '/lib/common.php';
require_once INSTALLDIR . '/lib/daemon.php';
require_once dirname(__DIR__) . '/twitter.php';

/**
 * Fetch statuses from Twitter
 *
 * Fetches statuses from Twitter and inserts them as notices
 *
 * NOTE: an Avatar path MUST be set in config.php for this
 * script to work, e.g.:
 *     $config['avatar']['path'] = $config['site']['path'] . '/avatar/';
 *
 * @todo @fixme @gar Fix the above. For some reason $_path is always empty when
 * this script is run, so the default avatar path is always set wrong in
 * default.php. Therefore it must be set explicitly in config.php. --Z
 *
 * @category Twitter
 * @package  StatusNet
 * @author   Zach Copley <zach@status.net>
 * @author   Evan Prodromou <evan@status.net>
 * @license  http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
 * @link     http://status.net/
 */
class TwitterStatusFetcher extends ParallelizingDaemon
{
    /**
     *  Constructor
     *
     * @param string  $id           the name/id of this daemon
     * @param int     $interval     sleep this long before doing everything again
     * @param int     $max_children maximum number of child processes at a time
     * @param boolean $debug        debug output flag
     *
     * @return void
     *
     **/
    function __construct($id = null, $interval = 60,
                         $max_children = 2, $debug = null)
    {
        parent::__construct($id, $interval, $max_children, $debug);
    }

    /**
     * Name of this daemon
     *
     * @return string Name of the daemon.
     */
    function name()
    {
        return ('twitterstatusfetcher.'.$this->_id);
    }

    /**
     * Find all the Twitter foreign links for users who have requested
     * importing of their friends' timelines
     *
     * @return array flinks an array of Foreign_link objects
     */
    function getObjects()
    {
        global $_DB_DATAOBJECT;
        $flink = new Foreign_link();
        $conn = &$flink->getDatabaseConnection();

        $flink->service = TWITTER_SERVICE;
        $flink->orderBy('last_noticesync');
        $flink->find();

        $flinks = array();

        while ($flink->fetch()) {

            if (($flink->noticesync & FOREIGN_NOTICE_RECV) ==
                FOREIGN_NOTICE_RECV) {
                $flinks[] = clone($flink);
                common_log(LOG_INFO, "sync: foreign id $flink->foreign_id");
            } else {
                common_log(LOG_INFO, "nothing to sync");
            }
        }

        $flink->free();
        unset($flink);

        $conn->disconnect();
        unset($_DB_DATAOBJECT['CONNECTIONS']);

        return $flinks;
    }

    function childTask($flink) {
        // Each child ps needs its own DB connection

        // Note: DataObject::getDatabaseConnection() creates
        // a new connection if there isn't one already
        $conn = &$flink->getDatabaseConnection();

        $this->getTimeline($flink, 'home_timeline');
        $this->getTimeline($flink, 'mentions_timeline');

        $flink->last_friendsync = common_sql_now();
        $flink->update();

        $conn->disconnect();

        // XXX: Couldn't find a less brutal way to blow
        // away a cached connection
        global $_DB_DATAOBJECT;
        unset($_DB_DATAOBJECT['CONNECTIONS']);
    }

    function getTimeline($flink, $timelineUri = 'home_timeline')
    {
        if (empty($flink)) {
            common_log(LOG_ERR, $this->name() .
                       " - Can't retrieve Foreign_link for foreign ID $fid");
            return;
        }

        common_log(LOG_DEBUG, $this->name() . ' - Trying to get ' . $timelineUri .
                   ' timeline for Twitter user ' . $flink->foreign_id);

        $client = null;

        if (TwitterOAuthClient::isPackedToken($flink->credentials)) {
            $token = TwitterOAuthClient::unpackToken($flink->credentials);
            $client = new TwitterOAuthClient($token->key, $token->secret);
            common_log(LOG_DEBUG, $this->name() . ' - Grabbing ' . $timelineUri . ' timeline with OAuth.');
        } else {
            common_log(LOG_ERR, "Skipping " . $timelineUri . " timeline for " .
                       $flink->foreign_id . " since not OAuth.");
        }

        $timeline = null;

        $lastId = Twitter_synch_status::getLastId($flink->foreign_id, $timelineUri);

        common_log(LOG_DEBUG, "Got lastId value '" . $lastId . "' for foreign id '" .
                     $flink->foreign_id . "' and timeline '" . $timelineUri. "'");

        try {
            $timeline = $client->statusesTimeline($lastId, $timelineUri);
        } catch (Exception $e) {
            common_log(LOG_ERR, $this->name() .
                       ' - Unable to get ' . $timelineUri . ' timeline for user ' . $flink->user_id .
                       ' - code: ' . $e->getCode() . 'msg: ' . $e->getMessage());
        }

        if (empty($timeline)) {
            common_log(LOG_DEBUG, $this->name() .  " - Empty '" . $timelineUri . "' timeline.");
            return;
        }

        common_log(LOG_INFO, $this->name() .
                   ' - Retrieved ' . sizeof($timeline) . ' statuses from ' . $timelineUri . ' timeline' .
                   ' - for user ' . $flink->user_id);

        if (!empty($timeline)) {
            $qm = QueueManager::get();

            // Reverse to preserve order
            foreach (array_reverse($timeline) as $status) {
                $data = array(
                    'status' => $status,
                    'for_user' => $flink->foreign_id,
                );
                $qm->enqueue($data, 'tweetin');
            }

            $lastId = twitter_id($timeline[0]);
            Twitter_synch_status::setLastId($flink->foreign_id, $timelineUri, $lastId);
            common_debug("Set lastId value '$lastId' for foreign id '{$flink->foreign_id}' and timeline '" .
                         $timelineUri . "'");
        }

        // Okay, record the time we synced with Twitter for posterity
        $flink->last_noticesync = common_sql_now();
        $flink->update();
    }
}

$id    = null;
$debug = null;

if (have_option('i')) {
    $id = get_option_value('i');
} else if (have_option('--id')) {
    $id = get_option_value('--id');
} else if (count($args) > 0) {
    $id = $args[0];
} else {
    $id = null;
}

if (have_option('d') || have_option('debug')) {
    $debug = true;
}

$fetcher = new TwitterStatusFetcher($id, POLL_INTERVAL, MAXCHILDREN, $debug);
$fetcher->runOnce();


code:
<?php
/**
 * StatusNet, the distributed open-source microblogging tool
 *
 * PHP version 5
 *
 * LICENCE: This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * @category  Plugin
 * @package   StatusNet
 * @author    Brion Vibber <brion@status.net>
 * @copyright 2010 StatusNet, Inc.
 * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
 * @link      http://status.net/
 */

class OAuthData
{
    public $consumer_key, $consumer_secret, $token, $token_secret;
}

/**
 *
 */
abstract class JsonStreamReader
{
    const CRLF = "\r\n";

    public $id;
    protected $socket = null;
    protected $state = 'init'; // 'init', 'connecting', 'waiting', 'headers', 'active'

    public function __construct()
    {
        $this->id = get_class($this) . '.' . substr(md5(mt_rand()), 0, 8);
    }

    /**
     * Starts asynchronous connect operation...
     *
     * @fixme Can we do the open-socket fully async to? (need write select infrastructure)
     *
     * @param string $url
     */
    public function connect($url)
    {
        common_log(LOG_DEBUG, "$this->id opening connection to $url");

        $scheme = parse_url($url, PHP_URL_SCHEME);
        if ($scheme == 'http') {
            $rawScheme = 'tcp';
        } else if ($scheme == 'https') {
            $rawScheme = 'ssl';
        } else {
            // TRANS: Server exception thrown when an invalid URL scheme is detected.
            throw new ServerException(_m('Invalid URL scheme for HTTP stream reader.'));
        }

        $host = parse_url($url, PHP_URL_HOST);
        $port = parse_url($url, PHP_URL_PORT);
        if (!$port) {
            if ($scheme == 'https') {
                $port = 443;
            } else {
                $port = 80;
            }
        }

        $path = parse_url($url, PHP_URL_PATH);
        $query = parse_url($url, PHP_URL_QUERY);
        if ($query) {
            $path .= '?' . $query;
        }

        $errno = $errstr = null;
        $timeout = 5;
        //$flags = STREAM_CLIENT_CONNECT | STREAM_CLIENT_ASYNC_CONNECT;
        $flags = STREAM_CLIENT_CONNECT;
        // @fixme add SSL params
        $this->socket = stream_socket_client("$rawScheme://$host:$port", $errno, $errstr, $timeout, $flags);

        $this->send($this->httpOpen($host, $path));

        stream_set_blocking($this->socket, false);
        $this->state = 'waiting';
    }

    /**
     * Send some fun data off to the server.
     *
     * @param string $buffer
     */
    function send($buffer)
    {
        fwrite($this->socket, $buffer);
    }

    /**
     * Read next packet of data from the socket.
     *
     * @return string
     */
    function read()
    {
        $buffer = fread($this->socket, 65536);
        return $buffer;
    }

    /**
     * Build HTTP request headers.
     *
     * @param string $host
     * @param string $path
     * @return string
     */
    protected function httpOpen($host, $path)
    {
        $lines = array(
            "GET $path HTTP/1.1",
            "Host: $host",
            'User-Agent: ' . HTTPClient::userAgent() . ' (TwitterBridgePlugin)',
            "Connection: close",
            "",
            ""
        );
        return implode(self::CRLF, $lines);
    }

    /**
     * Close the current connection, if open.
     */
    public function close()
    {
        if ($this->isConnected()) {
            common_log(LOG_DEBUG, "$this->id closing connection.");
            fclose($this->socket);
            $this->socket = null;
        }
    }

    /**
     * Are we currently connected?
     *
     * @return boolean
     */
    public function isConnected()
    {
        return $this->socket !== null;
    }

    /**
     * Send any sockets we're listening on to the IO manager
     * to wait for input.
     *
     * @return array of resources
     */
    public function getSockets()
    {
        if ($this->isConnected()) {
            return array($this->socket);
        }
        return array();
    }

    /**
     * Take a chunk of input over the horn and go go go! :D
     *
     * @param string $buffer
     */
    public function handleInput($socket)
    {
        if ($this->socket !== $socket) {
            // TRANS: Exception thrown when input from an inexpected socket is encountered.
            throw new Exception(_m('Got input from unexpected socket!'));
        }

        try {
            $buffer = $this->read();
            $lines = explode(self::CRLF, $buffer);
            foreach ($lines as $line) {
                $this->handleLine($line);
            }
        } catch (Exception $e) {
            common_log(LOG_ERR, "$this->id aborting connection due to error: " . $e->getMessage());
            fclose($this->socket);
            throw $e;
        }
    }

    protected function handleLine($line)
    {
        switch ($this->state)
        {
            case 'waiting':
                $this->handleLineWaiting($line);
                break;
            case 'headers':
                $this->handleLineHeaders($line);
                break;
            case 'active':
                $this->handleLineActive($line);
                break;
            default:
                // TRANS: Exception thrown when an invalid state is encountered in handleLine.
                // TRANS: %s is the invalid state.
                throw new Exception(sprintf(_m('Invalid state in handleLine: %s.'),$this->state));
        }
    }

    /**
     *
     * @param <type> $line
     */
    protected function handleLineWaiting($line)
    {
        $bits = explode(' ', $line, 3);
        if (count($bits) != 3) {
            // TRANS: Exception thrown when an invalid response line is encountered.
            // TRANS: %s is the invalid line.
            throw new Exception(sprintf(_m('Invalid HTTP response line: %s.'),$line));
        }

        list($http, $status, $text) = $bits;
        if (substr($http, 0, 5) != 'HTTP/') {
            // TRANS: Exception thrown when an invalid response line part is encountered.
            // TRANS: %1$s is the chunk, %2$s is the line.
            throw new Exception(sprintf(_m('Invalid HTTP response line chunk "%1$s": %2$s.'),$http, $line));
        }
        if ($status != '200') {
            // TRANS: Exception thrown when an invalid response code is encountered.
            // TRANS: %1$s is the response code, %2$s is the line.
            throw new Exception(sprintf(_m('Bad HTTP response code %1$s: %2$s.'),$status,$line));
        }
        common_log(LOG_DEBUG, "$this->id $line");
        $this->state = 'headers';
    }

    protected function handleLineHeaders($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;
        }
    }

    protected function handleLineActive($line)
    {
        if ($line == "") {
            // Server sends empty lines as keepalive.
            return;
        }
        $data = json_decode($line);
        if ($data) {
            $this->handleJson($data);
        } else {
            common_log(LOG_ERR, "$this->id received bogus JSON data: " . var_export($line, true));
        }
    }

    abstract protected function handleJson(stdClass $data);
}


code:
<?php
/*
 * StatusNet - the distributed open-source microblogging tool
 * Copyright (C) 2010, StatusNet, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }

require_once dirname(__DIR__) . '/twitter.php';

/**
 * Queue handler to deal with incoming Twitter status updates, as retrieved by
 * TwitterDaemon (twitterdaemon.php).
 *
 * The queue handler passes the status through TwitterImporter for import into the
 * local database (if necessary), then adds the imported notice to the local inbox
 * of the attached Twitter user.
 *
 * Warning: the way we do inbox distribution manually means that realtime, XMPP, etc
 * don't work on Twitter-borne messages. When TwitterImporter is changed to handle
 * that correctly, we'll only need to do this once...?
 */
class TweetInQueueHandler extends QueueHandler
{
    function transport()
    {
        return 'tweetin';
    }

    function handle($data)
    {
        // JSON object with Twitter data
        $status = $data['status'];

        // Twitter user ID this incoming data belongs to.
        $receiver = $data['for_user'];

        $importer = new TwitterImport();
        $notice = $importer->importStatus($status);
        if ($notice instanceof Notice) {
            $flink = Foreign_link::getByForeignID($receiver, TWITTER_SERVICE);
            if ($flink instanceof Foreign_link) {
                common_log(LOG_DEBUG, "TweetInQueueHandler - Got flink so add notice ".
                           $notice->id." to attentions for user ".$flink->user_id);
                try {
                    Attention::saveNew($notice, $flink->getProfile());
                } catch (Exception $e) {
                    // Log the exception, but make sure we don't bail out, we
                    // still have a queue item to remove here-after.
                    common_log(LOG_ERR, "Failed adding notice {$notice->id} to attentions for user {$flink->user_id}: " .
                                        $e->getMessage());
                }
            } else {
               common_log(LOG_DEBUG, "TweetInQueueHandler - No flink found for foreign user ".$receiver);
            }
        }

        return true;
    }
}


code:
<?php
/**
 * StatusNet, the distributed open-source microblogging tool
 *
 * PHP version 5
 *
 * LICENCE: This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * @category  Plugin
 * @package   StatusNet
 * @author    Zach Copley <zach@status.net>
 * @author    Julien C <chaumond@gmail.com>
 * @author    Brion Vibber <brion@status.net>
 * @copyright 2009-2010 StatusNet, Inc.
 * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
 * @link      http://status.net/
 */

if (!defined('STATUSNET')) {
    exit(1);
}

require_once dirname(__DIR__) . '/twitter.php';

/**
 * Encapsulation of the Twitter status -> notice incoming bridge import.
 * Is used by both the polling twitterstatusfetcher.php daemon, and the
 * in-progress streaming import.
 *
 * @category Plugin
 * @package  StatusNet
 * @author   Zach Copley <zach@status.net>
 * @author   Julien C <chaumond@gmail.com>
 * @author   Brion Vibber <brion@status.net>
 * @license  http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
 * @link     http://status.net/
 * @link     http://twitter.com/
 */
class TwitterImport
{
    public $avatarsizename = 'reasonably_small'; // a Twitter size name for 128x128 px
    public $avatarsize = 128;   // they're square...

    public function importStatus($status)
    {
        // Hacktastic: filter out stuff coming from this StatusNet
        $source = mb_strtolower(common_config('integration', 'source'));

        if (preg_match("/$source/", mb_strtolower($status->source))) {
            common_debug(__METHOD__ . ' - Skipping import of status ' .
                         twitter_id($status) . " with source {$source}");
            return null;
        }

        // Don't save it if the user is protected
        // FIXME: save it but treat it as private
        if ($status->user->protected) {
            return null;
        }

        $notice = $this->saveStatus($status);

        return $notice;
    }

    function name()
    {
        return get_class($this);
    }

    function saveStatus($status)
    {
        $profile = $this->ensureProfile($status->user);

        if (empty($profile)) {
            common_log(LOG_ERR, __METHOD__ . ' - Problem saving notice. No associated Profile.');
            return null;
        }

        $statusId = twitter_id($status);
        $statusUri = $this->makeStatusURI($status->user->screen_name, $statusId);

        // check to see if we've already imported the status
        $n2s = Notice_to_status::getKV('status_id', $statusId);

        if (!empty($n2s)) {
            common_log(
                LOG_INFO,
                __METHOD__ . " - Ignoring duplicate import: {$statusId}"
            );
            return Notice::getKV('id', $n2s->notice_id);
        }

        // If it's a retweet, save it as a repeat!
        if (!empty($status->retweeted_status)) {
            common_log(LOG_INFO, "Status {$statusId} is a retweet of " . twitter_id($status->retweeted_status) . ".");
            $original = $this->saveStatus($status->retweeted_status);
            if (empty($original)) {
                return null;
            } else {
                $author = $original->getProfile();
                // TRANS: Message used to repeat a notice. RT is the abbreviation of 'retweet'.
                // TRANS: %1$s is the repeated user's name, %2$s is the repeated notice.
                $content = sprintf(_m('RT @%1$s %2$s'),
                                   $author->nickname,
                                   $original->content);

                if (Notice::contentTooLong($content)) {
                    $contentlimit = Notice::maxContent();
                    $content = mb_substr($content, 0, $contentlimit - 4) . ' ...';
                }

                $repeat = Notice::saveNew($profile->id,
                                          $content,
                                          'twitter',
                                          array('repeat_of' => $original->id,
                                                'uri' => $statusUri,
                                                'is_local' => Notice::GATEWAY));
                common_log(LOG_INFO, "Saved {$repeat->id} as a repeat of {$original->id}");
                Notice_to_status::saveNew($repeat->id, $statusId);
                return $repeat;
            }
        }

        $notice = new Notice();

        $notice->profile_id = $profile->id;
        $notice->uri        = $statusUri;
        $notice->url        = $statusUri;
        $notice->created    = strftime(
            '%Y-%m-%d %H:%M:%S',
            strtotime($status->created_at)
        );

        $notice->source     = 'twitter';

        $notice->reply_to   = null;

        $replyTo = twitter_id($status, 'in_reply_to_status_id');
        if (!empty($replyTo)) {
            common_log(LOG_INFO, "Status {$statusId} is a reply to status {$replyTo}");
            $n2s = Notice_to_status::getKV('status_id', $replyTo);
            if (empty($n2s)) {
                common_log(LOG_INFO, "Couldn't find local notice for status {$replyTo}");
            } else {
                $reply = Notice::getKV('id', $n2s->notice_id);
                if (empty($reply)) {
                    common_log(LOG_INFO, "Couldn't find local notice for status {$replyTo}");
                } else {
                    common_log(LOG_INFO, "Found local notice {$reply->id} for status {$replyTo}");
                    $notice->reply_to     = $reply->id;
                    $notice->conversation = $reply->conversation;
                }
            }
        }

        $notice->is_local   = Notice::GATEWAY;

        $notice->content  = html_entity_decode($this->linkify($status, FALSE), ENT_QUOTES, 'UTF-8');
        $notice->rendered = $this->linkify($status, TRUE);

        if (Event::handle('StartNoticeSave', array(&$notice))) {

            $id = $notice->insert();

            if ($id === false) {
                common_log_db_error($notice, 'INSERT', __FILE__);
                common_log(LOG_ERR, __METHOD__ . ' - Problem saving notice.');
            }

            if (empty($notice->conversation)) {
                $orig = clone($notice);
                $conv = Conversation::create($notice);
                common_log(LOG_INFO, "No known conversation for status {$statusId} so a new one ({$conv->id}) was created.");
                $notice->conversation = $conv->id;
                $notice->update($orig);
            }

            Event::handle('EndNoticeSave', array($notice));
        }

        Notice_to_status::saveNew($notice->id, $statusId);

        $this->saveStatusMentions($notice, $status);
        $this->saveStatusAttachments($notice, $status);

        $notice->blowOnInsert();

        return $notice;
    }

    /**
     * Make an URI for a status.
     *
     * @param object $status status object
     *
     * @return string URI
     */
    function makeStatusURI($username, $id)
    {
        return 'http://twitter.com/#!/'
          . $username
          . '/status/'
          . $id;
    }

    /**
     * Look up a Profile by profileurl field.  Profile::getKV() was
     * not working consistently.
     *
     * @param string $nickname   local nickname of the Twitter user
     * @param string $profileurl the profile url
     *
     * @return mixed value the first Profile with that url, or null
     */
    protected function getProfileByUrl($nickname, $profileurl)
    {
        $profile = new Profile();
        $profile->nickname = $nickname;
        $profile->profileurl = $profileurl;
        $profile->limit(1);

        if (!$profile->find(true)) {
            throw new NoResultException($profile);
        }
        return $profile;
    }

    protected function ensureProfile($twuser)
    {
        // check to see if there's already a profile for this user
        $profileurl = 'http://twitter.com/' . $twuser->screen_name;
        try {
            $profile = $this->getProfileByUrl($twuser->screen_name, $profileurl);
            $this->updateAvatar($twuser, $profile);
            return $profile;
        } catch (NoResultException $e) {
            common_debug(__METHOD__ . ' - Adding profile and remote profile ' .
                         "for Twitter user: $profileurl.");
        }

        $profile = new Profile();
        $profile->query("BEGIN");
        $profile->nickname   = $twuser->screen_name;
        $profile->fullname   = $twuser->name;
        $profile->homepage   = $twuser->url;
        $profile->bio        = $twuser->description;
        $profile->location   = $twuser->location;
        $profile->profileurl = $profileurl;
        $profile->created    = common_sql_now();

        try {
            $id = $profile->insert();   // insert _should_ throw exception on failure
            if (empty($id)) {
                throw new Exception('Failed insert');
            }
        } catch(Exception $e) {
            common_log(LOG_WARNING, __METHOD__ . " Couldn't insert profile: " . $e->getMessage());
            common_log_db_error($profile, 'INSERT', __FILE__);
            $profile->query("ROLLBACK");
            return false;
        }

        $profile->query("COMMIT");
        $this->updateAvatar($twuser, $profile);
        return $profile;
    }

    /*
     * Checks whether we have to update the profile's avatar
     *
     * @return true when updated, false on failure, null when no action taken
     */
    protected function updateAvatar($twuser, Profile $profile)
    {
        $path_parts = pathinfo($twuser->profile_image_url);
        $ext        = isset($path_parts['extension'])
                        ? '.'.$path_parts['extension']
                        : '';	// some lack extension
        $img_root   = basename($path_parts['basename'], '_normal'.$ext);	// cut off extension
        $filename   = "Twitter_{$twuser->id}_{$img_root}_{$this->avatarsizename}{$ext}";

        try {
            $avatar = Avatar::getUploaded($profile);
            if ($avatar->filename === $filename) {
                return null;
            }
            common_debug(__METHOD__ . " - Updating profile avatar (profile_id={$profile->id}) " .
			"from {$avatar->filename} to {$filename}");
            // else we continue with creating a new avatar
        } catch (NoAvatarException $e) {
            // Avatar was not found. We can catch NoAvatarException or FileNotFoundException
            // but generally we just want to continue creating a new avatar.
            common_debug(__METHOD__ . " - No avatar found for (profile_id={$profile->id})");
        }

        $url        = "{$path_parts['dirname']}/{$img_root}_{$this->avatarsizename}{$ext}";
        $mediatype  = $this->getMediatype(mb_substr($ext, 1));

        try {
            $this->newAvatar($profile, $url, $filename, $mediatype);
        } catch (Exception $e) {
            if (file_exists(Avatar::path($filename))) {
                unlink(Avatar::path($filename));
            }
            return false;
        }

        return true;
    }

    protected function getMediatype($ext)
    {
        $mediatype = null;

        switch (strtolower($ext)) {
        case 'jpeg':
        case 'jpg':
            $mediatype = 'image/jpeg';
            break;
        case 'gif':
            $mediatype = 'image/gif';
            break;
        default:
            $mediatype = 'image/png';
        }

        return $mediatype;
    }

    protected function newAvatar(Profile $profile, $url, $filename, $mediatype)
    {
        // Clear out old avatars, won't do anything if there are none
        Avatar::deleteFromProfile($profile);

        // throws exception if unable to fetch
        $this->fetchRemoteUrl($url, Avatar::path($filename));

        $avatar = new Avatar();
        $avatar->profile_id = $profile->id;
        $avatar->original   = 1; // this is an original/"uploaded" avatar
        $avatar->mediatype  = $mediatype;
        $avatar->filename   = $filename;
        $avatar->url        = Avatar::url($filename);
        $avatar->width      = $this->avatarsize;
        $avatar->height     = $this->avatarsize;

        $avatar->created = common_sql_now();

        $id = $avatar->insert();

        if (empty($id)) {
            common_log(LOG_WARNING, __METHOD__ . " Couldn't insert avatar - " . $e->getMessage());
            common_log_db_error($avatar, 'INSERT', __FILE__);
            throw new ServerException('Could not insert avatar');
        }

        common_debug(__METHOD__ . " - Saved new avatar for {$profile->id}.");

        return $avatar;
    }

    /**
     * Fetch a remote avatar image and save to local storage.
     *
     * @param string $url avatar source URL
     * @param string $filename bare local filename for download
     * @return bool true on success, false on failure
     */
    protected function fetchRemoteUrl($url, $filename)
    {
        common_debug(__METHOD__ . " - Fetching Twitter avatar: {$url} to {$filename}");
        $request = HTTPClient::start();
        $request->setConfig('connect_timeout', 3);  // I had problems with throttling
        $request->setConfig('timeout', 6);          // and locking the process sucks.
        $response = $request->get($url);
        if ($response->isOk()) {
            if (!file_put_contents($filename, $response->getBody())) {
                throw new ServerException('Failed saving fetched file');
            }
        } else {
            throw new Exception('Unexpected HTTP status code');
        }
        return true;
    }

    const URL = 1;
    const HASHTAG = 2;
    const MENTION = 3;

    function linkify($status, $html = FALSE)
    {
        $text = $status->text;

        if (empty($status->entities)) {
            $statusId = twitter_id($status);
            common_log(LOG_WARNING, "No entities data for {$statusId}; trying to fake up links ourselves.");
            $text = common_replace_urls_callback($text, 'common_linkify');
            $text = preg_replace_callback('/(^|\&quot\;|\'|\(|\[|\{|\s+)#([\pL\pN_\-\.]{1,64})/',
                        function ($m) { return $m[1].'#'.TwitterStatusFetcher::tagLink($m[2]); }, $text);
            $text = preg_replace_callback('/(^|\s+)@([a-z0-9A-Z_]{1,64})/',
                        function ($m) { return $m[1].'@'.TwitterStatusFetcher::atLink($m[2]); }, $text);
            return $text;
        }

        // Move all the entities into order so we can
        // replace them and escape surrounding plaintext
        // in order

        $toReplace = array();

        if (!empty($status->entities->urls)) {
            foreach ($status->entities->urls as $url) {
                $toReplace[$url->indices[0]] = array(self::URL, $url);
            }
        }

        if (!empty($status->entities->hashtags)) {
            foreach ($status->entities->hashtags as $hashtag) {
                $toReplace[$hashtag->indices[0]] = array(self::HASHTAG, $hashtag);
            }
        }

        if (!empty($status->entities->user_mentions)) {
            foreach ($status->entities->user_mentions as $mention) {
                $toReplace[$mention->indices[0]] = array(self::MENTION, $mention);
            }
        }

        // sort in forward order by key

        ksort($toReplace);

        $result = '';
        $cursor = 0;

        foreach ($toReplace as $part) {
            list($type, $object) = $part;
            $start = $object->indices[0];
            $end = $object->indices[1];
            if ($cursor < $start) {
                // Copy in the preceding plaintext
                $result .= $this->twitEscape(mb_substr($text, $cursor, $start - $cursor));
                $cursor = $start;
            }
            $orig = $this->twitEscape(mb_substr($text, $start, $end - $start));
            switch($type) {
            case self::URL:
                $linkText = $this->makeUrlLink($object, $orig, $html);
                break;
            case self::HASHTAG:
                if ($html) {
                    $linkText = $this->makeHashtagLink($object, $orig);
                }else{
                    $linkText = $orig;
                }
                break;
            case self::MENTION:
                if ($html) {
                    $linkText = $this->makeMentionLink($object, $orig);
                }else{
                    $linkText = $orig;
                }
                break;
            default:
                $linkText = $orig;
                continue;
            }
            $result .= $linkText;
            $cursor = $end;
        }
        $last = $this->twitEscape(mb_substr($text, $cursor));
        $result .= $last;

        return $result;
    }

    function twitEscape($str)
    {
        // Twitter seems to preemptive turn < and > into < and >
        // but doesn't for &, so while you may have some magic protection
        // against XSS by not bothing to escape manually, you still get
        // invalid XHTML. Thanks!
        //
        // Looks like their web interface pretty much sends anything
        // through intact, so.... to do equivalent, decode all entities
        // and then re-encode the special ones.
        return htmlspecialchars(html_entity_decode($str, ENT_COMPAT, 'UTF-8'));
    }

    function makeUrlLink($object, $orig, $html)
    {
        if ($html) {
            return 'expanded_url).'" class="extlink">'.htmlspecialchars($object->display_url).'';
        }else{
            return htmlspecialchars($object->expanded_url);
        }
    }

    function makeHashtagLink($object, $orig)
    {
        return "#" . self::tagLink($object->text, substr($orig, 1));
    }

    function makeMentionLink($object, $orig)
    {
        return "@".self::atLink($object->screen_name, $object->name, substr($orig, 1));
    }

    static function tagLink($tag, $orig)
    {
        return "{$orig}";
    }

    static function atLink($screenName, $fullName, $orig)
    {
        if (!empty($fullName)) {
            return "{$orig}";
        } else {
            return "{$orig}";
        }
    }

    function saveStatusMentions($notice, $status)
    {
        $mentions = array();

        if (empty($status->entities) || empty($status->entities->user_mentions)) {
            return;
        }

        foreach ($status->entities->user_mentions as $mention) {
            $flink = Foreign_link::getByForeignID($mention->id, TWITTER_SERVICE);
            if (!empty($flink)) {
                $user = User::getKV('id', $flink->user_id);
                if (!empty($user)) {
                    $reply = new Reply();
                    $reply->notice_id  = $notice->id;
                    $reply->profile_id = $user->id;
                    $reply->modified   = $notice->created;
                    common_log(LOG_INFO, __METHOD__ . ": saving reply: notice {$notice->id} to profile {$user->id}");
                    $id = $reply->insert();
                }
            }
        }
    }

    /**
     * Record URL links from the notice. Needed to get thumbnail records
     * for referenced photo and video posts, etc.
     *
     * @param Notice $notice
     * @param object $status
     */
    function saveStatusAttachments($notice, $status)
    {
        if (common_config('attachments', 'process_links')) {
            if (!empty($status->entities) && !empty($status->entities->urls)) {
                foreach ($status->entities->urls as $url) {
                    try {
                        File::processNew($url->url, $notice->id);
                    } catch (ServerException $e) {
                        // Could not process attached URL
                    }
                }
            }
        }
    }
}


code:
<?php
/**
 * StatusNet, the distributed open-source microblogging tool
 *
 * Class for doing OAuth calls against Twitter
 *
 * PHP version 5
 *
 * LICENCE: This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * @category  Integration
 * @package   StatusNet
 * @author    Zach Copley <zach@status.net>
 * @copyright 2009-2010 StatusNet, Inc.
 * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
 * @link      http://status.net/
 */

if (!defined('STATUSNET') && !defined('LACONICA')) {
    exit(1);
}

/**
 * Class for talking to the Twitter API with OAuth.
 *
 * @category Integration
 * @package  StatusNet
 * @author   Zach Copley <zach@status.net>
 * @license  http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
 * @link     http://status.net/
 *
 */
class TwitterOAuthClient extends OAuthClient
{
    public static $requestTokenURL = 'https://api.twitter.com/oauth/request_token';
    public static $authorizeURL    = 'https://api.twitter.com/oauth/authorize';
    public static $signinUrl       = 'https://api.twitter.com/oauth/authenticate';
    public static $accessTokenURL  = 'https://api.twitter.com/oauth/access_token';

    /**
     * Constructor
     *
     * @param string $oauth_token        the user's token
     * @param string $oauth_token_secret the user's token secret
     *
     * @return nothing
     */
    function __construct($oauth_token = null, $oauth_token_secret = null)
    {
        $consumer_key    = common_config('twitter', 'consumer_key');
        $consumer_secret = common_config('twitter', 'consumer_secret');

        if (empty($consumer_key) && empty($consumer_secret)) {
            $consumer_key = common_config(
                'twitter',
                'global_consumer_key'
            );
            $consumer_secret = common_config(
                'twitter',
                'global_consumer_secret'
            );
        }

        parent::__construct(
            $consumer_key,
            $consumer_secret,
            $oauth_token,
            $oauth_token_secret
        );
    }

    // XXX: the following two functions are to support the horrible hack
    // of using the credentils field in Foreign_link to store both
    // the access token and token secret.  This hack should go away with
    // 0.9, in which we can make DB changes and add a new column for the
    // token itself.

    static function packToken($token)
    {
        return implode(chr(0), array($token->key, $token->secret));
    }

    static function unpackToken($str)
    {
        $vals = explode(chr(0), $str);
        return new OAuthToken($vals[0], $vals[1]);
    }

    static function isPackedToken($str)
    {
        if (strpos($str, chr(0)) === false) {
            return false;
        } else {
            return true;
        }
    }

    /**
     * Gets a request token from Twitter
     *
     * @return OAuthToken $token the request token
     */
    function getRequestToken()
    {
        return parent::getRequestToken(
            self::$requestTokenURL,
            common_local_url('twitterauthorization')
        );
    }

    /**
     * Builds a link to Twitter's endpoint for authorizing a request token
     *
     * @param OAuthToken $request_token token to authorize
     *
     * @return the link
     */
    function getAuthorizeLink($request_token, $signin = false)
    {
        $url = ($signin) ? self::$signinUrl : self::$authorizeURL;

        return parent::getAuthorizeLink($url,
                                        $request_token,
                                        common_local_url('twitterauthorization'));
    }

    /**
     * Fetches an access token from Twitter
     *
     * @param string $verifier 1.0a verifier
     *
     * @return OAuthToken $token the access token
     */
    function getAccessToken($verifier = null)
    {
        return parent::getAccessToken(
            self::$accessTokenURL,
            $verifier
        );
    }

    /**
     * Calls Twitter's /account/verify_credentials API method
     *
     * @return mixed the Twitter user
     */
    function verifyCredentials()
    {
        $url          = 'https://api.twitter.com/1.1/account/verify_credentials.json';
        $response     = $this->oAuthGet($url);
        $twitter_user = json_decode($response);
        return $twitter_user;
    }

    /**
     * Calls Twitter's /statuses/update API method
     *
     * @param string $status  text of the status
     * @param mixed  $params  optional other parameters to pass to Twitter,
     *                        as defined. For back-compatibility, if an int
     *                        is passed we'll consider it a reply-to ID.
     *
     * @return mixed the status
     */
    function statusesUpdate($status, $params=array())
    {
        $url      = 'https://api.twitter.com/1.1/statuses/update.json';
        if (is_numeric($params)) {
            $params = array('in_reply_to_status_id' => intval($params));
        }
        $params['status'] = $status;
        // We don't have to pass 'source' as the oauth key is tied to an app.

        $response = $this->oAuthPost($url, $params);
        $status   = json_decode($response);
        return $status;
    }

    /**
     * Calls Twitter's /statuses/home_timeline API method
     *
     * @param int    $since_id    show statuses after this id
     * @param string $timelineUri timeline to poll statuses from
     * @param int    $max_id      show statuses before this id
     * @param int    $cnt         number of statuses to show
     * @param int    $page        page number
     *
     * @return mixed an array of statuses
     */
    function statusesTimeline($since_id = null, $timelineUri = 'home_timeline',
                              $max_id = null, $cnt = 200, $page = null)
    {
        $url    = 'https://api.twitter.com/1.1/statuses/'.$timelineUri.'.json';

        $params = array('include_entities' => 'true',
                        'include_rts'      => 'true');

        if (!empty($since_id)) {
            $params['since_id'] = $since_id;
        }
        if (!empty($max_id)) {
            $params['max_id'] = $max_id;
        }
        if (!empty($cnt)) {
            $params['count'] = $cnt;
        }
        if (!empty($page)) {
            $params['page'] = $page;
        }

        $response = $this->oAuthGet($url, $params);
        $statuses = json_decode($response);
        return $statuses;
    }

    /**
     * Calls Twitter's /statuses/friends API method
     *
     * @param int $id          id of the user whom you wish to see friends of
     * @param int $user_id     numerical user id
     * @param int $screen_name screen name
     * @param int $page        page number
     *
     * @return mixed an array of twitter users and their latest status
     */
    function statusesFriends($id = null, $user_id = null, $screen_name = null,
                             $page = null)
    {
        $url = "https://api.twitter.com/1.1/friends/list.json";

        $params = array();

        if (!empty($id)) {
            $params['id'] = $id;
        }

        if (!empty($user_id)) {
            $params['user_id'] = $user_id;
        }

        if (!empty($screen_name)) {
            $params['screen_name'] = $screen_name;
        }

        if (!empty($page)) {
            $params['page'] = $page;
        }

        $response = $this->oAuthGet($url, $params);
        $friends  = json_decode($response);
        return $friends;
    }

    /**
     * Calls Twitter's /statuses/friends/ids API method
     *
     * @param int $id          id of the user whom you wish to see friends of
     * @param int $user_id     numerical user id
     * @param int $screen_name screen name
     * @param int $page        page number
     *
     * @return mixed a list of ids, 100 per page
     */
    function friendsIds($id = null, $user_id = null, $screen_name = null,
                         $page = null)
    {
        $url = "https://api.twitter.com/1.1/friends/ids.json";

        $params = array();

        if (!empty($id)) {
            $params['id'] = $id;
        }

        if (!empty($user_id)) {
            $params['user_id'] = $user_id;
        }

        if (!empty($screen_name)) {
            $params['screen_name'] = $screen_name;
        }

        if (!empty($page)) {
            $params['page'] = $page;
        }

        $response = $this->oAuthGet($url, $params);
        $ids      = json_decode($response);
        return $ids;
    }

    /**
     * Calls Twitter's /statuses/retweet/id.json API method
     *
     * @param int $id id of the notice to retweet
     *
     * @return retweeted status
     */

    function statusesRetweet($id)
    {
        $url = "https://api.twitter.com/1.1/statuses/retweet/$id.json";
        $response = $this->oAuthPost($url);
        $status = json_decode($response);
        return $status;
    }

    /**
     * Calls Twitter's /favorites/create API method
     *
     * @param int $id ID of the status to favorite
     *
     * @return object faved status
     */

    function favoritesCreate($id)
    {
        $url = "https://api.twitter.com/1.1/favorites/create.json";
        $params=array();
        $params['id'] = $id;
        $response = $this->oAuthPost($url, $params);
        $status = json_decode($response);
        return $status;
    }

    /**
     * Calls Twitter's /favorites/destroy API method
     *
     * @param int $id ID of the status to unfavorite
     *
     * @return object unfaved status
     */

    function favoritesDestroy($id)
    {
        $url = "https://api.twitter.com/1.1/favorites/destroy.json";
        $params=array();
        $params['id'] = $id;
        $response = $this->oAuthPost($url,$params);
        $status = json_decode($response);
        return $status;
    }

    /**
     * Calls Twitter's /statuses/destroy API method
     *
     * @param int $id ID of the status to destroy
     *
     * @return object destroyed
     */

    function statusesDestroy($id)
    {
        $url = "https://api.twitter.com/1.1/statuses/destroy/$id.json";
        $response = $this->oAuthPost($url);
        $status = json_decode($response);
        return $status;
    }
}


code:
<?php
/*
 * StatusNet - the distributed open-source microblogging tool
 * Copyright (C) 2008, 2009, StatusNet, Inc.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 */

if (!defined('STATUSNET') && !defined('LACONICA')) { exit(1); }

require_once dirname(__DIR__) . '/twitter.php';

class TwitterQueueHandler extends QueueHandler
{
    function transport()
    {
        return 'twitter';
    }

    function handle($notice)
    {
        $ok = broadcast_twitter($notice);
        return $ok || common_config('twitter', 'ignore_errors');
    }
}


code:
<?php
/**
 * StatusNet, the distributed open-source microblogging tool
 *
 * PHP version 5
 *
 * LICENCE: This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * @category  Plugin
 * @package   StatusNet
 * @author    Brion Vibber <brion@status.net>
 * @copyright 2010 StatusNet, Inc.
 * @license   http://www.fsf.org/licensing/licenses/agpl-3.0.html GNU Affero General Public License version 3.0
 * @link      http://status.net/
 */

/**
 * Multiuser stream listener for Twitter Site Streams API
 * http://dev.twitter.com/pages/site_streams
 *
 * The site streams API allows listening to updates for multiple users.
 * Pass in the user IDs to listen to in via followUser() -- note they
 * must each have a valid OAuth token for the application ID we're
 * connecting as.
 *
 * You'll need to be connecting with the auth keys for the user who
 * owns the application registration.
 *
 * The user each message is destined for will be passed to event handlers
 * in $context['for_user_id'].
 */
class TwitterSiteStream extends TwitterStreamReader
{
    protected $userIds;

    public function __construct(TwitterOAuthClient $auth, $baseUrl='https://sitestream.twitter.com')
    {
        parent::__construct($auth, $baseUrl);
    }

    public function connect($method='2b/site.json')
    {
        $params = array();
        if ($this->userIds) {
            $params['follow'] = implode(',', $this->userIds);
        }
        return parent::connect($method, $params);
    }

    /**
     * Set the users whose home streams should be pulled.
     * They all must have valid oauth tokens for this application.
     *
     * Must be called before connect().
     *
     * @param array $userIds
     */
    function followUsers($userIds)
    {
        $this->userIds = $userIds;
    }

    /**
     * Each message in the site stream tells us which user ID it should be
     * routed to; we'll need that to let the caller know what to do.
     *
     * @param array $data
     */
    function routeMessage(stdClass $data)
    {
        $context = array(
            'source' => 'sitestream',
            'for_user' => $data->for_user
        );
        parent::handleMessage($data->message, $context);
    }
}