xgdb-connect -- Connector Framework for xrayGraphDB

Standalone Python processes that consume data from external sources and write to xrayGraphDB over xrayProtocol (port 7689).

Architecture

xgdb_connect/           Core framework
  protocol.py           xrayProtocol v1 binary client
  base.py               BaseConnector (connect, transform, send, checkpoint)
  config.py             YAML config loader
  transform.py          Message-to-Cypher transformer

connectors/             Source-specific implementations
  nats_connector.py     NATS/JetStream
  redis_connector.py    Redis Streams (XREADGROUP)
  kafka_connector.py    Apache Kafka (consumer groups)
  postgres_connector.py PostgreSQL logical replication (pgoutput CDC)
  mysql_connector.py    MySQL binlog CDC

Installation

# Core only
pip install .

# With specific connector dependencies
pip install ".[nats]"
pip install ".[redis]"
pip install ".[kafka]"
pip install ".[postgres]"
pip install ".[mysql]"

# Everything
pip install ".[all]"

Usage

Each connector reads a YAML config file and runs as a standalone process:

python -m connectors.nats_connector --config nats.yaml
python -m connectors.redis_connector --config redis.yaml
python -m connectors.kafka_connector --config kafka.yaml
python -m connectors.postgres_connector --config postgres.yaml
python -m connectors.mysql_connector --config mysql.yaml

Config Format

source: nats
connection:
  url: nats://localhost:4222
  subject: "events.>"

xraygraphdb:
  host: localhost
  port: 7689
  user: admin
  password: secret

transform:
  node:
    label: Event
    id_field: event_id
    properties:
      - source_field: timestamp
        target: ts
      - source_field: type
        target: event_type
  edge:
    type: TRIGGERED_BY
    from_field: event_id
    to_field: parent_id

batch_size: 1000
checkpoint_file: /var/lib/xgdb-connect/nats.checkpoint

Features

• Batch writes -- configurable batch size, UNWIND-based Cypher for efficiency
• Checkpoint tracking -- saves last offset/position to disk for restart recovery
• Backpressure -- batches are flushed when full, with configurable size
• Graceful shutdown -- handles SIGTERM/SIGINT, flushes remaining batch, saves checkpoint
• Structured logging -- timestamped log output with configurable level
• Metrics -- messages/sec, latency, error counts logged at configurable intervals
• Reconnection -- automatic reconnect to xrayGraphDB with configurable retry (max 2 attempts)
• Transform engine -- configurable field mappings with dot-path extraction and type casting

Copyright

Copyright 2026 eMTAi LLC. All rights reserved. Patent pending.