Archived: This repository is a historical academic artifact and is no longer actively maintained. It is public solely for reference purposes and is published without an open-source license. Public availability should not be interpreted as permission to use, modify, or redistribute the material.

Bus Location Handler

The Bus Location Handler is a Python background service that consumes bus location messages from RabbitMQ, validates them, and fans them out to Redis, Parquet storage, and a downstream RabbitMQ exchange.

Architecture

flowchart LR
   producer[BusAPI] --> rabbitmq[(RabbitMQ)]
   schedule[(PostgreSQL/PostGIS)] --> handler[BusLocationHandler]
   rabbitmq --> handler
   handler --> redis[(Redis)]
   handler --> parquet[(Archiving Parquet files)]
   handler --> rabbitmq
   rabbitmq --> prediction[BusPredictionHandler]

Loading

Processing

1. The service receives bus location messages from RabbitMQ.
2. It validates the messages and collects them into batches.
3. Each batch updates the latest locations in Redis and archives the data in Parquet files.
4. The processed locations are published to RabbitMQ for BusPredictionHandler.

Inputs

The service receives messages from the RabbitMQ exchange bus.location with routing key bus.location.<route_id>. Each message contains location information for a specific bus.

Outputs

Redis

The latest location is stored as a Redis hash using key bus_location:<route_id>:<direction>:<vehicle_id>. Keys expire after 180 seconds.

Parquet

Parquet storage is enabled by default. A background worker buffers records and writes files when PARQUET_BATCH_SIZE is reached or PARQUET_FLUSH_INTERVAL seconds have elapsed. Files are partitioned by event date and hour:

/var/data/bus_locations/YYYY/MM/DD/bus_locations_YYYY-MM-DD_HH.parquet

RabbitMQ

After a batch is processed, the validated data is serialized as JSON and published to the topic exchange bus_location_handler.updated using routing key bus_location_handler.updated.<route_id>.<vehicle_id>

Configuration

Configuration is loaded from environment variables.

Common Settings

Variable	Default	Purpose
LOG_LEVEL	INFO	Logging level, including the custom TRACE level.
LOG_DIR	./logs	Directory for timestamped log files.
LOG_FILENAME_PREFIX	app	Prefix for log filenames.
RABBITMQ_HOST	localhost	RabbitMQ hostname.
RABBITMQ_PORT	5672	RabbitMQ AMQP port.
RABBITMQ_USER	guest	RabbitMQ username.
RABBITMQ_PASSWORD	guest	RabbitMQ password. Change this outside development.
RABBITMQ_VHOST	/	RabbitMQ virtual host.
POSTGRES_HOST	localhost	PostgreSQL hostname.
POSTGRES_PORT	5432	PostgreSQL port.
POSTGRES_USER	postgres	PostgreSQL username.
POSTGRES_PASSWORD	postgres	PostgreSQL password. Change this outside development.
POSTGRES_DB	bus_transit_system	PostgreSQL database name.
REDIS_HOST	redis_stack	Redis hostname.
REDIS_PORT	6379	Redis port.
REDIS_PASSWORD	empty	Redis password.
REDIS_DB	0	Redis database number.

Handler Settings

Variable	Default	Purpose
BUS_LOCATION_EXCHANGE	bus.location	Input RabbitMQ exchange.
BUS_LOCATION_QUEUE_PREFIX	bus_location_handler	Prefix for the input queue name.
BUS_LOCATION_UPDATED_EXCHANGE	bus_location_handler.updated	Output exchange for downstream notifications.
WORKER_THREADS	4	Configuration value reported in logs; the current processor starts one message-processing thread.
BATCH_SIZE	100	Maximum messages collected per batch and RabbitMQ prefetch count.
BATCH_TIMEOUT_SECONDS	5	Defined and logged, but not currently used by the batch collector.
HEALTH_CHECK_INTERVAL	30	Seconds between statistics reports.
BUS_LOCATION_REDIS_KEY_PREFIX	bus_location	Redis key prefix.
BUS_LOCATION_REDIS_TTL	180	Redis location expiration in seconds.
PARQUET_STORAGE_ENABLED	true	Enables background Parquet storage.
PARQUET_DATA_DIR	/var/data/bus_locations	Root directory for Parquet files.
PARQUET_BATCH_SIZE	1000	Records buffered before a Parquet flush.
PARQUET_FLUSH_INTERVAL	60	Maximum buffer age in seconds before a Parquet flush.