forked from GNUsocial/gnu-social
		
	Key changes: * Initialization code moved from common.php to StatusNet class; can now switch configurations during runtime. * As a consequence, configuration files must now be idempotent... Be careful with constant, function or class definitions. * Control structure for daemons/QueueManager/QueueHandler has been refactored; the run loop is now managed by IoMaster run via scripts/queuedaemon.php IoManager subclasses are woken to handle socket input or polling, and may cover multiple sites. * Plugins can implement notice queue handlers more easily by registering a QueueHandler class; no more need to add a daemon. The new QueueDaemon runs from scripts/queuedaemon.php: * This replaces most of the old *handler.php scripts; they've been refactored to the bare handler classes. * Spawns multiple child processes to spread load; defaults to CPU count on Linux and Mac OS X systems, or override with --threads=N * When multithreaded, child processes are automatically respawned on failure. * Threads gracefully shut down and restart when passing a soft memory limit (defaults to 90% of memory_limit), limiting damage from memory leaks. * Support for UDP-based monitoring: http://www.gitorious.org/snqmon Rough control flow diagram: QueueDaemon -> IoMaster -> IoManager QueueManager [listen or poll] -> QueueHandler XmppManager [ping & keepalive] XmppConfirmManager [poll updates] Todo: * Respawning features not currently available running single-threaded. * When running single-site, configuration changes aren't picked up. * New sites or config changes affecting queue subscriptions are not yet handled without a daemon restart. * SNMP monitoring output to integrate with general tools (nagios, ganglia) * Convert XMPP confirmation message sends to use stomp queue instead of polling * Convert xmppdaemon.php to IoManager? * Convert Twitter status, friends import polling daemons to IoManager * Clean up some error reporting and failure modes * May need to adjust queue priorities for best perf in backlog/flood cases Detailed code history available in my daemon-work branch: http://www.gitorious.org/~brion/statusnet/brion-fixes/commits/daemon-work
		
			
				
	
	
		
			134 lines
		
	
	
		
			4.2 KiB
		
	
	
	
		
			PHP
		
	
	
	
	
	
			
		
		
	
	
			134 lines
		
	
	
		
			4.2 KiB
		
	
	
	
		
			PHP
		
	
	
	
	
	
| <?php
 | |
| 
 | |
| /**
 | |
|  * Based on code from Stomp PHP library, working around bugs in the base class.
 | |
|  *
 | |
|  * Original code is copyright 2005-2006 The Apache Software Foundation
 | |
|  * Modifications copyright 2009 StatusNet Inc by Brion Vibber <brion@status.net>
 | |
|  *
 | |
|  * Licensed under the Apache License, Version 2.0 (the "License");
 | |
|  * you may not use this file except in compliance with the License.
 | |
|  * You may obtain a copy of the License at
 | |
|  *
 | |
|  * http://www.apache.org/licenses/LICENSE-2.0
 | |
|  *
 | |
|  * Unless required by applicable law or agreed to in writing, software
 | |
|  * distributed under the License is distributed on an "AS IS" BASIS,
 | |
|  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 | |
|  * See the License for the specific language governing permissions and
 | |
|  * limitations under the License.
 | |
|  */
 | |
| 
 | |
| class LiberalStomp extends Stomp
 | |
| {
 | |
|     /**
 | |
|      * We need to be able to get the socket so advanced daemons can
 | |
|      * do a select() waiting for input both from the queue and from
 | |
|      * other sources such as an XMPP connection.
 | |
|      *
 | |
|      * @return resource
 | |
|      */
 | |
|     function getSocket()
 | |
|     {
 | |
|         return $this->_socket;
 | |
|     }
 | |
| 
 | |
|     /**
 | |
|      * Make socket connection to the server
 | |
|      * We also set the stream to non-blocking mode, since we'll be
 | |
|      * select'ing to wait for updates. In blocking mode it seems
 | |
|      * to get confused sometimes.
 | |
|      *
 | |
|      * @throws StompException
 | |
|      */
 | |
|     protected function _makeConnection ()
 | |
|     {
 | |
|         parent::_makeConnection();
 | |
|         stream_set_blocking($this->_socket, 0);
 | |
|     }
 | |
| 
 | |
|     /**
 | |
|      * Version 1.0.0 of the Stomp library gets confused if messages
 | |
|      * come in too fast over the connection. This version will read
 | |
|      * out as many frames as are ready to be read from the socket.
 | |
|      *
 | |
|      * Modified from Stomp::readFrame()
 | |
|      *
 | |
|      * @return StompFrame False when no frame to read
 | |
|      */
 | |
|     public function readFrames ()
 | |
|     {
 | |
|         if (!$this->hasFrameToRead()) {
 | |
|             return false;
 | |
|         }
 | |
|         
 | |
|         $rb = 1024;
 | |
|         $data = '';
 | |
|         $end = false;
 | |
|         $frames = array();
 | |
| 
 | |
|         do {
 | |
|             // @fixme this sometimes hangs in blocking mode...
 | |
|             // shouldn't we have been idle until we found there's more data?
 | |
|             $read = fread($this->_socket, $rb);
 | |
|             if ($read === false) {
 | |
|                 $this->_reconnect();
 | |
|                 // @fixme this will lose prior items
 | |
|                 return $this->readFrames();
 | |
|             }
 | |
|             $data .= $read;
 | |
|             if (strpos($data, "\x00") !== false) {
 | |
|                 // Frames are null-delimited, but some servers
 | |
|                 // may append an extra \n according to old bug reports.
 | |
|                 $data = str_replace("\x00\n", "\x00", $data);
 | |
|                 $chunks = explode("\x00", $data);
 | |
| 
 | |
|                 $data = array_pop($chunks);
 | |
|                 $frames = array_merge($frames, $chunks);
 | |
|                 if ($data == '') {
 | |
|                     // We're at the end of a frame; stop reading.
 | |
|                     break;
 | |
|                 } else {
 | |
|                     // In the middle of a frame; keep going.
 | |
|                 }
 | |
|             }
 | |
|             // @fixme find out why this len < 2 check was there
 | |
|             //$len = strlen($data);
 | |
|         } while (true);//$len < 2 || $end == false);
 | |
| 
 | |
|         return array_map(array($this, 'parseFrame'), $frames);
 | |
|     }
 | |
|     
 | |
|     /**
 | |
|      * Parse a raw Stomp frame into an object.
 | |
|      * Extracted from Stomp::readFrame()
 | |
|      *
 | |
|      * @param string $data
 | |
|      * @return StompFrame
 | |
|      */
 | |
|     function parseFrame($data)
 | |
|     {
 | |
|         list ($header, $body) = explode("\n\n", $data, 2);
 | |
|         $header = explode("\n", $header);
 | |
|         $headers = array();
 | |
|         $command = null;
 | |
|         foreach ($header as $v) {
 | |
|             if (isset($command)) {
 | |
|                 list ($name, $value) = explode(':', $v, 2);
 | |
|                 $headers[$name] = $value;
 | |
|             } else {
 | |
|                 $command = $v;
 | |
|             }
 | |
|         }
 | |
|         $frame = new StompFrame($command, $headers, trim($body));
 | |
|         if (isset($frame->headers['transformation']) && $frame->headers['transformation'] == 'jms-map-json') {
 | |
|             require_once 'Stomp/Message/Map.php';
 | |
|             return new StompMessageMap($frame);
 | |
|         } else {
 | |
|             return $frame;
 | |
|         }
 | |
|         return $frame;
 | |
|     }
 | |
| }
 | |
| 
 |