protected function getNewPipe()
    {
        $cmd = 'mysqlbinlog ' . escapeshellarg($this->logDir . '/' . $this->baseName) . '.[0-9]*';

        if ($this->from !== null) {
            if ($this->startType == self::START_TYPE_DATE) {
                $cmd .= ' --start-datetime=' . escapeshellarg($this->from->format('Y-m-d H:i:s'));
            } elseif ($this->startType == self::START_TYPE_POSITION) {
                $cmd .= ' --start-position=' . escapeshellarg($this->from);
            }
        }

        $cmd .= " --base64-output=DECODE-ROWS -v 2>&1 | grep -E '###|#[0-9]|Errcode|ERROR'";

        $this->pipe = popen($cmd, 'r');

        if (empty($this->pipe)) {
            throw new \RuntimeException('Error while executing mysqlbinlog');
        }
    }