Bläddra i källkod

PuSH: Publishing moved to a new process. OStatus comments are now published to all subscribers.

tags/3.4.1^2
Michael Vogel 5 år sedan
förälder
incheckning
e66489d67f
6 ändrade filer med 184 tillägg och 100 borttagningar
  1. +1
    -1
      include/delivery.php
  2. +16
    -4
      include/items.php
  3. +36
    -21
      include/notifier.php
  4. +114
    -0
      include/pubsubpublish.php
  5. +4
    -61
      include/queue.php
  6. +13
    -13
      include/salmon.php

+ 1
- 1
include/delivery.php Visa fil

@@ -388,7 +388,7 @@ function delivery_run(&$argv, &$argc){
continue;

// For OStatus don't notify all contacts in the thread
if (!$top_level AND ($parent["network"] == NETWORK_OSTATUS) AND ($item["id"] != $item["parent"]))
if (!$top_level AND ($parent["network"] == NETWORK_OSTATUS) AND ($item["id"] != $item_id))
continue;

if(($top_level OR ($parent["network"] == NETWORK_OSTATUS)) && ($public_message) && ($item['author-link'] === $item['owner-link']) && (! $expire))


+ 16
- 4
include/items.php Visa fil

@@ -14,7 +14,7 @@ require_once('include/threads.php');
require_once('include/socgraph.php');
require_once('mod/share.php');

function get_feed_for(&$a, $dfrn_id, $owner_nick, $last_update, $direction = 0) {
function get_feed_for(&$a, $dfrn_id, $owner_nick, $last_update, $direction = 0, $forpubsub = false) {


$sitefeed = ((strlen($owner_nick)) ? false : true); // not yet implemented, need to rewrite huge chunks of following logic
@@ -55,6 +55,7 @@ function get_feed_for(&$a, $dfrn_id, $owner_nick, $last_update, $direction = 0)
$birthday = feed_birthday($owner_id,$owner['timezone']);

$sql_post_table = "";
$visibility = "";

if(! $public_feed) {

@@ -114,6 +115,17 @@ function get_feed_for(&$a, $dfrn_id, $owner_nick, $last_update, $direction = 0)
else
$sort = 'ASC';

// Include answers to status.net posts in public feeds
if($forpubsub) {
$sql_post_table = "INNER JOIN `thread` ON `thread`.`iid` = `item`.`parent` ";
$visibility = "OR (`item`.`network` = 'dfrn' AND `thread`.`network`='stat')";
$date_field = "`received`";
$sql_order = "`item`.`received` DESC";
} else {
$date_field = "`changed`";
$sql_order = "`item`.`parent` ".$sort.", `item`.`created` ASC";
}

if(! strlen($last_update))
$last_update = 'now -30 days';

@@ -133,7 +145,7 @@ function get_feed_for(&$a, $dfrn_id, $owner_nick, $last_update, $direction = 0)
// AND ( `item`.`edited` > '%s' OR `item`.`changed` > '%s' )
// dbesc($check_date),

$r = q("SELECT `item`.*, `item`.`id` AS `item_id`,
$r = q("SELECT STRAIGHT_JOIN `item`.*, `item`.`id` AS `item_id`,
`contact`.`name`, `contact`.`network`, `contact`.`photo`, `contact`.`url`,
`contact`.`name-date`, `contact`.`uri-date`, `contact`.`avatar-date`,
`contact`.`thumb`, `contact`.`dfrn-id`, `contact`.`self`,
@@ -144,9 +156,9 @@ function get_feed_for(&$a, $dfrn_id, $owner_nick, $last_update, $direction = 0)
AND `contact`.`blocked` = 0 AND `contact`.`pending` = 0
LEFT JOIN `sign` ON `sign`.`iid` = `item`.`id`
WHERE `item`.`uid` = %d AND `item`.`visible` = 1 and `item`.`moderated` = 0 AND `item`.`parent` != 0
AND `item`.`wall` = 1 AND `item`.`changed` > '%s'
AND ((`item`.`wall` = 1) $visibility) AND `item`.$date_field > '%s'
$sql_extra
ORDER BY `parent` %s, `created` ASC LIMIT 0, 300",
ORDER BY $sql_order LIMIT 0, 300",
intval($owner_id),
dbesc($check_date),
dbesc($sort)


+ 36
- 21
include/notifier.php Visa fil

@@ -221,6 +221,9 @@ function notifier_run(&$argv, &$argc){
// If this is a public conversation, notify the feed hub
$public_message = true;

// Do a PuSH
$push_notify = false;

// fill this in with a single salmon slap if applicable
$slap = '';

@@ -296,7 +299,11 @@ function notifier_run(&$argv, &$argc){
$recipients = array($parent['contact-id']);

if ($parent['network'] == NETWORK_OSTATUS) {
$ostatus_recipients = array();
logger('Parent is OStatus', LOGGER_DEBUG);

$push_notify = true;

/* $ostatus_recipients = array();

$r = q("SELECT * FROM `contact` WHERE `uid` = %d AND `network` = '%s'", intval($uid), dbesc(NETWORK_OSTATUS));
if(count($r)) {
@@ -305,7 +312,7 @@ function notifier_run(&$argv, &$argc){

$ostatus_recip_str = ", ".implode(', ', $ostatus_recipients);
}
*/
// Check if the recipient isn't in your contact list
$r = q("SELECT `url` FROM `contact` WHERE `id` = %d", $parent['contact-id']);
if (count($r)) {
@@ -978,32 +985,40 @@ function notifier_run(&$argv, &$argc){
}
}

$push_notify = true;

if(strlen($hub)) {
$hubs = explode(',', $hub);
if(count($hubs)) {
foreach($hubs as $h) {
$h = trim($h);
if(! strlen($h))
continue;
}

if ($h === '[internal]') {
// Set push flag for PuSH subscribers to this topic,
// they will be notified in queue.php
q("UPDATE `push_subscriber` SET `push` = 1 " .
"WHERE `nickname` = '%s'", dbesc($owner['nickname']));
} else {

$params = 'hub.mode=publish&hub.url=' . urlencode( $a->get_baseurl() . '/dfrn_poll/' . $owner['nickname'] );
post_url($h,$params);
logger('pubsub: publish: ' . $h . ' ' . $params . ' returned ' . $a->get_curl_code());
}
if(count($hubs) > 1)
sleep(7); // try and avoid multiple hubs responding at precisely the same time
if($push_notify AND strlen($hub)) {
$hubs = explode(',', $hub);
if(count($hubs)) {
foreach($hubs as $h) {
$h = trim($h);
if(! strlen($h))
continue;

if ($h === '[internal]') {
// Set push flag for PuSH subscribers to this topic,
// they will be notified in queue.php
q("UPDATE `push_subscriber` SET `push` = 1 " .
"WHERE `nickname` = '%s'", dbesc($owner['nickname']));

logger('Activating internal PuSH for item '.$item_id, LOGGER_DEBUG);

} else {

$params = 'hub.mode=publish&hub.url=' . urlencode( $a->get_baseurl() . '/dfrn_poll/' . $owner['nickname'] );
post_url($h,$params);
logger('publish for item '.$item_id.' ' . $h . ' ' . $params . ' returned ' . $a->get_curl_code());
}
if(count($hubs) > 1)
sleep(7); // try and avoid multiple hubs responding at precisely the same time
}
}

// Handling the pubsubhubbub requests
proc_run('php','include/pubsubpublish.php');
}

// If the item was deleted, clean up the `sign` table


+ 114
- 0
include/pubsubpublish.php Visa fil

@@ -0,0 +1,114 @@
<?php
require_once("boot.php");

function handle_pubsubhubbub() {
global $a, $db;

logger('start');

// We'll push to each subscriber that has push > 0,
// i.e. there has been an update (set in notifier.php).

$r = q("SELECT * FROM `push_subscriber` WHERE `push` > 0");

foreach($r as $rr) {
$params = get_feed_for($a, '', $rr['nickname'], $rr['last_update'], 0, true);
$hmac_sig = hash_hmac("sha1", $params, $rr['secret']);

$headers = array("Content-type: application/atom+xml",
sprintf("Link: <%s>;rel=hub," .
"<%s>;rel=self",
$a->get_baseurl() . '/pubsubhubbub',
$rr['topic']),
"X-Hub-Signature: sha1=" . $hmac_sig);

logger('POST '. print_r($headers, true)."\n".$params, LOGGER_DEBUG);

post_url($rr['callback_url'], $params, $headers);
$ret = $a->get_curl_code();

if ($ret >= 200 && $ret <= 299) {
logger('successfully pushed to '.$rr['callback_url']);

// set last_update to "now", and reset push=0
$date_now = datetime_convert('UTC','UTC','now','Y-m-d H:i:s');
q("UPDATE `push_subscriber` SET `push` = 0, last_update = '%s' WHERE id = %d",
dbesc($date_now),
intval($rr['id']));

} else {
logger('error when pushing to '.$rr['callback_url'].' HTTP: '.$ret);

// we use the push variable also as a counter, if we failed we
// increment this until some upper limit where we give up
$new_push = intval($rr['push']) + 1;

if ($new_push > 30) // OK, let's give up
$new_push = 0;

q("UPDATE `push_subscriber` SET `push` = %d, last_update = '%s' WHERE id = %d",
$new_push,
dbesc($date_now),
intval($rr['id']));
}
}

logger('done');
}


function pubsubpublish_run(&$argv, &$argc){
global $a, $db;

if(is_null($a)){
$a = new App;
}

if(is_null($db)){
@include(".htconfig.php");
require_once("include/dba.php");
$db = new dba($db_host, $db_user, $db_pass, $db_data);
unset($db_host, $db_user, $db_pass, $db_data);
};

require_once('include/items.php');
require_once('include/pidfile.php');

load_config('config');
load_config('system');

$lockpath = get_lockpath();
if ($lockpath != '') {
$pidfile = new pidfile($lockpath, 'pubsubpublish');
if($pidfile->is_already_running()) {
logger("Already running");
if ($pidfile->running_time() > 9*60) {
$pidfile->kill();
logger("killed stale process");
// Calling a new instance
proc_run('php',"include/pubsubpublish.php");
}
return;
}
}

$a->set_baseurl(get_config('system','url'));

load_hooks();

if($argc > 1)
$pubsubpublish_id = intval($argv[1]);
else
$pubsubpublish_id = 0;

handle_pubsubhubbub();

return;

}

if (array_search(__file__,get_included_files())===0){
pubsubpublish_run($_SERVER["argv"],$_SERVER["argc"]);
killme();
}


+ 4
- 61
include/queue.php Visa fil

@@ -2,64 +2,6 @@
require_once("boot.php");
require_once('include/queue_fn.php');

function handle_pubsubhubbub() {
global $a, $db;

logger('queue [pubsubhubbub]: start');

// We'll push to each subscriber that has push > 0,
// i.e. there has been an update (set in notifier.php).

$r = q("SELECT * FROM `push_subscriber` WHERE `push` > 0");

foreach($r as $rr) {
$params = get_feed_for($a, '', $rr['nickname'], $rr['last_update']);
$hmac_sig = hash_hmac("sha1", $params, $rr['secret']);

$headers = array("Content-type: application/atom+xml",
sprintf("Link: <%s>;rel=hub," .
"<%s>;rel=self",
$a->get_baseurl() . '/pubsubhubbub',
$rr['topic']),
"X-Hub-Signature: sha1=" . $hmac_sig);

logger('queue [pubsubhubbub]: POST', $headers);

post_url($rr['callback_url'], $params, $headers);
$ret = $a->get_curl_code();

if ($ret >= 200 && $ret <= 299) {
logger('queue [pubsubhubbub]: successfully pushed to ' .
$rr['callback_url']);

// set last_update to "now", and reset push=0
$date_now = datetime_convert('UTC','UTC','now','Y-m-d H:i:s');
q("UPDATE `push_subscriber` SET `push` = 0, last_update = '%s' " .
"WHERE id = %d",
dbesc($date_now),
intval($rr['id']));

} else {
logger('queue [pubsubhubbub]: error when pushing to ' .
$rr['callback_url'] . 'HTTP: ', $ret);

// we use the push variable also as a counter, if we failed we
// increment this until some upper limit where we give up
$new_push = intval($rr['push']) + 1;

if ($new_push > 30) // OK, let's give up
$new_push = 0;

q("UPDATE `push_subscriber` SET `push` = %d, last_update = '%s' " .
"WHERE id = %d",
$new_push,
dbesc($date_now),
intval($rr['id']));
}
}
}


function queue_run(&$argv, &$argc){
global $a, $db;

@@ -112,7 +54,8 @@ function queue_run(&$argv, &$argc){

logger('queue: start');

handle_pubsubhubbub();
// Handling the pubsubhubbub requests
proc_run('php','include/pubsubpublish.php');

$interval = ((get_config('system','delivery_interval') === false) ? 2 : intval(get_config('system','delivery_interval')));

@@ -126,8 +69,8 @@ function queue_run(&$argv, &$argc){
}
}

$r = q("SELECT `queue`.*, `contact`.`name`, `contact`.`uid` FROM `queue`
INNER JOIN `contact` ON `queue`.`cid` = `contact`.`id`
$r = q("SELECT `queue`.*, `contact`.`name`, `contact`.`uid` FROM `queue`
INNER JOIN `contact` ON `queue`.`cid` = `contact`.`id`
WHERE `queue`.`created` < UTC_TIMESTAMP() - INTERVAL 3 DAY");
if($r) {
foreach($r as $rr) {


+ 13
- 13
include/salmon.php Visa fil

@@ -62,13 +62,13 @@ function get_salmon_key($uri,$keyhash) {
return '';
}

function slapper($owner,$url,$slap) {

logger('slapper called for '.$owner['url'].' to '.$url.' . Data: ' . $slap);

// does contact have a salmon endpoint?
// does contact have a salmon endpoint?

if(! strlen($url))
return;
@@ -87,16 +87,16 @@ $namespaces = <<< EOT
xmlns:thr="http://purl.org/syndication/thread/1.0"
xmlns:at="http://purl.org/atompub/tombstones/1.0"
xmlns:media="http://purl.org/syndication/atommedia"
xmlns:dfrn="http://purl.org/macgirvin/dfrn/1.0"
xmlns:dfrn="http://purl.org/macgirvin/dfrn/1.0"
xmlns:as="http://activitystrea.ms/spec/1.0/"
xmlns:georss="http://www.georss.org/georss"
xmlns:poco="http://portablecontacts.net/spec/1.0"
xmlns:ostatus="http://ostatus.org/schema/1.0"
xmlns:georss="http://www.georss.org/georss"
xmlns:poco="http://portablecontacts.net/spec/1.0"
xmlns:ostatus="http://ostatus.org/schema/1.0"
xmlns:statusnet="http://status.net/schema/api/1/" > >
EOT;

$slap = str_replace('<entry>',$namespaces,$slap);
// create a magic envelope

$data = base64url_encode($slap);
@@ -125,7 +125,7 @@ EOT;
'$signature' => $signature
));

// slap them
// slap them
post_url($url,$salmon, array(
'Content-type: application/magic-envelope+xml',
'Content-length: ' . strlen($salmon)
@@ -152,7 +152,7 @@ EOT;
'$signature' => $signature2
));

// slap them
// slap them
post_url($url,$salmon, array(
'Content-type: application/magic-envelope+xml',
'Content-length: ' . strlen($salmon)
@@ -166,7 +166,7 @@ EOT;

// Entirely likely that their salmon implementation is
// non-compliant. Let's try once more, this time only signing
// the data, without the precomputed blob
// the data, without the precomputed blob

$salmon = replace_macros($salmon_tpl,array(
'$data' => $data,
@@ -176,7 +176,7 @@ EOT;
'$signature' => $signature3
));

// slap them
// slap them
post_url($url,$salmon, array(
'Content-type: application/magic-envelope+xml',
'Content-length: ' . strlen($salmon)
@@ -184,7 +184,7 @@ EOT;
$return_code = $a->get_curl_code();
}
}
logger('slapper returned ' . $return_code);
logger('slapper returned ' . $return_code);
if(! $return_code)
return(-1);
if(($return_code == 503) && (stristr($a->get_curl_headers(),'retry-after')))


Laddar…
Avbryt
Spara