SquidPurgeClient.php

Go to the documentation of this file.

<?php
class SquidPurgeClient {
    protected $host;
 
    protected $port;
 
    protected $ip;
 
    protected $readState = 'idle';
 
    protected $writeBuffer = '';
 
    protected $requests = [];
 
    protected $currentRequestIndex;
 
    const EINTR = 4;
    const EAGAIN = 11;
    const EINPROGRESS = 115;
    const BUFFER_SIZE = 8192;
 
    protected $socket;
 
    protected $readBuffer;
 
    protected $bodyRemaining;
 
    public function __construct( $server, $options = [] ) {
        $parts = explode( ':', $server, 2 );
        $this->host = $parts[0];
        $this->port = $parts[1] ?? 80;
    }
 
    protected function getSocket() {
        if ( $this->socket !== null ) {
            return $this->socket;
        }
 
        $ip = $this->getIP();
        if ( !$ip ) {
            $this->log( "DNS error" );
            $this->markDown();
            return false;
        }
        $this->socket = socket_create( AF_INET, SOCK_STREAM, SOL_TCP );
        socket_set_nonblock( $this->socket );
        Wikimedia\suppressWarnings();
        $ok = socket_connect( $this->socket, $ip, $this->port );
        Wikimedia\restoreWarnings();
        if ( !$ok ) {
            $error = socket_last_error( $this->socket );
            if ( $error !== self::EINPROGRESS ) {
                $this->log( "connection error: " . socket_strerror( $error ) );
                $this->markDown();
                return false;
            }
        }
 
        return $this->socket;
    }
 
    public function getReadSocketsForSelect() {
        if ( $this->readState == 'idle' ) {
            return [];
        }
        $socket = $this->getSocket();
        if ( $socket === false ) {
            return [];
        }
        return [ $socket ];
    }
 
    public function getWriteSocketsForSelect() {
        if ( !strlen( $this->writeBuffer ) ) {
            return [];
        }
        $socket = $this->getSocket();
        if ( $socket === false ) {
            return [];
        }
        return [ $socket ];
    }
 
    protected function getIP() {
        if ( $this->ip === null ) {
            if ( IP::isIPv4( $this->host ) ) {
                $this->ip = $this->host;
            } elseif ( IP::isIPv6( $this->host ) ) {
                throw new MWException( '$wgSquidServers does not support IPv6' );
            } else {
                Wikimedia\suppressWarnings();
                $this->ip = gethostbyname( $this->host );
                if ( $this->ip === $this->host ) {
                    $this->ip = false;
                }
                Wikimedia\restoreWarnings();
            }
        }
        return $this->ip;
    }
 
    protected function markDown() {
        $this->close();
        $this->socket = false;
    }
 
    public function close() {
        if ( $this->socket ) {
            Wikimedia\suppressWarnings();
            socket_set_block( $this->socket );
            socket_shutdown( $this->socket );
            socket_close( $this->socket );
            Wikimedia\restoreWarnings();
        }
        $this->socket = null;
        $this->readBuffer = '';
        // Write buffer is kept since it may contain a request for the next socket
    }
 
    public function queuePurge( $url ) {
        global $wgSquidPurgeUseHostHeader;
        $url = CdnCacheUpdate::expand( str_replace( "\n", '', $url ) );
        $request = [];
        if ( $wgSquidPurgeUseHostHeader ) {
            $url = wfParseUrl( $url );
            $host = $url['host'];
            if ( isset( $url['port'] ) && strlen( $url['port'] ) > 0 ) {
                $host .= ":" . $url['port'];
            }
            $path = $url['path'];
            if ( isset( $url['query'] ) && is_string( $url['query'] ) ) {
                $path = wfAppendQuery( $path, $url['query'] );
            }
            $request[] = "PURGE $path HTTP/1.1";
            $request[] = "Host: $host";
        } else {
            $request[] = "PURGE $url HTTP/1.0";
        }
        $request[] = "Connection: Keep-Alive";
        $request[] = "Proxy-Connection: Keep-Alive";
        $request[] = "User-Agent: " . Http::userAgent() . ' ' . __CLASS__;
        // Two ''s to create \r\n\r\n
        $request[] = '';
        $request[] = '';
 
        $this->requests[] = implode( "\r\n", $request );
        if ( $this->currentRequestIndex === null ) {
            $this->nextRequest();
        }
    }
 
    public function isIdle() {
        return strlen( $this->writeBuffer ) == 0 && $this->readState == 'idle';
    }
 
    public function doWrites() {
        if ( !strlen( $this->writeBuffer ) ) {
            return;
        }
        $socket = $this->getSocket();
        if ( !$socket ) {
            return;
        }
 
        if ( strlen( $this->writeBuffer ) <= self::BUFFER_SIZE ) {
            $buf = $this->writeBuffer;
            $flags = MSG_EOR;
        } else {
            $buf = substr( $this->writeBuffer, 0, self::BUFFER_SIZE );
            $flags = 0;
        }
        Wikimedia\suppressWarnings();
        $bytesSent = socket_send( $socket, $buf, strlen( $buf ), $flags );
        Wikimedia\restoreWarnings();
 
        if ( $bytesSent === false ) {
            $error = socket_last_error( $socket );
            if ( $error != self::EAGAIN && $error != self::EINTR ) {
                $this->log( 'write error: ' . socket_strerror( $error ) );
                $this->markDown();
            }
            return;
        }
 
        $this->writeBuffer = substr( $this->writeBuffer, $bytesSent );
    }
 
    public function doReads() {
        $socket = $this->getSocket();
        if ( !$socket ) {
            return;
        }
 
        $buf = '';
        Wikimedia\suppressWarnings();
        $bytesRead = socket_recv( $socket, $buf, self::BUFFER_SIZE, 0 );
        Wikimedia\restoreWarnings();
        if ( $bytesRead === false ) {
            $error = socket_last_error( $socket );
            if ( $error != self::EAGAIN && $error != self::EINTR ) {
                $this->log( 'read error: ' . socket_strerror( $error ) );
                $this->markDown();
                return;
            }
        } elseif ( $bytesRead === 0 ) {
            // Assume EOF
            $this->close();
            return;
        }
 
        $this->readBuffer .= $buf;
        while ( $this->socket && $this->processReadBuffer() === 'continue' );
    }
 
    protected function processReadBuffer() {
        switch ( $this->readState ) {
            case 'idle':
                return 'done';
            case 'status':
            case 'header':
                $lines = explode( "\r\n", $this->readBuffer, 2 );
                if ( count( $lines ) < 2 ) {
                    return 'done';
                }
                if ( $this->readState == 'status' ) {
                    $this->processStatusLine( $lines[0] );
                } else { // header
                    $this->processHeaderLine( $lines[0] );
                }
                $this->readBuffer = $lines[1];
                return 'continue';
            case 'body':
                if ( $this->bodyRemaining !== null ) {
                    if ( $this->bodyRemaining > strlen( $this->readBuffer ) ) {
                        $this->bodyRemaining -= strlen( $this->readBuffer );
                        $this->readBuffer = '';
                        return 'done';
                    } else {
                        $this->readBuffer = substr( $this->readBuffer, $this->bodyRemaining );
                        $this->bodyRemaining = 0;
                        $this->nextRequest();
                        return 'continue';
                    }
                } else {
                    // No content length, read all data to EOF
                    $this->readBuffer = '';
                    return 'done';
                }
            default:
                throw new MWException( __METHOD__ . ': unexpected state' );
        }
    }
 
    protected function processStatusLine( $line ) {
        if ( !preg_match( '!^HTTP/(\d+)\.(\d+) (\d{3}) (.*)$!', $line, $m ) ) {
            $this->log( 'invalid status line' );
            $this->markDown();
            return;
        }
        list( , , , $status, $reason ) = $m;
        $status = intval( $status );
        if ( $status !== 200 && $status !== 404 ) {
            $this->log( "unexpected status code: $status $reason" );
            $this->markDown();
            return;
        }
        $this->readState = 'header';
    }
 
    protected function processHeaderLine( $line ) {
        if ( preg_match( '/^Content-Length: (\d+)$/i', $line, $m ) ) {
            $this->bodyRemaining = intval( $m[1] );
        } elseif ( $line === '' ) {
            $this->readState = 'body';
        }
    }
 
    protected function nextRequest() {
        if ( $this->currentRequestIndex !== null ) {
            unset( $this->requests[$this->currentRequestIndex] );
        }
        if ( count( $this->requests ) ) {
            $this->readState = 'status';
            $this->currentRequestIndex = key( $this->requests );
            $this->writeBuffer = $this->requests[$this->currentRequestIndex];
        } else {
            $this->readState = 'idle';
            $this->currentRequestIndex = null;
            $this->writeBuffer = '';
        }
        $this->bodyRemaining = null;
    }
 
    protected function log( $msg ) {
        wfDebugLog( 'squid', __CLASS__ . " ($this->host): $msg" );
    }
}