# Feeds Stream each feed's real-time events and download its hourly and daily Parquet exports. Each feed comes from a `Client` method that returns a `Feed`, where `T` is the feed's event type. A `Feed` owns a clone of the client, so it is cheap to create and does not borrow the `Client`. Every feed has a real-time [stream](#streams) and hourly Parquet [exports](#exports) that roll up into daily exports. Streaming and exports need separate scopes on the API key. The event types are listed under [Event types](https://docs.synthient.com/sdks/rust/types#event-types). ```rust pub fn proxies(&self) -> Feed pub fn anonymizers(&self) -> Feed pub fn torrents(&self) -> Feed pub fn ja4t(&self) -> Feed pub fn helios_http(&self) -> Feed pub fn helios_https(&self) -> Feed pub fn helios_dns(&self) -> Feed pub fn helios_adb(&self) -> Feed ``` | Method | Event type | Stream scope | Export scope | | -------------- | ------------------ | ----------------------- | --------------------- | | `proxies` | `ProxyEvent` | `PROXY_FIREHOSE` | `PROXY_FEEDS` | | `anonymizers` | `AnonymizerEvent` | `ANONYMIZERS_STREAM` | `ANONYMIZERS_FEED` | | `torrents` | `TorrentEvent` | `TORRENTS_STREAM` | `TORRENTS_FEED` | | `ja4t` | `Ja4tEvent` | `JA4T_STREAM` | `JA4T_FEED` | | `helios_http` | `HeliosHttpEvent` | `HONEYPOT_HTTP_STREAM` | `HONEYPOT_HTTP_FEED` | | `helios_https` | `HeliosHttpsEvent` | `HONEYPOT_HTTPS_STREAM` | `HONEYPOT_HTTPS_FEED` | | `helios_dns` | `HeliosDnsEvent` | `HONEYPOT_DNS_STREAM` | `HONEYPOT_DNS_FEED` | | `helios_adb` | `HeliosAdbEvent` | `HONEYPOT_ADB_STREAM` | `HONEYPOT_ADB_FEED` | What each feed carries: - `proxies` carries residential, mobile, and datacenter proxy sightings. - `anonymizers` carries VPN and anonymizer IP ranges. - `torrents` carries torrents seen on the BitTorrent DHT and trackers. - `ja4t` carries JA4T TCP fingerprints. - The four Helios feeds carry honeypot captures: HTTP requests, TLS ClientHellos, tunnel destinations, and Android Debug Bridge commands. Despite its name, `helios_dns` does not carry DNS queries. See [Firehose](https://docs.synthient.com/enterprise/firehose), [Feeds](https://docs.synthient.com/enterprise/feeds), and [Helios](https://docs.synthient.com/enterprise/helios) for the data behind each feed. ## Streams ### stream Opens the feed's real-time event stream. ```rust pub async fn stream(&self) -> Result> ``` - **Connection:** Each call opens a new connection. The server keeps it open for up to 30 minutes and then closes it, which ends the stream with `None`. - **Errors:** A non-success response returns an `Error::Api`, and a 429 without `retry_after` means the key's monthly request budget is spent or it already has too many open streams. - **Timeout:** The client's `timeout` covers opening the stream but not reading it. - **Reconnecting:** The API has no resume cursor, so events sent between connections are missed. Call `stream` again to reconnect: ```rust loop { let mut events = client.proxies().stream().await?; while let Some(event) = events.next().await { let event = event?; println!( "{} {}", event.ip.unwrap_or_default(), event.provider.unwrap_or_default() ); } } ``` The loop reconnects immediately after a clean close. Opening a connection already uses the SDK's bounded retries, including `Retry-After` handling. Any error that remains after those retries, or occurs while reading the stream, stops this consumer. A `429` without `Retry-After` can indicate an exhausted monthly request budget or too many open streams; resolve that condition before reconnecting. Every feed streams the same way. Only the event type changes: **Streams** ```rust title="Proxies" let mut events = client.proxies().stream().await?; while let Some(event) = events.next().await { let event = event?; println!( "{:?} {:?} {:?} {:?} {:?}", event.ip, event.provider, event.r#type, event.country_code, event.asn ); } ``` ```rust title="Anonymizers" let mut events = client.anonymizers().stream().await?; while let Some(event) = events.next().await { let event = event?; println!( "{:?} {:?} {:?} {:?}", event.range_start, event.range_end, event.provider, event.r#type ); } ``` ```rust title="Torrents" let mut events = client.torrents().stream().await?; while let Some(event) = events.next().await { let event = event?; let peers = event.peers.as_ref().map_or(0, Vec::len); println!("{:?} {:?} {peers}", event.info_hash, event.name); } ``` ```rust title="JA4T" let mut events = client.ja4t().stream().await?; while let Some(event) = events.next().await { let event = event?; println!("{:?} {:?} {:?}", event.ip, event.ja4t, event.provider); } ``` ```rust title="Helios HTTP" let mut events = client.helios_http().stream().await?; while let Some(event) = events.next().await { let event = event?; if let Some(details) = &event.details { println!( "{} {:?} {:?} {:?}", event.domain, details.method, details.uri, event.meta.provider ); } } ``` ```rust title="Helios HTTPS" let mut events = client.helios_https().stream().await?; while let Some(event) = events.next().await { let event = event?; let Some(details) = &event.details else { continue; }; println!( "{} {:?} {:?}", event.domain, details.sni, details.handshake_version ); } ``` ```rust title="Helios DNS" let mut events = client.helios_dns().stream().await?; while let Some(event) = events.next().await { let event = event?; println!("{} {} {:?}", event.domain, event.port, event.meta.provider); } ``` ```rust title="Helios ADB" let mut events = client.helios_adb().stream().await?; while let Some(event) = events.next().await { let event = event?; println!( "{} {} {}", event.session, event.sequential_id, event.command ); } ``` Every field of a proxy, anonymizer, torrent, or JA4T event is an `Option`, because the API may omit any of them. Helios events require their core fields: `timestamp`, `tunnel_id`, `domain`, `port`, `protocol`, and `meta` on HTTP, HTTPS, and DNS events, and `session`, `sequential_id`, `command`, and `hash` on ADB events. An event that lacks one returns `Error::Decode`, which ends the stream like any other failure. `HeliosHttpsEvent::details` is `None` when the sensor could not parse the handshake, so check it before use. ### EventStream `EventStream` yields a feed's events from `stream`, or its exports from `all_exports`. `next` returns the next item, or `None` once the stream has ended. ```rust pub async fn next(&mut self) -> Option> ``` - **Results:** Blank keepalive lines are skipped. - **Errors:** A failure, such as a dropped connection, an event that doesn't decode, or a failed page request, is returned once, and the stream then ends. - **Cleanup:** Dropping the stream closes its connection or stops fetching pages. - **Traits:** `EventStream` implements the `futures` `Stream` trait with `Item = Result`, so it works with combinators from `futures` or `tokio-stream`. ## Exports Each feed has hourly Parquet exports that roll up into daily exports. - **Hourly exports:** An hourly export is available until its day's daily export is rolled up, shortly after 00:30 UTC the next day. After that, requests for the hour return a 400 `Error::Api`. - **Daily exports:** While a daily export is being rolled up, requests for it return a 404 `Error::Api` with `retry_after` set. ### Export IDs An export ID is one of: - `"latest"` - a UTC day as `"YYYY-MM-DD"` - a UTC hour as `"YYYY-MM-DD/H"`, where a zero-padded hour is also accepted `"latest"` is the feed's freshest export. `Export::id` values can be passed as they are. Any other string returns `Error::InvalidInput` before a request is sent. The Rust SDK has no constant or helper functions for export IDs, so build the string yourself. ### list\_exports Returns one page of the feed's exports, newest first. ```rust pub async fn list_exports(&self, limit: Option, cursor: Option<&str>) -> Result ``` ```rust let page = client.proxies().list_exports(Some(24), None).await?; for export in &page.exports { println!("{} {} {}", export.id, export.kind, export.row_count); } println!("next cursor: {:?}", page.next_cursor); ``` - **Input:** `limit` sets the page size, from 1 to 500, and `None` uses the API default of 100. - **Paging:** Pass a page's `next_cursor` as `cursor` to fetch the page after it, or call `all_exports` to read every export across pages. ### all\_exports Returns a stream of every export, newest first, starting at `cursor` when it is set. ```rust pub fn all_exports(&self, limit: Option, cursor: Option<&str>) -> ExportStream where T: 'static, ``` ```rust pub type ExportStream = EventStream; ``` ```rust let mut exports = client.proxies().all_exports(None, None); while let Some(export) = exports.next().await { let export = export?; println!("{} {} {}", export.id, export.kind, export.size_bytes); } ``` - **Paging:** It is not `async`: pages of `limit` exports are fetched lazily as the stream is read, and dropping the stream stops fetching. - **Errors:** An error is returned once and ends the stream, including an `Error::Protocol` when the API returns a cursor it already returned. ### export\_meta Returns the checksum, size, row count, and Parquet schema of an export. ```rust pub async fn export_meta(&self, id: &str) -> Result ``` ```rust let meta = client.proxies().export_meta("latest").await?; println!("{} {} {} {}", meta.id, meta.rows, meta.size, meta.checksum); for column in &meta.schema.fields { println!("{} {}", column.name, column.r#type); } ``` ### download\_url Returns a presigned URL for an export's Parquet file, as a `reqwest::Url`. ```rust pub async fn download_url(&self, id: &str) -> Result ``` ```rust let url = client.proxies().download_url("latest").await?; println!("{:?}", url.host_str()); ``` - **Lifetime:** The URL is valid for 24 hours. - **Security:** The URL grants access without an API key, so treat it as a secret. - **Use:** It is useful for tools that read Parquet over HTTPS directly. - **Errors:** A 4xx or 5xx response returns an `Error::Api`. A successful response that isn't a redirect, or a redirect without a valid `Location`, returns `Error::Protocol`. ### download Opens an export's Parquet file for reading. ```rust pub async fn download(&self, id: &str) -> Result ``` - **Security:** The API key is sent only to the Synthient API, never to the storage host that serves the file, and errors never include the presigned URL. - **Errors:** A non-success response from either returns an `Error::Api`. Up to 10 redirects from the storage host are followed, and more return `Error::Protocol`. Requests to the storage host are not retried. - **Timeout:** The client's `timeout` covers opening the download but not reading it. ### Download `Download` holds the contents of an export's Parquet file, read in chunks as they arrive. - `content_length` returns the size of the file in bytes, if the storage host sent it. - `next` returns the next chunk, or `None` once the file has been read. Each chunk is a `bytes::Bytes`, which dereferences to `[u8]`. ```rust pub fn content_length(&self) -> Option pub async fn next(&mut self) -> Option> ``` - **Traits:** `Download` implements the `futures` `Stream` trait with `Item = Result`. - **Cleanup:** Dropping it closes the connection. The SDK does not verify the file. Compare its SHA-256 with `Export::checksum` or `ExportMeta::checksum`. This program, adapted from the repository's `download_export` example, also uses the `sha2` and `hex` crates and Tokio's `fs` and `io-util` features: ```rust use sha2::{Digest, Sha256}; use synthient::{Client, ExportKind}; use tokio::fs::File; use tokio::io::{AsyncWriteExt, BufWriter}; #[tokio::main] async fn main() -> Result<(), Box> { let client = Client::new(std::env::var("SYNTHIENT_API_KEY")?)?; let feed = client.proxies(); let mut exports = feed.all_exports(None, None); let mut daily = None; while let Some(export) = exports.next().await { let export = export?; if export.kind == ExportKind::Daily { daily = Some(export); break; } } let daily = daily.ok_or("no daily export found")?; let mut download = feed.download(&daily.id).await?; if let Some(size) = download.content_length() { println!("downloading {} ({size} bytes)", daily.id); } let mut file = BufWriter::new(File::create("proxies.parquet").await?); let mut hash = Sha256::new(); while let Some(chunk) = download.next().await { let chunk = chunk?; hash.update(&chunk); file.write_all(&chunk).await?; } file.flush().await?; if hex::encode(hash.finalize()) != daily.checksum { return Err("checksum mismatch".into()); } Ok(()) } ```