# Streaming search results > Receive flight search results supplier by supplier as server-sent events, instead of waiting for the slowest supplier. Source: https://fx-port.com/docs/streaming `POST /api/v1/get_flights?stream=true` runs the same search as the JSON mode, but answers with **server-sent events**: the first offers arrive as soon as the first supplier has answered, and the rest follow supplier by supplier, instead of one response after the slowest supplier. The request body, the key permission (Read), the offers and the errors that are found before the search starts are the ones of [Search flights](https://fx-port.com/docs/flights/search.md). **POST** `/api/v1/get_flights?stream=true` ## When to use it A search fans out to every compatible supplier and the suppliers answer at different speeds. Measured on the sandbox through `api.fx-port.com` on 2026-11-01 (ALG to CDG, one adult, two suppliers): | | JSON mode | Stream | |---|---|---| | First byte | After the slowest supplier: 2.7 to 8 s | **0.3 to 0.7 s** (the `start` event) | | First offers | Same moment as the last offers | **1.5 to 2.6 s** (the fastest supplier) | | Last offers | 2.7 to 8 s | 2.9 to 8 s (the slowest supplier) | | Bytes on the wire | About 26 KB with `Accept-Encoding: gzip` (700 KB without) | About 660 KB: **streams are never compressed** | | Code | One `json.loads` | An event loop: see [the complete client](#a-complete-client) | Use the stream for a **search page that a person is looking at**: the first fares appear in about two and a half seconds and the list grows while the slow suppliers finish. Use the JSON mode for **everything else**: batch jobs, price comparison, caching a search for later, and any client that cannot read a stream. On a slow link (a phone on mobile data) the 660 KB of an uncompressed stream can take longer to download than the 26 KB of a gzipped JSON response: measure before you choose. ## Request Same body as [Search flights](https://fx-port.com/docs/flights/search.md#request-fields), same Read permission. Ask for a stream with **either** of these (each works alone): - the query parameter `?stream=true` (`1` and `yes` also work); - the header `Accept: text/event-stream`. | Header | Value | | |---|---|---| | `Authorization` | `Bearer fxp_test_...` | Required | | `Content-Type` | `application/json` | Required: the search is a JSON body | | `Accept` | `text/event-stream` | Recommended | | `User-Agent` | Your own, for example `my-app/1.0` | Cloudflare blocks the default agents of Python's `urllib` and Perl's `libwww-perl` with a `403` and the text `error code: 1010` | **Request** ```bash curl 'https://api.fx-port.com/api/v1/get_flights?stream=true' \ -H "Authorization: Bearer fxp_test_YOUR_KEY" \ -H "Content-Type: application/json" \ -H "Accept: text/event-stream" \ -d '{ "origin": "ALG", "destination": "CDG", "departure_date": "2026-12-15", "cabin_class": "economy", "passengers": { "adults": 1 } }' ``` **A problem with the request is answered before the stream starts, as an ordinary JSON error** with its own status code, not `200`: a missing or invalid key (`401`), a key that is not allowed (`403`), a malformed body, an invalid passenger mix or an unknown supplier (`400`), too many requests (`429`). Always check that the status is `200` and the `Content-Type` is `text/event-stream` before you parse events. Once the stream has started the status is `200` and failures arrive as events. ## Response headers Verified through Cloudflare on 2026-11-01: | Header | Value | Meaning | |---|---|---| | `Content-Type` | `text/event-stream` | Server-sent events in UTF-8 (the header has no `charset`: decode the bytes as UTF-8) | | `Cache-Control` | `no-cache, no-transform` | Never cached, and proxies are asked not to compress or rewrite it | | `Transfer-Encoding` | `chunked` | The length is unknown (on HTTP/2 the body is sent as frames and the header is absent) | | `X-Environment` | `sandbox` or `live` | The environment of the key | | `X-Request-Id` | The id of the request | Quote it to support | | `cf-cache-status` | `DYNAMIC` | Cloudflare does not cache or hold the stream | There is **no `Content-Encoding`**: the stream is not compressed, even when you send `Accept-Encoding: gzip`. ## The wire format A stream is text. Each event is a few lines ended by an empty line: ~~~text id: 4 event: supplier_complete data: {"type":"supplier_complete","supplierId":"duffel_1","totalOffers":92,"hasFlexPrices":false,"flexPricesCount":0,"complete":true,"timestamp":"2026-11-01T15:22:59.175148"} ~~~ | Field | Content | |---|---| | `id` | The number of the event, from `0`, increasing by one. For logging: a stream **cannot be resumed** (see [Timeouts and reconnecting](#timeouts-and-reconnecting)) | | `event` | The name of the event: `start`, `supplier_start`, `offers_chunk`, `supplier_complete`, `supplier_result`, `flex_prices`, `error`, `summary`, `done` | | `data` | One line of compact JSON. The same name is in its `type` key | | an empty line | Ends the event: dispatch it now | Rules every client must follow: 1. **Ignore lines that start with a colon.** They are comments, such as the keep-alive line `: keep-alive` ([below](#keep-alive-lines)). 2. **Ignore a field or an event you do not know.** More of them can appear in the future. 3. **A line can be huge.** One `offers_chunk` carries up to 50 offers and its `data` line was up to 210 KB in our measurements. A reader with a small line buffer (Go's `bufio.Scanner` has 64 KB by default) fails with a cryptic error. 4. **Timestamps** (`timestamp`) are ISO 8601 in UTC **without** a `Z`: `2026-11-01T15:22:59.175148`. 5. Only the `start` event carries `environment` (as its **first key**); the header `X-Environment` says it too. ## Events A search with two suppliers produced these 12 events (the offers are not shown): ~~~text 0 start 1 supplier_start duffel_1 92 offers coming in 2 chunks 2 offers_chunk duffel_1 chunk 1 of 2, 50 offers 3 offers_chunk duffel_1 chunk 2 of 2, 42 offers 4 supplier_complete duffel_1 complete: true 5 supplier_start amadeus_aqc_dz_1 68 offers coming in 2 chunks 6 offers_chunk amadeus_aqc_dz_1 chunk 1 of 2, 50 offers 7 offers_chunk amadeus_aqc_dz_1 chunk 2 of 2, 18 offers 8 supplier_complete amadeus_aqc_dz_1 complete: false <- another batch of this supplier follows 9 supplier_result amadeus_aqc_dz_1 2 more offers, complete: true, another searchId 10 summary 11 done ~~~ | Event | When | What it carries | |---|---|---| | `start` | First, at once | The request id and the suppliers that will answer | | `supplier_result` | A supplier, or a later batch of it, answered with **50 offers or fewer** | All the offers of one result set, in `data.results.offers` | | `supplier_start` | A supplier answered with **more than 50 offers** | The result set without its offers: the `searchId`, the currencies, `results.totalOffers`, `chunkSize`, `totalChunks` | | `offers_chunk` | After `supplier_start` | Up to 50 offers, numbered `chunkNumber` of `totalChunks` | | `flex_prices` | A chunked result set that carries a date matrix (`flexible: true`) | The matrix of the supplier. Not seen in the sandbox: see [flexible dates](#flexible-dates) | | `supplier_complete` | After the last chunk | The counters of the result set and `complete` | | `error` | A supplier failed | Which supplier, and the error text | | `summary` | After every supplier has answered or failed | The counters and the list of failures | | `done` | Last | The end of the stream | ### start ~~~json { "environment": "sandbox", "type": "start", "requestId": "d36b6918", "searchType": "ONE WAY", "search_type": "ONE_WAY", "flexible": false, "timestamp": "2026-11-01T15:22:57.762071", "expectedSuppliers": ["amadeus_aqc_dz_1", "duffel_1"] } ~~~ `expectedSuppliers` are the suppliers called for this search: the compatible ones, a supplier skipped for the route or the passenger mix is not listed. `searchType` here is the raw classification of the route (`ONE WAY`, `ROUNDTRIP`, `MULTIPLE CITY`, `OPEN`), **not** the `ONE_WAY`, `ROUND_TRIP`, `OPEN_JAW`, `MULTI_CITY` of the JSON mode. Branch on `search_type`: it has the vocabulary of the JSON mode. ### supplier_result and supplier_start A supplier answered. `data` is the same object as an entry of `suppliers[]` in the JSON mode, arranged a little differently: ~~~json { "type": "supplier_start", "supplierId": "duffel_1", "success": true, "data": { "success": true, "requestId": "d36b6918", "searchId": "fs_dfx_59ce98b7-ce5a-4911-abb8-b43d4786e6c9", "searchType": "ONE_WAY", "processingTime": 1.381, "supplierCurrency": "USD", "agencyCurrency": "DZD", "results": { "totalOffers": 92, "supplierCurrency": "USD", "agencyCurrency": "DZD", "offers": [], "chunked": true, "chunkSize": 50, "totalChunks": 2 }, "meta": { "filtersApplied": {}, "warnings": [], "supplierId": "duffel_1", "technologyType": "duffel", "supportHold": false, "supportVoid": false, "voidDeadline": null }, "exchangeRate": 250.0 }, "processingTime": 1.381, "timestamp": "2026-11-01T15:22:59.161692" } ~~~ `supplier_result` has the same shape, plus a top-level `complete`, and the offers in `data.results.offers` (no `chunked`, `chunkSize` or `totalChunks`). Compared with the JSON mode: `searchId` and `requestId` are inside `data`; `supportHold`, `supportVoid`, `voidDeadline`, `filtersApplied` and `warnings` stay inside `data.meta` (the JSON mode moves them to the supplier entry); `meta` has no `supplierName`, `processingTime` or currencies (the last two are next to it, in `data`). The offers themselves are exactly those of the JSON mode: see [the Flight offer object](https://fx-port.com/docs/api/objects/flight-offer.md). ### offers_chunk ~~~json { "type": "offers_chunk", "supplierId": "duffel_1", "chunkNumber": 1, "totalChunks": 2, "offersInChunk": 50, "offers": [{ "id": "fx_offer_9d08a93c", "type": "flight_offer", "...": "an offer, as in the JSON mode" }], "timestamp": "2026-11-01T15:22:59.161771" } ~~~ A chunk carries only the `supplierId`: add its offers to the result set opened by the last `supplier_start` of that supplier. ### supplier_complete ~~~json { "type": "supplier_complete", "supplierId": "amadeus_aqc_dz_1", "totalOffers": 68, "hasFlexPrices": false, "flexPricesCount": 0, "complete": false, "timestamp": "2026-11-01T15:22:59.819079" } ~~~ ### error ~~~json { "type": "error", "supplierId": "duffel_1", "success": false, "error": "The supplier rejected the search request.", "cause": "supplier_error", "processingTime": 0, "timestamp": "2026-11-01T15:23:32.838533" } ~~~ One supplier failed: it timed out, refused the search, was skipped for a short while or was busy. The other suppliers go on. `cause` is a stable code: `supplier_timeout`, `supplier_api_error`, `supplier_auth_error`, `supplier_error`, `supplier_temporarily_unavailable`, `supplier_busy` (FX-Port's capacity towards the supplier is full for a moment: the supplier was not called, search again in a few seconds) or `service_overloaded` (FX-Port was too busy to prepare that supplier's results: search again in a few seconds). `error` is a fixed sentence about the cause: it never carries the supplier's own text (addresses, stack text), and its wording can change, so branch on `cause`. ### summary and done ~~~json { "type": "summary", "requestId": "d36b6918", "success": true, "processingTime": 2.581, "summary": { "totalSuppliers": 2, "successfulSuppliers": 2, "failedSuppliers": 0, "totalOffers": 162 }, "successfulSuppliers": [ { "supplierId": "duffel_1", "offersCount": 92, "processingTime": 1.381 }, { "supplierId": "amadeus_aqc_dz_1", "offersCount": 70, "processingTime": 2.516 } ], "failures": null, "timestamp": "2026-11-01T15:23:00.286463" } ~~~ ~~~json { "type": "done", "requestId": "d36b6918", "timestamp": "2026-11-01T15:23:00.286605" } ~~~ `success` is `true` when at least one supplier returned offers. `failures` is `null`, or a list of `{ "supplierId", "error", "cause" }` (the same `cause` values as the `error` event). `successfulSuppliers` is a **number** inside `summary` and a **list** next to it. In the stream `summary.totalSuppliers` is the number of suppliers that were **called** (in the JSON mode it counts the suppliers that answered). When the stream reaches its time limit (see [Timeouts and reconnecting](#timeouts-and-reconnecting)) the `summary` event has `"truncated": true`, and the suppliers that had not answered are in `failures` with the cause `supplier_timeout`. ### Flexible dates With `"flexible": true` the date matrix (`flexPrices`, see [Flexible-date search](https://fx-port.com/docs/flights/flexible-search.md)) is the `data.flexPrices` of the result set that has it. For Amadeus Algeria that is the **second, slower batch**: the first batch has the offers and no matrix, and the matrix arrives with the `supplier_result` that has `complete: true` (7 seconds into the stream in our test). A chunked result set that carries a matrix sends it in a separate `flex_prices` event (`flexPrices` and a `count`) before `supplier_complete`; we have not seen that event in the sandbox, so treat it as optional. ## Ordering guarantees - `start` is the first event (`id` 0) and `done` the last. `summary` is just before `done`. - The events of a result set are in order: `supplier_start`, then `offers_chunk` 1 to `totalChunks`, then `flex_prices` if there is one, then `supplier_complete`. A small result set is a single `supplier_result`. - Suppliers are independent and arrive in the order they finish. There is no fixed order between suppliers, and it can change from one search to the next. - The events of two result sets are not mixed in practice, but do not rely on it: file every `offers_chunk` under its `supplierId`. - An `error` can come at any point after `start`. The stream still ends with `summary` and `done`: `done` is **always** the last event, also when the search failed half-way. - The stream is over **only** at `done`. A stream that ends without it (a dropped connection, a proxy that closed it) is incomplete. ## Merging the offers A supplier can answer **more than once**. Keep one list of offers per `searchId`: **Step 1: Open a result set** On `supplier_start` or `supplier_result`, read `data.searchId`, remember it as the current `searchId` of that `supplierId`, and start a list with `data.results.offers` (empty for `supplier_start`). **Step 2: Fill it** On every `offers_chunk`, append `offers` to the list of the current `searchId` of its `supplierId`. `chunkNumber` and `totalChunks` let you check that none is missing. **Step 3: Know when a supplier is finished** A supplier is finished at its first `supplier_complete` or `supplier_result` with `complete: true`, or at its `error`. `complete: false` means that **another batch of this supplier follows**. > [!WARNING] > **One supplier, two searchIds** > > Amadeus Algeria answers in two batches, a fast one and a slow one. The first batch has `complete: false`. The second arrives later as a `supplier_result` with `complete: true`, with **only the extra offers** and **a different `searchId`**. In the example above the first batch had 68 offers under `fs_509846e6-...` and the second 2 offers under `fs_35418d61-...`, and `summary` counted 70 for the supplier. > > Price and book an offer with the `searchId` of **the result set it arrived in**. Do not merge two result sets into one list that has a single `searchId`. `duffel_1` answered in a single result set. `supplier_complete.totalOffers` and `results.totalOffers` are the number of offers of **that result set**. The total of the search is `summary.summary.totalOffers`. ## Keep-alive lines While a supplier is slow the server writes a comment line about **every 15 seconds**, so that proxies, load balancers and your own read timeout see traffic: ~~~text : keep-alive ~~~ A comment starts with a colon and carries no event. **Skip every line that starts with a colon**: the SSE standard says so too. Do not count a comment as an event and do not use it as an end condition. ## Errors in the stream | Situation | What you see | What to do | |---|---|---| | The request is wrong (invalid key, malformed body, unknown supplier, a passenger mix no supplier can serve) | An ordinary JSON error with a `4xx` status, **no stream** | Fix the request. See [Errors](https://fx-port.com/docs/errors.md) | | One supplier fails | An `error` event for it, then the stream goes on | Show the offers of the others. Search again later if the missing supplier matters | | Every supplier fails | `error` events, then `summary` with `success: false`, then `done` | Treat it like the `503` of the JSON mode (`all_suppliers_failed`, or `supplier_busy` or `service_overloaded` when every `cause` says so): wait, then retry once or twice with backoff (a few seconds are enough for `supplier_busy` and `service_overloaded`) | | The connection drops, or the stream ends without `done` | The stream is cut | The offers you have are valid but incomplete. Run the search again: see below | | `429` | An ordinary JSON error before the stream: `rate_limit_exceeded`, or `concurrency_limit_exceeded` when too many streams of your key are open ([Limits](#limits)) | Wait for `Retry-After` ([Rate limits](https://fx-port.com/docs/rate-limits.md)) | ## Timeouts and reconnecting - **A stream cannot be resumed.** There is no `Last-Event-ID` support and no `retry:` field. If a connection drops, **start a new search**: it is a new request and counts once in your [rate limit](https://fx-port.com/docs/rate-limits.md). - **Do not use the browser's `EventSource`.** It cannot send a `POST` body or an `Authorization` header, and it reconnects on its own and would repeat the search. Read the stream with `fetch` on your server, or relay it to the browser with an endpoint of yours. - **Never put your key in a browser.** Call the API from your backend and forward the events to the page. - **A stream lasts 300 seconds at most.** At that point the server cancels the suppliers that are still running, sends an `error` event with `cause: "supplier_timeout"` for each, then `summary` with `"truncated": true`, then `done`. A normal search ends within about 10 seconds. - **Set a read timeout of at least 60 seconds** (the time you accept without receiving a byte) and a **total deadline of about two minutes**. A slow supplier can take more than 15 seconds, and the keep-alive lines keep the connection busy meanwhile. - **Reconnect with backoff.** After a failure wait 1 s, then 2 s, then 4 s (with jitter) and give up after three tries. Do not reconnect in a tight loop. - **Offers expire.** A search stays usable for 24 hours, but a supplier can withdraw an offer much earlier. Price an offer soon after you show it ([Price a flight offer](https://fx-port.com/docs/flights/price.md)). ## Limits | Limit | Value | |---|---| | Streams open at the same time | **Keep at most 4 streams open per API key.** A stream holds its connection and its supplier calls for as long as it runs, so the open streams of one key are capped at 4. Every search with `stream=true` counts while it runs; a JSON search never does. Once the cap is enforced, a fifth stream is refused before it starts with `429 concurrency_limit_exceeded` and `Retry-After: 5`. Until then the streams are only counted: do not rely on the extra ones. | | Duration of one stream | 300 seconds: see [Timeouts and reconnecting](#timeouts-and-reconnecting). | | Requests | A stream is one request of the search group ([Rate limits](https://fx-port.com/docs/rate-limits.md)). | | Request body | The body of the search: 64 KB at most (`413 request_too_large`). | When many of your users search at the same moment, their streams all use **your** key. Open a stream only for a search that a person is waiting for, close the connection when that person leaves the page (the server stops the search and frees the slot), and use the JSON mode for batch jobs and for anything that is not shown while it arrives. After a `429`, wait for `Retry-After` and open the stream again. ## Proxies, CDNs and compression - **Cloudflare passes the stream through unbuffered**: the first event arrived after 0.3 to 0.7 s and the others as the suppliers answered. The response has `cf-cache-status: DYNAMIC`. - **Streams are not compressed.** `Content-Encoding` is absent whatever you send in `Accept-Encoding`. The JSON mode is compressed (26 KB instead of 700 KB for the same search). - **Do not buffer in your own proxy.** In nginx set `proxy_buffering off;` for the route that forwards the stream, turn off response compression for `text/event-stream` and raise the proxy read timeout above 60 seconds. - **Use your own `User-Agent`.** Cloudflare answers the default agents of Python's `urllib` and Perl's `libwww-perl` with `403` and the text `error code: 1010`: see [Errors that do not come from the API](https://fx-port.com/docs/errors.md#errors-that-do-not-come-from-the-api). `curl`, Go, Java, Node.js, PHP and Python `requests` are not blocked. - **Decode as UTF-8.** Some HTTP libraries assume ISO-8859-1 for `text/*` types without a charset (Python `requests` does): set the encoding to UTF-8 or read bytes, or the names of airports and cities come out garbled. - **A line can be 200 KB or more.** Do not use a reader with a small fixed line limit. ## A complete client One program per language. Each sends a search, ignores comment lines, keeps the offers of every result set under its `searchId`, prints every event with the time it arrived and ends at `done`. They read the key from `FXPORT_API_KEY` and search for a date 30 days from today. The Java and Kotlin versions count the offers with regular expressions because the JDK has no JSON parser: parse the `data` lines with Jackson or Gson to keep the offers. **Read the search stream, merge the offers and stop at done** **cURL** ```bash # --no-buffer shows every event when it arrives instead of when the program exits. DEPARTURE=$(date -d '+30 days' +%F 2>/dev/null || date -v+30d +%F) curl --no-buffer --silent --show-error \ --request POST "https://api.fx-port.com/api/v1/get_flights?stream=true" \ --header "Authorization: Bearer fxp_test_YOUR_KEY" \ --header "Content-Type: application/json" \ --header "Accept: text/event-stream" \ --header "User-Agent: my-flight-app/1.0" \ --data "{\"origin\":\"ALG\",\"destination\":\"CDG\",\"departure_date\":\"$DEPARTURE\",\"cabin_class\":\"economy\",\"passengers\":{\"adults\":1}}" # Only the event names, each with the time it arrived: # ... | grep --line-buffered '^event:' | while read -r line; do echo "$(date +%T) $line"; done ``` **Node.js** ```javascript // Node.js 18 or later. No library is needed: fetch() gives a stream and a few lines split it into events. const API_KEY = process.env.FXPORT_API_KEY ?? "fxp_test_YOUR_KEY"; const BASE_URL = "https://api.fx-port.com/api/v1"; const departure = new Date(Date.now() + 30 * 86_400_000).toISOString().slice(0, 10); const startedAt = Date.now(); const stamp = () => "[" + ((Date.now() - startedAt) / 1000).toFixed(2) + "s]"; const response = await fetch(BASE_URL + "/get_flights?stream=true", { method: "POST", headers: { Authorization: "Bearer " + API_KEY, "Content-Type": "application/json", Accept: "text/event-stream", "User-Agent": "my-flight-app/1.0", }, body: JSON.stringify({ origin: "ALG", destination: "CDG", departure_date: departure, cabin_class: "economy", passengers: { adults: 1 }, }), signal: AbortSignal.timeout(120_000), }); if (!response.ok) { // A problem with the request is reported before the stream starts, as an ordinary JSON error. console.error(response.status, await response.text()); process.exit(1); } const offersBySearch = new Map(); // searchId -> offers: price an offer with the searchId of the result set it came in const currentSearch = new Map(); // supplierId -> searchId of the result set being received let done = false; function handle(event, data) { switch (event) { case "start": console.log(stamp(), "start, suppliers:", data.expectedSuppliers.join(", ")); break; case "supplier_start": // the offers follow in offers_chunk events case "supplier_result": { // every offer of the result set is in this event const searchId = data.data.searchId; currentSearch.set(data.supplierId, searchId); const offers = offersBySearch.get(searchId) ?? []; offers.push(...data.data.results.offers); offersBySearch.set(searchId, offers); console.log(stamp(), event, data.supplierId, searchId, "totalOffers", data.data.results.totalOffers); break; } case "offers_chunk": offersBySearch.get(currentSearch.get(data.supplierId)).push(...data.offers); console.log(stamp(), " chunk", data.chunkNumber + "/" + data.totalChunks, data.supplierId, "+" + data.offersInChunk); break; case "supplier_complete": console.log(stamp(), data.supplierId, "complete:", data.complete); break; case "error": console.warn(stamp(), data.supplierId, "failed:", data.error); break; case "summary": console.log(stamp(), "summary", JSON.stringify(data.summary)); break; case "done": done = true; break; } } const decoder = new TextDecoder(); let buffer = ""; let eventName = "message"; let dataLines = []; for await (const chunk of response.body) { buffer += decoder.decode(chunk, { stream: true }); let newline; while ((newline = buffer.indexOf("\n")) >= 0) { const line = buffer.slice(0, newline).replace(/\r$/, ""); buffer = buffer.slice(newline + 1); if (line === "") { // A blank line ends the event. if (dataLines.length) handle(eventName, JSON.parse(dataLines.join("\n"))); eventName = "message"; dataLines = []; } else if (line.startsWith(":")) { // A comment such as ": keep-alive": ignore it. } else { const colon = line.indexOf(":"); const field = colon < 0 ? line : line.slice(0, colon); const value = colon < 0 ? "" : line.slice(colon + 1).replace(/^ /, ""); if (field === "event") eventName = value; else if (field === "data") dataLines.push(value); } } } const total = [...offersBySearch.values()].reduce((sum, offers) => sum + offers.length, 0); console.log(stamp(), total + " offers in " + offersBySearch.size + " result sets"); if (!done) { console.error("The stream ended without a done event: run the search again."); process.exit(1); } ``` **Python** ```python # pip install requests import json import os import sys import time from datetime import date, timedelta import requests API_KEY = os.environ.get("FXPORT_API_KEY", "fxp_test_YOUR_KEY") BASE_URL = "https://api.fx-port.com/api/v1" started = time.monotonic() def stamp(): return "[%5.2fs]" % (time.monotonic() - started) body = { "origin": "ALG", "destination": "CDG", "departure_date": (date.today() + timedelta(days=30)).isoformat(), "cabin_class": "economy", "passengers": {"adults": 1}, } headers = {"Authorization": "Bearer " + API_KEY, "Accept": "text/event-stream", "User-Agent": "my-flight-app/1.0"} offers_by_search = {} # searchId -> offers: price an offer with the searchId of the result set it came in current_search = {} # supplierId -> searchId of the result set being received done = False def handle(event, data): global done if event == "start": print(stamp(), "start, suppliers:", ", ".join(data["expectedSuppliers"])) elif event in ("supplier_start", "supplier_result"): # the offers are here, or follow in offers_chunk events search_id = data["data"]["searchId"] current_search[data["supplierId"]] = search_id offers_by_search.setdefault(search_id, []).extend(data["data"]["results"]["offers"]) print(stamp(), event, data["supplierId"], search_id, "totalOffers", data["data"]["results"]["totalOffers"]) elif event == "offers_chunk": offers_by_search[current_search[data["supplierId"]]].extend(data["offers"]) print(stamp(), " chunk %s/%s" % (data["chunkNumber"], data["totalChunks"]), data["supplierId"], "+%s" % data["offersInChunk"]) elif event == "supplier_complete": print(stamp(), data["supplierId"], "complete:", data["complete"]) elif event == "error": print(stamp(), data["supplierId"], "failed:", data["error"], file=sys.stderr) elif event == "summary": print(stamp(), "summary", json.dumps(data["summary"])) elif event == "done": done = True url = BASE_URL + "/get_flights?stream=true" # timeout = (connect, read): the read timeout counts the silence between two chunks, not the whole stream. with requests.post(url, json=body, headers=headers, stream=True, timeout=(10, 90)) as response: if response.status_code != 200: # A problem with the request is reported before the stream starts, as an ordinary JSON error. sys.exit("%s %s" % (response.status_code, response.text)) response.encoding = "utf-8" # requests assumes ISO-8859-1 for text/* types without a charset event, data_lines = "message", [] for line in response.iter_lines(decode_unicode=True): if line == "": # A blank line ends the event. if data_lines: handle(event, json.loads("\n".join(data_lines))) event, data_lines = "message", [] elif line.startswith(":"): continue # A comment such as ": keep-alive": ignore it. else: field, _, value = line.partition(":") value = value[1:] if value.startswith(" ") else value if field == "event": event = value elif field == "data": data_lines.append(value) total = sum(len(offers) for offers in offers_by_search.values()) print(stamp(), "%d offers in %d result sets" % (total, len(offers_by_search))) if not done: sys.exit("The stream ended without a done event: run the search again.") ``` **PHP** ```php sprintf('[%5.2fs]', microtime(true) - $started); $body = json_encode([ 'origin' => 'ALG', 'destination' => 'CDG', 'departure_date' => date('Y-m-d', strtotime('+30 days')), 'cabin_class' => 'economy', 'passengers' => ['adults' => 1], ]); $offersBySearch = []; // searchId => offers: price an offer with the searchId of the result set it came in $currentSearch = []; // supplierId => searchId of the result set being received $done = false; $handle = function (string $event, array $data) use (&$offersBySearch, &$currentSearch, &$done, $stamp) { switch ($event) { case 'start': echo $stamp(), ' start, suppliers: ', implode(', ', $data['expectedSuppliers']), "\n"; break; case 'supplier_start': // the offers follow in offers_chunk events case 'supplier_result': // every offer of the result set is in this event $searchId = $data['data']['searchId']; $currentSearch[$data['supplierId']] = $searchId; $offersBySearch[$searchId] = array_merge($offersBySearch[$searchId] ?? [], $data['data']['results']['offers']); echo $stamp(), " $event {$data['supplierId']} $searchId totalOffers {$data['data']['results']['totalOffers']}\n"; break; case 'offers_chunk': $searchId = $currentSearch[$data['supplierId']]; $offersBySearch[$searchId] = array_merge($offersBySearch[$searchId], $data['offers']); echo $stamp(), " chunk {$data['chunkNumber']}/{$data['totalChunks']} {$data['supplierId']} +{$data['offersInChunk']}\n"; break; case 'supplier_complete': echo $stamp(), " {$data['supplierId']} complete: ", json_encode($data['complete']), "\n"; break; case 'error': fwrite(STDERR, $stamp() . " {$data['supplierId']} failed: {$data['error']}\n"); break; case 'summary': echo $stamp(), ' summary ', json_encode($data['summary']), "\n"; break; case 'done': $done = true; break; } }; $buffer = ''; $errorBody = ''; $eventName = 'message'; $dataLines = []; $ch = curl_init($baseUrl . '/get_flights?stream=true'); curl_setopt_array($ch, [ CURLOPT_POST => true, CURLOPT_POSTFIELDS => $body, CURLOPT_HTTPHEADER => [ 'Authorization: Bearer ' . $apiKey, 'Content-Type: application/json', 'Accept: text/event-stream', 'User-Agent: my-flight-app/1.0', ], CURLOPT_TIMEOUT => 120, CURLOPT_WRITEFUNCTION => function ($ch, string $chunk) use (&$buffer, &$errorBody, &$eventName, &$dataLines, $handle) { if (curl_getinfo($ch, CURLINFO_RESPONSE_CODE) !== 200) { $errorBody .= $chunk; // a problem with the request is an ordinary JSON error return strlen($chunk); } $buffer .= $chunk; while (($newline = strpos($buffer, "\n")) !== false) { $line = rtrim(substr($buffer, 0, $newline), "\r"); $buffer = substr($buffer, $newline + 1); if ($line === '') { // A blank line ends the event. if ($dataLines) { $handle($eventName, json_decode(implode("\n", $dataLines), true)); } $eventName = 'message'; $dataLines = []; } elseif ($line[0] === ':') { continue; // A comment such as ": keep-alive": ignore it. } else { [$field, $value] = array_pad(explode(':', $line, 2), 2, ''); $value = ltrim($value, ' '); if ($field === 'event') { $eventName = $value; } elseif ($field === 'data') { $dataLines[] = $value; } } } return strlen($chunk); }, ]); curl_exec($ch); $status = curl_getinfo($ch, CURLINFO_RESPONSE_CODE); curl_close($ch); if ($status !== 200) { fwrite(STDERR, "$status $errorBody\n"); exit(1); } $total = array_sum(array_map('count', $offersBySearch)); echo $stamp(), " $total offers in ", count($offersBySearch), " result sets\n"; if (!$done) { fwrite(STDERR, "The stream ended without a done event: run the search again.\n"); exit(1); } ``` **Go** ```go package main import ( "bufio" "bytes" "context" "encoding/json" "fmt" "io" "net/http" "os" "strings" "time" ) const baseURL = "https://api.fx-port.com/api/v1" func text(v any) string { s, _ := v.(string); return s } func object(v any) map[string]any { m, _ := v.(map[string]any); return m } func main() { apiKey := os.Getenv("FXPORT_API_KEY") if apiKey == "" { apiKey = "fxp_test_YOUR_KEY" } started := time.Now() stamp := func() string { return fmt.Sprintf("[%5.2fs]", time.Since(started).Seconds()) } payload, _ := json.Marshal(map[string]any{ "origin": "ALG", "destination": "CDG", "departure_date": time.Now().AddDate(0, 0, 30).Format("2006-01-02"), "cabin_class": "economy", "passengers": map[string]any{"adults": 1}, }) // A context deadline bounds the whole stream; http.Client.Timeout would cut a long stream too. ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) defer cancel() req, _ := http.NewRequestWithContext(ctx, http.MethodPost, baseURL+"/get_flights?stream=true", bytes.NewReader(payload)) req.Header.Set("Authorization", "Bearer "+apiKey) req.Header.Set("Content-Type", "application/json") req.Header.Set("Accept", "text/event-stream") req.Header.Set("User-Agent", "my-flight-app/1.0") resp, err := http.DefaultClient.Do(req) if err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { // A problem with the request is reported before the stream starts, as an ordinary JSON error. body, _ := io.ReadAll(resp.Body) fmt.Fprintln(os.Stderr, resp.StatusCode, string(body)) os.Exit(1) } offersBySearch := map[string][]any{} // searchId -> offers: price an offer with the searchId of the result set it came in currentSearch := map[string]string{} // supplierId -> searchId of the result set being received done := false handle := func(event string, data map[string]any) { supplier := text(data["supplierId"]) switch event { case "start": fmt.Println(stamp(), "start, suppliers:", data["expectedSuppliers"]) case "supplier_start", "supplier_result": // the offers are here, or follow in offers_chunk events inner := object(data["data"]) searchID := text(inner["searchId"]) results := object(inner["results"]) currentSearch[supplier] = searchID offers, _ := results["offers"].([]any) offersBySearch[searchID] = append(offersBySearch[searchID], offers...) fmt.Println(stamp(), event, supplier, searchID, "totalOffers", results["totalOffers"]) case "offers_chunk": offers, _ := data["offers"].([]any) searchID := currentSearch[supplier] offersBySearch[searchID] = append(offersBySearch[searchID], offers...) fmt.Printf("%s chunk %v/%v %s +%d\n", stamp(), data["chunkNumber"], data["totalChunks"], supplier, len(offers)) case "supplier_complete": fmt.Println(stamp(), supplier, "complete:", data["complete"]) case "error": fmt.Fprintln(os.Stderr, stamp(), supplier, "failed:", data["error"]) case "summary": summary, _ := json.Marshal(data["summary"]) fmt.Println(stamp(), "summary", string(summary)) case "done": done = true } } // A data line can be 200 KB or more: read lines with bufio.Reader, not bufio.Scanner (64 KB limit by default). reader := bufio.NewReaderSize(resp.Body, 1<<20) event, dataLines := "message", []string{} for { line, readErr := reader.ReadString('\n') line = strings.TrimRight(line, "\r\n") switch { case line == "" && readErr == nil: // A blank line ends the event. if len(dataLines) > 0 { var data map[string]any if err := json.Unmarshal([]byte(strings.Join(dataLines, "\n")), &data); err == nil { handle(event, data) } } event, dataLines = "message", nil case strings.HasPrefix(line, ":"): // A comment such as ": keep-alive": ignore it. case line != "": field, value, _ := strings.Cut(line, ":") value = strings.TrimPrefix(value, " ") if field == "event" { event = value } else if field == "data" { dataLines = append(dataLines, value) } } if readErr != nil { break } } total := 0 for _, offers := range offersBySearch { total += len(offers) } fmt.Println(stamp(), total, "offers in", len(offersBySearch), "result sets") if !done { fmt.Fprintln(os.Stderr, "The stream ended without a done event: run the search again.") os.Exit(1) } } ``` **Java** ```java import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; import java.time.Duration; import java.time.LocalDate; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.regex.Matcher; import java.util.regex.Pattern; import java.util.stream.Collectors; // Java 11 or later: java FxPortExample.java // The JDK has no JSON parser, so this sample reads the few fields it needs with regular expressions and counts the // offers. To keep the offers themselves, parse each data line with Jackson or Gson. public class FxPortExample { static final String BASE_URL = "https://api.fx-port.com/api/v1"; static long started = System.nanoTime(); static String stamp() { return String.format("[%5.2fs]", (System.nanoTime() - started) / 1e9); } static String field(String json, String name) { Matcher m = Pattern.compile("\"" + name + "\":(\"([^\"]*)\"|[0-9.]+|true|false)").matcher(json); return m.find() ? (m.group(2) != null ? m.group(2) : m.group(1)) : ""; } static int countOffers(String json) { Matcher m = Pattern.compile("\"type\":\"flight_offer\"").matcher(json); int count = 0; while (m.find()) count++; return count; } static final Map offersBySearch = new HashMap<>(); // searchId -> offers received static final Map currentSearch = new HashMap<>(); // supplierId -> searchId being received static boolean done = false; static void handle(String event, String data) { String supplier = field(data, "supplierId"); switch (event) { case "start": System.out.println(stamp() + " start: " + data.substring(0, Math.min(data.length(), 160))); break; case "supplier_start": // the offers follow in offers_chunk events case "supplier_result": { // every offer of the result set is in this event String searchId = field(data, "searchId"); currentSearch.put(supplier, searchId); offersBySearch.merge(searchId, countOffers(data), Integer::sum); System.out.println(stamp() + " " + event + " " + supplier + " " + searchId + " totalOffers " + field(data, "totalOffers")); break; } case "offers_chunk": offersBySearch.merge(currentSearch.get(supplier), countOffers(data), Integer::sum); System.out.println(stamp() + " chunk " + field(data, "chunkNumber") + "/" + field(data, "totalChunks") + " " + supplier + " +" + field(data, "offersInChunk")); break; case "supplier_complete": System.out.println(stamp() + " " + supplier + " complete: " + field(data, "complete")); break; case "error": System.err.println(stamp() + " " + supplier + " failed: " + field(data, "error")); break; case "summary": System.out.println(stamp() + " summary " + data.replaceAll(".*\"summary\":(\\{[^}]*\\}).*", "$1")); break; case "done": done = true; break; } } public static void main(String[] args) throws Exception { String apiKey = System.getenv("FXPORT_API_KEY") != null ? System.getenv("FXPORT_API_KEY") : "fxp_test_YOUR_KEY"; String body = "{\"origin\":\"ALG\",\"destination\":\"CDG\",\"departure_date\":\"" + LocalDate.now().plusDays(30) + "\",\"cabin_class\":\"economy\",\"passengers\":{\"adults\":1}}"; HttpRequest request = HttpRequest.newBuilder(URI.create(BASE_URL + "/get_flights?stream=true")) .timeout(Duration.ofMinutes(2)) .header("Authorization", "Bearer " + apiKey) .header("Content-Type", "application/json") .header("Accept", "text/event-stream") .header("User-Agent", "my-flight-app/1.0") .POST(HttpRequest.BodyPublishers.ofString(body)) .build(); // ofLines() hands over each line as soon as it has arrived. started = System.nanoTime(); HttpResponse> response = HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofLines()); if (response.statusCode() != 200) { // A problem with the request is reported before the stream starts, as an ordinary JSON error. System.err.println(response.statusCode() + " " + response.body().collect(Collectors.joining("\n"))); System.exit(1); } String[] event = {"message"}; List dataLines = new ArrayList<>(); response.body().forEach(line -> { if (line.isEmpty()) { // A blank line ends the event. if (!dataLines.isEmpty()) handle(event[0], String.join("\n", dataLines)); event[0] = "message"; dataLines.clear(); } else if (line.startsWith(":")) { // A comment such as ": keep-alive": ignore it. } else { int colon = line.indexOf(':'); String name = colon < 0 ? line : line.substring(0, colon); String value = colon < 0 ? "" : line.substring(colon + 1).replaceFirst("^ ", ""); if (name.equals("event")) event[0] = value; else if (name.equals("data")) dataLines.add(value); } }); int total = offersBySearch.values().stream().mapToInt(Integer::intValue).sum(); System.out.println(stamp() + " " + total + " offers in " + offersBySearch.size() + " result sets"); if (!done) { System.err.println("The stream ended without a done event: run the search again."); System.exit(1); } } } ``` **C#** ```csharp // .NET 8 or later: dotnet run using System.Diagnostics; using System.Text; using System.Text.Json; var apiKey = Environment.GetEnvironmentVariable("FXPORT_API_KEY") ?? "fxp_test_YOUR_KEY"; const string BaseUrl = "https://api.fx-port.com/api/v1"; var clock = Stopwatch.StartNew(); string Stamp() => $"[{clock.Elapsed.TotalSeconds,5:F2}s]"; var body = JsonSerializer.Serialize(new { origin = "ALG", destination = "CDG", departure_date = DateTime.UtcNow.AddDays(30).ToString("yyyy-MM-dd"), cabin_class = "economy", passengers = new { adults = 1 }, }); using var http = new HttpClient { Timeout = Timeout.InfiniteTimeSpan }; using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2)); // bounds the whole stream using var request = new HttpRequestMessage(HttpMethod.Post, BaseUrl + "/get_flights?stream=true") { Content = new StringContent(body, Encoding.UTF8, "application/json"), }; request.Headers.TryAddWithoutValidation("Authorization", "Bearer " + apiKey); request.Headers.TryAddWithoutValidation("Accept", "text/event-stream"); request.Headers.TryAddWithoutValidation("User-Agent", "my-flight-app/1.0"); // ResponseHeadersRead returns as soon as the headers arrive, so the body can be read while it is still being sent. using var response = await http.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cts.Token); if (!response.IsSuccessStatusCode) { // A problem with the request is reported before the stream starts, as an ordinary JSON error. Console.Error.WriteLine($"{(int)response.StatusCode} {await response.Content.ReadAsStringAsync(cts.Token)}"); return 1; } var offersBySearch = new Dictionary>(); // searchId -> offers: price an offer with the searchId of the result set it came in var currentSearch = new Dictionary(); // supplierId -> searchId of the result set being received var done = false; void Handle(string eventName, JsonElement data) { var supplier = data.TryGetProperty("supplierId", out var s) ? s.GetString() ?? "" : ""; switch (eventName) { case "start": Console.WriteLine($"{Stamp()} start, suppliers: {data.GetProperty("expectedSuppliers")}"); break; case "supplier_start": // the offers follow in offers_chunk events case "supplier_result": // every offer of the result set is in this event var inner = data.GetProperty("data"); var searchId = inner.GetProperty("searchId").GetString()!; var results = inner.GetProperty("results"); currentSearch[supplier] = searchId; if (!offersBySearch.TryGetValue(searchId, out var list)) offersBySearch[searchId] = list = new(); list.AddRange(results.GetProperty("offers").EnumerateArray().Select(o => o.Clone())); Console.WriteLine($"{Stamp()} {eventName} {supplier} {searchId} totalOffers {results.GetProperty("totalOffers")}"); break; case "offers_chunk": offersBySearch[currentSearch[supplier]].AddRange(data.GetProperty("offers").EnumerateArray().Select(o => o.Clone())); Console.WriteLine($"{Stamp()} chunk {data.GetProperty("chunkNumber")}/{data.GetProperty("totalChunks")} {supplier} +{data.GetProperty("offersInChunk")}"); break; case "supplier_complete": Console.WriteLine($"{Stamp()} {supplier} complete: {data.GetProperty("complete")}"); break; case "error": Console.Error.WriteLine($"{Stamp()} {supplier} failed: {data.GetProperty("error")}"); break; case "summary": Console.WriteLine($"{Stamp()} summary {data.GetProperty("summary")}"); break; case "done": done = true; break; } } await using var stream = await response.Content.ReadAsStreamAsync(cts.Token); using var reader = new StreamReader(stream, Encoding.UTF8); var currentEvent = "message"; var dataLines = new List(); while (await reader.ReadLineAsync(cts.Token) is string line) { if (line.Length == 0) { // A blank line ends the event. if (dataLines.Count > 0) { using var document = JsonDocument.Parse(string.Join("\n", dataLines)); Handle(currentEvent, document.RootElement); } currentEvent = "message"; dataLines.Clear(); } else if (line.StartsWith(':')) { // A comment such as ": keep-alive": ignore it. } else { var colon = line.IndexOf(':'); var name = colon < 0 ? line : line[..colon]; var value = colon < 0 ? "" : line[(colon + 1)..].TrimStart(' '); if (name == "event") currentEvent = value; else if (name == "data") dataLines.Add(value); } } Console.WriteLine($"{Stamp()} {offersBySearch.Values.Sum(offers => offers.Count)} offers in {offersBySearch.Count} result sets"); if (!done) { Console.Error.WriteLine("The stream ended without a done event: run the search again."); return 1; } return 0; ``` **Swift** ```swift // macOS 12 / iOS 15 or later: swift main.swift import Foundation let apiKey = ProcessInfo.processInfo.environment["FXPORT_API_KEY"] ?? "fxp_test_YOUR_KEY" let baseURL = "https://api.fx-port.com/api/v1" let startedAt = Date() func stamp() -> String { String(format: "[%5.2fs]", Date().timeIntervalSince(startedAt)) } let dayFormat = ISO8601DateFormatter() dayFormat.formatOptions = [.withFullDate] var request = URLRequest(url: URL(string: baseURL + "/get_flights?stream=true")!) request.httpMethod = "POST" request.timeoutInterval = 120 // the silence allowed between two chunks request.setValue("Bearer " + apiKey, forHTTPHeaderField: "Authorization") request.setValue("application/json", forHTTPHeaderField: "Content-Type") request.setValue("text/event-stream", forHTTPHeaderField: "Accept") request.setValue("my-flight-app/1.0", forHTTPHeaderField: "User-Agent") request.httpBody = try JSONSerialization.data(withJSONObject: [ "origin": "ALG", "destination": "CDG", "departure_date": dayFormat.string(from: Date().addingTimeInterval(30 * 86_400)), "cabin_class": "economy", "passengers": ["adults": 1], ] as [String: Any]) let (bytes, response) = try await URLSession.shared.bytes(for: request) guard let http = response as? HTTPURLResponse, http.statusCode == 200 else { // A problem with the request is reported before the stream starts, as an ordinary JSON error. var text = "" for try await line in bytes.lines { text += line } FileHandle.standardError.write(Data(("HTTP error: " + text + "\n").utf8)) exit(1) } var offersBySearch: [String: [Any]] = [:] // searchId -> offers: price an offer with the searchId of the result set it came in var currentSearch: [String: String] = [:] // supplierId -> searchId of the result set being received var done = false func handle(_ event: String, _ data: [String: Any]) { let supplier = data["supplierId"] as? String ?? "" switch event { case "start": print(stamp(), "start, suppliers:", (data["expectedSuppliers"] as? [String] ?? []).joined(separator: ", ")) case "supplier_start", "supplier_result": // the offers are here, or follow in offers_chunk events let inner = data["data"] as? [String: Any] ?? [:] let results = inner["results"] as? [String: Any] ?? [:] let searchId = inner["searchId"] as? String ?? "" currentSearch[supplier] = searchId offersBySearch[searchId, default: []] += results["offers"] as? [Any] ?? [] print(stamp(), event, supplier, searchId, "totalOffers", results["totalOffers"] ?? "?") case "offers_chunk": let offers = data["offers"] as? [Any] ?? [] offersBySearch[currentSearch[supplier] ?? "", default: []] += offers print(stamp(), " chunk", data["chunkNumber"] ?? "?", "of", data["totalChunks"] ?? "?", supplier, "+\(offers.count)") case "supplier_complete": print(stamp(), supplier, "complete:", data["complete"] as? Bool ?? false) case "error": print(stamp(), supplier, "failed:", data["error"] ?? "?") case "summary": let summary = try? JSONSerialization.data(withJSONObject: data["summary"] ?? [:], options: [.sortedKeys]) print(stamp(), "summary", summary.flatMap { String(data: $0, encoding: .utf8) } ?? "") case "done": done = true default: break } } // bytes.lines drops empty lines, so an event is handled when its data line arrives (each event has exactly one). var eventName = "message" for try await line in bytes.lines { if line.hasPrefix(":") { continue } // A comment such as ": keep-alive": ignore it. guard let colon = line.firstIndex(of: ":") else { continue } let name = String(line[..() // searchId -> offers received val currentSearch = mutableMapOf() // supplierId -> searchId being received var done = false fun handle(event: String, data: String) { val supplier = field(data, "supplierId") when (event) { "start" -> println(stamp() + " start: " + data.take(160)) "supplier_start", "supplier_result" -> { // the offers are here, or follow in offers_chunk events val searchId = field(data, "searchId") currentSearch[supplier] = searchId offersBySearch[searchId] = (offersBySearch[searchId] ?: 0) + countOffers(data) println(stamp() + " " + event + " " + supplier + " " + searchId + " totalOffers " + field(data, "totalOffers")) } "offers_chunk" -> { val searchId = currentSearch[supplier] ?: "" offersBySearch[searchId] = (offersBySearch[searchId] ?: 0) + countOffers(data) println(stamp() + " chunk " + field(data, "chunkNumber") + "/" + field(data, "totalChunks") + " " + supplier + " +" + field(data, "offersInChunk")) } "supplier_complete" -> println(stamp() + " " + supplier + " complete: " + field(data, "complete")) "error" -> System.err.println(stamp() + " " + supplier + " failed: " + field(data, "error")) "summary" -> println(stamp() + " summary " + Regex("\"summary\":(\\{[^}]*\\})").find(data)?.groupValues?.get(1)) "done" -> done = true } } fun main() { val apiKey = System.getenv("FXPORT_API_KEY") ?: "fxp_test_YOUR_KEY" val body = """{"origin":"ALG","destination":"CDG","departure_date":"${LocalDate.now().plusDays(30)}","cabin_class":"economy","passengers":{"adults":1}}""" val request = HttpRequest.newBuilder(URI.create("$BASE_URL/get_flights?stream=true")) .timeout(Duration.ofMinutes(2)) .header("Authorization", "Bearer $apiKey") .header("Content-Type", "application/json") .header("Accept", "text/event-stream") .header("User-Agent", "my-flight-app/1.0") .POST(HttpRequest.BodyPublishers.ofString(body)) .build() // ofLines() hands over each line as soon as it has arrived. startedAt = System.nanoTime() val response = HttpClient.newHttpClient().send(request, HttpResponse.BodyHandlers.ofLines()) if (response.statusCode() != 200) { // A problem with the request is reported before the stream starts, as an ordinary JSON error. System.err.println(response.statusCode().toString() + " " + response.body().iterator().asSequence().joinToString("\n")) exitProcess(1) } var event = "message" val dataLines = mutableListOf() response.body().forEach { line -> when { line.isEmpty() -> { // A blank line ends the event. if (dataLines.isNotEmpty()) handle(event, dataLines.joinToString("\n")) event = "message" dataLines.clear() } line.startsWith(":") -> {} // A comment such as ": keep-alive": ignore it. else -> { val colon = line.indexOf(':') val name = if (colon < 0) line else line.substring(0, colon) val value = if (colon < 0) "" else line.substring(colon + 1).removePrefix(" ") if (name == "event") event = value else if (name == "data") dataLines.add(value) } } } println(stamp() + " " + offersBySearch.values.sum() + " offers in " + offersBySearch.size + " result sets") if (!done) { System.err.println("The stream ended without a done event: run the search again.") exitProcess(1) } } ``` A run of the Node.js program against the sandbox (two suppliers, 184 offers; the first offers arrived after 1.8 seconds and the stream took 7 seconds): ~~~text [0.40s] start, suppliers: amadeus_aqc_dz_1, duffel_1 [1.45s] supplier_start duffel_1 fs_dfx_3f7927a9-e248-40ec-96ce-1453f3831f22 totalOffers 108 [1.81s] chunk 1/3 duffel_1 +50 [1.94s] chunk 2/3 duffel_1 +50 [1.94s] chunk 3/3 duffel_1 +8 [1.94s] duffel_1 complete: true [6.57s] supplier_start amadeus_aqc_dz_1 fs_4c684e2a-5de8-447e-9d5e-2a02ab8d8575 totalOffers 76 [6.93s] chunk 1/2 amadeus_aqc_dz_1 +50 [7.05s] chunk 2/2 amadeus_aqc_dz_1 +26 [7.05s] amadeus_aqc_dz_1 complete: true [7.05s] summary {"totalSuppliers":2,"successfulSuppliers":2,"failedSuppliers":0,"totalOffers":184} [7.05s] 184 offers in 2 result sets ~~~ ## Test it in the sandbox | You want to see | Send | |---|---| | Chunks | Any search that returns more than 50 offers from a supplier: `ALG` to `CDG` in 30 days returns 90 to 110 from `duffel_1` | | A result set with `complete: false`, then a second `searchId` | A route that touches Algeria, so that `amadeus_aqc_dz_1` answers (`ALG` to `CDG`); it happens on most searches, not on all | | One supplier only | `"supplier_id": "duffel_1"` | | An `error` event | A well-formed code that does not exist, for example `"origin": "XXX"` with `"supplier_id": "duffel_1"`: the supplier refuses the request, the stream carries one `error` event (`cause: supplier_error`), then `summary` (`success: false`) and `done`. A refusal of a request does not count as an outage; a supplier that keeps timing out or answering 5xx is skipped for a short while, for every agency of the environment (`cause: supplier_temporarily_unavailable`) | | A cut stream | Close your connection while the search runs: nothing is kept on the server | ## Related - [Search flights](https://fx-port.com/docs/flights/search.md): the request, the filters and the offers. - [API reference: Search flights](https://fx-port.com/docs/api/get-flights.md): every field, with the events listed. - [Price a flight offer](https://fx-port.com/docs/flights/price.md): the next step, with the `searchId` and the offer `id`. - [Rate limits](https://fx-port.com/docs/rate-limits.md) and [Errors](https://fx-port.com/docs/errors.md).