public function read()
    {
        $read = $this->getStreams();
        $write = null;
        $except = null;
        $modified = stream_select($read, $write, $except, 0, 200000);

        if ($modified === false) {
            throw new \RuntimeException("stream_select() returned an error");
        }

        foreach ($read as $resource) {
            foreach ($this->messages as $message) {
                $this->readResource($message, $resource);
            }
        }
    }