diff --git a/dexcom-library/build.gradle b/dexcom-library/build.gradle new file mode 100644 index 00000000..56f0c6d2 --- /dev/null +++ b/dexcom-library/build.gradle @@ -0,0 +1,49 @@ + +group = 'org.radarbase' +version = '0.0.1' + +apply plugin: 'maven-publish' + +repositories { + // Use jcenter for resolving dependencies. + // You can declare any Maven/Ivy/file repository here. + mavenCentral() +} + +dependencies { + // Use the Kotlin JDK 8 standard library. + implementation libs.kotlin.stdlib + + implementation libs.okhttp + + implementation libs.radar.schemas.commons + + implementation libs.jackson.annotations + + implementation libs.jackson.databind + + implementation libs.avro + + implementation libs.jackson.datatype.jsr310 + + implementation libs.slf4j.api + + // Use the Kotlin test library. + testImplementation libs.kotlin.test + + // Use the Kotlin JUnit integration. + testImplementation libs.kotlin.test.junit +} + +project.afterEvaluate { + publishing { + publications { + library(MavenPublication) { + setGroupId "$group" + setArtifactId "dexcom-library" + version "$version" + from components.java + } + } + } +} \ No newline at end of file diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomAlertsConverter.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomAlertsConverter.kt new file mode 100644 index 00000000..83dbe2af --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomAlertsConverter.kt @@ -0,0 +1,46 @@ +package org.radarbase.dexcom.converter + +import com.fasterxml.jackson.databind.JsonNode +import org.radarbase.dexcom.user.User +import org.radarcns.connector.dexcom.DexcomAlert +import java.time.Instant + +class DexcomAlertsConverter( + private val topic: String = "connect_dexcom_alert", +) : DexcomDataConverter { + override fun processRecords( + root: JsonNode, + user: User, + ): Sequence> { + val array = root.get("records") + ?: return emptySequence() + return array.asSequence() + .mapCatching { + val systemTimeInstant = DexcomEGVConverter.parseDexcomTime(it.get("systemTime").asText()) + TopicData( + key = user.observationKey, + topic = topic, + offset = systemTimeInstant.epochSecond, + value = it.toDexcomAlert(systemTimeInstant), + ) + } + } + + private fun JsonNode.toDexcomAlert(systemTimeInstant: Instant): DexcomAlert = + DexcomAlert.newBuilder().apply { + recordId = get("recordId").asText() + systemTime = systemTimeInstant.epochSecond.toDouble() + displayTime = textOrNull("displayTime") + alertName = get("alertName").asText() + alertState = get("alertState").asText() + displayDevice = textOrNull("displayDevice") + transmitterGeneration = textOrNull("transmitterGeneration") + transmitterGenerationVariant = textOrNull("transmitterGenerationVariant") + transmitterId = textOrNull("transmitterId") + displayApp = textOrNull("displayApp") + timeReceived = System.currentTimeMillis() / 1000.0 + }.build() + + private fun JsonNode.textOrNull(field: String): String? = + get(field)?.takeIf { !it.isNull }?.asText() +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomCalibrationsConverter.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomCalibrationsConverter.kt new file mode 100644 index 00000000..08daa423 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomCalibrationsConverter.kt @@ -0,0 +1,52 @@ +package org.radarbase.dexcom.converter + +import com.fasterxml.jackson.databind.JsonNode +import org.radarbase.dexcom.user.User +import org.radarcns.connector.dexcom.DexcomCalibration +import java.time.Instant + +class DexcomCalibrationsConverter( + private val topic: String = "connect_dexcom_calibration", +) : DexcomDataConverter { + override fun processRecords( + root: JsonNode, + user: User, + ): Sequence> { + val array = root.get("records") + ?: return emptySequence() + return array.asSequence() + .mapCatching { + val systemTimeInstant = DexcomEGVConverter.parseDexcomTime(it.get("systemTime").asText()) + TopicData( + key = user.observationKey, + topic = topic, + offset = systemTimeInstant.epochSecond, + value = it.toDexcomCalibration(systemTimeInstant), + ) + } + } + + private fun JsonNode.toDexcomCalibration(systemTimeInstant: Instant): DexcomCalibration = + DexcomCalibration.newBuilder().apply { + recordId = get("recordId").asText() + systemTime = systemTimeInstant.epochSecond.toDouble() + displayTime = textOrNull("displayTime") + unit = textOrNull("unit") + value = intOrNull("value") + displayDevice = textOrNull("displayDevice") + transmitterId = textOrNull("transmitterId") + transmitterTicks = longOrNull("transmitterTicks") + transmitterGeneration = textOrNull("transmitterGeneration") + transmitterGenerationVariant = textOrNull("transmitterGenerationVariant") + timeReceived = System.currentTimeMillis() / 1000.0 + }.build() + + private fun JsonNode.textOrNull(field: String): String? = + get(field)?.takeIf { !it.isNull }?.asText() + + private fun JsonNode.intOrNull(field: String): Int? = + get(field)?.takeIf { !it.isNull }?.asInt() + + private fun JsonNode.longOrNull(field: String): Long? = + get(field)?.takeIf { !it.isNull }?.asLong() +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomDataConverter.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomDataConverter.kt new file mode 100644 index 00000000..e92ee722 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomDataConverter.kt @@ -0,0 +1,37 @@ +package org.radarbase.dexcom.converter + +import com.fasterxml.jackson.databind.JsonNode +import okhttp3.Headers +import org.radarbase.dexcom.request.DexcomRequestGenerator.Companion.JSON_READER +import org.radarbase.dexcom.request.RestRequest +import org.radarbase.dexcom.user.User +import java.time.Instant + +interface DexcomDataConverter : RecordConverter { + fun processRecords( + root: JsonNode, + user: User, + ): Sequence> + + override fun convert( + request: RestRequest, + headers: Headers, + data: ByteArray, + ): List { + val node = JSON_READER.readTree(data) + + return processRecords(node, request.user) + .mapNotNull { result -> + result.fold( + { it }, + { + RecordConverter.logger.error("Data conversion failed: ${it.message}") + null + }, + ) + } + .toList() + } + + fun Instant.toEpoch(): Long = toEpochMilli() / 1000 +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomDataRangeConverter.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomDataRangeConverter.kt new file mode 100644 index 00000000..dd2e3f9f --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomDataRangeConverter.kt @@ -0,0 +1,142 @@ +package org.radarbase.dexcom.converter + +import com.fasterxml.jackson.databind.JsonNode +import org.radarbase.dexcom.user.User +import org.radarcns.connector.dexcom.DexcomDataRange +import org.radarcns.connector.dexcom.DexcomDataRangeMoment +import org.radarcns.connector.dexcom.DexcomDataRangeMomentEnd +import org.radarcns.connector.dexcom.DexcomDataRangeMomentEgvsEnd +import org.radarcns.connector.dexcom.DexcomDataRangeMomentEgvsStart +import org.radarcns.connector.dexcom.DexcomDataRangeMomentEventsEnd +import org.radarcns.connector.dexcom.DexcomDataRangeMomentEventsStart +import org.radarcns.connector.dexcom.DexcomDataRangeWindow +import org.radarcns.connector.dexcom.DexcomDataRangeWindowEgvs +import org.radarcns.connector.dexcom.DexcomDataRangeWindowEvents +import java.time.Instant + +/** + * /dataRange returns a single object (not records[]). + */ +class DexcomDataRangeConverter( + private val topic: String = "connect_dexcom_data_range", +) : DexcomDataConverter { + override fun processRecords( + root: JsonNode, + user: User, + ): Sequence> { + return sequenceOf(root).mapCatching { + val value = it.toDexcomDataRange() + val offset = it.offsetFromResponse() + TopicData( + key = user.observationKey, + topic = topic, + offset = offset, + value = value, + ) + } + } + + private fun JsonNode.offsetFromResponse(): Long { + val candidates = listOfNotNull( + get("egvs")?.get("end")?.get("systemTime")?.asText(), + get("events")?.get("end")?.get("systemTime")?.asText(), + get("calibrations")?.get("end")?.get("systemTime")?.asText(), + ) + return candidates.firstNotNullOfOrNull { text -> + runCatching { DexcomEGVConverter.parseDexcomTime(text).epochSecond }.getOrNull() + } ?: Instant.now().epochSecond + } + + private fun JsonNode.toDexcomDataRange(): DexcomDataRange = + DexcomDataRange.newBuilder().apply { + calibrations = get("calibrations")?.takeIf { !it.isNull }?.toCalibrationsWindow() + egvs = get("egvs")?.takeIf { !it.isNull }?.toEgvsWindow() + events = get("events")?.takeIf { !it.isNull }?.toEventsWindow() + timeReceived = System.currentTimeMillis() / 1000.0 + }.build() + + private fun JsonNode.toCalibrationsWindow(): DexcomDataRangeWindow = + DexcomDataRangeWindow.newBuilder().apply { + start = get("start")?.toMoment() ?: emptyMoment() + end = get("end")?.toMomentEnd() ?: emptyMomentEnd() + }.build() + + private fun JsonNode.toEgvsWindow(): DexcomDataRangeWindowEgvs = + DexcomDataRangeWindowEgvs.newBuilder().apply { + start = get("start")?.toEgvsStart() ?: emptyEgvsStart() + end = get("end")?.toEgvsEnd() ?: emptyEgvsEnd() + }.build() + + private fun JsonNode.toEventsWindow(): DexcomDataRangeWindowEvents = + DexcomDataRangeWindowEvents.newBuilder().apply { + start = get("start")?.toEventsStart() ?: emptyEventsStart() + end = get("end")?.toEventsEnd() ?: emptyEventsEnd() + }.build() + + private fun JsonNode.toMoment(): DexcomDataRangeMoment = + DexcomDataRangeMoment.newBuilder().apply { + systemTime = systemTimeEpochOrNull() + displayTime = textOrNull("displayTime") + }.build() + + private fun JsonNode.toMomentEnd(): DexcomDataRangeMomentEnd = + DexcomDataRangeMomentEnd.newBuilder().apply { + systemTime = systemTimeEpochOrNull() + displayTime = textOrNull("displayTime") + }.build() + + private fun JsonNode.toEgvsStart(): DexcomDataRangeMomentEgvsStart = + DexcomDataRangeMomentEgvsStart.newBuilder().apply { + systemTime = systemTimeEpochOrNull() + displayTime = textOrNull("displayTime") + }.build() + + private fun JsonNode.toEgvsEnd(): DexcomDataRangeMomentEgvsEnd = + DexcomDataRangeMomentEgvsEnd.newBuilder().apply { + systemTime = systemTimeEpochOrNull() + displayTime = textOrNull("displayTime") + }.build() + + private fun JsonNode.toEventsStart(): DexcomDataRangeMomentEventsStart = + DexcomDataRangeMomentEventsStart.newBuilder().apply { + systemTime = systemTimeEpochOrNull() + displayTime = textOrNull("displayTime") + }.build() + + private fun JsonNode.toEventsEnd(): DexcomDataRangeMomentEventsEnd = + DexcomDataRangeMomentEventsEnd.newBuilder().apply { + systemTime = systemTimeEpochOrNull() + displayTime = textOrNull("displayTime") + }.build() + + private fun JsonNode.systemTimeEpochOrNull(): Double? = + get("systemTime") + ?.takeIf { !it.isNull } + ?.asText() + ?.let { text -> + runCatching { + DexcomEGVConverter.parseDexcomTime(text).epochSecond.toDouble() + }.getOrNull() + } + + private fun JsonNode.textOrNull(field: String): String? = + get(field)?.takeIf { !it.isNull }?.asText() + + private fun emptyMoment() = + DexcomDataRangeMoment.newBuilder().build() + + private fun emptyMomentEnd() = + DexcomDataRangeMomentEnd.newBuilder().build() + + private fun emptyEgvsStart() = + DexcomDataRangeMomentEgvsStart.newBuilder().build() + + private fun emptyEgvsEnd() = + DexcomDataRangeMomentEgvsEnd.newBuilder().build() + + private fun emptyEventsStart() = + DexcomDataRangeMomentEventsStart.newBuilder().build() + + private fun emptyEventsEnd() = + DexcomDataRangeMomentEventsEnd.newBuilder().build() +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomDevicesConverter.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomDevicesConverter.kt new file mode 100644 index 00000000..23558606 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomDevicesConverter.kt @@ -0,0 +1,119 @@ +package org.radarbase.dexcom.converter + +import com.fasterxml.jackson.databind.JsonNode +import org.radarbase.dexcom.user.User +import org.radarcns.connector.dexcom.DexcomAlertSchedule +import org.radarcns.connector.dexcom.DexcomAlertScheduleOverride +import org.radarcns.connector.dexcom.DexcomAlertScheduleSettings +import org.radarcns.connector.dexcom.DexcomDevice +import org.radarcns.connector.dexcom.DexcomDeviceAlertSetting +import java.time.Instant + +class DexcomDevicesConverter( + private val topic: String = "connect_dexcom_device", +) : DexcomDataConverter { + override fun processRecords( + root: JsonNode, + user: User, + ): Sequence> { + val array = root.get("records") + ?: return emptySequence() + return array.asSequence() + .mapCatching { + val lastUpload = textOrNull(it, "lastUploadDate") + val offsetInstant = lastUpload + ?.let { value -> runCatching { DexcomEGVConverter.parseDexcomTime(value) }.getOrNull() } + ?: Instant.now() + TopicData( + key = user.observationKey, + topic = topic, + offset = offsetInstant.epochSecond, + value = it.toDexcomDevice(), + ) + } + } + + private fun JsonNode.toDexcomDevice(): DexcomDevice = + DexcomDevice.newBuilder().apply { + transmitterGeneration = textOrNull(this@toDexcomDevice, "transmitterGeneration") + transmitterGenerationVariant = textOrNull(this@toDexcomDevice, "transmitterGenerationVariant") + displayDevice = textOrNull(this@toDexcomDevice, "displayDevice") + displayApp = textOrNull(this@toDexcomDevice, "displayApp") + lastUploadDate = textOrNull(this@toDexcomDevice, "lastUploadDate") + alertSchedules = get("alertSchedules") + ?.takeIf { it.isArray } + ?.map { it.toAlertSchedule() } + ?: emptyList() + transmitterId = textOrNull(this@toDexcomDevice, "transmitterId") + timeReceived = System.currentTimeMillis() / 1000.0 + }.build() + + private fun JsonNode.toAlertSchedule(): DexcomAlertSchedule = + DexcomAlertSchedule.newBuilder().apply { + alertScheduleSettings = get("alertScheduleSettings")?.toAlertScheduleSettings() + ?: DexcomAlertScheduleSettings.newBuilder() + .setAlertScheduleName("") + .setIsEnabled(false) + .setStartTime("00:00") + .setEndTime("00:00") + .setDaysOfWeek(emptyList()) + .build() + alertSettings = get("alertSettings") + ?.takeIf { it.isArray } + ?.map { it.toDeviceAlertSetting() } + ?: emptyList() + }.build() + + private fun JsonNode.toAlertScheduleSettings(): DexcomAlertScheduleSettings { + val builder = DexcomAlertScheduleSettings.newBuilder() + builder.alertScheduleName = get("alertScheduleName")?.asText().orEmpty() + builder.isEnabled = get("isEnabled")?.asBoolean() ?: false + builder.startTime = get("startTime")?.asText().orEmpty() + builder.endTime = get("endTime")?.asText().orEmpty() + builder.isActive = booleanOrNull("isActive") + builder.setOverride(get("override")?.takeIf { !it.isNull }?.toOverride()) + builder.daysOfWeek = get("daysOfWeek") + ?.takeIf { it.isArray } + ?.map { it.asText() } + ?: emptyList() + return builder.build() + } + + private fun JsonNode.toOverride(): DexcomAlertScheduleOverride = + DexcomAlertScheduleOverride.newBuilder().apply { + isOverrideEnabled = booleanOrNull("isOverrideEnabled") + mode = textOrNull(this@toOverride, "mode") + endTime = textOrNull(this@toOverride, "endTime") + }.build() + + private fun JsonNode.toDeviceAlertSetting(): DexcomDeviceAlertSetting = + DexcomDeviceAlertSetting.newBuilder().apply { + systemTime = get("systemTime") + ?.takeIf { !it.isNull } + ?.asText() + ?.let { value -> + runCatching { + DexcomEGVConverter.parseDexcomTime(value).epochSecond.toDouble() + }.getOrNull() + } + displayTime = textOrNull(this@toDeviceAlertSetting, "displayTime") + alertName = get("alertName").asText() + value = intOrNull("value") + unit = textOrNull(this@toDeviceAlertSetting, "unit") + snooze = intOrNull("snooze") + enabled = get("enabled")?.asBoolean() ?: false + secondaryTriggerCondition = intOrNull("SecondaryTriggerCondition") + ?: intOrNull("secondaryTriggerCondition") + soundTheme = textOrNull(this@toDeviceAlertSetting, "soundTheme") + soundOutputMode = textOrNull(this@toDeviceAlertSetting, "soundOutputMode") + }.build() + + private fun textOrNull(node: JsonNode, field: String): String? = + node.get(field)?.takeIf { !it.isNull }?.asText() + + private fun JsonNode.booleanOrNull(field: String): Boolean? = + get(field)?.takeIf { !it.isNull }?.asBoolean() + + private fun JsonNode.intOrNull(field: String): Int? = + get(field)?.takeIf { !it.isNull }?.asInt() +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomEGVConverter.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomEGVConverter.kt new file mode 100644 index 00000000..e48d30bc --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomEGVConverter.kt @@ -0,0 +1,72 @@ +package org.radarbase.dexcom.converter + +import com.fasterxml.jackson.databind.JsonNode +import org.radarbase.dexcom.user.User +import org.radarcns.connector.dexcom.DexcomEgv +import java.time.Instant +import java.time.OffsetDateTime +import java.time.format.DateTimeParseException + +class DexcomEGVConverter( + private val topic: String = "connect_dexcom_egv", +) : DexcomDataConverter { + override fun processRecords( + root: JsonNode, + user: User, + ): Sequence> { + val array = root.get("records") + ?: return emptySequence() + return array.asSequence() + .mapCatching { + val systemTimeInstant = parseDexcomTime(it.get("systemTime").asText()) + TopicData( + key = user.observationKey, + topic = topic, + offset = systemTimeInstant.epochSecond, + value = it.toDexcomEgv(systemTimeInstant), + ) + } + } + + private fun JsonNode.toDexcomEgv(systemTimeInstant: Instant): DexcomEgv = + DexcomEgv.newBuilder().apply { + recordId = get("recordId").asText() + systemTime = systemTimeInstant.epochSecond.toDouble() + displayTime = textOrNull("displayTime") + transmitterId = textOrNull("transmitterId") + transmitterTicks = longOrNull("transmitterTicks") + value = intOrNull("value") + status = textOrNull("status") + trend = textOrNull("trend") + trendRate = doubleOrNull("trendRate") + unit = get("unit")?.takeIf { !it.isNull }?.asText() ?: "unknown" + rateUnit = textOrNull("rateUnit") + displayDevice = textOrNull("displayDevice") + transmitterGeneration = textOrNull("transmitterGeneration") + transmitterGenerationVariant = textOrNull("transmitterGenerationVariant") + displayApp = textOrNull("displayApp") + timeReceived = System.currentTimeMillis() / 1000.0 + }.build() + + private fun JsonNode.textOrNull(field: String): String? = + get(field)?.takeIf { !it.isNull }?.asText() + + private fun JsonNode.intOrNull(field: String): Int? = + get(field)?.takeIf { !it.isNull }?.asInt() + + private fun JsonNode.longOrNull(field: String): Long? = + get(field)?.takeIf { !it.isNull }?.asLong() + + private fun JsonNode.doubleOrNull(field: String): Double? = + get(field)?.takeIf { !it.isNull }?.asDouble() + + companion object { + fun parseDexcomTime(value: String): Instant { + return try { + OffsetDateTime.parse(value).toInstant() + } catch (_: DateTimeParseException) { + Instant.parse(value) + } + } + } +} \ No newline at end of file diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomEventsConverter.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomEventsConverter.kt new file mode 100644 index 00000000..5e344bcb --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/DexcomEventsConverter.kt @@ -0,0 +1,57 @@ +package org.radarbase.dexcom.converter + +import com.fasterxml.jackson.databind.JsonNode +import org.radarbase.dexcom.user.User +import org.radarcns.connector.dexcom.DexcomEvent +import java.time.Instant + +class DexcomEventsConverter( + private val topic: String = "connect_dexcom_event", +) : DexcomDataConverter { + override fun processRecords( + root: JsonNode, + user: User, + ): Sequence> { + val array = root.get("records") + ?: return emptySequence() + return array.asSequence() + .mapCatching { + val systemTimeInstant = DexcomEGVConverter.parseDexcomTime(it.get("systemTime").asText()) + TopicData( + key = user.observationKey, + topic = topic, + offset = systemTimeInstant.epochSecond, + value = it.toDexcomEvent(systemTimeInstant), + ) + } + } + + private fun JsonNode.toDexcomEvent(systemTimeInstant: Instant): DexcomEvent = + DexcomEvent.newBuilder().apply { + recordId = get("recordId").asText() + systemTime = systemTimeInstant.epochSecond.toDouble() + displayTime = textOrNull("displayTime") + eventStatus = get("eventStatus").asText() + eventType = get("eventType").asText() + eventSubType = textOrNull("eventSubType") + value = valueAsStringOrNull("value") + unit = textOrNull("unit") + transmitterId = textOrNull("transmitterId") + transmitterGeneration = textOrNull("transmitterGeneration") + transmitterGenerationVariant = textOrNull("transmitterGenerationVariant") + displayDevice = textOrNull("displayDevice") + timeReceived = System.currentTimeMillis() / 1000.0 + }.build() + + private fun JsonNode.textOrNull(field: String): String? = + get(field)?.takeIf { !it.isNull }?.asText() + + /** Dexcom may return value as string or number; Avro field is string. */ + private fun JsonNode.valueAsStringOrNull(field: String): String? { + val node = get(field)?.takeIf { !it.isNull } ?: return null + return when { + node.isNumber -> node.numberValue().toString() + else -> node.asText() + } + } +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/RecordConverter.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/RecordConverter.kt new file mode 100644 index 00000000..b05ac022 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/RecordConverter.kt @@ -0,0 +1,19 @@ +package org.radarbase.dexcom.converter + +import okhttp3.Headers +import org.radarbase.dexcom.request.RestRequest +import org.slf4j.LoggerFactory +import java.io.IOException + +interface RecordConverter { + @Throws(IOException::class) + fun convert( + request: RestRequest, + headers: Headers, + data: ByteArray, + ): List + + companion object { + var logger = LoggerFactory.getLogger(RecordConverter::class.java) + } +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/SequenceExtensions.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/SequenceExtensions.kt new file mode 100644 index 00000000..0952551b --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/SequenceExtensions.kt @@ -0,0 +1,18 @@ +package org.radarbase.dexcom.converter + +import org.slf4j.LoggerFactory + +val logger = LoggerFactory.getLogger("org.radarbase.oura.converter.SequenceExtensions") + +fun Sequence.mapCatching(fn: (T) -> S): Sequence> = map { t -> + runCatching { + fn(t) + } +} + +fun Sequence.mapIndexedCatching(fn: (index: Int, T) -> S): Sequence> = + mapIndexed { index, t -> + runCatching { + fn(index, t) + } + } diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/TopicData.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/TopicData.kt new file mode 100644 index 00000000..859e634f --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/converter/TopicData.kt @@ -0,0 +1,11 @@ +package org.radarbase.dexcom.converter + +import org.apache.avro.specific.SpecificRecord + +/** Single value for a topic. */ +data class TopicData( + val topic: String, + val key: SpecificRecord, + val value: SpecificRecord, + val offset: Long, +) \ No newline at end of file diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/offset/Offset.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/offset/Offset.kt new file mode 100644 index 00000000..d2279dd4 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/offset/Offset.kt @@ -0,0 +1,11 @@ +package org.radarbase.dexcom.offset + +import org.radarbase.dexcom.route.Route +import org.radarbase.dexcom.user.User +import java.time.Instant + +data class Offset( + val user: User, + val route: Route, + val offset: Instant, +) \ No newline at end of file diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/offset/Offsets.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/offset/Offsets.kt new file mode 100644 index 00000000..3681411b --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/offset/Offsets.kt @@ -0,0 +1,7 @@ +package org.radarbase.dexcom.offset + +import org.radarbase.dexcom.request.Offset + +data class Offsets( + val offsets: List, +) diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/DexcomOffsetManager.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/DexcomOffsetManager.kt new file mode 100644 index 00000000..842a4b36 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/DexcomOffsetManager.kt @@ -0,0 +1,18 @@ +package org.radarbase.dexcom.request + +import org.radarbase.dexcom.route.Route +import org.radarbase.dexcom.user.User +import java.time.Instant + +data class Offset( + val user: User, + val route: Route, + val offset: Instant, +) + +interface DexcomOffsetManager { + + fun getOffset(route: Route, user: User): Offset? + + fun updateOffsets(route: Route, user: User, offset: Instant) +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/DexcomRequestGenerator.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/DexcomRequestGenerator.kt new file mode 100644 index 00000000..3cd00e58 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/DexcomRequestGenerator.kt @@ -0,0 +1,304 @@ +package org.radarbase.dexcom.request + +import com.fasterxml.jackson.core.JsonFactory +import com.fasterxml.jackson.databind.ObjectMapper +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule +import okhttp3.Response +import okhttp3.ResponseBody +import org.radarbase.dexcom.converter.TopicData +import org.radarbase.dexcom.route.DexcomRouteFactory +import org.radarbase.dexcom.route.Route +import org.radarbase.dexcom.user.User +import org.radarbase.dexcom.user.UserRepository +import org.slf4j.LoggerFactory +import java.io.IOException +import java.time.Duration +import java.time.Instant + +class DexcomRequestGenerator +@JvmOverloads +constructor( + private val userRepository: UserRepository, + private val dexcomOffsetManager: DexcomOffsetManager, + val routes: List = DexcomRouteFactory.getRoutes(userRepository), + private val defaultQueryRange: Duration = Duration.ofDays(15), +) : RequestGenerator { + private val routeNextRequest: MutableMap = mutableMapOf() + + var nextRequestTime: Instant = Instant.MIN + + private val shouldBackoff: Boolean + get() = Instant.now() < nextRequestTime + + override fun requests( + user: User, + max: Int, + ): Sequence { + return routes.asSequence() + .flatMap { route -> + if (routeReady(user, route)) { + generateRequests(route, user) + } else { + logger.info( + "Skip {} for {}: route in backoff until {}", + route, + user.versionedId, + routeNextRequest[routeKey(route, user)], + ) + emptySequence() + } + } + } + + override fun requests( + route: Route, + max: Int, + ): Sequence { + return userRepository + .stream() + .flatMap { user -> + if (routeReady(user, route)) { + generateRequests(route, user) + } else { + logger.info( + "Skip {} for {}: route in backoff until {}", + route, + user.versionedId, + routeNextRequest[routeKey(route, user)], + ) + emptySequence() + } + } + } + + override fun requests( + route: Route, + user: User, + max: Int, + ): Sequence { + return if (routeReady(user, route)) { + generateRequests(route, user) + } else { + logger.info( + "Skip {} for {}: route in backoff until {}", + route, + user.versionedId, + routeNextRequest[routeKey(route, user)], + ) + emptySequence() + } + } + + fun generateRequests( + route: Route, + user: User, + ): Sequence { + val offset = dexcomOffsetManager.getOffset(route, user) + val startDate = user.startDate + val startOffset: Instant = + if (offset == null) { + logger.info("No offsets found for $user, using the start date.") + startDate + } else { + val offsetTime = offset.offset + logger.info("Offsets found in persistence: $offsetTime") + offsetTime.coerceAtLeast(startDate) + } + val endDate = user.endDate?.coerceAtMost(Instant.now()) ?: Instant.now() + if (!startOffset.isBefore(endDate)) { + val userEnd = user.endDate + if (userEnd != null && endDate == userEnd && + Duration.between(userEnd, Instant.now()) > Duration.ofDays(30) + ) { + val key = routeKey(route, user) + routeNextRequest[key] = Instant.MAX + logger.info( + "Disable future requests for {}: user={}, endDate={} (>30d ago), startOffset={}", + route, + user.versionedId, + userEnd, + startOffset, + ) + } + logger.info( + "Skip {} for {}: interval empty (startOffset={} >= endDate={}), " + + "persistedOffset={}, userStartDate={}", + route, + user.versionedId, + startOffset, + endDate, + offset?.offset, + startDate, + ) + return emptySequence() + } + val timeSinceStart = Duration.between(startOffset, Instant.now()) + return if (timeSinceStart > HISTORICAL_DATA_THRESHOLD) { + val endTime = (startOffset + HISTORICAL_QUERY_RANGE).coerceAtMost(endDate) + route.generateRequests(user, startOffset, endTime) + } else { + route.generateRequests(user, startOffset, endDate, USER_MAX_REQUESTS) + } + } + + fun handleResponse( + req: RestRequest, + response: Response, + ): DexcomResult> { + if (response.isSuccessful) { + return DexcomResult.Success(requestSuccessful(req, response)) + } + return try { + DexcomResult.Error(requestFailed(req, response)) + } catch (e: TooManyRequestsException) { + DexcomResult.Success(emptyList()) + } + } + + override fun requestSuccessful( + request: RestRequest, + response: Response, + ): List { + logger.debug("Request successful: {}..", request.request) + val body: ResponseBody = response.body ?: return emptyList() + val data = body.bytes() + val records = + request.route.converters.flatMap { it.convert(request, response.headers, data) } + val offset = records.maxByOrNull { it.offset }?.offset + if (offset != null) { + logger.info("Writing ${records.size} records to offsets...") + val maxOffsetTime = Instant.ofEpochSecond(offset) + val dataAge = Duration.between(maxOffsetTime, Instant.now()) + val nextOffset = if (dataAge <= Duration.ofDays(7)) { + maxOffsetTime.plus(OFFSET_BUFFER) + } else { + maxOf(maxOffsetTime.plus(OFFSET_BUFFER), request.endDate) + } + dexcomOffsetManager.updateOffsets( + request.route, + request.user, + nextOffset, + ) + val nextRequestTime = Instant.now().plus(SUCCESS_BACK_OFF_TIME) + val key = routeKey(request.route, request.user) + routeNextRequest[key] = + routeNextRequest[key]?.let { if (it > nextRequestTime) it else nextRequestTime } + ?: nextRequestTime + } else { + if (request.startDate.plus(TIME_AFTER_REQUEST).isBefore(Instant.now())) { + logger.info("No records found, updating offsets to end date..") + dexcomOffsetManager.updateOffsets( + request.route, + request.user, + request.endDate, + ) + val key = routeKey(request.route, request.user) + routeNextRequest[key] = Instant.now().plus(SUCCESS_BACK_OFF_TIME) + } else { + val key = routeKey(request.route, request.user) + routeNextRequest[key] = Instant.now().plus(BACK_OFF_TIME) + } + } + return records + } + + override fun requestFailed( + request: RestRequest, + response: Response, + ): DexcomError { + return when (response.code) { + 429 -> { + logger.info("Too many requests, rate limit reached. Backing off...") + nextRequestTime = Instant.now().plus(BACK_OFF_TIME) + DexcomRateLimitError("Rate limit reached.", TooManyRequestsException(), "429") + } + 403 -> { + logger.warn( + "User ${request.user} has expired. Please renew the subscription.", + ) + routeNextRequest[routeKey(request.route, request.user)] = + Instant.now().plus(USER_BACK_OFF_TIME) + DexcomAccessForbiddenError( + "Dexcom subscription has expired or API data not available.", + IOException("Forbidden"), + "403", + ) + } + 401 -> { + logger.warn( + "User ${request.user} access token is expired, malformed, or revoked. " + + response.body?.string(), + ) + routeNextRequest[routeKey(request.route, request.user)] = + Instant.now().plus(USER_BACK_OFF_TIME) + DexcomUnauthorizedAccessError( + "Access token expired or revoked.", + IOException("Unauthorized"), + "401", + ) + } + 400 -> { + logger.warn("Client exception.") + nextRequestTime = Instant.now().plus(BACK_OFF_TIME) + routeNextRequest[routeKey(request.route, request.user)] = + Instant.now().plus(BACK_OFF_TIME) + DexcomClientException( + "Client unsupported or unauthorized.", + IOException("Invalid client"), + "400", + ) + } + 422 -> { + logger.warn("Request failed: {}, {}", request, response) + routeNextRequest[routeKey(request.route, request.user)] = + Instant.now().plus(BACK_OFF_TIME) + DexcomValidationError( + response.body?.string().orEmpty(), + IOException("Validation error"), + "422", + ) + } + 404 -> { + logger.warn("Not found.") + routeNextRequest[routeKey(request.route, request.user)] = + Instant.now().plus(BACK_OFF_TIME) + DexcomNotFoundError( + response.body?.string().orEmpty(), + IOException("Data not found"), + "404", + ) + } + else -> { + logger.warn("Request failed: {}, {}", request, response) + routeNextRequest[routeKey(request.route, request.user)] = + Instant.now().plus(BACK_OFF_TIME) + DexcomGenericError( + response.body?.string().orEmpty(), + IOException("Unknown error"), + response.code.toString(), + ) + } + } + } + + private fun routeReady(user: User, route: Route): Boolean { + val key = routeKey(route, user) + return routeNextRequest[key]?.let { Instant.now().isAfter(it) } ?: true + } + + private fun routeKey(route: Route, user: User): String = user.versionedId + "#" + route + + companion object { + private val logger = LoggerFactory.getLogger(DexcomRequestGenerator::class.java) + private val BACK_OFF_TIME = Duration.ofMinutes(10L) + private val TIME_AFTER_REQUEST = Duration.ofDays(30) + private val USER_BACK_OFF_TIME = Duration.ofHours(12L) + private val SUCCESS_BACK_OFF_TIME = Duration.ofSeconds(10L) + private val OFFSET_BUFFER = Duration.ofHours(12) + private val USER_MAX_REQUESTS = 1000 + private val HISTORICAL_DATA_THRESHOLD = Duration.ofDays(365L) + private val HISTORICAL_QUERY_RANGE = Duration.ofDays(365L) + val JSON_FACTORY = JsonFactory() + val JSON_READER = ObjectMapper(JSON_FACTORY).registerModule(JavaTimeModule()).reader() + } +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/DexcomResult.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/DexcomResult.kt new file mode 100644 index 00000000..2f153de9 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/DexcomResult.kt @@ -0,0 +1,64 @@ +package org.radarbase.dexcom.request + +sealed class DexcomResult { + data class Success(val value: T) : DexcomResult() + data class Error(val error: DexcomError) : DexcomResult() +} + +sealed interface DexcomError + +sealed class DexcomErrorBase( + val message: String, + val cause: Exception? = null, + val code: String, +) : DexcomError + +class DexcomRateLimitError(message: String, cause: Exception? = null, code: String) : DexcomErrorBase( + message, + cause, + code, +) + +class DexcomClientException(message: String, cause: Exception? = null, code: String) : DexcomErrorBase( + message, + cause, + code, +) + +class DexcomUnauthorizedAccessError( + message: String, + cause: Exception? = null, + code: String, +) : DexcomErrorBase( + message, + cause, + code, +) + +class DexcomAccessForbiddenError( + message: String, + cause: Exception? = null, + code: String, +) : DexcomErrorBase( + message, + cause, + code, +) + +class DexcomValidationError(message: String, cause: Exception? = null, code: String) : DexcomErrorBase( + message, + cause, + code, +) + +class DexcomGenericError(message: String, cause: Exception? = null, code: String) : DexcomErrorBase( + message, + cause, + code, +) + +class DexcomNotFoundError(message: String, cause: Exception? = null, code: String) : DexcomErrorBase( + message, + cause, + code, +) diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/RequestGenerator.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/RequestGenerator.kt new file mode 100644 index 00000000..2a3f00fa --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/RequestGenerator.kt @@ -0,0 +1,23 @@ +package org.radarbase.dexcom.request + +import okhttp3.Response +import org.radarbase.dexcom.converter.TopicData +import org.radarbase.dexcom.route.Route +import org.radarbase.dexcom.user.User + + +interface RequestGenerator { + + fun requests(user: User, max: Int): Sequence + + fun requests(route: Route, user: User, max: Int): Sequence + + fun requests(route: Route, max: Int): Sequence + + fun requestSuccessful(request: RestRequest, response: Response): List + + fun requestFailed(request: RestRequest, response: Response): DexcomError +} + + + diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/RestRequest.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/RestRequest.kt new file mode 100644 index 00000000..acb988d9 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/RestRequest.kt @@ -0,0 +1,14 @@ +package org.radarbase.dexcom.request + +import okhttp3.Request +import org.radarbase.dexcom.route.DexcomRoute +import org.radarbase.dexcom.user.User +import java.time.Instant + +data class RestRequest( + val request: Request, + val user: User, + val route: DexcomRoute, + val startDate: Instant, + val endDate: Instant, +) diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/TooManyRequestsException.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/TooManyRequestsException.kt new file mode 100644 index 00000000..fe37406f --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/request/TooManyRequestsException.kt @@ -0,0 +1,3 @@ +package org.radarbase.dexcom.request + +class TooManyRequestsException : RuntimeException() diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomAlertsRoute.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomAlertsRoute.kt new file mode 100644 index 00000000..7c10ae8f --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomAlertsRoute.kt @@ -0,0 +1,16 @@ +package org.radarbase.dexcom.route + +import org.radarbase.dexcom.converter.DexcomAlertsConverter +import org.radarbase.dexcom.converter.DexcomDataConverter +import org.radarbase.dexcom.user.UserRepository + +class DexcomAlertsRoute( + userRepository: UserRepository, + apiBaseUrl: String = DEFAULT_API_BASE_URL, +) : DexcomRoute(userRepository, apiBaseUrl) { + override fun subPath(): String = "alerts" + + override fun toString(): String = "dexcom_alert" + + override val converters: List = listOf(DexcomAlertsConverter()) +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomCalibrationsRoute.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomCalibrationsRoute.kt new file mode 100644 index 00000000..d4dca169 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomCalibrationsRoute.kt @@ -0,0 +1,16 @@ +package org.radarbase.dexcom.route + +import org.radarbase.dexcom.converter.DexcomCalibrationsConverter +import org.radarbase.dexcom.converter.DexcomDataConverter +import org.radarbase.dexcom.user.UserRepository + +class DexcomCalibrationsRoute( + userRepository: UserRepository, + apiBaseUrl: String = DEFAULT_API_BASE_URL, +) : DexcomRoute(userRepository, apiBaseUrl) { + override fun subPath(): String = "calibrations" + + override fun toString(): String = "dexcom_calibration" + + override val converters: List = listOf(DexcomCalibrationsConverter()) +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomDataRangeRoute.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomDataRangeRoute.kt new file mode 100644 index 00000000..18ba384c --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomDataRangeRoute.kt @@ -0,0 +1,38 @@ +package org.radarbase.dexcom.route + +import org.radarbase.dexcom.converter.DexcomDataConverter +import org.radarbase.dexcom.converter.DexcomDataRangeConverter +import org.radarbase.dexcom.request.RestRequest +import org.radarbase.dexcom.user.User +import org.radarbase.dexcom.user.UserRepository +import java.time.Instant + +/** + * /dataRange has no startDate/endDate query params. + */ +class DexcomDataRangeRoute( + userRepository: UserRepository, + private val dataRangeApiBaseUrl: String = DEFAULT_API_BASE_URL, +) : DexcomRoute(userRepository, dataRangeApiBaseUrl) { + override fun subPath(): String = "dataRange" + + override fun toString(): String = "dexcom_data_range" + + override val converters: List = listOf(DexcomDataRangeConverter()) + + override fun generateRequests( + user: User, + start: Instant, + end: Instant, + ): Sequence { + val request = createRequest(user, "$dataRangeApiBaseUrl/${subPath()}", "") + return sequenceOf(RestRequest(request, user, this, start, end)) + } + + override fun generateRequests( + user: User, + start: Instant, + end: Instant, + max: Int, + ): Sequence = generateRequests(user, start, end).take(max) +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomDevicesRoute.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomDevicesRoute.kt new file mode 100644 index 00000000..b55018cd --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomDevicesRoute.kt @@ -0,0 +1,38 @@ +package org.radarbase.dexcom.route + +import org.radarbase.dexcom.converter.DexcomDataConverter +import org.radarbase.dexcom.converter.DexcomDevicesConverter +import org.radarbase.dexcom.request.RestRequest +import org.radarbase.dexcom.user.User +import org.radarbase.dexcom.user.UserRepository +import java.time.Instant + +/** + * Devices endpoint has no startDate/endDate query params. + */ +class DexcomDevicesRoute( + userRepository: UserRepository, + private val devicesApiBaseUrl: String = DEFAULT_API_BASE_URL, +) : DexcomRoute(userRepository, devicesApiBaseUrl) { + override fun subPath(): String = "devices" + + override fun toString(): String = "dexcom_device" + + override val converters: List = listOf(DexcomDevicesConverter()) + + override fun generateRequests( + user: User, + start: Instant, + end: Instant, + ): Sequence { + val request = createRequest(user, "$devicesApiBaseUrl/${subPath()}", "") + return sequenceOf(RestRequest(request, user, this, start, end)) + } + + override fun generateRequests( + user: User, + start: Instant, + end: Instant, + max: Int, + ): Sequence = generateRequests(user, start, end).take(max) +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomEGVRoute.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomEGVRoute.kt new file mode 100644 index 00000000..d14bb1fb --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomEGVRoute.kt @@ -0,0 +1,15 @@ +package org.radarbase.dexcom.route + +import org.radarbase.dexcom.converter.DexcomDataConverter +import org.radarbase.dexcom.converter.DexcomEGVConverter +import org.radarbase.dexcom.user.UserRepository + +class DexcomEGVRoute( + userRepository: UserRepository, +) : DexcomRoute(userRepository) { + override fun subPath(): String = "egvs" + + override fun toString(): String = "dexcom_egv" + + override val converters: List = listOf(DexcomEGVConverter()) +} \ No newline at end of file diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomEventsRoute.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomEventsRoute.kt new file mode 100644 index 00000000..d3349da4 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomEventsRoute.kt @@ -0,0 +1,16 @@ +package org.radarbase.dexcom.route + +import org.radarbase.dexcom.converter.DexcomDataConverter +import org.radarbase.dexcom.converter.DexcomEventsConverter +import org.radarbase.dexcom.user.UserRepository + +class DexcomEventsRoute( + userRepository: UserRepository, + apiBaseUrl: String = DEFAULT_API_BASE_URL, +) : DexcomRoute(userRepository, apiBaseUrl) { + override fun subPath(): String = "events" + + override fun toString(): String = "dexcom_event" + + override val converters: List = listOf(DexcomEventsConverter()) +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomRoute.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomRoute.kt new file mode 100644 index 00000000..263a48a8 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomRoute.kt @@ -0,0 +1,72 @@ +package org.radarbase.dexcom.route + +import okhttp3.Request +import org.radarbase.dexcom.converter.DexcomDataConverter +import org.radarbase.dexcom.request.RestRequest +import org.radarbase.dexcom.user.User +import org.radarbase.dexcom.user.UserRepository +import java.time.Duration +import java.time.Instant +import java.time.LocalDateTime +import java.time.ZoneOffset +import java.time.format.DateTimeFormatter + +abstract class DexcomRoute( + private val userRepository: UserRepository, + override val maxIntervalPerRequest: Duration = DEFAULT_INTERVAL_PER_REQUEST, +) : Route { + abstract val converters: List + + fun createRequest(user: User, baseUrl: String, queryParams: String): Request { + val accessToken = userRepository.getAccessToken(user) + return Request.Builder() + .url(baseUrl + queryParams) + .header("Authorization", "Bearer $accessToken") + .get() + .build() + } + + override fun generateRequests( + user: User, + start: Instant, + end: Instant, + ): Sequence { + val request = createRequest( + user, + "$DEXCOM_API_BASE_URL/${subPath()}", + "?startDate=${start.toDexcomDate()}&endDate=${end.toDexcomDate()}", + ) + return sequenceOf(RestRequest(request, user, this, start, end)) + } + + override fun generateRequests( + user: User, + start: Instant, + end: Instant, + max: Int, + ): Sequence { + return generateSequence(start) { it + maxIntervalPerRequest } + .takeWhile { it < end } + .take(max) + .map { startRange -> + val endRange = (startRange + maxIntervalPerRequest).coerceAtMost(end) + val request = createRequest( + user, + "$DEXCOM_API_BASE_URL/${subPath()}", + "?startDate=${startRange.toDexcomDate()}&endDate=${endRange.toDexcomDate()}", + ) + RestRequest(request, user, this, startRange, endRange) + } + } + + abstract fun subPath(): String + + fun Instant.toDexcomDate(): String = + LocalDateTime.ofInstant(this, ZoneOffset.UTC).format(DEXCOM_DATE_FORMAT) + + companion object { + const val DEXCOM_API_BASE_URL = "https://api.dexcom.com/v3/users/self" + private val DEXCOM_DATE_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss") + private val DEFAULT_INTERVAL_PER_REQUEST = Duration.ofDays(30L) + } +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomRouteFactory.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomRouteFactory.kt new file mode 100644 index 00000000..4e869745 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/DexcomRouteFactory.kt @@ -0,0 +1,17 @@ +package org.radarbase.dexcom.route + +import org.radarbase.dexcom.user.UserRepository + +object DexcomRouteFactory { + + fun getRoutes(userRepository: UserRepository): List { + return listOf( + DexcomEGVRoute(userRepository), + DexcomEventsRoute(userRepository), + DexcomCalibrationsRoute(userRepository), + DexcomAlertsRoute(userRepository), + DexcomDevicesRoute(userRepository), + DexcomDataRangeRoute(userRepository), + ) + } +} \ No newline at end of file diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/Route.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/Route.kt new file mode 100644 index 00000000..e33d3a6a --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/route/Route.kt @@ -0,0 +1,23 @@ +package org.radarbase.dexcom.route + +import org.radarbase.dexcom.request.RestRequest +import org.radarbase.dexcom.user.User +import java.time.Duration +import java.time.Instant + +interface Route { + + fun generateRequests(user: User, start: Instant, end: Instant): Sequence + + fun generateRequests(user: User, start: Instant, end: Instant, max: Int): Sequence + + /** + * This is how it would appear in the offsets + */ + override fun toString(): String + + /** + * The number of days to request in a single request of this route. + */ + val maxIntervalPerRequest: Duration +} \ No newline at end of file diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/DexcomUser.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/DexcomUser.kt new file mode 100644 index 00000000..0c559113 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/DexcomUser.kt @@ -0,0 +1,28 @@ +package org.radarbase.dexcom.user + +import com.fasterxml.jackson.annotation.JsonIgnoreProperties +import com.fasterxml.jackson.annotation.JsonProperty +import org.radarcns.kafka.ObservationKey +import java.time.Instant + +@JsonIgnoreProperties(ignoreUnknown = true) +data class DexcomUser( + @JsonProperty("id") override val id: String, + @JsonProperty("createdAt") override val createdAt: Instant, + @JsonProperty("projectId") override val projectId: String, + @JsonProperty("userId") override val userId: String, + @JsonProperty("humanReadableUserId") override val humanReadableUserId: String?, + @JsonProperty("sourceId") override val sourceId: String, + @JsonProperty("externalId") override val externalId: String?, + @JsonProperty("isAuthorized") override val isAuthorized: Boolean, + @JsonProperty("startDate") override val startDate: Instant, + @JsonProperty("endDate") override val endDate: Instant? = null, + @JsonProperty("version") override val version: String? = null, + @JsonProperty("serviceUserId") override val serviceUserId: String? = null, +) : User { + override val observationKey: ObservationKey = ObservationKey(projectId, userId, sourceId) + override val versionedId: String = "$id${version?.let { "#$it" } ?: ""}" + + fun isComplete() = + isAuthorized && (endDate == null || startDate.isBefore(endDate)) && serviceUserId != null +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/User.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/User.kt new file mode 100644 index 00000000..da7414b6 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/User.kt @@ -0,0 +1,22 @@ +package org.radarbase.dexcom.user + +import org.radarcns.kafka.ObservationKey +import java.time.Instant + +interface User { + val id: String + val projectId: String + val userId: String + val sourceId: String + val externalId: String? + val startDate: Instant + val endDate: Instant? + val createdAt: Instant + val humanReadableUserId: String? + val serviceUserId: String? + val version: String? + val isAuthorized: Boolean + + val observationKey: ObservationKey + val versionedId: String +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/UserNotAuthorizedException.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/UserNotAuthorizedException.kt new file mode 100644 index 00000000..d22bb2f7 --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/UserNotAuthorizedException.kt @@ -0,0 +1,5 @@ +package org.radarbase.dexcom.user + +class UserNotAuthorizedException(message: String) : Exception(message) { + constructor(user: User) : this("User ${user.id} is not authorized") +} diff --git a/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/UserRepository.kt b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/UserRepository.kt new file mode 100644 index 00000000..ff4c6bae --- /dev/null +++ b/dexcom-library/src/main/kotlin/org/radarbase/dexcom/user/UserRepository.kt @@ -0,0 +1,32 @@ +package org.radarbase.dexcom.user + +import java.io.IOException + +interface UserRepository { + /** + * Get specified user. + * + * @throws IOException if the user cannot be retrieved from the repository. + */ + @Throws(IOException::class) + operator fun get(key: String): User? + + /** + * Get all relevant users. + * + * @throws IOException if the list cannot be retrieved from the repository. + */ + @Throws(IOException::class) + fun stream(): Sequence + + /** + * Get the current access token of given user. + * + * @throws IOException if the new access token cannot be retrieved from the repository. + * @throws NotAuthorizedException if the refresh token is no longer valid. Manual action should + * be taken to get a new refresh token. + * @throws NoSuchElementException if the user does not exists in this repository. + */ + @Throws(IOException::class, UserNotAuthorizedException::class) + fun getAccessToken(user: User): String +} diff --git a/docker-compose.yml b/docker-compose.yml index 53c62257..087720d8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -4,6 +4,7 @@ version: "2.4" volumes: fitbit-logs: {} oura-logs: {} + dexcom-logs: {} services: #---------------------------------------------------------------------------# @@ -231,3 +232,49 @@ services: # SENTRY_DSN: 'https://000000000000.ingest.de.sentry.io/000000000000' # SENTRY_ATTACHSTACKTRACE: true # SENTRY_STACKTRACE_APP_PACKAGES: io.confluent.connect,org.radarbase.connect.rest + + #---------------------------------------------------------------------------# + # RADAR Dexcom connector # + #---------------------------------------------------------------------------# + radar-dexcom-connector: + build: + context: . + dockerfile: ./kafka-connect-dexcom-source/Dockerfile + image: radarbase/radar-connect-dexcom-source + restart: on-failure + volumes: + - ./docker/source-dexcom.properties:/etc/kafka-connect/source-dexcom.properties + - ./docker/users:/var/lib/kafka-connect-dexcom-source/users + - dexcom-logs:/var/lib/kafka-connect-dexcom-source/logs + depends_on: + - zookeeper-1 + - zookeeper-2 + - zookeeper-3 + - kafka-1 + - kafka-2 + - kafka-3 + - schema-registry-1 + environment: + CONNECT_BOOTSTRAP_SERVERS: PLAINTEXT://kafka-1:9092,PLAINTEXT://kafka-2:9092,PLAINTEXT://kafka-3:9092 + CONNECT_REST_PORT: 8083 + CONNECT_GROUP_ID: "default" + CONNECT_CONFIG_STORAGE_TOPIC: "default.config" + CONNECT_OFFSET_STORAGE_TOPIC: "default.offsets" + CONNECT_STATUS_STORAGE_TOPIC: "default.status" + CONNECT_KEY_CONVERTER: "io.confluent.connect.avro.AvroConverter" + CONNECT_VALUE_CONVERTER: "io.confluent.connect.avro.AvroConverter" + CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry-1:8081" + CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry-1:8081" + CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter" + CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter" + CONNECT_OFFSET_STORAGE_FILE_FILENAME: "/var/lib/kafka-connect-dexcom-source/logs/connect.offsets" + CONNECT_REST_ADVERTISED_HOST_NAME: "radar-dexcom-connector" + CONNECT_ZOOKEEPER_CONNECT: zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181 + CONNECTOR_PROPERTY_FILE_PREFIX: "source-dexcom" + KAFKA_HEAP_OPTS: "-Xms256m -Xmx768m" + KAFKA_BROKERS: 3 + CONNECT_LOG4J_LOGGERS: "org.reflections=ERROR" + # SENTRY_LOG_LEVEL: 'ERROR' + # SENTRY_DSN: 'https://000000000000.ingest.de.sentry.io/000000000000' + # SENTRY_ATTACHSTACKTRACE: true + # SENTRY_STACKTRACE_APP_PACKAGES: io.confluent.connect,org.radarbase.connect.rest diff --git a/docker/source-dexcom.properties.template b/docker/source-dexcom.properties.template new file mode 100644 index 00000000..c55c2037 --- /dev/null +++ b/docker/source-dexcom.properties.template @@ -0,0 +1,12 @@ +name=radar-dexcom-source +connector.class=org.radarbase.connect.rest.dexcom.DexcomSourceConnector +tasks.max=4 +rest.source.base.url=https://sandbox-api.dexcom.com +rest.source.poll.interval.ms=5000 +dexcom.api.client=? +dexcom.api.secret=? +dexcom.user.repository.class=org.radarbase.connect.rest.dexcom.user.DexcomServiceUserRepository +dexcom.user.repository.url= +dexcom.user.repository.client.id=radar_dexcom_connector +dexcom.user.repository.client.secret= +dexcom.user.repository.oauth2.token.url= diff --git a/kafka-connect-dexcom-source/Dockerfile b/kafka-connect-dexcom-source/Dockerfile new file mode 100644 index 00000000..807f89ed --- /dev/null +++ b/kafka-connect-dexcom-source/Dockerfile @@ -0,0 +1,61 @@ +# Copyright 2018 The Hyve +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +FROM --platform=$BUILDPLATFORM gradle:8.14-jdk17 AS builder + +RUN mkdir /code +WORKDIR /code + +ENV GRADLE_USER_HOME=/code/.gradlecache \ + GRADLE_OPTS="-Dorg.gradle.vfs.watch=false -Djdk.lang.Process.launchMechanism=vfork" + +COPY ./gradle/libs.versions.toml /code/gradle/ +COPY ./build.gradle.kts ./settings.gradle.kts ./gradle.properties /code/ +COPY kafka-connect-dexcom-source/build.gradle.kts /code/kafka-connect-dexcom-source/ +COPY dexcom-library/build.gradle /code/dexcom-library/ + +RUN gradle downloadDependencies copyDependencies + +COPY ./kafka-connect-dexcom-source/src/ /code/kafka-connect-dexcom-source/src +COPY ./dexcom-library/src/ /code/dexcom-library/src + +RUN gradle jar + +FROM confluentinc/cp-kafka-connect-base:7.8.7 + + +LABEL org.opencontainers.image.authors="pauline.conde@kcl.ac.uk" + +LABEL description="Kafka Dexcom REST API Source connector" + +ENV CONNECT_PLUGIN_PATH="/usr/share/java/kafka-connect/plugins" \ + WAIT_FOR_KAFKA="1" + +# To isolate the classpath from the plugin path as recommended +COPY --from=builder /code/kafka-connect-dexcom-source/build/third-party/*.jar ${CONNECT_PLUGIN_PATH}/kafka-connect-dexcom-source/ +COPY --from=builder /code/dexcom-library/build/third-party/*.jar ${CONNECT_PLUGIN_PATH}/kafka-connect-dexcom-source/ + +COPY --from=builder /code/kafka-connect-dexcom-source/build/libs/*.jar ${CONNECT_PLUGIN_PATH}/kafka-connect-dexcom-source/ +COPY --from=builder /code/dexcom-library/build/libs/*.jar ${CONNECT_PLUGIN_PATH}/kafka-connect-dexcom-source/ + +# Load topics validator +COPY --chown=appuser:appuser ./docker/ensure /etc/confluent/docker/ensure + +# Load modified launcher +COPY --chown=appuser:appuser ./docker/launch /etc/confluent/docker/launch + +# Overwrite the log4j configuration to include Sentry monitoring. +COPY ./docker/log4j.properties.template /etc/confluent/docker/log4j.properties.template +# Copy Sentry monitoring jars. +COPY --from=builder /code/kafka-connect-dexcom-source/build/third-party/sentry-* /etc/kafka-connect/jars \ No newline at end of file diff --git a/kafka-connect-dexcom-source/build.gradle.kts b/kafka-connect-dexcom-source/build.gradle.kts new file mode 100644 index 00000000..904f25eb --- /dev/null +++ b/kafka-connect-dexcom-source/build.gradle.kts @@ -0,0 +1,36 @@ +description = "Kafka connector for Dexcom API source" + +dependencies { + + /* The entries in the block below are added here to force the version of + * transitive dependencies and mitigate reported vulnerabilities + */ + implementation(libs.netty.handler.proxy) + implementation(libs.netty.handler) + + api(project(":dexcom-library")) + api(libs.kafka.connect.avro.converter) + api(libs.radar.schemas.commons) + implementation(libs.radar.commons.kotlin) + + api(libs.okhttp) + implementation(platform(libs.jackson.bom)) + implementation(libs.jackson.dataformat.yaml) + implementation(libs.jackson.datatype.jsr310) + implementation(libs.firebase.admin) + implementation(libs.kotlin.stdlib) + + implementation(libs.ktor.client.auth) + implementation(libs.ktor.client.content.negotiation) + implementation(libs.ktor.serialization.jackson) + implementation(libs.ktor.client.cio) + implementation(libs.ktor.serialization.kotlinx.json) + implementation(libs.jackson.module.kotlin) + + // Included in connector runtime + compileOnly(libs.kafka.connect.api) + compileOnly(platform(libs.jackson.bom)) + compileOnly(libs.jackson.databind) + + testImplementation(libs.kafka.connect.api) +} \ No newline at end of file diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/AbstractRestSourceConnector.java b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/AbstractRestSourceConnector.java new file mode 100644 index 00000000..e41b87a9 --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/AbstractRestSourceConnector.java @@ -0,0 +1,57 @@ +package org.radarbase.connect.rest.dexcom; + +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.kafka.connect.connector.Task; +import org.apache.kafka.connect.source.SourceConnector; +import org.radarbase.connect.rest.dexcom.util.VersionUtil; + +@SuppressWarnings("unused") +public abstract class AbstractRestSourceConnector extends SourceConnector { + protected DexcomRestSourceConnectorConfig config; + + @Override + public String version() { + return VersionUtil.getVersion(); + } + + @Override + public Class taskClass() { + return DexcomSourceTask.class; + } + + @Override + public List> taskConfigs(int maxTasks) { + return Collections.nCopies(maxTasks, new HashMap<>(config.originalsStrings())); + } + + @Override + public void start(Map props) { + config = getConfig(props); + } + + public abstract DexcomRestSourceConnectorConfig getConfig(Map conf); + + @Override + public void stop() { + } +} diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/DexcomRestSourceConnectorConfig.java b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/DexcomRestSourceConnectorConfig.java new file mode 100644 index 00000000..4a4022c1 --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/DexcomRestSourceConnectorConfig.java @@ -0,0 +1,202 @@ +package org.radarbase.connect.rest.dexcom; + +import java.lang.reflect.InvocationTargetException; +import java.net.MalformedURLException; +import java.net.URL; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import okhttp3.HttpUrl; +import org.apache.kafka.common.config.AbstractConfig; +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigDef.Importance; +import org.apache.kafka.common.config.ConfigDef.Type; +import org.apache.kafka.common.config.ConfigException; +import org.apache.kafka.connect.errors.ConnectException; +import org.radarbase.connect.rest.dexcom.user.DexcomServiceUserRepository; +import org.radarbase.connect.rest.dexcom.user.DexcomUserRepository; +import org.radarbase.dexcom.route.DexcomRoute; + +public class DexcomRestSourceConnectorConfig extends AbstractConfig { + + private static final String DEXCOM_EGV_ENABLED_CONFIG = "dexcom.egv.enabled"; + private static final String DEXCOM_CALIBRATION_ENABLED_CONFIG = "dexcom.calibration.enabled"; + private static final String DEXCOM_EVENT_ENABLED_CONFIG = "dexcom.event.enabled"; + private static final String DEXCOM_ALERT_ENABLED_CONFIG = "dexcom.alert.enabled"; + private static final String DEXCOM_DATA_RANGE_ENABLED_CONFIG = "dexcom.datarange.enabled"; + private static final String DEXCOM_DEVICE_ENABLED_CONFIG = "dexcom.device.enabled"; + static final String SOURCE_URL_CONFIG = "rest.source.base.url"; + public static final String DEXCOM_USERS_CONFIG = "dexcom.users"; + public static final String DEXCOM_API_CLIENT_CONFIG = "dexcom.api.client"; + public static final String DEXCOM_API_SECRET_CONFIG = "dexcom.api.secret"; + public static final String DEXCOM_USER_REPOSITORY_CONFIG = "dexcom.user.repository.class"; + public static final String DEXCOM_USER_REPOSITORY_URL_CONFIG = "dexcom.user.repository.url"; + public static final String DEXCOM_USER_REPOSITORY_CLIENT_ID_CONFIG = + "dexcom.user.repository.client.id"; + public static final String DEXCOM_USER_REPOSITORY_CLIENT_SECRET_CONFIG = + "dexcom.user.repository.client.secret"; + public static final String DEXCOM_USER_REPOSITORY_TOKEN_URL_CONFIG = + "dexcom.user.repository.oauth2.token.url"; + + private static final String USERS_SELF_PATH = "/v3/users/self"; + + private DexcomUserRepository userRepository; + + public DexcomRestSourceConnectorConfig(ConfigDef config, Map originals, boolean doLog) { + super(config, originals, doLog); + } + + public DexcomRestSourceConnectorConfig(Map originals, boolean doLog) { + this(conf(), originals, doLog); + } + + public DexcomRestSourceConnectorConfig(Map originals) { + this(originals, true); + } + + public static ConfigDef conf() { + return new ConfigDef() + .define( + SOURCE_URL_CONFIG, + Type.STRING, + DexcomRoute.DEFAULT_API_BASE_URL, + Importance.HIGH, + "Dexcom API base URL (host or full .../v3/users/self path).") + .define(DEXCOM_USERS_CONFIG, Type.LIST, Collections.emptyList(), Importance.HIGH, "...") + .define( + DEXCOM_USER_REPOSITORY_CONFIG, + Type.CLASS, + DexcomServiceUserRepository.class, + Importance.MEDIUM, + "...") + .define(DEXCOM_USER_REPOSITORY_URL_CONFIG, Type.STRING, "", Importance.LOW, "...") + .define(DEXCOM_USER_REPOSITORY_CLIENT_ID_CONFIG, Type.STRING, "", Importance.MEDIUM, "...") + .define( + DEXCOM_USER_REPOSITORY_CLIENT_SECRET_CONFIG, + Type.PASSWORD, + "", + Importance.MEDIUM, + "...") + .define(DEXCOM_USER_REPOSITORY_TOKEN_URL_CONFIG, Type.STRING, "", Importance.MEDIUM, "...") + .define(DEXCOM_EGV_ENABLED_CONFIG, Type.BOOLEAN, true, Importance.LOW, "...") + .define(DEXCOM_CALIBRATION_ENABLED_CONFIG, Type.BOOLEAN, true, Importance.LOW, "...") + .define(DEXCOM_EVENT_ENABLED_CONFIG, Type.BOOLEAN, true, Importance.LOW, "...") + .define(DEXCOM_ALERT_ENABLED_CONFIG, Type.BOOLEAN, true, Importance.LOW, "...") + .define(DEXCOM_DATA_RANGE_ENABLED_CONFIG, Type.BOOLEAN, true, Importance.LOW, "...") + .define(DEXCOM_DEVICE_ENABLED_CONFIG, Type.BOOLEAN, true, Importance.LOW, "..."); + } + + public List getDexcomUsers() { + return getList(DEXCOM_USERS_CONFIG); + } + + public HttpUrl getDexcomUserRepositoryUrl() { + String urlString = getString(DEXCOM_USER_REPOSITORY_URL_CONFIG).trim(); + if (urlString.isEmpty()) { + throw new ConfigException( + DEXCOM_USER_REPOSITORY_URL_CONFIG, urlString, "User repository URL is required."); + } + if (urlString.charAt(urlString.length() - 1) != '/') { + urlString += '/'; + } + HttpUrl url = HttpUrl.parse(urlString); + if (url == null) { + throw new ConfigException( + DEXCOM_USER_REPOSITORY_URL_CONFIG, + getString(DEXCOM_USER_REPOSITORY_URL_CONFIG), + "User repository URL " + urlString + " cannot be parsed as URL."); + } + return url; + } + + public String getDexcomUserRepositoryClientId() { + return getString(DEXCOM_USER_REPOSITORY_CLIENT_ID_CONFIG); + } + + public String getDexcomUserRepositoryClientSecret() { + return getPassword(DEXCOM_USER_REPOSITORY_CLIENT_SECRET_CONFIG).value(); + } + + public URL getDexcomUserRepositoryTokenUrl() { + String value = getString(DEXCOM_USER_REPOSITORY_TOKEN_URL_CONFIG); + if (value == null || value.isEmpty()) { + return null; + } + try { + return new URL(value); + } catch (MalformedURLException e) { + throw new ConfigException("Dexcom user repository token URL is invalid."); + } + } + + public boolean getDexcomEgvEnabled() { + return getBoolean(DEXCOM_EGV_ENABLED_CONFIG); + } + + public boolean getDexcomCalibrationEnabled() { + return getBoolean(DEXCOM_CALIBRATION_ENABLED_CONFIG); + } + + public boolean getDexcomEventEnabled() { + return getBoolean(DEXCOM_EVENT_ENABLED_CONFIG); + } + + public boolean getDexcomAlertEnabled() { + return getBoolean(DEXCOM_ALERT_ENABLED_CONFIG); + } + + public boolean getDexcomDataRangeEnabled() { + return getBoolean(DEXCOM_DATA_RANGE_ENABLED_CONFIG); + } + + public boolean getDexcomDeviceEnabled() { + return getBoolean(DEXCOM_DEVICE_ENABLED_CONFIG); + } + + /** + * Base URL passed into Dexcom routes. Accepts either a host ({@code + * https://sandbox-api.dexcom.com}) or a full path ending in {@code /v3/users/self}. + */ + public String getDexcomApiBaseUrl() { + String url = getString(SOURCE_URL_CONFIG).trim(); + while (url.endsWith("/")) { + url = url.substring(0, url.length() - 1); + } + if (!url.endsWith(USERS_SELF_PATH)) { + url = url + USERS_SELF_PATH; + } + return url; + } + + public DexcomUserRepository getUserRepository(DexcomUserRepository reuse) { + if (reuse != null && reuse.getClass().equals(getClass(DEXCOM_USER_REPOSITORY_CONFIG))) { + userRepository = reuse; + } else { + userRepository = createUserRepository(); + } + userRepository.initialize(this); + return userRepository; + } + + public DexcomUserRepository getUserRepository() { + if (userRepository == null) { + userRepository = createUserRepository(); + } + userRepository.initialize(this); + return userRepository; + } + + @SuppressWarnings("unchecked") + public DexcomUserRepository createUserRepository() { + try { + return ((Class) getClass(DEXCOM_USER_REPOSITORY_CONFIG)) + .getDeclaredConstructor() + .newInstance(); + } catch (IllegalAccessException + | InstantiationException + | InvocationTargetException + | NoSuchMethodException e) { + throw new ConnectException("Invalid user repository class. " + e); + } + } +} diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/DexcomSourceConnector.java b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/DexcomSourceConnector.java new file mode 100644 index 00000000..8fbba01b --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/DexcomSourceConnector.java @@ -0,0 +1,146 @@ +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.radarbase.connect.rest.dexcom; + +import static org.radarbase.connect.rest.dexcom.DexcomRestSourceConnectorConfig.DEXCOM_USERS_CONFIG; + +import java.io.IOException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; +import kotlin.sequences.Sequence; +import kotlin.sequences.SequencesKt; +import kotlin.streams.jdk8.StreamsKt; +import org.apache.kafka.common.config.ConfigDef; +import org.apache.kafka.common.config.ConfigException; +import org.radarbase.connect.rest.dexcom.user.DexcomUserRepository; +import org.radarbase.dexcom.user.User; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DexcomSourceConnector extends AbstractRestSourceConnector { + + private static final Logger logger = LoggerFactory.getLogger(DexcomSourceConnector.class); + private ScheduledExecutorService executor; + private Set configuredUsers; + private DexcomUserRepository repository; + + @Override + public void start(Map props) { + super.start(props); + executor = Executors.newSingleThreadScheduledExecutor(); + + executor.scheduleAtFixedRate(() -> { + if (repository.hasPendingUpdates()) { + try { + logger.info("Requesting latest user details..."); + repository.applyPendingUpdates(); + Set newUsers = + SequencesKt.toSet(getConfig(props, false).getUserRepository(repository).stream()); + if (configuredUsers != null && !newUsers.equals(configuredUsers)) { + logger.info("User info mismatch found. Requesting reconfiguration..."); + reconfigure(); + } + } catch (IOException e) { + logger.warn("Failed to refresh users: {}", e.toString()); + } + } else { + logger.info("No pending updates found. Not attempting to refresh users."); + } + }, 0, 5, TimeUnit.MINUTES); + } + + @Override + public void stop() { + super.stop(); + executor.shutdown(); + + configuredUsers = null; + } + + private DexcomRestSourceConnectorConfig getConfig(Map conf, boolean doLog) { + return new DexcomRestSourceConnectorConfig(conf, doLog); + } + + public DexcomRestSourceConnectorConfig getConfig(Map conf) { + DexcomRestSourceConnectorConfig connectorConfig = getConfig(conf, true); + repository = connectorConfig.getUserRepository(repository); + return connectorConfig; + } + + @Override + public ConfigDef config() { + return DexcomRestSourceConnectorConfig.conf(); + } + + @Override + public List> taskConfigs(int maxTasks) { + return configureTasks(maxTasks); + } + + private List> configureTasks(int maxTasks) { + Map baseConfig = config.originalsStrings(); + DexcomRestSourceConnectorConfig dexcomConfig = getConfig(baseConfig); + if (repository == null) { + repository = dexcomConfig.getUserRepository(null); + } + // Divide the users over tasks + try { + Sequence ids = + SequencesKt.map( + dexcomConfig.getUserRepository(repository).stream(), u -> u.getVersionedId()); + List> userTasks = + StreamsKt.asStream(ids) + // group users based on their hashCode, in principle, this allows for more efficient + // reconfigurations for a fixed number of tasks, since that allows existing tasks to + // only handle small modifications users to handle. + .collect( + Collectors.groupingBy( + u -> Math.abs(u.hashCode()) % maxTasks, Collectors.joining(","))) + .values() + .stream() + .map( + u -> { + Map config = new HashMap<>(baseConfig); + config.put(DEXCOM_USERS_CONFIG, u); + return config; + }) + .collect(Collectors.toList()); + this.configuredUsers = SequencesKt.toSet(dexcomConfig.getUserRepository().stream()); + logger.info("Received userTask Configs {}", userTasks); + return userTasks; + } catch (Exception ex) { + throw new ConfigException("Cannot read users", ex); + } + } + + public void reconfigure() { + new Thread( + () -> { + logger.info("Requesting reconfiguration"); + context.requestTaskReconfiguration(); + logger.info("Requested reconfiguration"); + }) + .start(); + } +} diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/DexcomSourceTask.java b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/DexcomSourceTask.java new file mode 100644 index 00000000..a555dec8 --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/DexcomSourceTask.java @@ -0,0 +1,225 @@ +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.radarbase.connect.rest.dexcom; + +import io.confluent.connect.avro.AvroData; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import kotlin.streams.jdk8.StreamsKt; +import okhttp3.OkHttpClient; +import okhttp3.Response; +import org.apache.kafka.connect.data.SchemaAndValue; +import org.apache.kafka.connect.source.SourceRecord; +import org.apache.kafka.connect.source.SourceTask; +import org.apache.kafka.connect.storage.OffsetStorageReader; +import org.radarbase.connect.rest.dexcom.offset.KafkaOffsetManager; +import org.radarbase.connect.rest.dexcom.user.DexcomUserRepository; +import org.radarbase.connect.rest.dexcom.util.VersionUtil; +import org.radarbase.dexcom.converter.TopicData; +import org.radarbase.dexcom.request.DexcomErrorBase; +import org.radarbase.dexcom.request.DexcomRequestGenerator; +import org.radarbase.dexcom.request.DexcomResult; +import org.radarbase.dexcom.request.RestRequest; +import org.radarbase.dexcom.route.DexcomAlertsRoute; +import org.radarbase.dexcom.route.DexcomCalibrationsRoute; +import org.radarbase.dexcom.route.DexcomDataRangeRoute; +import org.radarbase.dexcom.route.DexcomDevicesRoute; +import org.radarbase.dexcom.route.DexcomEGVRoute; +import org.radarbase.dexcom.route.DexcomEventsRoute; +import org.radarbase.dexcom.route.Route; +import org.radarbase.dexcom.user.User; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class DexcomSourceTask extends SourceTask { + private static final Logger logger = LoggerFactory.getLogger(DexcomSourceTask.class); + + private OkHttpClient baseClient; + private DexcomUserRepository userRepository; + private List routes; + private DexcomRequestGenerator dexcomRequestGenerator; + private final AvroData avroData = new AvroData(20); + private KafkaOffsetManager offsetManager; + String TIMESTAMP_OFFSET_KEY = "timestamp"; + long TIMEOUT = 60000L; + private int routeStartIndex = 0; + + public void initialize( + DexcomRestSourceConnectorConfig config, OffsetStorageReader offsetStorageReader) { + this.baseClient = new OkHttpClient(); + this.userRepository = config.getUserRepository(); + this.offsetManager = new KafkaOffsetManager(offsetStorageReader); + this.routes = this.getRoutes(config); + this.dexcomRequestGenerator = + new DexcomRequestGenerator(this.userRepository, this.offsetManager, this.routes); + this.offsetManager.initialize(getPartitions()); + } + + private List getRoutes(DexcomRestSourceConnectorConfig config) { + List routes = new ArrayList<>(); + String apiBaseUrl = config.getDexcomApiBaseUrl(); + + if (config.getDexcomEgvEnabled()) { + routes.add(new DexcomEGVRoute(userRepository, apiBaseUrl)); + } + if (config.getDexcomCalibrationEnabled()) { + routes.add(new DexcomCalibrationsRoute(userRepository, apiBaseUrl)); + } + if (config.getDexcomEventEnabled()) { + routes.add(new DexcomEventsRoute(userRepository, apiBaseUrl)); + } + if (config.getDexcomAlertEnabled()) { + routes.add(new DexcomAlertsRoute(userRepository, apiBaseUrl)); + } + if (config.getDexcomDataRangeEnabled()) { + routes.add(new DexcomDataRangeRoute(userRepository, apiBaseUrl)); + } + if (config.getDexcomDeviceEnabled()) { + routes.add(new DexcomDevicesRoute(userRepository, apiBaseUrl)); + } + return routes; + } + + public List> getPartitions() { + try { + return StreamsKt.asStream(userRepository.stream()) + .flatMap(u -> this.routes.stream().map(r -> getPartition(r.toString(), u))) + .collect(Collectors.toList()); + } catch (Exception e) { + logger.warn("Failed to initialize user partitions.."); + return Collections.emptyList(); + } + } + + public Map getPartition(String route, User user) { + Map partition = new HashMap<>(4); + partition.put("user", user.getVersionedId()); + partition.put("route", route); + return partition; + } + + public Stream requests() { + if (this.routes == null || this.routes.isEmpty()) { + return Stream.empty(); + } + + List rotatedRoutes = getRotatedRoutes(); + return rotatedRoutes.stream() + .flatMap((Route r) -> StreamsKt.asStream(dexcomRequestGenerator.requests(r, 100))); + } + + private List getRotatedRoutes() { + List rotatedRoutes = new ArrayList<>(this.routes); + Collections.rotate(rotatedRoutes, routeStartIndex % this.routes.size()); + routeStartIndex = (routeStartIndex + 1) % this.routes.size(); + return rotatedRoutes; + } + + @SuppressWarnings("unchecked") + public Stream handleRequest(RestRequest req) throws IOException { + try (Response response = baseClient.newCall(req.getRequest()).execute()) { + DexcomResult result = this.dexcomRequestGenerator.handleResponse(req, response); + if (result instanceof DexcomResult.Success) { + DexcomResult.Success> success = + (DexcomResult.Success>) result; + return success.getValue().stream() + .map( + r -> { + SchemaAndValue avro = + avroData.toConnectData(r.getValue().getSchema(), r.getValue()); + SchemaAndValue key = avroData.toConnectData(r.getKey().getSchema(), r.getKey()); + Map partition = + getPartition(req.getRoute().toString(), req.getUser()); + Map offset = + Collections.singletonMap(TIMESTAMP_OFFSET_KEY, r.getOffset()); + + return new SourceRecord( + partition, + offset, + r.getTopic(), + key.schema(), + key.value(), + avro.schema(), + avro.value()); + }); + } else { + DexcomErrorBase e = (DexcomErrorBase) ((DexcomResult.Error) result).getError(); + logger.warn( + "Failed to make request: {} {} {}", + e.getMessage(), + e.getCause() != null ? e.getCause().toString() : "null", + e.getCode()); + return Stream.empty(); + } + } + } + + @Override + public void start(Map map) { + DexcomRestSourceConnectorConfig connectorConfig = new DexcomRestSourceConnectorConfig(map); + this.initialize(connectorConfig, context.offsetStorageReader()); + } + + @Override + public List poll() throws InterruptedException { + long requestsGenerated = 0; + List sourceRecords = Collections.emptyList(); + + do { + Thread.sleep(TIMEOUT); + + Iterator requestIterator = this.requests().iterator(); + + while (sourceRecords.isEmpty() && requestIterator.hasNext()) { + RestRequest request = requestIterator.next(); + + logger.info( + "Requesting for user {}, url: {}", + request.getUser().getUserId(), + request.getRequest().url()); + requestsGenerated++; + + try { + sourceRecords = this.handleRequest(request).collect(Collectors.toList()); + } catch (IOException ex) { + logger.warn("Failed to make request: {}", ex.toString()); + } + } + } while (sourceRecords.isEmpty()); + + logger.info("Processed {} records from {} URLs", sourceRecords.size(), requestsGenerated); + + return sourceRecords; + } + + @Override + public void stop() { + logger.debug("Stopping source task"); + } + + @Override + public String version() { + return VersionUtil.getVersion(); + } +} diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/offset/KafkaOffsetManager.java b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/offset/KafkaOffsetManager.java new file mode 100644 index 00000000..55e44629 --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/offset/KafkaOffsetManager.java @@ -0,0 +1,79 @@ +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.radarbase.connect.rest.dexcom.offset; + +import static java.time.temporal.ChronoUnit.NANOS; + +import java.time.Duration; +import java.time.Instant; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import org.apache.kafka.connect.storage.OffsetStorageReader; +import org.radarbase.dexcom.request.DexcomOffsetManager; +import org.radarbase.dexcom.request.Offset; +import org.radarbase.dexcom.route.Route; +import org.radarbase.dexcom.user.User; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class KafkaOffsetManager implements DexcomOffsetManager { + private final OffsetStorageReader offsetStorageReader; + private Map offsets = new HashMap<>(); + private static final Logger logger = LoggerFactory.getLogger(KafkaOffsetManager.class); + + String TIMESTAMP_OFFSET_KEY = "timestamp"; + protected static final Duration ONE_NANO = NANOS.getDuration(); + + public KafkaOffsetManager(OffsetStorageReader offsetStorageReader) { + this.offsetStorageReader = offsetStorageReader; + } + + public void initialize(List> partitions) { + if (this.offsetStorageReader != null) { + this.offsets = + this.offsetStorageReader.offsets(partitions).entrySet().stream() + .filter(e -> e.getValue() != null && e.getValue().containsKey(TIMESTAMP_OFFSET_KEY)) + .collect( + Collectors.toMap( + e -> (String) e.getKey().get("user") + "-" + e.getKey().get("route"), + e -> + Instant.ofEpochSecond( + ((Number) e.getValue().get(TIMESTAMP_OFFSET_KEY)).longValue()))); + } else { + logger.warn("Offset storage reader is null, will resume from an empty state."); + this.offsets = new HashMap<>(); + } + } + + @Override + public Offset getOffset(Route route, User user) { + Instant offset = + offsets.getOrDefault(getOffsetKey(route, user), user.getStartDate().minus(ONE_NANO)); + return new Offset(user, route, offset); + } + + @Override + public void updateOffsets(Route route, User user, Instant offset) { + offsets.put(getOffsetKey(route, user), offset); + } + + private String getOffsetKey(Route route, User user) { + return user.getVersionedId() + "-" + route.toString(); + } +} diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/DexcomServiceUserRepository.kt b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/DexcomServiceUserRepository.kt new file mode 100644 index 00000000..ae7f10be --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/DexcomServiceUserRepository.kt @@ -0,0 +1,307 @@ +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.radarbase.connect.rest.dexcom.user + +import com.fasterxml.jackson.databind.ObjectMapper +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule +import com.fasterxml.jackson.module.kotlin.readValue +import com.fasterxml.jackson.module.kotlin.registerKotlinModule +import io.ktor.client.HttpClient +import io.ktor.client.engine.cio.CIO +import io.ktor.client.plugins.HttpTimeout +import io.ktor.client.plugins.auth.Auth +import io.ktor.client.plugins.auth.providers.BasicAuthCredentials +import io.ktor.client.plugins.auth.providers.basic +import io.ktor.client.plugins.contentnegotiation.ContentNegotiation +import io.ktor.client.plugins.defaultRequest +import io.ktor.client.request.HttpRequestBuilder +import io.ktor.client.request.request +import io.ktor.client.request.setBody +import io.ktor.client.request.url +import io.ktor.client.statement.bodyAsText +import io.ktor.client.statement.request +import io.ktor.http.ContentType +import io.ktor.http.HttpMethod +import io.ktor.http.HttpStatusCode +import io.ktor.http.URLBuilder +import io.ktor.http.Url +import io.ktor.http.contentLength +import io.ktor.http.contentType +import io.ktor.http.isSuccess +import io.ktor.http.takeFrom +import io.ktor.serialization.jackson.jackson +import io.ktor.serialization.kotlinx.json.json +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import kotlinx.serialization.json.Json +import org.radarbase.connect.rest.dexcom.DexcomRestSourceConnectorConfig +import org.radarbase.dexcom.user.DexcomUser +import org.radarbase.dexcom.user.User +import org.radarbase.kotlin.coroutines.CacheConfig +import org.radarbase.kotlin.coroutines.CachedSet +import org.radarbase.kotlin.coroutines.CachedValue +import org.radarbase.ktor.auth.ClientCredentialsConfig +import org.radarbase.ktor.auth.clientCredentials +import org.slf4j.LoggerFactory +import java.io.IOException +import java.util.concurrent.ConcurrentHashMap +import kotlin.streams.asSequence +import kotlin.time.Duration.Companion.days +import kotlin.time.Duration.Companion.hours +import kotlin.time.Duration.Companion.minutes +import kotlin.time.Duration.Companion.seconds + +@Suppress("unused") +class DexcomServiceUserRepository : DexcomUserRepository() { + private lateinit var userCache: CachedSet + private lateinit var client: HttpClient + private val credentialCaches = ConcurrentHashMap>() + private val credentialCacheConfig = + CacheConfig(refreshDuration = 1.days, retryDuration = 1.minutes) + private val mapper = ObjectMapper().registerKotlinModule().registerModule(JavaTimeModule()) + + @Throws(IOException::class) + override fun get(key: String): User = + runBlocking(Dispatchers.Default) { + makeRequest { url("users/$key") } + } + + override fun initialize(config: DexcomRestSourceConnectorConfig) { + val containedUsers = config.dexcomUsers.toHashSet() + + client = + createClient( + baseUrl = URLBuilder(config.dexcomUserRepositoryUrl.toString()).build(), + tokenUrl = config.dexcomUserRepositoryTokenUrl?.let { + URLBuilder(it.toString()).build() + }, + clientId = config.dexcomUserRepositoryClientId, + clientSecret = config.dexcomUserRepositoryClientSecret, + scope = "SUBJECT.READ MEASUREMENT.CREATE", + audience = "res_restAuthorizer", + ) + + userCache = + CachedSet( + CacheConfig(refreshDuration = 1.hours, retryDuration = 1.minutes), + ) { + makeRequest { url("users?source-type=Dexcom") } + .users + .toHashSet() + .filterTo(HashSet()) { u -> + u.isComplete() && + (containedUsers.isEmpty() || u.versionedId in containedUsers) + } + } + } + + private fun createClient( + baseUrl: Url, + tokenUrl: Url?, + clientId: String?, + clientSecret: String?, + scope: String?, + audience: String?, + ): HttpClient = + HttpClient(CIO) { + if (tokenUrl != null) { + install(Auth) { + clientCredentials( + ClientCredentialsConfig( + tokenUrl.toString(), + clientId, + clientSecret, + scope, + audience, + ).copyWithEnv("MANAGEMENT_PORTAL"), + baseUrl.host, + ) + } + install(ContentNegotiation) { + json( + Json { + ignoreUnknownKeys = true + }, + ) + } + } else if (clientId != null && clientSecret != null) { + install(Auth) { + basic { + credentials { + BasicAuthCredentials(username = clientId, password = clientSecret) + } + realm = "Access to the '/' path" + sendWithoutRequest { + it.url.host == baseUrl.host + } + } + } + } + + defaultRequest { + url.takeFrom(baseUrl) + } + + install(ContentNegotiation) { + jackson { + registerModule(JavaTimeModule()) + } + } + + install(HttpTimeout) { + connectTimeoutMillis = 60.seconds.inWholeMilliseconds + requestTimeoutMillis = 90.seconds.inWholeMilliseconds + } + } + + override fun stream(): Sequence = + runBlocking(Dispatchers.Default) { + val valueInCache = + userCache.getFromCache() + .takeIf { it is CachedValue.CacheValue } + ?.getOrThrow() + + (valueInCache ?: userCache.get()) + .stream() + .filter { it.isComplete() } + .asSequence() + } + + @Throws(IOException::class, UserNotAuthorizedException::class) + override fun getAccessToken(user: User): String { + if (!user.isAuthorized) { + throw UserNotAuthorizedException("User is not authorized") + } + return runBlocking(Dispatchers.Default) { + credentialCache(user) + .get { !it.isAccessTokenExpired } + .value + .accessToken + } + } + + @Throws(IOException::class, UserNotAuthorizedException::class) + override fun refreshAccessToken(user: User): String { + if (!user.isAuthorized) { + throw UserNotAuthorizedException("User is not authorized") + } + return runBlocking(Dispatchers.Default) { + val token = + requestAccessToken(user) { + url("users/${user.id}/token") + method = HttpMethod.Post + setBody("{}") + contentType(ContentType.Application.Json) + } + credentialCache(user).set(token) + token.accessToken + } + } + + private suspend fun credentialCache(user: User): CachedValue = + credentialCaches.computeIfAbsent(user.id) { + CachedValue(credentialCacheConfig) { + requestAccessToken(user) { url("users/${user.id}/token") } + } + } + + @Throws(UserNotAuthorizedException::class, IOException::class) + private suspend fun requestAccessToken( + user: User, + builder: HttpRequestBuilder.() -> Unit, + ): OAuth2UserCredentials = + try { + makeRequest(builder) + } catch (ex: HttpResponseException) { + if (ex.statusCode == 407) { + credentialCaches -= user.id + throw UserNotAuthorizedException(ex.message) + } + throw ex + } + + override fun hasPendingUpdates(): Boolean = + runBlocking(Dispatchers.Default) { + userCache.isStale(1.hours) + } + + @Throws(IOException::class) + override fun applyPendingUpdates() { + logger.info("Requesting user information from webservice") + + runBlocking(Dispatchers.Default) { + userCache.get() + } + } + + private suspend inline fun makeRequest( + crossinline builder: HttpRequestBuilder.() -> Unit, + ): T = + withContext(Dispatchers.IO) { + val requestBuilder = HttpRequestBuilder() + builder(requestBuilder) + logger.info("Making HTTP request: ${requestBuilder.method} ${requestBuilder.url}") + + val response = client.request(builder) + logger.info("Response status: ${response.status}") + val contentLength = response.contentLength() + val transferEncoding = response.headers["Transfer-Encoding"] + val hasBody = (contentLength != null && contentLength > 0) || + (transferEncoding != null && transferEncoding.contains("chunked")) + val responseBody = try { + response.bodyAsText() + } catch (e: Exception) { + "Error reading body: ${e.message}" + } + + if (response.status == HttpStatusCode.NotFound) { + logger.error("HTTP 404 Not Found: ${response.request.url}") + throw NoSuchElementException("URL " + response.request.url + " does not exist") + } else if (!response.status.isSuccess()) { + val message = "HTTP ${response.status.value} error: $responseBody" + logger.error(message) + throw HttpResponseException(message, response.status.value) + } else if (!hasBody) { + logger.warn( + "HTTP ${response.status.value} OK but no body content. Returning empty result.", + ) + @Suppress("UNCHECKED_CAST") + return@withContext when (T::class) { + String::class -> "" as T + List::class -> emptyList() as T + else -> mapper.readValue("{}") + } + } + + try { + val result = mapper.readValue(responseBody) + logger.info("Successfully parsed response as ${T::class.simpleName}") + result + } catch (e: Exception) { + logger.error( + "Failed to parse response body as ${T::class.simpleName}: ${e.message}", + ) + logger.error("Response body that failed to parse: $responseBody") + throw e + } + } + + companion object { + private val logger = LoggerFactory.getLogger(DexcomServiceUserRepository::class.java) + } +} diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/DexcomUserRepository.kt b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/DexcomUserRepository.kt new file mode 100644 index 00000000..5ad2208d --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/DexcomUserRepository.kt @@ -0,0 +1,41 @@ +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ +package org.radarbase.connect.rest.dexcom.user + +import org.radarbase.connect.rest.dexcom.DexcomRestSourceConnectorConfig +import org.radarbase.dexcom.user.User +import org.radarbase.dexcom.user.UserNotAuthorizedException +import org.radarbase.dexcom.user.UserRepository +import org.slf4j.LoggerFactory +import java.io.IOException + +@Suppress("unused") +abstract class DexcomUserRepository : UserRepository { + abstract fun initialize(config: DexcomRestSourceConnectorConfig) + + @Throws(IOException::class, UserNotAuthorizedException::class) + abstract fun refreshAccessToken(user: User): String + + @Throws(IOException::class) + abstract fun applyPendingUpdates() + + abstract fun hasPendingUpdates(): Boolean + + companion object { + private val logger = LoggerFactory.getLogger(DexcomUserRepository::class.java) + } +} diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/DexcomUsers.java b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/DexcomUsers.java new file mode 100644 index 00000000..7eb69f0d --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/DexcomUsers.java @@ -0,0 +1,39 @@ +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.radarbase.connect.rest.dexcom.user; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import java.util.ArrayList; +import java.util.List; +import org.radarbase.dexcom.user.DexcomUser; + +@JsonIgnoreProperties(ignoreUnknown = true) +public class DexcomUsers { + private final List users; + + @JsonCreator + public DexcomUsers(@JsonProperty("users") List users) { + this.users = new ArrayList<>(users); + } + + public List getUsers() { + return users; + } +} diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/HttpResponseException.java b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/HttpResponseException.java new file mode 100644 index 00000000..10eb4b38 --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/HttpResponseException.java @@ -0,0 +1,33 @@ +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.radarbase.connect.rest.dexcom.user; + +import java.io.IOException; + +public class HttpResponseException extends IOException { + private final int statusCode; + + public HttpResponseException(String message, int statusCode) { + super(message); + this.statusCode = statusCode; + } + + public int getStatusCode() { + return statusCode; + } +} diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/OAuth2UserCredentials.java b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/OAuth2UserCredentials.java new file mode 100644 index 00000000..6dd3f0ce --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/OAuth2UserCredentials.java @@ -0,0 +1,79 @@ +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.radarbase.connect.rest.dexcom.user; + +import com.fasterxml.jackson.annotation.JsonIgnore; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; +import com.fasterxml.jackson.annotation.JsonProperty; +import com.fasterxml.jackson.annotation.JsonSetter; +import java.time.Duration; +import java.time.Instant; + +@JsonIgnoreProperties(ignoreUnknown = true) +public class OAuth2UserCredentials { + private static final Duration DEFAULT_EXPIRY = Duration.ofHours(1); + private static final Duration EXPIRY_TIME_MARGIN = Duration.ofMinutes(5); + + @JsonProperty + private String accessToken; + @JsonProperty + private String refreshToken; + @JsonProperty + private Instant expiresAt; + + public OAuth2UserCredentials() { + } + + public OAuth2UserCredentials(String refreshToken, String accessToken, Long expiresIn) { + this.refreshToken = refreshToken; + this.accessToken = accessToken; + this.expiresAt = getExpiresAt(expiresIn != null && expiresIn > 0L + ? Duration.ofSeconds(expiresIn) : DEFAULT_EXPIRY); + } + + public String getAccessToken() { + return accessToken; + } + + @JsonSetter + public void setAccessToken(String accessToken) { + this.accessToken = accessToken; + if (expiresAt == null) { + expiresAt = getExpiresAt(DEFAULT_EXPIRY); + } + } + + public boolean hasRefreshToken() { + return refreshToken != null && !refreshToken.isEmpty(); + } + + public String getRefreshToken() { + return refreshToken; + } + + protected static Instant getExpiresAt(Duration expiresIn) { + return Instant.now() + .plus(expiresIn) + .minus(EXPIRY_TIME_MARGIN); + } + + @JsonIgnore + public boolean isAccessTokenExpired() { + return expiresAt == null || Instant.now().isAfter(expiresAt); + } +} diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/UserNotAuthorizedException.java b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/UserNotAuthorizedException.java new file mode 100644 index 00000000..752490f3 --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/user/UserNotAuthorizedException.java @@ -0,0 +1,24 @@ +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.radarbase.connect.rest.dexcom.user; + +public class UserNotAuthorizedException extends RuntimeException { + public UserNotAuthorizedException(String message) { + super(message); + } +} \ No newline at end of file diff --git a/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/util/VersionUtil.java b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/util/VersionUtil.java new file mode 100644 index 00000000..d86a104c --- /dev/null +++ b/kafka-connect-dexcom-source/src/main/java/org/radarbase/connect/rest/dexcom/util/VersionUtil.java @@ -0,0 +1,32 @@ +/* + * Copyright 2018 The Hyve + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.radarbase.connect.rest.dexcom.util; + +public final class VersionUtil { + private VersionUtil() { + // utility class + } + + public static String getVersion() { + try { + return VersionUtil.class.getPackage().getImplementationVersion(); + } catch (Exception ex) { + return "0.0.0.0"; + } + } +} \ No newline at end of file diff --git a/settings.gradle.kts b/settings.gradle.kts index 64f23944..2a682abe 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -3,6 +3,8 @@ include(":kafka-connect-fitbit-source") include(":kafka-connect-rest-source") include(":kafka-connect-oura-source") include(":oura-library") +include(":dexcom-library") +include(":kafka-connect-dexcom-source") pluginManagement { repositories {