Search.php

<?php

namespace App\Jobs;

use App\Jobs\Job;
use Illuminate\Queue\SerializesModels;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Http\Request;
use Redis;

class Search extends Job implements ShouldQueue
{
    use InteractsWithQueue, SerializesModels;
    
    protected $hash, $host, $port, $name, $getString, $useragent, $fp, $sumaFile;
    protected $buffer_length = 8192;

    /**
     * Create a new job instance.
     *
     * @return void
     */
    public function __construct($hash, $host, $port, $name, $getString, $useragent, $sumaFile)
    {
        $this->hash = $hash;
        $this->host = $host;
        $this->port = $port;
        $this->name = $name;
        $this->getString = $getString;
        $this->useragent = $useragent;
        $this->sumaFile = $sumaFile;
    }

    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle(Request $request)
    {
        $this->fp = $this->getFreeSocket();
        
        if(!$this->fp)
        {
            $this->disable($this->sumaFile, "Die Suchmaschine " . $this->name . " wurde für 1h deaktiviert, weil keine Verbindung aufgebaut werden konnte");
        }else
        {
            if($this->writeRequest())
            {
                $this->readAnswer();
            }
        }
    }

    public function disable($sumaFile, $message)
    {
        $xml = simplexml_load_file($sumaFile);
        $xml->xpath("//sumas/suma[@name='" . $this->name . "']")['0']['disabled'] = date(DATE_RFC822, mktime(date("H")+1,date("i"), date("s"), date("m"), date("d"), date("Y")));
        $xml->saveXML($sumaFile);
    }

    private function readAnswer ()
    {
        $time = microtime(true);
        $headers = '';
        $body = '';
        $length = 0;

        if(!$this->fp)
        {
            return;
        }

        // get headers FIRST
        $c = 0;
        stream_set_blocking($this->fp, 1);
        do
        {
            // use fgets() not fread(), fgets stops reading at first newline
            // or buffer which ever one is reached first
            $data = fgets($this->fp, 8192);
            // a sincle CRLF indicates end of headers
            if ($data === false || $data == "\r\n" || feof($this->fp) ) {
                // break BEFORE OUTPUT
                break;
            }
            if( sizeof(($tmp = explode(": ", $data))) === 2 )
                $headers[trim($tmp[0])] = trim($tmp[1]);
            $c++;
        }
        while (true);

        // end of headers
        if(sizeof($headers) > 1){
            $bodySize = 0;
            if( isset($headers["Transfer-Encoding"]) && $headers["Transfer-Encoding"] === "chunked" )
            {
                $body = $this->readChunked();
                
            }elseif( isset($headers['Content-Length']) )
            {
                $length = trim($headers['Content-Length']);
                if(is_numeric($length) && $length >= 1)
                    $body = $this->readBody($length);
                $bodySize = strlen($body);
            }else
            {
                exit;
            }
        }else
        {
            return;
        }

        Redis::del($this->host . "." . $this->socketNumber);
        if( isset($headers["Content-Encoding"]) && $headers['Content-Encoding'] === "gzip")
        {
            $body = $this->gunzip($body);
        }
        Redis::hset('search.' . $this->hash, $this->name, $body);
        Redis::expire('search.' . $this->hash, 5);
    }

    private function readBody($length)
    {
        $theData = '';
        $done = false;
        stream_set_blocking($this->fp, 0);
        $startTime = time();
        $lastTime = $startTime;
        while (!feof($this->fp) && !$done && (($startTime + 1) > time()) && $length !== 0)
        {
            usleep(100);
            $theNewData = fgets($this->fp, 8192);
            $theData .= $theNewData;
            $length -= strlen($theNewData);
            $done = (trim($theNewData) === '0');

        }
        return $theData;
    }

    private function readChunked()
    {
        $body = '';
        // read from chunked stream
        // loop though the stream
        do
        {
            // NOTE: for chunked encoding to work properly make sure
            // there is NOTHING (besides newlines) before the first hexlength

            // get the line which has the length of this chunk (use fgets here)
            $line = fgets($this->fp, 8192);

            // if it's only a newline this normally means it's read
            // the total amount of data requested minus the newline
            // continue to next loop to make sure we're done
            if ($line == "\r\n") {
                continue;
            }

            // the length of the block is sent in hex decode it then loop through
            // that much data get the length
            // NOTE: hexdec() ignores all non hexadecimal chars it finds
            $length = hexdec($line);

            if (!is_int($length)) {
                trigger_error('Most likely not chunked encoding', E_USER_ERROR);
            }

            // zero is sent when at the end of the chunks
            // or the end of the stream or error
            if ($line === false || $length < 1 || feof($this->fp)) {
                if($length <= 0)
                        fgets($this->fp, 8192);
                // break out of the streams loop
                break;
            }

            // loop though the chunk
            do
            {
                // read $length amount of data
                // (use fread here)
                $data = fread($this->fp, $length);

                // remove the amount received from the total length on the next loop
                // it'll attempt to read that much less data
                $length -= strlen($data);

                // PRINT out directly
                // you could also save it directly to a file here

                // store in string for later use
                $body .= $data;

                // zero or less or end of connection break
                if ($length <= 0 || feof($this->fp))
                {
                    // break out of the chunk loop
                    if($length <= 0)
                        fgets($this->fp, 8192);
                    break;
                }
            }
            while (true);
            // end of chunk loop
        }
        while (true);
        // end of stream loop
        return $body;
    }

    private function gunzip($zipped) {
      $offset = 0;
      if (substr($zipped,0,2) == "\x1f\x8b")
         $offset = 2;
      if (substr($zipped,$offset,1) == "\x08")  
      {
        try
        {
         return gzinflate(substr($zipped, $offset + 8));
        } catch (\Exception $e)
        {
            abort(500, "Fehler beim unzip des Ergebnisses von folgendem Anbieter: " . $this->name);
        }
      }
      return "Unknown Format";
   }  

    private function writeRequest ()
    {
        
        $out = "GET " . $this->getString . " HTTP/1.1\r\n";
        $out .= "Host: " . $this->host . "\r\n";
        $out .= "User-Agent: " . $this->useragent . "\r\n";
        $out .= "Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8\r\n";
        $out .= "Accept-Language: de,en-US;q=0.7,en;q=0.3\r\n";
        $out .= "Accept-Encoding: gzip, deflate, br\r\n";
        $out .= "Connection: keep-alive\r\n\r\n";
        # Anfrage senden:
        $sent = 0; $string = $out; $time = microtime(true);
        while(true)
        {   
            try{
                $tmp = fwrite($this->fp, $string);
            }catch(\ErrorException $e)
            {
                # Irgendwas ist mit unserem Socket passiert. Wir brauchen einen neuen:
                fclose($this->fp);
                Redis::del($this->name . "." . $this->socketNumber);
                $this->fp = $this->getFreeSocket();
                $sent = 0;
                $string = $out;
                continue;
            }
            if($tmp){
                $sent += $tmp;
                $string = substr($string, $tmp);
            }

            if($sent >= strlen($out))
                break;
        }
        if( $sent === strlen($out) )
        {
            return true;
        }
        return false;
    }

    private function getFreeSocket()
    {
        # Je nach Auslastung des Servers ( gleichzeitige Abfragen ), kann es sein, dass wir mehrere Sockets benötigen um die Abfragen ohne Wartezeit beantworten zu können.
        # pfsockopen öffnet dabei einen persistenten Socket, der also auch zwischen den verschiedenen php Prozessen geteilt werden kann. 
        # Wenn der Hostname mit einem bereits erstellten Socket übereinstimmt, wird die Verbindung also aufgegriffen und fortgeführt.
        # Allerdings dürfen wir diesen nur verwenden, wenn er nicht bereits von einem anderen Prozess zur Kommunikation verwendet wird.
        # Wenn dem so ist, probieren wir den nächsten Socket zu verwenden.
        # Dies festzustellen ist komplizierter, als man sich das vorstellt. Folgendes System sollte funktionieren:
        # 1. Stelle fest, ob dieser Socket neu erstellt wurde, oder ob ein existierender geöffnet wurde.
        $counter = 0; $fp = null;
        do
        {
            
            if( intval(Redis::exists($this->host . ".$counter")) === 0 )              
            {
                Redis::set($this->host . ".$counter", 1);
                Redis::expire($this->host . ".$counter", 5);
                $this->socketNumber = $counter;

                try
                {
                    $fp = pfsockopen($this->getHost() . ":" . $this->port . "/$counter", $this->port, $errstr, $errno, 1);
                }catch(\ErrorException $e)
                {
                    break;
                }
                # Wir gucken, ob der Lesepuffer leer ist:
                stream_set_blocking($fp, 0);
                $string = fgets($fp, 8192);
                if( $string !== false || feof($fp) )
                {
                    fclose($fp);
                    continue;
                }
                break;
            }
            $counter++;
        }while(true);
        return $fp;
    }

    private function getHost()
    {
        $return = "";
        if( $this->port === "443" )
        {
            $return .= "tls://";
        }else
        {
            $return .= "tcp://";
        }
        $return .= $this->host;
        return $return;
    }
}