protected function handleFeedJob(ServerRequestInterface $request)
    {
        $server = $this->server;
        $loop = $this->loop;
        $fs = $this->fs;
        $jm = $this->jobManager;
        $collection = $this->collection;

        $body = $request->getBody();
        \assert($body instanceof \React\Stream\ReadableStreamInterface);
        $lock = $server->acquireLock(0, static function () use ($server, $body) {
            $server->error('Feed reached timeout');
            $body->emit('end');
        });

        return new \React\Promise\Promise(static function ($resolve, $reject) use ($server, $loop, $fs, $jm, $collection, $body, $lock) {
            $tmpfile = $fs->tempnam('m2mql_feed/m2mql_feed');
            $fh = \fopen($tmpfile, 'w');
            if (false === $fh) {
                $reject(new \RuntimeException("Cannot fopen {$tmpfile}"));

                return;
            }

            $stream = new WritableResourceStream($fh, $loop, 8192);

            $bytes = 0;
            $body->on('data', static function ($data) use (&$bytes) {
                $bytes += \strlen($data);
            });

            $uploaded = false;
            $body->on('end', static function () use (&$uploaded) {
                // it's ok
                $uploaded = true;
            });

            $stream->on('close', static function () use ($server, $fs, $jm, $collection, $lock, $resolve, &$bytes, $tmpfile, &$uploaded, $fh) {
                if (!$uploaded) {
                    $server->releaseLock($lock);

                    return;
                }

                // on ferme le fichier
                \fclose($fh);
                $feedname = $collection . '_' . \bin2hex(\random_bytes(16)) . '.ndjson';
                // le tableau permet de préciser un dossier préfix (cad avant le hash)
                $feedname = [$collection . '/feed', $feedname];
                $feedname = $fs->move($tmpfile, $feedname);
                $feedname = $fs->magicpath($feedname);

                /** @var FeedTask $job */
                $job = $jm->createAndDispatchJob(FeedTask::class, $collection, $feedname, $bytes);

                $server->releaseLock($lock);
                $resolve($jm->toApi($job));
            });

            // an error occures e.g. on invalid chunked encoded data or an unexpected 'end' event
            $body->on('error', static function (\Exception $exception) use ($server, $lock, $reject) {
                // NB ici throw casse le daemon.
                $server->releaseLock($lock);
                $reject(new ApiException(ApiException::IO_ERROR, 'Error reading stream'));
            });

            $body->pipe($stream);
        });
    }