123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277 |
- <?php
- if (!defined('STATUSNET')) {
- exit(1);
- }
- class OStatusQueueHandler extends QueueHandler
- {
-
-
-
-
- const MAX_UNBATCHED = 50;
-
-
-
-
- const BATCH_SIZE = 1000;
- function transport()
- {
- return 'ostatus';
- }
- function handle($notice)
- {
- assert($notice instanceof Notice);
- $this->notice = $notice;
- $this->user = User::getKV('id', $notice->profile_id);
- try {
- $profile = $this->notice->getProfile();
- } catch (Exception $e) {
- common_log(LOG_ERR, "Can't get profile for notice; skipping: " . $e->getMessage());
- return true;
- }
- if ($notice->isLocal()) {
-
-
- $this->pushUser();
- }
- foreach ($notice->getGroups() as $group) {
- $oprofile = Ostatus_profile::getKV('group_id', $group->id);
- if ($oprofile) {
-
- if ($notice->isLocal()) {
- $this->pingReply($oprofile);
- }
- } else {
-
- $this->pushGroup($group->id);
- }
- }
- if ($notice->isLocal()) {
-
-
- foreach ($notice->getReplies() as $profile_id) {
- $oprofile = Ostatus_profile::getKV('profile_id', $profile_id);
- if ($oprofile) {
- $this->pingReply($oprofile);
- }
- }
- if (!empty($this->notice->reply_to)) {
- $replyTo = Notice::getKV('id', $this->notice->reply_to);
- if (!empty($replyTo)) {
- foreach($replyTo->getReplies() as $profile_id) {
- $oprofile = Ostatus_profile::getKV('profile_id', $profile_id);
- if ($oprofile) {
- $this->pingReply($oprofile);
- }
- }
- }
- }
- foreach ($notice->getProfileTags() as $ptag) {
- $oprofile = Ostatus_profile::getKV('peopletag_id', $ptag->id);
- if (!$oprofile) {
- $this->pushPeopletag($ptag);
- }
- }
- }
- return true;
- }
- function pushUser()
- {
- if ($this->user) {
-
-
- $feed = common_local_url('ApiTimelineUser',
- array('id' => $this->user->id,
- 'format' => 'atom'));
- $this->pushFeed($feed, array($this, 'userFeedForNotice'));
- }
- }
- function pushGroup($group_id)
- {
-
-
- $feed = common_local_url('ApiTimelineGroup',
- array('id' => $group_id,
- 'format' => 'atom'));
- $this->pushFeed($feed, array($this, 'groupFeedForNotice'), $group_id);
- }
- function pushPeopletag($ptag)
- {
-
-
- $feed = common_local_url('ApiTimelineList',
- array('id' => $ptag->id,
- 'user' => $ptag->tagger,
- 'format' => 'atom'));
- $this->pushFeed($feed, array($this, 'peopletagFeedForNotice'), $ptag);
- }
- function pingReply($oprofile)
- {
- if ($this->user) {
-
-
-
-
- $oprofile->notifyDeferred($this->notice, $this->user);
- }
- }
-
- function pushFeed($feed, $callback)
- {
- $hub = common_config('ostatus', 'hub');
- if ($hub) {
- $this->pushFeedExternal($feed, $hub);
- }
- $sub = new HubSub();
- $sub->topic = $feed;
- if ($sub->find()) {
- $args = array_slice(func_get_args(), 2);
- $atom = call_user_func_array($callback, $args);
- $this->pushFeedInternal($atom, $sub);
- } else {
- common_log(LOG_INFO, "No PuSH subscribers for $feed");
- }
- return true;
- }
-
- function pushFeedExternal($feed, $hub)
- {
- $client = new HTTPClient();
- try {
- $data = array('hub.mode' => 'publish',
- 'hub.url' => $feed);
- $response = $client->post($hub, array(), $data);
- if ($response->getStatus() == 204) {
- common_log(LOG_INFO, "PuSH ping to hub $hub for $feed ok");
- return true;
- } else {
- common_log(LOG_ERR, "PuSH ping to hub $hub for $feed failed with HTTP " .
- $response->getStatus() . ': ' .
- $response->getBody());
- }
- } catch (Exception $e) {
- common_log(LOG_ERR, "PuSH ping to hub $hub for $feed failed: " . $e->getMessage());
- return false;
- }
- }
-
- function pushFeedInternal($atom, $sub)
- {
- common_log(LOG_INFO, "Preparing $sub->N PuSH distribution(s) for $sub->topic");
- $n = 0;
- $batch = array();
- while ($sub->fetch()) {
- $n++;
- if ($n < self::MAX_UNBATCHED) {
- $sub->distribute($atom);
- } else {
- $batch[] = $sub->callback;
- if (count($batch) >= self::BATCH_SIZE) {
- $sub->bulkDistribute($atom, $batch);
- $batch = array();
- }
- }
- }
- if (count($batch) >= 0) {
- $sub->bulkDistribute($atom, $batch);
- }
- }
-
- function userFeedForNotice()
- {
- $atom = new AtomUserNoticeFeed($this->user);
- $atom->addEntryFromNotice($this->notice);
- $feed = $atom->getString();
- return $feed;
- }
- function groupFeedForNotice($group_id)
- {
- $group = User_group::getKV('id', $group_id);
- $atom = new AtomGroupNoticeFeed($group);
- $atom->addEntryFromNotice($this->notice);
- $feed = $atom->getString();
- return $feed;
- }
- function peopletagFeedForNotice($ptag)
- {
- $atom = new AtomListNoticeFeed($ptag);
- $atom->addEntryFromNotice($this->notice);
- $feed = $atom->getString();
- return $feed;
- }
- }
|