forked from GNUsocial/gnu-social
many jabber queue management changes
Added a method to QueueManager to let subclasses do stuff when idle. Needed so that XMPP queue manager can service its message queue. Cleaned up jabber_broadcast_message quite a bit. Use custom joins instead of loop-and-query, should fix some problems with users who are getting messages even after turning off notification. Only build $msg and $entry once, and use the XMPPHP function for messages with a payload, rather than rolling our own. darcs-hash:20080829200352-84dde-427e4ca8c81d4222a36f78e7c580b611ff0bf765.gz
This commit is contained in:
Evan Prodromou
parent
a0f726fb6c
commit
f5fe013657
130
lib/jabber.php
130
lib/jabber.php
@ -25,25 +25,6 @@ require_once('XMPPHP/XMPP.php');
|
||||
|
||||
class Laconica_XMPP extends XMPPHP_XMPP {
|
||||
|
||||
function messageplus($to, $body, $type = 'chat', $subject = null, $payload = null) {
|
||||
$to = htmlspecialchars($to);
|
||||
$body = htmlspecialchars($body);
|
||||
$subject = htmlspecialchars($subject);
|
||||
|
||||
$jid = jabber_daemon_address();
|
||||
|
||||
$out = "<message from='$jid' to='$to' type='$type'>";
|
||||
if($subject) $out .= "<subject>$subject</subject>";
|
||||
$out .= "<body>$body</body>";
|
||||
if($payload) $out .= $payload;
|
||||
$out .= "</message>";
|
||||
|
||||
$cnt = strlen($out);
|
||||
common_log(LOG_DEBUG, "Sending $cnt chars to $to");
|
||||
$this->send($out);
|
||||
common_log(LOG_DEBUG, 'Done.');
|
||||
}
|
||||
|
||||
public function presence($status = null, $show = 'available', $to = null, $type='available', $priority=NULL) {
|
||||
if($type == 'available') $type = '';
|
||||
$to = htmlspecialchars($to);
|
||||
@ -134,7 +115,7 @@ function jabber_send_notice($to, $notice) {
|
||||
}
|
||||
$msg = jabber_format_notice($profile, $notice);
|
||||
$entry = jabber_format_entry($profile, $notice);
|
||||
$conn->messageplus($to, $msg, 'chat', NULL, $entry);
|
||||
$conn->message($to, $msg, 'chat', NULL, $entry);
|
||||
return true;
|
||||
}
|
||||
|
||||
@ -142,6 +123,8 @@ function jabber_send_notice($to, $notice) {
|
||||
|
||||
function jabber_format_entry($profile, $notice) {
|
||||
|
||||
# FIXME: notice url might be remote
|
||||
|
||||
$noticeurl = common_local_url('shownotice',
|
||||
array('notice' => $notice->id));
|
||||
$msg = jabber_format_notice($profile, $notice);
|
||||
@ -168,6 +151,10 @@ function jabber_format_entry($profile, $notice) {
|
||||
$html .= "\n</body>\n";
|
||||
$html .= "\n</html>\n";
|
||||
|
||||
$address = "<addresses xmlns='http://jabber.org/protocol/address'>\n";
|
||||
$address .= "<address type='replyto' jid='" . jabber_daemon_address() . "' />\n";
|
||||
$address .= "</addresses>\n";
|
||||
|
||||
$event = "<event xmlns='http://jabber.org/protocol/pubsub#event'>\n";
|
||||
$event .= "<items xmlns='http://jabber.org/protocol/pubsub' ";
|
||||
$event .= "node='" . common_local_url('public') . "'>\n";
|
||||
@ -175,7 +162,7 @@ function jabber_format_entry($profile, $notice) {
|
||||
$event .= "</items>\n";
|
||||
$event .= "</event>\n";
|
||||
# FIXME: include the pubsub event, too.
|
||||
return $html . $entry;
|
||||
return $html . $entry . $address;
|
||||
# return $entry . "\n" . $event;
|
||||
}
|
||||
|
||||
@ -234,63 +221,54 @@ function jabber_broadcast_notice($notice) {
|
||||
return true;
|
||||
}
|
||||
$profile = Profile::staticGet($notice->profile_id);
|
||||
|
||||
if (!$profile) {
|
||||
common_log(LOG_WARNING, 'Refusing to broadcast notice with ' .
|
||||
'unknown profile ' . common_log_objstring($notice),
|
||||
__FILE__);
|
||||
return false;
|
||||
}
|
||||
$sent_to = array();
|
||||
# First, get users who this is a direct reply to
|
||||
$reply = new Reply();
|
||||
$reply->notice_id = $notice->id;
|
||||
if ($reply->find()) {
|
||||
while ($reply->fetch()) {
|
||||
$user = User::staticGet($reply->profile_id);
|
||||
if ($user && $user->jabber && $user->jabbernotify && $user->jabberreplies) {
|
||||
common_log(LOG_INFO,
|
||||
'Sending reply notice ' . $notice->id . ' to ' . $user->jabber,
|
||||
__FILE__);
|
||||
$success = jabber_send_notice($user->jabber, $notice);
|
||||
if ($success) {
|
||||
# Remember so we don't send twice
|
||||
$sent_to[$user->id] = true;
|
||||
} else {
|
||||
# XXX: Not sure, but I think that's the right thing to do
|
||||
common_log(LOG_WARNING,
|
||||
'Sending reply notice ' . $notice->id . ' to ' . $user->jabber . ' FAILED, cancelling.',
|
||||
__FILE__);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
# Now, get users subscribed to this profile
|
||||
# XXX: use a join here rather than looping through results
|
||||
$sub = new Subscription();
|
||||
$sub->subscribed = $notice->profile_id;
|
||||
|
||||
if ($sub->find()) {
|
||||
while ($sub->fetch()) {
|
||||
$user = User::staticGet($sub->subscriber);
|
||||
if ($user && $user->jabber && $user->jabbernotify && !array_key_exists($user->id,$sent_to)) {
|
||||
common_log(LOG_INFO,
|
||||
'Sending notice ' . $notice->id . ' to ' . $user->jabber,
|
||||
__FILE__);
|
||||
$success = jabber_send_notice($user->jabber, $notice);
|
||||
if ($success) {
|
||||
$sent_to[$user->id] = true;
|
||||
} else {
|
||||
# XXX: Not sure, but I think that's the right thing to do
|
||||
common_log(LOG_WARNING,
|
||||
'Sending notice ' . $notice->id . ' to ' . $user->jabber . ' FAILED, cancelling.',
|
||||
__FILE__);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
$msg = jabber_format_notice($profile, $notice);
|
||||
$entry = jabber_format_entry($profile, $notice);
|
||||
|
||||
$sent_to = array();
|
||||
$conn = jabber_connect();
|
||||
|
||||
# First, get users to whom this is a direct reply
|
||||
$user = new User();
|
||||
$user->query('SELECT user.id, user.jabber ' .
|
||||
'FROM user JOIN reply ON user.id = reply.profile_id ' .
|
||||
'WHERE reply.notice_id = ' . $notice->id . ' ' .
|
||||
'AND user.jabber is not null ' .
|
||||
'AND user.jabbernotify = 1 ' .
|
||||
'AND user.jabberreplies = 1 ');
|
||||
|
||||
while ($user->fetch()) {
|
||||
common_log(LOG_INFO,
|
||||
'Sending reply notice ' . $notice->id . ' to ' . $user->jabber,
|
||||
__FILE__);
|
||||
$conn->message($user->jabber, $msg, 'chat', NULL, $entry);
|
||||
}
|
||||
|
||||
# Now, get users subscribed to this profile
|
||||
|
||||
$user = new User();
|
||||
$user->query('SELECT user.id, user.jabber ' .
|
||||
'FROM user JOIN subscription ON user.id = subscription.subscriber ' .
|
||||
'WHERE subscription.subscribed = ' . $notice->profile_id . ' ' .
|
||||
'AND user.jabber is not null ' .
|
||||
'AND user.jabbernotify = 1 ');
|
||||
|
||||
while ($user->fetch()) {
|
||||
if (!array_key_exists($user->id, $sent_to)) {
|
||||
common_log(LOG_INFO,
|
||||
'Sending notice ' . $notice->id . ' to ' . $user->jabber,
|
||||
__FILE__);
|
||||
$conn->message($user->jabber, $msg, 'chat', NULL, $entry);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
@ -305,11 +283,17 @@ function jabber_public_notice($notice) {
|
||||
# = false? I think not
|
||||
|
||||
if ($public && $notice->is_local) {
|
||||
$msg = jabber_format_notice($profile, $notice);
|
||||
$entry = jabber_format_entry($profile, $notice);
|
||||
|
||||
$sent_to = array();
|
||||
$conn = jabber_connect();
|
||||
|
||||
foreach ($public as $address) {
|
||||
common_log(LOG_INFO,
|
||||
'Sending notice ' . $notice->id . ' to public listener ' . $address,
|
||||
__FILE__);
|
||||
jabber_send_notice($address, $notice);
|
||||
common_log(LOG_INFO,
|
||||
'Sending notice ' . $notice->id . ' to public listener ' . $address,
|
||||
__FILE__);
|
||||
$conn->message($address, $msg, 'chat', NULL, $entry);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -57,7 +57,6 @@ class QueueHandler {
|
||||
|
||||
function handle_queue() {
|
||||
$this->log(LOG_INFO, 'checking for queued notices');
|
||||
$cnt = 0;
|
||||
$transport = $this->transport();
|
||||
do {
|
||||
$qi = Queue_item::top($transport);
|
||||
@ -82,14 +81,23 @@ class QueueHandler {
|
||||
$this->log(LOG_WARNING, 'queue item for notice that does not exist');
|
||||
}
|
||||
$qi->delete();
|
||||
$cnt++;
|
||||
$this->idle();
|
||||
} else {
|
||||
$this->clear_old_claims();
|
||||
sleep(10);
|
||||
$start = microtime();
|
||||
$this->idle();
|
||||
$used = microtime() - $start;
|
||||
if ($used < 5000000) {
|
||||
usleep(5000000 - $used);
|
||||
}
|
||||
}
|
||||
} while (true);
|
||||
}
|
||||
|
||||
function idle() {
|
||||
return true;
|
||||
}
|
||||
|
||||
function clear_old_claims() {
|
||||
$qi = new Queue_item();
|
||||
$qi->transport = $this->transport();
|
||||
|
||||
@ -34,6 +34,8 @@ require_once(INSTALLDIR . '/lib/queuehandler.php');
|
||||
set_error_handler('common_error_handler');
|
||||
|
||||
class XmppQueueHandler extends QueueHandler {
|
||||
|
||||
var $conn = NULL;
|
||||
|
||||
function transport() {
|
||||
return 'jabber';
|
||||
@ -42,15 +44,24 @@ class XmppQueueHandler extends QueueHandler {
|
||||
function start() {
|
||||
# Low priority; we don't want to receive messages
|
||||
$this->conn = jabber_connect($this->_id, NULL, -1);
|
||||
$this->conn->addEventHandler('message', 'forward_message', $this);
|
||||
return !is_null($this->conn);
|
||||
}
|
||||
|
||||
function handle_notice($notice) {
|
||||
return jabber_broadcast_notice($notice);
|
||||
}
|
||||
|
||||
function idle() {
|
||||
# Process the queue for a second
|
||||
$this->conn->processTime(1);
|
||||
}
|
||||
|
||||
function finish() {
|
||||
}
|
||||
|
||||
function forward_message(&$pl) {
|
||||
}
|
||||
}
|
||||
|
||||
mb_internal_encoding('UTF-8');
|
||||
|
||||
Loading…
Reference in New Issue
Block a user