Crate drasi_source_http 

HTTP Source Plugin for Drasi

This plugin exposes HTTP endpoints for receiving data change events. It includes adaptive batching for optimized throughput and supports both single-event and batch submission modes.

Endpoints

The HTTP source exposes the following endpoints:

• POST /sources/{source_id}/events - Submit a single event
• POST /sources/{source_id}/events/batch - Submit multiple events
• GET /health - Health check endpoint

Data Format

Events are submitted as JSON using the HttpSourceChange format:

Insert Operation

{
    "operation": "insert",
    "element": {
        "type": "node",
        "id": "user-123",
        "labels": ["User"],
        "properties": {
            "name": "Alice",
            "email": "alice@example.com"
        }
    },
    "timestamp": 1699900000000000000
}

Update Operation

{
    "operation": "update",
    "element": {
        "type": "node",
        "id": "user-123",
        "labels": ["User"],
        "properties": {
            "name": "Alice Updated"
        }
    }
}

Delete Operation

{
    "operation": "delete",
    "id": "user-123",
    "labels": ["User"]
}

Relation Element

{
    "operation": "insert",
    "element": {
        "type": "relation",
        "id": "follows-1",
        "labels": ["FOLLOWS"],
        "from": "user-123",
        "to": "user-456",
        "properties": {}
    }
}

Batch Submission

{
    "events": [
        { "operation": "insert", ... },
        { "operation": "update", ... }
    ]
}

Adaptive Batching

The HTTP source includes adaptive batching to optimize throughput. Events are buffered and dispatched in batches, with batch size and timing adjusted based on throughput patterns.

Parameter	Default	Description
adaptive_enabled	true	Enable/disable adaptive batching
adaptive_max_batch_size	1000	Maximum events per batch
adaptive_min_batch_size	1	Minimum events per batch
adaptive_max_wait_ms	100	Maximum wait time before dispatching
adaptive_min_wait_ms	10	Minimum wait time between batches

Configuration

Field	Type	Default	Description
host	string	required	Host address to bind to
port	u16	8080	Port to listen on
endpoint	string	None	Optional custom path prefix
timeout_ms	u64	10000	Request timeout in milliseconds

Example Configuration (YAML)

source_type: http
properties:
  host: "0.0.0.0"
  port: 8080
  adaptive_enabled: true
  adaptive_max_batch_size: 500

Usage Examples

Rust

use drasi_source_http::{HttpSource, HttpSourceBuilder};

let config = HttpSourceBuilder::new()
    .with_host("0.0.0.0")
    .with_port(8080)
    .with_adaptive_enabled(true)
    .build();

let source = Arc::new(HttpSource::new("http-source", config)?);
drasi.add_source(source).await?;

curl (Single Event)

curl -X POST http://localhost:8080/sources/my-source/events \
  -H "Content-Type: application/json" \
  -d '{"operation":"insert","element":{"type":"node","id":"1","labels":["Test"],"properties":{}}}'

curl (Batch)

curl -X POST http://localhost:8080/sources/my-source/events/batch \
  -H "Content-Type: application/json" \
  -d '{"events":[...]}'

Re-exports

pub use config::HttpSourceConfig;

Modules

config

Configuration for the HTTP source.

Structs

BatchEventRequest

Batch event request that can accept multiple events

EventResponse

Response for event submission

HttpSource

HTTP source with configurable adaptive batching.

HttpSourceBuilder

Builder for HttpSource instances.

Enums

HttpElement

Element that can be either a Node or Relation

HttpSourceChange

Data schema for HTTP source events

Functions

convert_http_to_source_change

Convert HttpSourceChange to drasi_core::models::SourceChange