nzb-nntp

Async NNTP client library for Rust, designed for high-performance Usenet article downloading.

Features

• Async/await — Built on Tokio for non-blocking I/O
• TLS — Implicit TLS via rustls with optional certificate verification bypass
• Authentication — AUTHINFO USER/PASS (RFC 4643)
• Pipelining — Send multiple ARTICLE/STAT commands before reading responses, dramatically improving throughput on high-latency links
• Connection pooling — Per-server pools with health checking, idle timeouts, and automatic reconnection
• Multi-server failover — Priority-based server selection with automatic retry on next server
• Penalty system — Temporarily backs off servers experiencing errors (auth, timeout, connection)
• GZIP compression — XFEATURE COMPRESS GZIP negotiation and transparent decompression
• SOCKS5 proxy — Route connections through SOCKS5 proxies with optional authentication
• Bandwidth limiting — Global bandwidth throttling
• Pause/resume/shutdown — Runtime control of download operations

Quick Start

Add to your Cargo.toml:

[dependencies]
nzb-nntp = "0.1"
tokio = { version = "1", features = ["full"] }

Single Connection

use nzb_nntp::{NntpConnection, ServerConfig};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let config = ServerConfig {
        id: "my-server".into(),
        name: "Usenet Provider".into(),
        host: "news.example.com".into(),
        port: 563,
        ssl: true,
        ssl_verify: true,
        username: Some("user".into()),
        password: Some("pass".into()),
        ..ServerConfig::default()
    };

    let mut conn = NntpConnection::new("conn-1".into());
    conn.connect(&config).await?;

    // Fetch an article by message-id
    let response = conn.fetch_article("abc123@example.com").await?;
    println!("Article size: {} bytes", response.data.unwrap().len());

    // Check if an article exists (no data transfer)
    let stat = conn.stat_article("def456@example.com").await?;
    println!("Article exists: code {}", stat.code);

    conn.quit().await?;
    Ok(())
}

Group Browsing

use nzb_nntp::{NntpConnection, ServerConfig, ArticleRange};

async fn browse(conn: &mut NntpConnection) -> nzb_nntp::NntpResult<()> {
    // Select a newsgroup
    let group = conn.group("alt.binaries.test").await?;
    println!("{}: {} articles ({}-{})", group.name, group.count, group.first, group.last);

    // Get article overview data
    let overview = conn.xover(group.last - 100, group.last).await?;
    for entry in &overview {
        println!("  [{}] {} ({}B)", entry.article_num, entry.subject, entry.bytes);
    }

    // Search by subject pattern
    let matches = conn.xpat(
        "subject",
        ArticleRange::Range(group.first, group.last),
        &["*linux*"],
    ).await?;
    println!("Found {} matching articles", matches.len());

    // Get specific headers
    let subjects = conn.xhdr(
        "subject",
        ArticleRange::Range(group.last - 10, group.last),
    ).await?;
    for h in &subjects {
        println!("  #{}: {}", h.article_num, h.value);
    }

    // List newsgroups
    let groups = conn.list_active(Some("alt.binaries.*")).await?;
    for g in &groups {
        println!("  {} ({}-{}) [{}]", g.name, g.low, g.high, g.status);
    }

    Ok(())
}

Pipelining

Pipelining sends multiple commands before reading responses, reducing round-trip overhead:

use nzb_nntp::{NntpConnection, Pipeline, StatPipeline};

async fn pipelined_download(conn: &mut NntpConnection) -> nzb_nntp::NntpResult<()> {
    // Pipeline ARTICLE commands (depth = 5 concurrent requests)
    let mut pipeline = Pipeline::new(5);
    pipeline.submit("msg1@example.com".into(), 1);
    pipeline.submit("msg2@example.com".into(), 2);
    pipeline.submit("msg3@example.com".into(), 3);

    let results = pipeline.process_all(conn).await?;
    for r in &results {
        match &r.result {
            Ok(resp) => println!("Tag {}: {} bytes", r.request.tag, resp.data.as_ref().map_or(0, |d| d.len())),
            Err(e) => println!("Tag {}: {}", r.request.tag, e),
        }
    }

    // Batch STAT checks (up to 100 per batch)
    let mut stat_pipe = StatPipeline::new();
    stat_pipe.add("check1@example.com".into());
    stat_pipe.add("check2@example.com".into());

    let stat_results = stat_pipe.execute(conn).await?;
    for s in &stat_results {
        println!("{}: {}", s.message_id, if s.exists { "exists" } else { "missing" });
    }

    Ok(())
}

Multi-Server Downloading

The Downloader coordinates article fetching across multiple servers with automatic failover:

use nzb_nntp::{Article, Downloader, ServerConfig};
use tokio::sync::mpsc;

async fn download_articles() -> nzb_nntp::NntpResult<()> {
    let servers = vec![
        ServerConfig {
            id: "primary".into(),
            name: "Primary Provider".into(),
            host: "news.primary.com".into(),
            port: 563,
            ssl: true,
            priority: 0, // Tried first
            connections: 8,
            pipelining: 5,
            ..ServerConfig::default()
        },
        ServerConfig {
            id: "backup".into(),
            name: "Backup Provider".into(),
            host: "news.backup.com".into(),
            port: 563,
            ssl: true,
            priority: 1, // Tried if primary fails
            connections: 4,
            ..ServerConfig::default()
        },
    ];

    let downloader = Downloader::new(servers, 0); // 0 = no bandwidth limit

    let articles = vec![
        Article {
            message_id: "part1@example.com".into(),
            segment_number: 1,
            bytes: 768000,
            downloaded: false,
            data_begin: None,
            data_size: None,
            crc32: None,
            tried_servers: Vec::new(),
            tries: 0,
        },
        // ... more articles
    ];

    let (tx, mut rx) = mpsc::channel(100);

    // Start download (runs to completion)
    tokio::spawn({
        let downloader_ref = &downloader;
        async move {
            downloader_ref.download(articles, tx).await
        }
    });

    // Process results as they arrive
    while let Some(result) = rx.recv().await {
        match result.result {
            Ok(data) => println!(
                "Segment {}: {} bytes from {}",
                result.article.segment_number,
                data.len(),
                result.server_id.unwrap_or_default()
            ),
            Err(e) => eprintln!(
                "Segment {} failed: {}",
                result.article.segment_number, e
            ),
        }
    }

    Ok(())
}

Connection Pooling

use nzb_nntp::{ConnectionPool, ServerConfig};
use std::sync::Arc;

async fn pooled_usage() -> nzb_nntp::NntpResult<()> {
    let config = Arc::new(ServerConfig {
        id: "pooled-server".into(),
        host: "news.example.com".into(),
        port: 563,
        ssl: true,
        connections: 4, // Max 4 concurrent connections
        ..ServerConfig::default()
    });

    let pool = ConnectionPool::new(config);

    // Acquire a connection (creates new or reuses idle)
    let mut pooled = pool.acquire().await?;

    // Use the connection
    let resp = pooled.conn.fetch_article("msg@example.com").await?;
    println!("Got {} bytes", resp.data.unwrap().len());

    // Return to pool for reuse
    pool.release(pooled);

    // Pool tracks idle connections
    println!("Idle connections: {}", pool.idle_count());

    // Clean up
    pool.close_idle().await;

    Ok(())
}

Architecture

                 ┌──────────────┐
                 │  Downloader   │  Orchestrates multi-server downloads
                 │  (failover)   │  with priority, pause/resume, bandwidth
                 └──────┬───────┘
                        │
              ┌─────────┼─────────┐
              ▼         ▼         ▼
        ┌───────────┐         ┌───────────┐
        │ServerState│   ...   │ServerState│  Per-server health tracking,
        │ + Pool    │         │ + Pool    │  penalties, speed metrics
        └─────┬─────┘         └─────┬─────┘
              │                     │
              ▼                     ▼
     ┌─────────────────┐   ┌─────────────────┐
     │ConnectionPool   │   │ConnectionPool   │  Semaphore-limited pools,
     │ (idle conns)    │   │ (idle conns)    │  health checks, reconnect
     └────────┬────────┘   └────────┬────────┘
              │                     │
              ▼                     ▼
     ┌─────────────────┐   ┌─────────────────┐
     │NntpConnection   │   │NntpConnection   │  TCP/TLS state machine,
     │ (state machine) │   │ (state machine) │  auth, commands, compression
     └────────┬────────┘   └────────┬────────┘
              │                     │
              ▼                     ▼
     ┌──────────────┐      ┌──────────────┐
     │  Transport   │      │  Transport   │    Plain TCP or TLS stream
     │ (TCP / TLS)  │      │ (TCP / TLS)  │    with buffered I/O
     └──────────────┘      └──────────────┘

Module Summary

Configuration

ServerConfig supports serde serialization for easy config file integration:

{
  "id": "news-provider",
  "name": "My Usenet Provider",
  "host": "news.example.com",
  "port": 563,
  "ssl": true,
  "ssl_verify": true,
  "username": "myuser",
  "password": "mypass",
  "connections": 8,
  "priority": 0,
  "enabled": true,
  "retention": 3600,
  "pipelining": 5,
  "optional": false,
  "compress": true,
  "proxy_url": null
}

Error Handling

All operations return NntpResult<T> (alias for Result<T, NntpError>). Errors are structured by cause:

Fatal errors (Auth, ServiceUnavailable, Connection, I/O) transition the connection to Error state, requiring a new connection. Non-fatal errors (ArticleNotFound, NoSuchGroup) keep the connection in Ready state.

NNTP Commands Supported

For detailed spec compliance analysis, see NNTP_COMPLIANCE.md.

Testing

cargo test

The test suite uses an in-process mock NNTP server (testutil::MockNntpServer) that supports all implemented commands with configurable responses and error injection. No external NNTP server is required.

License

MIT