public function onError($error, Message $message)
    {
        $worker = $this->getWorkerForStream($message->getResource());

        if ($worker === null) {
            return;
        }

        $this->eventEmitter->emit('peridot.concurrency.worker.error', [$error, $worker]);
        $this->detach($worker);
        $this->broker->removeMessage($message);

        $worker = new Worker($this->command, $this->eventEmitter, $this->resourceOpen);
        $this->attach($worker);
    }