Wait to have reached length to complete
@param $data
559638
public function parseContentLength($data)
    {
        $this->buffer .= $data;
        if (strlen($this->buffer) >= $this->contentLength) {
            $this->completed();
        }
    }
1093954
public function writeAsync($data)
    {
        $this->dataBuffer .= $this->buffer->encodeMessage(serialize($data));
        if ($this->writeEvent === false)
        {
            $this->loop->addWriteStream($this->connection, function ($stream, LoopInterface $loop)
            {
                if (strlen($this->dataBuffer) > 0)
                {
                    $dataWritten = fwrite($stream,$this->dataBuffer);
                    $this->dataBuffer = substr($this->dataBuffer,$dataWritten);
                    if(strlen($this->dataBuffer) <= 0)
                    {
                        $loop->removeWriteStream($stream);
                        $this->writeEvent = false;
                    }
                }
            });
            $this->writeEvent = true;
        }
    }
229156
public function handle(string $data)
    {
        if (!$this->payload) {
            $this->payload = $this->protocol->getPayload();
        }

        while ($data) { // Each iteration pulls off a single payload chunk
            $size = strlen($data);
            $remaining = $this->payload->getRemainingData();

            // If we don't yet know how much data is remaining, read data into
            // the payload in two byte chunks (the size of a WebSocket frame
            // header to get the initial length)
            //
            // Then re-loop. For extended lengths, this will happen once or four
            // times extra, as the extended length is read in.
            if ($remaining === null) {
                $chunkSize = 2;
            } elseif ($remaining > 0) {
                $chunkSize = $remaining;
            } elseif ($remaining === 0) {
                $chunkSize = 0;
            }

            $chunkSize = min(strlen($data), $chunkSize);
            $chunk = substr($data, 0, $chunkSize);
            $data = substr($data, $chunkSize);

            $this->payload->receiveData($chunk);

            if ($remaining !== 0 && !$this->payload->isComplete()) {
                continue;
            }

            if ($this->payload->isComplete()) {
                $this->emit($this->payload);
                $this->payload = $this->protocol->getPayload();
            } else {
                throw new PayloadException('Payload will not complete');
            }
        }
    }
422236
private function waitUntilStopped(int $pid): void
    {
        $wait = 5000; // 5ms
        $timeout = 1000000; // 1 sec
        $elapsed = 0;
        while (posix_getpgid($pid) !== false) {
            usleep($wait);
            $elapsed += $wait;
            if ($elapsed > $timeout) {
                throw new \Exception('Timeout while waiting for PHP-FPM to stop');
            }
        }
    }
863899
public function data_postprocessing($data) {
        parent::data_postprocessing($data);
        if (!empty($data->completionunlocked)) {
            // Turn off completion settings if the checkboxes aren't ticked.
            $autocompletion = !empty($data->completion) &&
                $data->completion == COMPLETION_TRACKING_AUTOMATIC;
            if (!$autocompletion || empty($data->completionsubmit)) {
                $data->completionsubmit = 0;
            }
        }
    }
1297140
public function addLoop($data)
    {
        $length = is_array($data) || $data instanceof Countable ? count($data) : null;

        $parent = Arr::last($this->loopsStack);

        $this->loopsStack[] = [
            'iteration' => 0,
            'index'     => 0,
            'remaining' => isset($length) ? $length : null,
            'count'     => $length,
            'first'     => true,
            'last'      => isset($length) ? $length === 1 : null,
            'depth'     => count($this->loopsStack) + 1,
            'parent'    => $parent ? (object) $parent : null,
        ];
    }
501517
public function next($data)
    {
        if ($this->destinations!=null) {
            foreach ($this->destinations as $node) {
                $node->input($data, $this);
            }
        }
    }
839259
public function waitForFinishedJob()
    {
        if ($this->inExecution === 0) {
            return;
        }
        $tellsUsItHasFinished = false;
        stream_set_blocking($this->pipes[1], true);
        while ($line = fgets($this->pipes[1])) {
            if (strstr($line, "FINISHED\n")) {
                $tellsUsItHasFinished = true;
                --$this->inExecution;
                break;
            }
        }
        if (!$tellsUsItHasFinished) {
            throw new \RuntimeException('The Worker terminated without finishing the job.');
        }
    }
930585
protected function onInput($data)
    {
        //Process data here
        foreach ($this->params as $column) {
            if ($this->character_mask) {
                $data[$column] = trim($data[$column], $this->character_mask);
            } else {
                $data[$column] = trim($data[$column]);
            }
        }
        $this->next($data);
    }
1114525
public function writeData($data)
	{
		if (strlen($data) - 4 > $this->_limit) {
			$data = $this->_addContinue($data);
		}
		$this->_datasize += strlen($data);

		return $data;
	}
231609
public function onInput($data)
    {
        //Process data here
        foreach ($this->params as $copy => $original) {
            $data[$copy] = $data[$original];
        }
        $this->next($data);
    }
606268
protected function onInput($data)
    {
        //Process data here
        foreach ($this->params as $cName => $cParams) {
            $from = Utility::get($cParams, "from", "Y-m-d H:i:s");
            $to = Utility::get($cParams, "to", "Y-m-d H:i:s");
            if ($data[$cName] && $from != $to) {
                $obj = \DateTime::createFromFormat($from, $data[$cName]);
                $data[$cName] = ($obj) ? $obj->format($to) : null;
            }
        }
        $this->next($data);
    }
1164076
public void waitUntilComplete(long timeout) {
        long base=System.currentTimeMillis();
        long now=0;
        lock.lock();
        try {
            while(processing || count.get() > 0) {
                long delay=timeout-now;
                if(delay <= 0)
                    break;
                try {
                    processing_done.await(delay, TimeUnit.MILLISECONDS);
                    now=System.currentTimeMillis()-base;
                }
                catch(InterruptedException e) {
                }
            }
        }
        finally {
            lock.unlock();
        }
    }
837896
public function write(string $data) : void
    {
        $this->buffer .= $data;

        if (strlen($this->buffer) >= $this->size) {
            $this->flush();
        }
    }
1289096
public function fill($data)
    {
        // Field value is already setted.
//        if (!is_null($this->value)) {
//            return;
//        }

        $this->data = $data;

        if (is_array($this->column)) {
            foreach ($this->column as $key => $column) {
                $this->value[$key] = Arr::get($data, $column);
            }

            return;
        }

        $this->value = Arr::get($data, $this->column);
        if (isset($this->customFormat) && $this->customFormat instanceof \Closure) {
            $this->value = call_user_func($this->customFormat, $this->value);
        }
    }
289563
def finish(self, data=''):
        '''
        
        '''
        self.pending += data
        self.streaming = False
        for i in self:
            yield i
645842
protected function onInput($data)
    {
        $this->index++;
        if ($this->index >= $this->offset && $this->index < $this->offset + $this->limit) {
            $this->next($data);
        }
    }