feat: Support CRUD functionality (#7)

The station search implementation only works via DB and is very slow. For a simple query, the answer takes about 1s or longer. That is 100 times slower than it should be. For now, however, this solution is adequate to achieve a first prototype that includes the core functionality.

feat: Add station search
This commit is contained in:
2024-01-06 14:06:43 +01:00
committed by GitHub
parent b5681181fb
commit 7fbd270019
166 changed files with 2039 additions and 482 deletions
+23
View File
@@ -0,0 +1,23 @@
plugins {
id("twomartens.spring-boot-cloud-base")
id("twomartens.kotlin")
kotlin("kapt")
}
dependencies {
implementation(project(":common"))
implementation(project(":model"))
implementation(project(":support"))
implementation(libs.mapstruct.base)
annotationProcessor(libs.mapstruct.processor)
kapt(libs.mapstruct.processor)
implementation(libs.jaxb.impl)
implementation(libs.jakarta.xml.binding)
implementation(libs.spring.openapi)
implementation(libs.spring.boot.mongo)
implementation(libs.spring.cloud.leader.election)
implementation(libs.spring.cloud.starter.bus.kafka)
}
@@ -0,0 +1,19 @@
package de.twomartens.timetable.bahnApi.configuration
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
private const val CORE_POOL_SIZE = 1
@Configuration
open class ThreadPoolTaskExecutorConfig {
@Bean
open fun threadPoolTaskExecutor(): ThreadPoolTaskExecutor {
val executor = ThreadPoolTaskExecutor()
executor.corePoolSize = CORE_POOL_SIZE
executor.maxPoolSize = CORE_POOL_SIZE
return executor
}
}
@@ -0,0 +1,19 @@
package de.twomartens.timetable.bahnApi.configuration
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler
private const val POOL_SIZE = 1
private const val THREAD_NAME_PREFIX = "ThreadPoolTaskScheduler"
@Configuration
open class ThreadPoolTaskSchedulerConfig {
@Bean
open fun threadPoolTaskScheduler(): ThreadPoolTaskScheduler {
val scheduler = ThreadPoolTaskScheduler()
scheduler.poolSize = POOL_SIZE
scheduler.setThreadNamePrefix(THREAD_NAME_PREFIX)
return scheduler
}
}
@@ -0,0 +1,29 @@
package de.twomartens.timetable.bahnApi.events
import org.springframework.cloud.bus.event.Destination
import org.springframework.cloud.bus.event.PathDestinationFactory
import org.springframework.cloud.bus.event.RemoteApplicationEvent
import java.time.Instant
private const val TIMETABLE_SERVICE_NAME = "timetable"
class ScheduledTasksCreatedEvent private constructor(source: Instant, originService: String,
destination: Destination)
: RemoteApplicationEvent(source, originService, destination) {
private val creationTime: Instant = source
override fun getSource(): Instant {
return creationTime
}
companion object {
private val destinationFactory = PathDestinationFactory()
fun of(source: Instant): ScheduledTasksCreatedEvent {
return ScheduledTasksCreatedEvent(source,
TIMETABLE_SERVICE_NAME,
destinationFactory.getDestination(TIMETABLE_SERVICE_NAME))
}
}
}
@@ -0,0 +1,30 @@
package de.twomartens.timetable.bahnApi.jaxb
import jakarta.xml.bind.annotation.adapters.XmlAdapter
import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.time.format.DateTimeParseException
import java.util.*
class LocalDateTimeAdapter : XmlAdapter<String?, LocalDateTime>() {
@Throws(DateTimeParseException::class)
override fun unmarshal(xmlValue: String?): LocalDateTime {
return LocalDateTime.parse(xmlValue, dateFormatter.get())
}
@Throws(DateTimeParseException::class)
override fun marshal(kotlinValue: LocalDateTime): String? {
return dateFormatter.get().format(kotlinValue)
}
companion object {
private val dateFormatter: ThreadLocal<DateTimeFormatter> = object : ThreadLocal<DateTimeFormatter>() {
override fun initialValue(): DateTimeFormatter {
return DateTimeFormatter.ofPattern("yyMMddHHmm", Locale.GERMANY)
}
}
}
}
@@ -0,0 +1,40 @@
package de.twomartens.timetable.bahnApi.mapper
import de.twomartens.timetable.bahnApi.model.db.BahnStation
import de.twomartens.timetable.model.common.CountryCode
import de.twomartens.timetable.model.common.StationId
import de.twomartens.timetable.model.db.Station
import de.twomartens.timetable.types.NonEmptyString
import org.mapstruct.*
@Mapper(
collectionMappingStrategy = CollectionMappingStrategy.ADDER_PREFERRED,
nullValueCheckStrategy = NullValueCheckStrategy.ALWAYS,
unmappedTargetPolicy = ReportingPolicy.IGNORE
)
interface BahnStationMapper {
@Mapping(target = "id", ignore = true)
@Mapping(target = "created", ignore = true)
@Mapping(target = "lastModified", ignore = true)
fun mapToDB(dto: de.twomartens.timetable.bahnApi.model.dto.BahnStation): BahnStation {
return BahnStation(
dto.eva,
NonEmptyString(dto.name),
dto.ds100,
dto.db
)
}
@Mapping(target = "id", ignore = true)
@Mapping(target = "created", ignore = true)
@Mapping(target = "lastModified", ignore = true)
fun mapToCommonDB(db: BahnStation, countryCode: String): Station {
return Station(
StationId.of(NonEmptyString(countryCode + "-" + db.eva.value.toString())),
CountryCode(NonEmptyString(countryCode)),
db.name,
listOf()
)
}
}
@@ -0,0 +1,26 @@
package de.twomartens.timetable.bahnApi.mapper
import de.twomartens.timetable.bahnApi.model.db.BahnTimetable
import de.twomartens.timetable.types.HourAtDay
import de.twomartens.timetable.types.NonEmptyString
import org.mapstruct.*
@Mapper(
collectionMappingStrategy = CollectionMappingStrategy.ADDER_PREFERRED,
nullValueCheckStrategy = NullValueCheckStrategy.ALWAYS,
unmappedTargetPolicy = ReportingPolicy.IGNORE
)
interface BahnTimetableMapper {
@Mapping(target = "id", ignore = true)
@Mapping(target = "created", ignore = true)
@Mapping(target = "lastModified", ignore = true)
fun mapToDB(dto: de.twomartens.timetable.bahnApi.model.dto.BahnTimetable,
hourAtDay: HourAtDay): BahnTimetable {
return BahnTimetable(
dto.eva,
hourAtDay,
NonEmptyString(dto.station),
dto.stops
)
}
}
@@ -0,0 +1,22 @@
package de.twomartens.timetable.bahnApi.model
@JvmInline
value class Eva(val value: Int) {
init {
if (isKnown()) {
require(value in 8000000..8999999) {
"Eva number must start with an 8 and have 7 digits"
}
}
}
private fun isKnown() = value != -1
companion object {
val UNKNOWN = Eva(-1)
fun of(stationId: String): Eva {
return Eva(stationId.toInt())
}
}
}
@@ -0,0 +1,5 @@
package de.twomartens.timetable.bahnApi.model
import java.time.LocalDate
data class FetchDates(val previousDay: LocalDate, val fetchDate: LocalDate, val nextDate: LocalDate)
@@ -0,0 +1,65 @@
package de.twomartens.timetable.bahnApi.model
import de.twomartens.timetable.bahnApi.model.db.ScheduledFetchTask
import de.twomartens.timetable.types.HourAtDay
import org.springframework.stereotype.Component
import java.time.LocalDateTime
private const val SECONDS_BETWEEN_TASKS = 2
private const val ZERO = 0
@Component
class TaskFactory {
private val _tasksPerHour: MutableMap<HourAtDay, Int> = mutableMapOf()
fun updateCounterWith(existingTasks: List<ScheduledFetchTask>) {
countTasks(existingTasks)
}
fun createTaskAndUpdateCounter(eva: Eva, hourAtDay: HourAtDay): ScheduledFetchTask {
val fetchDateTime = calculateNextAvailableFetchDateTime(hourAtDay)
incrementTaskCountForHourAtDay(hourAtDay)
return ScheduledFetchTask(eva, hourAtDay, fetchDateTime)
}
private fun calculateNextAvailableFetchDateTime(hourAtDay: HourAtDay): LocalDateTime {
val secondInHourToExecute = calculateSecondInHourToCalculate(hourAtDay)
return calculateFetchDateTime(hourAtDay, secondInHourToExecute)
}
private fun calculateSecondInHourToCalculate(hourAtDay: HourAtDay): Int {
val numberOfTasksInHour = getNumberOfTasksInHour(hourAtDay)
return numberOfTasksInHour * SECONDS_BETWEEN_TASKS
}
private fun calculateFetchDateTime(
hourAtDay: HourAtDay,
secondInHourToExecute: Int
): LocalDateTime {
return hourAtDay.dateTime.plusSeconds(secondInHourToExecute.toLong())
}
private fun countTasks(existingTasks: List<ScheduledFetchTask>) {
existingTasks.forEach { countTask(it) }
}
private fun countTask(scheduledFetchTask: ScheduledFetchTask) {
val scheduledTime = scheduledFetchTask.scheduledExecutionDateTime
val hourAtDay = HourAtDay.of(scheduledTime)
incrementTaskCountForHourAtDay(hourAtDay)
}
private fun incrementTaskCountForHourAtDay(hourAtDay: HourAtDay) {
val numberOfTasksInHour = getNumberOfTasksInHour(hourAtDay)
_tasksPerHour[hourAtDay] = numberOfTasksInHour + 1
}
private fun getNumberOfTasksInHour(hourAtDay: HourAtDay) =
_tasksPerHour.computeIfAbsent(hourAtDay) { ZERO }
private fun resetCounter() {
_tasksPerHour.clear()
}
}
@@ -0,0 +1,29 @@
package de.twomartens.timetable.bahnApi.model.db
import de.twomartens.timetable.bahnApi.model.Eva
import de.twomartens.timetable.types.NonEmptyString
import org.bson.types.ObjectId
import org.springframework.data.annotation.CreatedDate
import org.springframework.data.annotation.Id
import org.springframework.data.annotation.LastModifiedDate
import org.springframework.data.mongodb.core.index.Indexed
import org.springframework.data.mongodb.core.index.TextIndexed
import org.springframework.data.mongodb.core.mapping.Document
import java.time.Instant
@Document
data class BahnStation(
@Indexed(unique = true) var eva: Eva,
@TextIndexed var name: NonEmptyString,
var ds100: String,
var db: Boolean
) {
@Id
var id: ObjectId = ObjectId()
@CreatedDate
lateinit var created: Instant
@LastModifiedDate
lateinit var lastModified: Instant
}
@@ -0,0 +1,32 @@
package de.twomartens.timetable.bahnApi.model.db
import de.twomartens.timetable.bahnApi.model.Eva
import de.twomartens.timetable.bahnApi.model.dto.BahnStationStop
import de.twomartens.timetable.types.HourAtDay
import de.twomartens.timetable.types.NonEmptyString
import org.bson.types.ObjectId
import org.springframework.data.annotation.CreatedDate
import org.springframework.data.annotation.Id
import org.springframework.data.annotation.LastModifiedDate
import org.springframework.data.mongodb.core.index.CompoundIndex
import org.springframework.data.mongodb.core.index.TextIndexed
import org.springframework.data.mongodb.core.mapping.Document
import java.time.Instant
@Document
@CompoundIndex(def = "{'eva': 1, 'dateTime': 1}", unique = true)
data class BahnTimetable(
var eva: Eva,
var hourAtDay: HourAtDay,
@TextIndexed var station: NonEmptyString,
var stops: List<BahnStationStop>
) {
@Id
var id: ObjectId = ObjectId()
@CreatedDate
lateinit var created: Instant
@LastModifiedDate
lateinit var lastModified: Instant
}
@@ -0,0 +1,29 @@
package de.twomartens.timetable.bahnApi.model.db
import de.twomartens.timetable.bahnApi.model.Eva
import de.twomartens.timetable.types.HourAtDay
import org.bson.types.ObjectId
import org.springframework.data.annotation.CreatedDate
import org.springframework.data.annotation.Id
import org.springframework.data.annotation.LastModifiedDate
import org.springframework.data.mongodb.core.index.CompoundIndex
import org.springframework.data.mongodb.core.mapping.Document
import java.time.Instant
import java.time.LocalDateTime
@Document
@CompoundIndex(def = "{'eva': 1, 'fetchedDateTime': 1}", unique = true)
data class ScheduledFetchTask(
var eva: Eva,
var fetchedDateTime: HourAtDay,
var scheduledExecutionDateTime: LocalDateTime
) {
@Id
var id: ObjectId = ObjectId()
@CreatedDate
lateinit var created: Instant
@LastModifiedDate
lateinit var lastModified: Instant
}
@@ -0,0 +1,16 @@
package de.twomartens.timetable.bahnApi.model.dto
import de.twomartens.timetable.bahnApi.model.Eva
import jakarta.xml.bind.annotation.XmlAccessType
import jakarta.xml.bind.annotation.XmlAccessorType
import jakarta.xml.bind.annotation.XmlAttribute
@XmlAccessorType(XmlAccessType.FIELD)
data class BahnStation(
@field:XmlAttribute var name: String,
@field:XmlAttribute var eva: Eva,
@field:XmlAttribute var ds100: String,
@field:XmlAttribute var db: Boolean
) {
constructor() : this("", Eva.UNKNOWN, "", true)
}
@@ -0,0 +1,18 @@
package de.twomartens.timetable.bahnApi.model.dto
import de.twomartens.timetable.bahnApi.model.Eva
import jakarta.xml.bind.annotation.XmlAccessType
import jakarta.xml.bind.annotation.XmlAccessorType
import jakarta.xml.bind.annotation.XmlAttribute
import jakarta.xml.bind.annotation.XmlElement
@XmlAccessorType(XmlAccessType.FIELD)
data class BahnStationStop(
@field:XmlAttribute var eva: Eva,
@field:XmlAttribute var id: String,
@field:XmlElement(name = "tl") var tripLabel: BahnTripLabel,
@field:XmlElement(name = "ar") var arrival: BahnStopEvent?,
@field:XmlElement(name = "dp") var departure: BahnStopEvent?
) {
constructor() : this(Eva.UNKNOWN, "", BahnTripLabel(), null, null)
}
@@ -0,0 +1,14 @@
package de.twomartens.timetable.bahnApi.model.dto
import jakarta.xml.bind.annotation.XmlAccessType
import jakarta.xml.bind.annotation.XmlAccessorType
import jakarta.xml.bind.annotation.XmlElement
import jakarta.xml.bind.annotation.XmlRootElement
@XmlRootElement(name = "stations")
@XmlAccessorType(XmlAccessType.FIELD)
data class BahnStations(
@field:XmlElement(name = "station") var stations: MutableList<BahnStation>
) {
constructor() : this(mutableListOf())
}
@@ -0,0 +1,32 @@
package de.twomartens.timetable.bahnApi.model.dto
import de.twomartens.timetable.bahnApi.jaxb.LocalDateTimeAdapter
import jakarta.xml.bind.annotation.XmlAccessType
import jakarta.xml.bind.annotation.XmlAccessorType
import jakarta.xml.bind.annotation.XmlAttribute
import jakarta.xml.bind.annotation.adapters.XmlJavaTypeAdapter
import java.time.LocalDateTime
@XmlAccessorType(XmlAccessType.FIELD)
data class BahnStopEvent(
@field:XmlAttribute(name = "pde") var plannedDistantEndpoint: String,
@field:XmlAttribute(name = "cde") var changedDistantEndpoint: String,
@field:XmlAttribute(name = "pt")
@field:XmlJavaTypeAdapter(value = LocalDateTimeAdapter::class)
var plannedTime: LocalDateTime?,
@field:XmlAttribute(name = "ct")
@field:XmlJavaTypeAdapter(value = LocalDateTimeAdapter::class)
var changedTime: LocalDateTime?,
@field:XmlAttribute(name = "ppth") var plannedPath: String,
@field:XmlAttribute(name = "cpth") var changedPath: String,
@field:XmlAttribute(name = "pp") var plannedPlatform: String,
@field:XmlAttribute(name = "cp") var changedPlatform: String,
@field:XmlAttribute(name = "wings") var wings: String,
@field:XmlAttribute(name = "l") var line: String
) {
constructor() : this("", "",
null,
null,
"", "", "", "",
"", "")
}
@@ -0,0 +1,14 @@
package de.twomartens.timetable.bahnApi.model.dto
import de.twomartens.timetable.bahnApi.model.Eva
import jakarta.xml.bind.annotation.*
@XmlRootElement(name = "timetable")
@XmlAccessorType(XmlAccessType.FIELD)
data class BahnTimetable(
@field:XmlAttribute var eva: Eva,
@field:XmlAttribute var station: String,
@field:XmlElement(name = "s") var stops: List<BahnStationStop>
) {
constructor() : this(Eva.UNKNOWN, "", listOf())
}
@@ -0,0 +1,17 @@
package de.twomartens.timetable.bahnApi.model.dto
import jakarta.xml.bind.annotation.XmlAccessType
import jakarta.xml.bind.annotation.XmlAccessorType
import jakarta.xml.bind.annotation.XmlAttribute
@XmlAccessorType(XmlAccessType.FIELD)
data class BahnTripLabel(
@field:XmlAttribute(name = "c") var category: String,
// S: S-Bahn, N: Nahverkehr, F: Fernverkehr
@field:XmlAttribute(name = "f") var filterFlags: String,
@field:XmlAttribute(name = "n") var trainNumber: String,
@field:XmlAttribute(name = "o") var owner: String,
@field:XmlAttribute(name = "t") var tripType: BahnTripType
) {
constructor() : this("", "", "", "", BahnTripType.p)
}
@@ -0,0 +1,10 @@
package de.twomartens.timetable.bahnApi.model.dto
enum class BahnTripType {
p, // passenger?
e,
z,
s,
h,
n
}
@@ -0,0 +1,14 @@
package de.twomartens.timetable.bahnApi.property
import io.swagger.v3.oas.annotations.media.Schema
import org.springframework.boot.context.properties.ConfigurationProperties
import org.springframework.boot.context.properties.bind.ConstructorBinding
import org.springframework.cloud.context.config.annotation.RefreshScope
@RefreshScope
@ConfigurationProperties(prefix = "de.twomartens.timetable.bahn-api")
@Schema(description = "Properties to configure the BahnAPI")
data class BahnApiProperties @ConstructorBinding constructor(
val clientId: String,
val clientSecret: String
)
@@ -0,0 +1,10 @@
package de.twomartens.timetable.bahnApi.repository
import de.twomartens.timetable.bahnApi.model.Eva
import de.twomartens.timetable.bahnApi.model.db.BahnStation
import org.bson.types.ObjectId
import org.springframework.data.mongodb.repository.MongoRepository
interface BahnStationRepository : MongoRepository<BahnStation, ObjectId> {
fun findByEva(eva: Eva): BahnStation?
}
@@ -0,0 +1,12 @@
package de.twomartens.timetable.bahnApi.repository
import de.twomartens.timetable.bahnApi.model.Eva
import de.twomartens.timetable.bahnApi.model.db.BahnTimetable
import de.twomartens.timetable.types.HourAtDay
import org.bson.types.ObjectId
import org.springframework.data.mongodb.repository.MongoRepository
// TODO: figure out what ID type should be
interface BahnTimetableRepository : MongoRepository<BahnTimetable, ObjectId> {
fun findByEvaAndHourAtDay(eva: Eva, hourAtDay: HourAtDay): BahnTimetable?
}
@@ -0,0 +1,14 @@
package de.twomartens.timetable.bahnApi.repository
import de.twomartens.timetable.bahnApi.model.Eva
import de.twomartens.timetable.bahnApi.model.db.ScheduledFetchTask
import de.twomartens.timetable.types.HourAtDay
import org.bson.types.ObjectId
import org.springframework.data.mongodb.repository.MongoRepository
import java.time.Instant
interface ScheduledFetchTaskRepository : MongoRepository<ScheduledFetchTask, ObjectId> {
fun findByEvaAndFetchedDateTime(eva: Eva, fetchedDateTime: HourAtDay): ScheduledFetchTask?
fun findAllByCreatedAfter(createdAt: Instant): List<ScheduledFetchTask>
}
@@ -0,0 +1,51 @@
package de.twomartens.timetable.bahnApi.service
import de.twomartens.timetable.bahnApi.model.Eva
import de.twomartens.timetable.bahnApi.model.dto.BahnStation
import de.twomartens.timetable.bahnApi.model.dto.BahnStations
import de.twomartens.timetable.bahnApi.model.dto.BahnTimetable
import de.twomartens.timetable.bahnApi.property.BahnApiProperties
import de.twomartens.timetable.types.HourAtDay
import org.springframework.http.MediaType
import org.springframework.stereotype.Service
import org.springframework.web.client.RestClient
import java.time.LocalTime
import java.time.format.DateTimeFormatter
@Service
class BahnApiService(
private val restClient: RestClient,
private val properties: BahnApiProperties
) {
fun fetchStations(pattern: String): List<BahnStation> {
val body = restClient.get()
.uri("https://apis.deutschebahn.com/db-api-marketplace/apis/timetables/v1/station/${pattern}")
.headers {
it.accept = mutableListOf(MediaType.APPLICATION_XML)
it.contentType = MediaType.APPLICATION_XML
it.set("DB-Client-Id", properties.clientId)
it.set("DB-Api-Key", properties.clientSecret)
}
.retrieve()
.body(BahnStations::class.java)
return body?.stations ?: listOf()
}
fun fetchTimetable(eva: Eva, hourAtDay: HourAtDay): BahnTimetable {
val dateFormatter = DateTimeFormatter.ofPattern("yyMMdd")
val timeFormatter = DateTimeFormatter.ofPattern("HH")
val time = LocalTime.of(hourAtDay.hour.value, 0)
val body = restClient.get()
.uri("https://apis.deutschebahn.com/db-api-marketplace/apis/timetables/v1/plan/" +
"${eva}/${hourAtDay.date.format(dateFormatter)}/${time.format(timeFormatter)}")
.headers {
it.accept = mutableListOf(MediaType.APPLICATION_XML)
it.contentType = MediaType.APPLICATION_XML
it.set("DB-Client-Id", properties.clientId)
it.set("DB-Api-Key", properties.clientSecret)
}
.retrieve()
.body(BahnTimetable::class.java)
return body!!
}
}
@@ -0,0 +1,97 @@
package de.twomartens.timetable.bahnApi.service
import de.twomartens.timetable.bahnApi.mapper.BahnStationMapper
import de.twomartens.timetable.bahnApi.mapper.BahnTimetableMapper
import de.twomartens.timetable.bahnApi.model.dto.BahnStation
import de.twomartens.timetable.bahnApi.model.dto.BahnTimetable
import de.twomartens.timetable.bahnApi.repository.BahnStationRepository
import de.twomartens.timetable.bahnApi.repository.BahnTimetableRepository
import de.twomartens.timetable.model.db.Station
import de.twomartens.timetable.model.repository.StationRepository
import de.twomartens.timetable.types.HourAtDay
import org.mapstruct.factory.Mappers
import org.springframework.data.mongodb.core.BulkOperations
import org.springframework.data.mongodb.core.MongoTemplate
import org.springframework.stereotype.Service
@Service
open class BahnDatabaseService(
private val bahnStationRepository: BahnStationRepository,
private val stationRepository: StationRepository,
private val bahnTimetableRepository: BahnTimetableRepository,
private val mongoTemplate: MongoTemplate
) {
private val bahnStationMapper = Mappers.getMapper(BahnStationMapper::class.java)
private val bahnTimetableMapper = Mappers.getMapper(BahnTimetableMapper::class.java)
fun storeStations(stations: List<BahnStation>) {
val existingStations = stationRepository.findAllByCountryCode(COUNTRY_CODE)
val commonStationMap = existingStations
.associateBy { it.stationId.stationIdWithinCountry }
val bahnStationMap = stations.asSequence()
.map(bahnStationMapper::mapToDB)
.associateBy { it.eva.toString() }
updateBahnStations(bahnStationMap)
deleteRemovedStations(existingStations, bahnStationMap)
updateStations(existingStations, bahnStationMap)
addNewStations(bahnStationMap, commonStationMap)
}
private fun updateBahnStations(bahnStationMap: Map<String, de.twomartens.timetable.bahnApi.model.db.BahnStation>) {
bahnStationRepository.deleteAll()
val bahnStations: List<de.twomartens.timetable.bahnApi.model.db.BahnStation> = buildList {
this.addAll(bahnStationMap.values)
}
mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED,
de.twomartens.timetable.bahnApi.model.db.BahnStation::class.java)
.insert(bahnStations)
.execute()
}
private fun deleteRemovedStations(
existingStations: List<Station>,
bahnStationMap: Map<String, de.twomartens.timetable.bahnApi.model.db.BahnStation>
) {
val deletedStations = existingStations
.filterNot { bahnStationMap.containsKey(it.stationId.stationIdWithinCountry) }
stationRepository.deleteAll(deletedStations)
}
private fun updateStations(
existingStations: List<Station>,
bahnStationMap: Map<String, de.twomartens.timetable.bahnApi.model.db.BahnStation>
) {
val updatedStations = existingStations
.filter { bahnStationMap.containsKey(it.stationId.stationIdWithinCountry) }
updatedStations.map {
it.name = bahnStationMap[it.stationId.stationIdWithinCountry]!!.name
}
stationRepository.saveAll(updatedStations)
}
private fun addNewStations(
bahnStations: Map<String, de.twomartens.timetable.bahnApi.model.db.BahnStation>,
commonStationMap: Map<String, Station>
) {
val newStations = bahnStations.asSequence()
.filterNot { commonStationMap.containsKey(it.key) }
.map { bahnStationMapper.mapToCommonDB(it.value, COUNTRY_CODE) }
.toList()
mongoTemplate.bulkOps(BulkOperations.BulkMode.UNORDERED,
Station::class.java)
.insert(newStations)
.execute()
}
fun storeTimetable(timetable: BahnTimetable, hourAtDay: HourAtDay) {
bahnTimetableRepository.save(bahnTimetableMapper.mapToDB(timetable, hourAtDay))
}
companion object {
private const val COUNTRY_CODE = "de"
}
}
@@ -0,0 +1,94 @@
package de.twomartens.timetable.bahnApi.service
import de.twomartens.timetable.bahnApi.model.db.ScheduledFetchTask
import de.twomartens.timetable.bahnApi.model.dto.BahnTimetable
import de.twomartens.timetable.bahnApi.repository.ScheduledFetchTaskRepository
import de.twomartens.timetable.bahnApi.tasks.DeleteScheduledTask
import de.twomartens.timetable.bahnApi.tasks.FetchTimetableTask
import de.twomartens.timetable.bahnApi.tasks.StoreTimetableTask
import de.twomartens.timetable.types.HourAtDay
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler
import org.springframework.stereotype.Service
import java.time.Clock
import java.time.ZonedDateTime
private const val PAST_TASK_EXECUTION_OFFSET = 1
@Service
class FetchTaskScheduler(
private val clock: Clock,
private val threadPoolTaskScheduler: ThreadPoolTaskScheduler,
private val threadPoolTaskExecutor: ThreadPoolTaskExecutor,
private val bahnApiService: BahnApiService,
private val bahnDatabaseService: BahnDatabaseService,
private val scheduledFetchTaskRepository: ScheduledFetchTaskRepository
) {
fun scheduleStoreTask(timetable: BahnTimetable, hourAtDay: HourAtDay) {
val storeTask = StoreTimetableTask(timetable, hourAtDay, bahnDatabaseService)
threadPoolTaskExecutor.execute(storeTask)
}
fun scheduleFetchTasks(scheduledTasks: List<ScheduledFetchTask>) {
val nowPlus1Minute = ZonedDateTime.now(clock).plusMinutes(1)
scheduleTasksFromThePast(scheduledTasks, nowPlus1Minute)
scheduleTasksFromTheFuture(scheduledTasks, nowPlus1Minute)
}
private fun scheduleTasksFromThePast(
scheduledTasks: List<ScheduledFetchTask>,
nowPlus1Minute: ZonedDateTime
) {
val tasksScheduledForPast = scheduledTasks.filter {
val zonedExecutionTime = it.scheduledExecutionDateTime.atZone(clock.zone)
nowPlus1Minute.isAfter(zonedExecutionTime)
}
tasksScheduledForPast.forEach {
val zonedExecutionTime = it.scheduledExecutionDateTime.atZone(clock.zone)
val adjustedExecutionTime = nowPlus1Minute.withOffsetAndSecondsFrom(zonedExecutionTime)
scheduleDeleteTask(it, adjustedExecutionTime)
scheduleFetchTimetableTask(it, adjustedExecutionTime)
}
}
private fun scheduleTasksFromTheFuture(
scheduledTasks: List<ScheduledFetchTask>,
nowPlus1Minute: ZonedDateTime
) {
val tasksScheduledForFuture = scheduledTasks.filter {
val zonedExecutionTime = it.scheduledExecutionDateTime.atZone(clock.zone)
nowPlus1Minute.isOnOrBefore(zonedExecutionTime)
}
tasksScheduledForFuture.forEach {
val zonedExecutionTime = it.scheduledExecutionDateTime.atZone(clock.zone)
scheduleDeleteTask(it, zonedExecutionTime)
scheduleFetchTimetableTask(it, zonedExecutionTime)
}
}
private fun scheduleDeleteTask(
scheduledFetchTask: ScheduledFetchTask,
zonedExecutionTime: ZonedDateTime
) {
val deleteTask = DeleteScheduledTask(scheduledFetchTaskRepository, scheduledFetchTask)
threadPoolTaskScheduler.schedule(deleteTask, zonedExecutionTime.toInstant())
}
private fun scheduleFetchTimetableTask(
scheduledFetchTask: ScheduledFetchTask,
zonedExecutionTime: ZonedDateTime
) {
val timetableTask = FetchTimetableTask(scheduledFetchTask.eva,
scheduledFetchTask.fetchedDateTime,
bahnApiService, this)
threadPoolTaskScheduler.schedule(timetableTask, zonedExecutionTime.toInstant())
}
companion object {
fun ZonedDateTime.withOffsetAndSecondsFrom(other: ZonedDateTime): ZonedDateTime =
this.withSecond(other.second + PAST_TASK_EXECUTION_OFFSET)
fun ZonedDateTime.isOnOrBefore(other: ZonedDateTime): Boolean =
this.isBefore(other) || this.isEqual(other)
}
}
@@ -0,0 +1,142 @@
package de.twomartens.timetable.bahnApi.service
import de.twomartens.support.model.LeadershipStatus
import de.twomartens.support.service.BusService
import de.twomartens.timetable.bahnApi.events.ScheduledTasksCreatedEvent
import de.twomartens.timetable.bahnApi.model.Eva
import de.twomartens.timetable.bahnApi.model.FetchDates
import de.twomartens.timetable.bahnApi.model.TaskFactory
import de.twomartens.timetable.bahnApi.model.db.ScheduledFetchTask
import de.twomartens.timetable.bahnApi.repository.ScheduledFetchTaskRepository
import de.twomartens.timetable.model.db.TswRoute
import de.twomartens.timetable.types.Hour
import de.twomartens.timetable.types.HourAtDay
import mu.KotlinLogging
import org.springframework.boot.context.event.ApplicationReadyEvent
import org.springframework.cloud.kubernetes.commons.leader.LeaderProperties
import org.springframework.context.event.EventListener
import org.springframework.integration.leader.event.OnGrantedEvent
import org.springframework.stereotype.Service
import java.time.Instant
import java.time.LocalDate
@Service
class ScheduledTaskService(
private val busService: BusService,
private val leadershipStatus: LeadershipStatus,
private val leaderProperties: LeaderProperties,
private val scheduledFetchTaskRepository: ScheduledFetchTaskRepository,
private val taskFactory: TaskFactory,
private val fetchTaskScheduler: FetchTaskScheduler
) {
private var createdTime: Instant = Instant.EPOCH
private var lastUpdate: Instant = Instant.EPOCH
@EventListener(ApplicationReadyEvent::class)
fun onApplicationReady(event: ApplicationReadyEvent) {
if (!leaderProperties.isEnabled) {
val updateTime = Instant.ofEpochMilli(event.timestamp)
updateTaskCounterAndScheduleTasksIfLeader(updateTime)
}
}
@EventListener(OnGrantedEvent::class)
fun onLeadershipGranted(event: OnGrantedEvent) {
val updateTime = Instant.ofEpochMilli(event.timestamp)
updateTaskCounterAndScheduleTasksIfLeader(updateTime)
}
@EventListener(ScheduledTasksCreatedEvent::class)
fun onScheduledTasksCreated(event: ScheduledTasksCreatedEvent) {
updateTaskCounterAndScheduleTasksIfLeader(event.source)
}
private fun updateTaskCounterAndScheduleTasksIfLeader(updateTime: Instant) {
log.info { "Update tasks from database and schedule if leader" }
val createdTasks = findTasksCreatedSince(lastUpdate)
updateCounterIfNotUpToDate(updateTime, createdTasks)
scheduleTasksIfLeader(createdTasks)
lastUpdate = updateTime
}
fun triggerTimetableFetch(tswRoute: TswRoute, fetchedDate: LocalDate) {
log.info {
"Trigger timetable fetch: [route ${tswRoute.name}]"
}
val fetchDates = calculateDatesToFetch(fetchedDate)
val newTasks = buildScheduledTasks(tswRoute, fetchDates)
storeTasksInDatabaseAndStoreCreationTime(newTasks)
publishTasksCreatedEvent()
}
private fun calculateDatesToFetch(fetchedDate: LocalDate): FetchDates {
val previousDay = fetchedDate.minusDays(1)
val nextDay = fetchedDate.plusDays(1)
return FetchDates(previousDay, fetchedDate, nextDay)
}
private fun buildScheduledTasks(
tswRoute: TswRoute,
fetchDates: FetchDates
): List<ScheduledFetchTask> {
val newTasks = mutableListOf<ScheduledFetchTask>()
tswRoute.stations.forEach {
val stationId = it.id
val eva = Eva.of(stationId)
var hourAtDay = HourAtDay.of(Hour.of(23), fetchDates.previousDay)
var newTask = taskFactory.createTaskAndUpdateCounter(eva, hourAtDay)
newTasks.add(newTask)
for (hour in 0..23) {
hourAtDay = HourAtDay.of(Hour.of(hour), fetchDates.fetchDate)
newTask = taskFactory.createTaskAndUpdateCounter(eva, hourAtDay)
newTasks.add(newTask)
}
for (hour in 0..3) {
hourAtDay = HourAtDay.of(Hour.of(hour), fetchDates.nextDate)
newTask = taskFactory.createTaskAndUpdateCounter(eva, hourAtDay)
newTasks.add(newTask)
}
}
return newTasks
}
private fun scheduleTasksIfLeader(tasksToSchedule: List<ScheduledFetchTask>) {
if (leadershipStatus.isLeader) {
fetchTaskScheduler.scheduleFetchTasks(tasksToSchedule)
}
}
private fun updateCounterIfNotUpToDate(updateTime: Instant,
createdTasks: List<ScheduledFetchTask>) {
if (notUpToDate(updateTime)) {
taskFactory.updateCounterWith(createdTasks)
}
}
private fun notUpToDate(updateTime: Instant) = createdTime != updateTime
private fun findTasksCreatedSince(lastScheduledTime: Instant): List<ScheduledFetchTask> {
return scheduledFetchTaskRepository.findAllByCreatedAfter(lastScheduledTime)
}
private fun storeTasksInDatabaseAndStoreCreationTime(newTasks: List<ScheduledFetchTask>) {
val savedEntities = scheduledFetchTaskRepository.saveAll(newTasks)
storeCreationTime(savedEntities)
}
private fun storeCreationTime(savedEntities: MutableList<ScheduledFetchTask>) {
if (savedEntities.isNotEmpty()) {
createdTime = savedEntities[savedEntities.lastIndex].created
}
}
private fun publishTasksCreatedEvent() {
val event = ScheduledTasksCreatedEvent.of(createdTime)
busService.publishEvent(event)
}
companion object {
private val log = KotlinLogging.logger {}
}
}
@@ -0,0 +1,26 @@
package de.twomartens.timetable.bahnApi.tasks
import de.twomartens.timetable.bahnApi.model.db.ScheduledFetchTask
import de.twomartens.timetable.bahnApi.repository.ScheduledFetchTaskRepository
import mu.KotlinLogging
import org.springframework.scheduling.annotation.Async
@Async
open class DeleteScheduledTask(
private val scheduledFetchTaskRepository: ScheduledFetchTaskRepository,
private val scheduledFetchTask: ScheduledFetchTask
) : Runnable {
override fun run() {
log.info {
"Delete scheduled fetch task after the fetch occurred: " +
"[eva: ${scheduledFetchTask.eva}], " +
"[date: ${scheduledFetchTask.fetchedDateTime.date}], " +
"[time: ${scheduledFetchTask.fetchedDateTime.hour}]"
}
scheduledFetchTaskRepository.delete(scheduledFetchTask)
}
companion object {
private val log = KotlinLogging.logger {}
}
}
@@ -0,0 +1,27 @@
package de.twomartens.timetable.bahnApi.tasks
import de.twomartens.timetable.bahnApi.model.Eva
import de.twomartens.timetable.bahnApi.service.BahnApiService
import de.twomartens.timetable.bahnApi.service.FetchTaskScheduler
import de.twomartens.timetable.types.HourAtDay
import mu.KotlinLogging
import org.springframework.scheduling.annotation.Async
@Async
open class FetchTimetableTask(
private val eva: Eva,
private val hourAtDay: HourAtDay,
private val bahnApiService: BahnApiService,
private val scheduler: FetchTaskScheduler) : Runnable {
override fun run() {
log.info {
"Fetch timetable: [eva: $eva], [date: ${hourAtDay.date}], [hour: ${hourAtDay.hour}]"
}
val timetable = bahnApiService.fetchTimetable(eva, hourAtDay)
scheduler.scheduleStoreTask(timetable, hourAtDay)
}
companion object {
private val log = KotlinLogging.logger {}
}
}
@@ -0,0 +1,23 @@
package de.twomartens.timetable.bahnApi.tasks
import de.twomartens.timetable.bahnApi.model.dto.BahnTimetable
import de.twomartens.timetable.bahnApi.service.BahnDatabaseService
import de.twomartens.timetable.types.HourAtDay
import mu.KotlinLogging
import org.springframework.scheduling.annotation.Async
@Async
open class StoreTimetableTask(
private val timetable: BahnTimetable,
private val hourAtDay: HourAtDay,
private val bahnDatabaseService: BahnDatabaseService
) : Runnable {
override fun run() {
log.info { "Store timetable: [eva: ${timetable.eva}], station: ${timetable.station}]" }
bahnDatabaseService.storeTimetable(timetable, hourAtDay)
}
companion object {
private val log = KotlinLogging.logger {}
}
}