Add pagination to datacontroller

Taskfile.yml

@@ -49,6 +49,7 @@ tasks:
     desc: Run all tests
     cmds:
       # - task: test:unit
+      - task: test:broker
       - task: test:geo
       - task: test:source
 
@@ -62,6 +63,7 @@ tasks:
       vars:
         TASK_ARGS: bin/phpunit tests/{{.TASK | replace "test:" "" | title }}
 
+  test:broker: *test_task
   test:geo: *test_task
   test:source: *test_task
 

src/Broker/PagedBrokerReader.php

@@ -0,0 +1,123 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Broker;
+
+use Psr\Log\LoggerInterface;
+use Symfony\Contracts\HttpClient\HttpClientInterface;
+use Symfony\Contracts\HttpClient\ResponseInterface;
+
+/**
+ * Reads a broker list endpoint on behalf of a client that cannot page itself.
+ *
+ * A map widget asks for a layer once and draws whatever comes back, so a
+ * result the broker splits across pages arrives silently truncated. The
+ * broker refuses a limit above its own maximum rather than returning
+ * everything, which leaves following the pages as the only way to get a
+ * complete set. Doing it here keeps that out of the widget configuration.
+ */
+final readonly class PagedBrokerReader
+{
+    /**
+     * Comfortably within the maximum brokers tend to impose; a larger page
+     * risks the 403 the broker answers an over-large limit with.
+     */
+    private const int PAGE_SIZE = 1000;
+
+    /**
+     * Stops a broker that keeps advertising a next page from looping forever.
+     */
+    private const int MAX_PAGES = 100;
+
+    public function __construct(
+        private HttpClientInterface $brokerClient,
+        private LoggerInterface $logger,
+    ) {
+    }
+
+    /**
+     * Every page the broker offers, merged into one payload.
+     *
+     * @param array<string, mixed> $query
+     * @param array<string, mixed> $headers
+     *
+     * @return array<mixed>
+     */
+    public function readAll(string $path, array $query, array $headers): array
+    {
+        // Asking for everything leaves the paging to us. A limit the caller
+        // named would only decide where the run gives up, which is the silent
+        // truncation this exists to avoid.
+        unset($query['limit'], $query['offset']);
+        $merged = null;
+
+        for ($page = 0; $page < self::MAX_PAGES; ++$page) {
+            $response = $this->brokerClient->request('GET', $path, [
+                'query' => [...$query, 'limit' => self::PAGE_SIZE, 'offset' => $page * self::PAGE_SIZE],
+                'headers' => $headers,
+            ]);
+
+            $data = $response->toArray();
+            $merged = null === $merged ? $data : $this->merge($merged, $data);
+
+            // A page the broker did not fill is the last one whatever its
+            // headers say, so the run stops without a further request.
+            if ($this->count($data) < self::PAGE_SIZE || !$this->hasNextPage($response)) {
+                return $merged;
+            }
+        }
+
+        // Returning a truncated set as though it were complete is the failure
+        // worth being loud about; the caller cannot tell from the payload.
+        $this->logger->warning('Stopped reading {path} after {pages} pages; the result is incomplete.', [
+            'path' => $path,
+            'pages' => self::MAX_PAGES,
+        ]);
+
+        return $merged;
+    }
+
+    /**
+     * @param array<mixed> $merged
+     * @param array<mixed> $page
+     *
+     * @return array<mixed>
+     */
+    private function merge(array $merged, array $page): array
+    {
+        // Entities come back as a bare list, GeoJSON as a FeatureCollection
+        // wrapping one. Everything outside the features belongs to the
+        // collection rather than the page, so the first page's copy stands.
+        if (isset($merged['features'], $page['features'])) {
+            $merged['features'] = [...$merged['features'], ...$page['features']];
+
+            return $merged;
+        }
+
+        return [...$merged, ...$page];
+    }
+
+    /**
+     * @param array<mixed> $data
+     */
+    private function count(array $data): int
+    {
+        return \count($data['features'] ?? $data);
+    }
+
+    /**
+     * The broker announces a further page in a Link header, alongside the
+     * ones it uses to point at the context.
+     */
+    private function hasNextPage(ResponseInterface $response): bool
+    {
+        foreach ($response->getHeaders()['link'] ?? [] as $link) {
+            if (str_contains($link, 'rel="next"')) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+}

src/Controller/DataController.php

@@ -2,6 +2,7 @@
 
 namespace App\Controller;
 
+use App\Broker\PagedBrokerReader;
 use Symfony\Bundle\FrameworkBundle\Controller\AbstractController;
 use Symfony\Component\HttpFoundation\JsonResponse;
 use Symfony\Component\HttpFoundation\Request;
@@ -14,6 +15,12 @@ final class DataController extends AbstractController
     private const string FORMAT_GEOJSON = 'geojson';
 
     private const string APPLICATION_GEOJSON = 'application/geo+json';
+    private const string APPLICATION_JSON = 'application/json';
+
+    /**
+     * Asks for the complete set rather than the page the broker defaults to.
+     */
+    private const string PARAMETER_ALL = 'all';
 
     #[Route(
         path: '/data/{path}.{_format}',
@@ -27,6 +34,7 @@ final class DataController extends AbstractController
     )]
     public function index(Request $request, string $path, string $_format,
         HttpClientInterface $brokerClient,
+        PagedBrokerReader $pagedReader,
     ): JsonResponse {
         $path = '/'.ltrim($path, '/');
         $headers = $request->headers->all();
@@ -41,8 +49,25 @@ public function index(Request $request, string $path, string $_format,
                 && !str_starts_with($name, 'x-forwarded-'),
             ARRAY_FILTER_USE_KEY
         );
+        $query = $request->query->all();
+
+        // Opting in is left to the caller: a client that pages for itself
+        // passes its own limit and offset, and answering those with the whole
+        // set instead would break it.
+        if ($request->query->getBoolean(self::PARAMETER_ALL)) {
+            unset($query[self::PARAMETER_ALL]);
+
+            return new JsonResponse(
+                data: $pagedReader->readAll($path, $query, $headers),
+                headers: ['content-type' => match ($_format) {
+                    self::FORMAT_GEOJSON => self::APPLICATION_GEOJSON,
+                    default => self::APPLICATION_JSON,
+                }],
+            );
+        }
+
         $response = $brokerClient->request($request->getMethod(), $path, [
-            'query' => $request->query->all(),
+            'query' => $query,
             'headers' => $headers,
         ]);
 

tests/Broker/PagedBrokerReaderTest.php

@@ -0,0 +1,173 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Tests\Broker;
+
+use App\Broker\PagedBrokerReader;
+use PHPUnit\Framework\TestCase;
+use Psr\Log\NullLogger;
+use Symfony\Component\HttpClient\MockHttpClient;
+use Symfony\Component\HttpClient\Response\MockResponse;
+
+class PagedBrokerReaderTest extends TestCase
+{
+    private const string PATH = '/ngsi-ld/v1/entities';
+
+    /**
+     * The page size the reader asks for when the caller names none. A page
+     * shorter than this tells it the run is over.
+     */
+    private const int PAGE_SIZE = 1000;
+
+    public function testItMergesTheFeaturesOfEveryPage(): void
+    {
+        $client = new MockHttpClient([
+            $this->geoJsonPage(self::PAGE_SIZE, hasNext: true),
+            $this->geoJsonPage(345, hasNext: false),
+        ]);
+
+        $data = new PagedBrokerReader($client, new NullLogger())->readAll(self::PATH, [], []);
+
+        $this->assertCount(1345, $data['features']);
+        $this->assertSame(2, $client->getRequestsCount());
+    }
+
+    public function testItKeepsTheCollectionAroundTheMergedFeatures(): void
+    {
+        $client = new MockHttpClient([
+            $this->geoJsonPage(self::PAGE_SIZE, hasNext: true),
+            $this->geoJsonPage(1, hasNext: false),
+        ]);
+
+        $data = new PagedBrokerReader($client, new NullLogger())->readAll(self::PATH, [], []);
+
+        $this->assertSame('FeatureCollection', $data['type']);
+        $this->assertArrayHasKey('@context', $data);
+    }
+
+    public function testItMergesEntitiesReturnedAsABareList(): void
+    {
+        $client = new MockHttpClient([
+            $this->listPage(self::PAGE_SIZE, hasNext: true),
+            $this->listPage(20, hasNext: false),
+        ]);
+
+        $data = new PagedBrokerReader($client, new NullLogger())->readAll(self::PATH, [], []);
+
+        $this->assertCount(1020, $data);
+    }
+
+    public function testItAsksForEachPageInTurn(): void
+    {
+        $offsets = [];
+        $client = new MockHttpClient(function (string $method, string $url) use (&$offsets): MockResponse {
+            parse_str((string) parse_url($url, PHP_URL_QUERY), $query);
+            $offsets[] = $query['offset'] ?? null;
+
+            return $this->geoJsonPage(count($offsets) < 3 ? self::PAGE_SIZE : 1, hasNext: count($offsets) < 3);
+        });
+
+        new PagedBrokerReader($client, new NullLogger())->readAll(self::PATH, [], []);
+
+        $this->assertSame(['0', '1000', '2000'], $offsets);
+    }
+
+    /**
+     * A broker that keeps claiming another page must not be followed for ever.
+     */
+    public function testItStopsFollowingAnEndlessRunOfPages(): void
+    {
+        $client = new MockHttpClient(fn (): MockResponse => $this->geoJsonPage(self::PAGE_SIZE, hasNext: true));
+
+        new PagedBrokerReader($client, new NullLogger())->readAll(self::PATH, [], []);
+
+        $this->assertSame(100, $client->getRequestsCount());
+    }
+
+    public function testItStopsOnAShortPageEvenWhenAnotherIsAnnounced(): void
+    {
+        $client = new MockHttpClient([
+            $this->geoJsonPage(3, hasNext: true),
+            $this->geoJsonPage(3, hasNext: false),
+        ]);
+
+        $data = new PagedBrokerReader($client, new NullLogger())->readAll(self::PATH, [], []);
+
+        $this->assertCount(3, $data['features']);
+        $this->assertSame(1, $client->getRequestsCount());
+    }
+
+    /**
+     * Paging with a limit the caller named would stop the run wherever that
+     * limit ran out, which is the truncation the reader exists to avoid.
+     */
+    public function testItPagesPastALimitTheCallerNamed(): void
+    {
+        $limits = [];
+        $client = new MockHttpClient(function (string $method, string $url) use (&$limits): MockResponse {
+            parse_str((string) parse_url($url, PHP_URL_QUERY), $query);
+            $limits[] = $query['limit'] ?? null;
+
+            return $this->geoJsonPage(count($limits) < 2 ? self::PAGE_SIZE : 345, hasNext: count($limits) < 2);
+        });
+
+        $data = new PagedBrokerReader($client, new NullLogger())
+            ->readAll(self::PATH, ['limit' => 10, 'offset' => 40], []);
+
+        $this->assertCount(1345, $data['features']);
+        $this->assertSame(['1000', '1000'], $limits);
+    }
+
+    public function testItWarnsWhenItGivesUpOnAnIncompleteResult(): void
+    {
+        $logger = new class extends NullLogger {
+            public int $warnings = 0;
+
+            public function warning(string|\Stringable $message, array $context = []): void
+            {
+                ++$this->warnings;
+            }
+        };
+        $client = new MockHttpClient(fn (): MockResponse => $this->geoJsonPage(self::PAGE_SIZE, hasNext: true));
+
+        new PagedBrokerReader($client, $logger)->readAll(self::PATH, [], []);
+
+        $this->assertSame(1, $logger->warnings);
+    }
+
+    private function geoJsonPage(int $features, bool $hasNext): MockResponse
+    {
+        return new MockResponse(
+            json_encode([
+                'type' => 'FeatureCollection',
+                'features' => array_fill(0, $features, ['type' => 'Feature']),
+                '@context' => 'https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.8.jsonld',
+            ]),
+            ['response_headers' => $this->headers($hasNext)]
+        );
+    }
+
+    private function listPage(int $entities, bool $hasNext): MockResponse
+    {
+        return new MockResponse(
+            json_encode(array_fill(0, $entities, ['id' => 'urn:ngsi-ld:OnStreetParking:x'])),
+            ['response_headers' => $this->headers($hasNext)]
+        );
+    }
+
+    /**
+     * @return array<string, list<string>>
+     */
+    private function headers(bool $hasNext): array
+    {
+        // The context link is always there, so a reader looking for the next
+        // page has to pick it out rather than trust that a link means more.
+        $links = ['<https://uri.etsi.org/ngsi-ld/v1/ngsi-ld-core-context-v1.8.jsonld>;rel="http://www.w3.org/ns/json-ld#context"'];
+        if ($hasNext) {
+            $links[] = '</ngsi-ld/v1/entities?offset=1000>;rel="next";type="application/ld+json"';
+        }
+
+        return ['content-type' => ['application/json'], 'link' => $links];
+    }
+}

tests/resources/config/Parking/OnStreetParking.yaml

@@ -13,7 +13,9 @@ map:
     layer:
         - namedlayer: "#septima_standard"
 
-        - features_host: "/data/ngsi-ld/v1/entities.geojson?type=https://smartdatamodels.org/dataModel.Parking/OnStreetParking"
+          # "all" makes the proxy follow the broker's paging; without it only
+          # the broker's first page of entities reaches the map.
+        - features_host: "/data/ngsi-ld/v1/entities.geojson?type=https://smartdatamodels.org/dataModel.Parking/OnStreetParking&all=1"
 
           # THe undocumented property src is needed to display GeoJSON!
           srs: "EPSG:4326"