code: 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');
          }
      }
docs: Initializes a new pipe.
  
  @throws \RuntimeException