GNUsocial/gnu-social
4
0
Fork 1
Code Issues Releases Wiki Activity

use updated xmpphp library

Browse Source 
darcs-hash:20080902163718-84dde-80ccca8a5ad66b9993ced2419aa76f686d098334.gz
This commit is contained in:
Evan Prodromou 2008-09-02 12:37:18 -04:00
parent 7ba8bce1d3
commit 695da8311a
2 changed files with 78 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
extlib/XMPPHP/XMLStream.php
View File

@ -253,29 +253,36 @@ class XMPPHP_XMLStream {
* @param boolean $sendinit * @param boolean $sendinit
*/ */
public function connect($timeout = 30, $persistent = false, $sendinit = true) { public function connect($timeout = 30, $persistent = false, $sendinit = true) {
$this->disconnected = false; $starttime = time();
$this->sent_disconnect = false;
if($persistent) {
$conflag = STREAM_CLIENT_CONNECT | STREAM_CLIENT_PERSISTENT;
} else {
$conflag = STREAM_CLIENT_CONNECT;
}
$conntype = 'tcp';
if($this->use_ssl) $conntype = 'ssl';
$this->log->log("Connecting to $conntype://{$this->host}:{$this->port}");
try {
$this->socket = @stream_socket_client("$conntype://{$this->host}:{$this->port}", $errno, $errstr, $timeout, $conflag);
} catch (Exception $e) {
throw new XMPPHP_Exception($e->getMessage());
}
if(!$this->socket) {
$this->log->log("Could not connect.", XMPPHP_Log::LEVEL_ERROR);
$this->disconnected = true;
throw new XMPPHP_Exception('Could not connect.'); do {
$this->disconnected = false;
$this->sent_disconnect = false;
if($persistent) {
$conflag = STREAM_CLIENT_CONNECT | STREAM_CLIENT_PERSISTENT;
} else {
$conflag = STREAM_CLIENT_CONNECT;
}
$conntype = 'tcp';
if($this->use_ssl) $conntype = 'ssl';
$this->log->log("Connecting to $conntype://{$this->host}:{$this->port}");
try {
$this->socket = @stream_socket_client("$conntype://{$this->host}:{$this->port}", $errno, $errstr, $timeout, $conflag);
} catch (Exception $e) {
throw new XMPPHP_Exception($e->getMessage());
}
if(!$this->socket) {
$this->log->log("Could not connect.", XMPPHP_Log::LEVEL_ERROR);
$this->disconnected = true;
}
} while (!$this->socket && (time() - $starttime) < $timeout);
if ($this->socket) {
stream_set_blocking($this->socket, 1);
if($sendinit) $this->send($this->stream_start);
} else {
throw new XMPPHP_Exception("Could not connect before timeout.");
} }
stream_set_blocking($this->socket, 1);
if($sendinit) $this->send($this->stream_start);
} }
/** /**
@ -286,6 +293,7 @@ class XMPPHP_XMLStream {
$this->log->log("Reconnecting...", XMPPHP_Log::LEVEL_WARNING); $this->log->log("Reconnecting...", XMPPHP_Log::LEVEL_WARNING);
$this->connect(30, false, false); $this->connect(30, false, false);
$this->reset(); $this->reset();
$this->event('reconnect');
} }
} }
@ -319,12 +327,10 @@ class XMPPHP_XMLStream {
private function __process($maximum=0) { private function __process($maximum=0) {
$this->log->log("__process($maximum)", XMPPHP_Log::LEVEL_VERBOSE);
$remaining = $maximum; $remaining = $maximum;
do { do {
$starttime = microtime(); $starttime = (microtime(true) * 1000000);
$read = array($this->socket); $read = array($this->socket);
$write = array(); $write = array();
$except = array(); $except = array();
@ -338,27 +344,25 @@ class XMPPHP_XMLStream {
$usecs = $remaining % 1000000; $usecs = $remaining % 1000000;
$secs = floor(($remaining - $usecs) / 1000000); $secs = floor(($remaining - $usecs) / 1000000);
} }
$this->log->log("stream_select(read, write, except, $secs, $usecs)", XMPPHP_Log::LEVEL_VERBOSE);
$updated = @stream_select($read, $write, $except, $secs, $usecs); $updated = @stream_select($read, $write, $except, $secs, $usecs);
if ($updated === false) { if ($updated === false) {
$this->log->log("Error on stream_select()", XMPPHP_Log::LEVEL_VERBOSE); $this->log->log("Error on stream_select()", XMPPHP_Log::LEVEL_VERBOSE);
if ($this->reconnect) { if ($this->reconnect) {
$this->doReconnect(); $this->doReconnect();
} else { } else {
$this->log->log("Giving up", XMPPHP_Log::LEVEL_VERBOSE);
fclose($this->socket); fclose($this->socket);
$this->socket = NULL;
return false; return false;
} }
} else if ($updated > 0) { } else if ($updated > 0) {
# XXX: Is this big enough? # XXX: Is this big enough?
$this->log->log("Reading from socket", XMPPHP_Log::LEVEL_VERBOSE);
$buff = @fread($this->socket, 4096); $buff = @fread($this->socket, 4096);
if(!$buff) { if(!$buff) {
if($this->reconnect) { if($this->reconnect) {
$this->doReconnect(); $this->doReconnect();
} else { } else {
$this->log->log("Error on fread(), reconnect", XMPPHP_Log::LEVEL_VERBOSE);
fclose($this->socket); fclose($this->socket);
$this->socket = NULL;
return false; return false;
} }
} }
@ -367,7 +371,9 @@ class XMPPHP_XMLStream {
} else { } else {
# $updated == 0 means no changes during timeout. # $updated == 0 means no changes during timeout.
} }
$remaining -= (microtime() - $starttime); $endtime = (microtime(true)*1000000);
$time_past = $endtime - $starttime;
$remaining = $remaining - $time_past;
} while (is_null($maximum) || $remaining > 0); } while (is_null($maximum) || $remaining > 0);
return true; return true;
} }
@ -595,34 +601,47 @@ class XMPPHP_XMLStream {
* *
* @param string $msg * @param string $msg
*/ */
public function send($msg, $rec=false) { public function send($msg, $timeout=NULL) {
if($this->time() - $this->last_send < .1) {
usleep(100000); if (is_null($timeout)) {
$secs = NULL;
$usecs = NULL;
} else if ($timeout == 0) {
$secs = 0;
$usecs = 0;
} else {
$maximum = $timeout * 1000000;
$usecs = $maximum % 1000000;
$secs = floor(($maximum - $usecs) / 1000000);
} }
$wait = true;
while($wait) { $read = array();
$read = null; $write = array($this->socket);
$write = array($this->socket); $except = array();
$except = null;
$select = @stream_select($read, $write, $except, 0, 0); $select = @stream_select($read, $write, $except, $secs, $usecs);
if($select === False) {
$this->doReconnect(); if($select === False) {
return false; $this->log->log("ERROR sending message; reconnecting.");
} elseif ($select > 0) { $this->doReconnect();
$wait = false; # TODO: retry send here
} else { return false;
usleep(100000); } elseif ($select > 0) {
//$this->processTime(.25); $this->log->log("Socket is ready; send it.");
} } else {
$this->log->log("Socket is not ready; break.");
return false;
} }
$sentbytes = @fwrite($this->socket, $msg, 1024);
$this->last_send = $this->time(); $sentbytes = @fwrite($this->socket, $msg);
$this->log->log("SENT: " . mb_substr($msg, 0, $sentbytes, '8bit'), XMPPHP_Log::LEVEL_VERBOSE); $this->log->log("SENT: " . mb_substr($msg, 0, $sentbytes, '8bit'), XMPPHP_Log::LEVEL_VERBOSE);
if($sentbytes === FALSE) { if($sentbytes === FALSE) {
$this->log->log("ERROR sending message; reconnecting.");
$this->doReconnect(); $this->doReconnect();
} elseif ($sentbytes != mb_strlen($msg, '8bit')) { return false;
$this->send(mb_substr($msg, $sentbytes, mb_strlen($msg, '8bit'), '8bit'), true);
} }
$this->log->log("Successfully sent $sentbytes bytes.");
return $sentbytes;
} }
public function time() { public function time() {

10
extlib/XMPPHP/XMPP.php
View File

@ -144,6 +144,12 @@ class XMPPHP_XMPP extends XMPPHP_XMLStream {
* @param string $subject * @param string $subject
*/ */
public function message($to, $body, $type = 'chat', $subject = null, $payload = null) { public function message($to, $body, $type = 'chat', $subject = null, $payload = null) {
if(is_null($type))
{
$type = 'chat';
}
$to = htmlspecialchars($to); $to = htmlspecialchars($to);
$body = htmlspecialchars($body); $body = htmlspecialchars($body);
$subject = htmlspecialchars($subject); $subject = htmlspecialchars($subject);
@ -154,7 +160,7 @@ class XMPPHP_XMPP extends XMPPHP_XMLStream {
if($payload) $out .= $payload; if($payload) $out .= $payload;
$out .= "</message>"; $out .= "</message>";
$this->send($out); return $this->send($out);
} }
/** /**
@ -183,7 +189,7 @@ class XMPPHP_XMPP extends XMPPHP_XMLStream {
$out .= "</presence>"; $out .= "</presence>";
} }
$this->send($out); return $this->send($out);
} }
/** /**