update for memory leak
- fix memory leak in MeasurementRepository - add logging in MeasurementService - update catalog version
This commit is contained in:
@@ -15,6 +15,8 @@ class MeasurementRepository(
|
||||
private val properties: InfluxDbProperties,
|
||||
) {
|
||||
|
||||
private val writeApi = influxDBClient.makeWriteApi()
|
||||
|
||||
companion object {
|
||||
const val BY_NODE_QUERY: String = """
|
||||
from(bucket: "%s")
|
||||
@@ -29,9 +31,7 @@ class MeasurementRepository(
|
||||
|
||||
suspend fun save(
|
||||
point: Point,
|
||||
) = withContext(Dispatchers.IO) {
|
||||
influxDBClient.makeWriteApi().writePoint(point)
|
||||
}
|
||||
) = withContext(Dispatchers.IO) { writeApi.writePoint(point) }
|
||||
|
||||
suspend fun getByNode(
|
||||
client: UUID,
|
||||
|
||||
@@ -4,10 +4,13 @@ import com.influxdb.client.write.Point
|
||||
import java.util.UUID
|
||||
import ltd.hlaeja.library.deviceData.MeasurementData
|
||||
import ltd.hlaeja.repository.MeasurementRepository
|
||||
import mu.KotlinLogging
|
||||
import org.springframework.http.HttpStatus.NOT_FOUND
|
||||
import org.springframework.stereotype.Service
|
||||
import org.springframework.web.server.ResponseStatusException
|
||||
|
||||
private val log = KotlinLogging.logger {}
|
||||
|
||||
@Service
|
||||
class MeasurementService(
|
||||
private val repository: MeasurementRepository,
|
||||
@@ -19,7 +22,10 @@ class MeasurementService(
|
||||
): MeasurementData.Response {
|
||||
val result = repository.getByNode(client, node)
|
||||
if (result.isEmpty()) {
|
||||
throw ResponseStatusException(NOT_FOUND, "No data for client: $client, device: $node")
|
||||
"No data for client: $client, node: $node".also {
|
||||
log.warn { it }
|
||||
throw ResponseStatusException(NOT_FOUND, it)
|
||||
}
|
||||
}
|
||||
val latestData = mutableMapOf<String, Number>()
|
||||
result.forEach { table ->
|
||||
@@ -27,6 +33,7 @@ class MeasurementService(
|
||||
latestData[record.getValueByKey("_field") as String] = record.value as Number
|
||||
}
|
||||
}
|
||||
log.info { "Load data for client $client" }
|
||||
return MeasurementData.Response(latestData)
|
||||
}
|
||||
|
||||
@@ -38,7 +45,10 @@ class MeasurementService(
|
||||
addTags(request.tags, point)
|
||||
addFields(request.fields, point)
|
||||
}
|
||||
.let { point -> repository.save(point) }
|
||||
.let { point ->
|
||||
repository.save(point)
|
||||
log.debug { "Save data for client $client" }
|
||||
}
|
||||
|
||||
private suspend fun addFields(
|
||||
measurements: Map<String, Number>,
|
||||
|
||||
@@ -19,16 +19,15 @@ import org.junit.jupiter.api.Assertions.assertEquals
|
||||
import org.junit.jupiter.api.BeforeEach
|
||||
|
||||
class MeasurementRepositoryTest {
|
||||
|
||||
private val client: InfluxDBClient = mockk()
|
||||
private val properties: InfluxDbProperties = mockk()
|
||||
private val writeApi: WriteApi = mockk()
|
||||
private val queryApi: QueryApi = mockk()
|
||||
|
||||
private lateinit var repository: MeasurementRepository
|
||||
|
||||
@BeforeEach
|
||||
fun setUp() {
|
||||
every { client.makeWriteApi() } returns writeApi
|
||||
repository = MeasurementRepository(client, properties)
|
||||
}
|
||||
|
||||
@@ -36,8 +35,6 @@ class MeasurementRepositoryTest {
|
||||
fun `save point to influxdb`() = runTest {
|
||||
// given
|
||||
val point = Point.measurement("test").addField("value", 12.3)
|
||||
|
||||
every { client.makeWriteApi() } returns writeApi
|
||||
every { writeApi.writePoint(any()) } just Runs
|
||||
|
||||
// when
|
||||
|
||||
@@ -65,7 +65,7 @@ class MeasurementServiceTest {
|
||||
// then
|
||||
assertEquals(NOT_FOUND, exception.statusCode)
|
||||
assertEquals(
|
||||
"No data for client: 00000000-0000-0000-0000-000000000000, device: 00000000-0000-0000-0000-000000000000",
|
||||
"No data for client: 00000000-0000-0000-0000-000000000000, node: 00000000-0000-0000-0000-000000000000",
|
||||
exception.reason,
|
||||
)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user