config.ini

[DEFAULT]
# Address of the RabbitMQ host (for all except client)
RabbitHost = rabbitmq
# DEBUG, INFO, WARNING, ERROR, CRITICAL
LogLevel = INFO
# Number of messages to prefetch from RabbitMQ
# (for all except input and client)
PrefetchCount = 2
# System name (must be the same as in compose.yaml)
SystemName = distribuidos-tp2

# For parsers, joiners & aggregators
FiltersExchangeBase = filters
FiltersQueueBase = filters_{host_id}
FiltersRoutingKeysFormat = ["check_processed_response.{host_id}", "check_processed", "remove_check.{host_id}"] 

# Heartbeat settings
HeartbeatExchange = 
HeartbeatRoutingKey = heartbeat
HeartbeatFrequency = 3

# --------------------   CLIENT    --------------------

[client]
BatchSize = 10000
# Address of the input host
InputAddress = tcp://input:5555
# Address of the output host
OutputAddress = tcp://output:5555
# Path to the data directory
DataPath = /tmp/data/medium
# Path to the results directory
ResultsPath = /tmp/results

# --------------------    INPUT    --------------------

[input]
# Address to bind to
Address = tcp://*:5555
MaxJobs = 2

# --------------------   PARSERS   --------------------

[parsers]

# Middleware settings
InExchange = raw_records
FiltersExchange = parser_${DEFAULT:FiltersExchangeBase}
FiltersQueueFormat = parser_${DEFAULT:FiltersQueueBase}
InTripLinesQueueFormat = raw_trip_lines_{job_id}
InTripLinesRoutingKeysFormat = ["{job_id}.trip"]
InWeatherStationLinesQueueFormat = raw_weather_station_lines_{job_id}
InWeatherStationLinesRoutingKeysFormat  = ["{job_id}.weather", "{job_id}.station"]
InOthersQueueFormat = parser_others_{host_id}
InOthersQueueRoutingKeys = ["*.start", "*.end", "*.trips_start"]

# --------------------   JOINERS   --------------------
[joiners]

# Middleware settings
InExchange = basic_records
FiltersExchangeFormat = {name}_joiner_${DEFAULT:FiltersExchangeBase}
FiltersQueueFormat = {name}_joiner_${DEFAULT:FiltersQueueBase}
InTripsQueueFormat = {name}_basic_trips_{job_id}
InOthersQueueFormat = {name}_joiner_others_{host_id}
InOthersQueueRoutingKeysBase = "*.trips_start", "*.end"

    [joiners.rain]
    # Drop trips on days with less than this mm of rain
    PrecipitationThreshold = 30

    # Middleware settings
    InOthersQueueRoutingKeys = [${joiners:InOthersQueueRoutingKeysBase}, "*.weather.#"]
    InTripsQueueRoutingKeysFormat = ["{job_id}.trip.#"]

    [joiners.year]
    # Base year for comparison
    YearBase = 2016
    # Year to compare to the base year
    YearCompared = 2017

    # Middleware settings
    InOthersQueueRoutingKeys = [${joiners:InOthersQueueRoutingKeysBase}, "*.station.*.${YearBase}", "*.station.*.${YearCompared}"]
    InTripsQueueRoutingKeysFormat = ["{job_id}.trip.*.${YearBase}", "{job_id}.trip.*.${YearCompared}"]

    [joiners.city]
    # City to filter on
    City = montreal

    # Middleware settings
    InOthersQueueRoutingKeys = [${joiners:InOthersQueueRoutingKeysBase}, "*.station.${City}.*"]
    InTripsQueueRoutingKeysFormat = ["{job_id}.trip.${City}.*"]

# -------------------- AGGREGATORS --------------------
[aggregators]
# Interval at which to send the aggregated data
SendIntervalSeconds = 10

# Middleware settings
InExchangeFormat = {name}_joined_records
FiltersExchangeFormat = {name}_aggregator_${DEFAULT:FiltersExchangeBase}
FiltersQueueFormat = {name}_aggregator_${DEFAULT:FiltersQueueBase}
InTripsQueueFormat = {name}_joined_trips_{job_id}
InTripsQueueRoutingKeysFormat = ["{job_id}.trip"]
InOthersQueueFormat = {name}_aggregator_others_{host_id}
InOthersQueueRoutingKeys = ["*.start", "*.end"]

    [aggregators.rain]
    
    [aggregators.year]
    YearBase = ${joiners.year:YearBase}
    YearCompared = ${joiners.year:YearCompared}

    [aggregators.city]


# --------------------  REDUCERS   --------------------
[reducers]

# Middleware settings
# No exchange for reducers, just the default nameless one
InExchange = 
InQueueFormat = {name}_aggregated

    [reducers.rain]

    [reducers.year]
    # Factor to filter with. Keeps stations if (Trips in compared year) > Factor * (Trips in base year) 
    Factor = 2

    [reducers.city]
    # Minimum distance between stations in km to filter on
    MinDistanceKm = 6

# --------------------   OUTPUT    --------------------

[output]
# Address to bind to
Address = tcp://*:5555
# Middleware settings
InExchange = 
InQueue = stats

# --------------------    MEDIC    --------------------

[medic]
# (Bully) How much to wait for an answer to an election message
AnswerTimeout = 10
# (Bully) How much to wait for a coordinator message after getting
# a response in an election
CoordinatorTimeout = 20
# How often medics send a heartbeat
HeartbeatInterval = ${DEFAULT:HeartbeatFrequency}
# How much time to wait for a heartbeat
HeartbeatTimeout = 15
# How much time to wait for the first heartbeat after startup
FirstHeartbeatTimeout = 15
# How much time to wait for the next heartbeat after restarting a node
RestartTimeout = 20
# How often the leader sends a heartbeat
LeaderHeartbeatInterval = 1
# How much time to wait for the leader to send a heartbeat
LeaderHeartbeatTimeout = 5

# Middleware settings
MedicExchange = medics
MedicQueueFormat = medic_{host_id}
MedicQueueRoutingKeysFormat = [ "#.{host_id}.#" ]