initial commit2
This commit is contained in:
@@ -0,0 +1,57 @@
|
||||
package database
|
||||
|
||||
import io.github.oshai.kotlinlogging.KotlinLogging
|
||||
import org.postgresql.PGConnection
|
||||
import javax.sql.DataSource
|
||||
|
||||
private val logger = KotlinLogging.logger {}
|
||||
|
||||
class CacheInvalidator(
|
||||
private val dataSource: DataSource,
|
||||
private val onInvalidated: () -> Unit
|
||||
) {
|
||||
@Volatile
|
||||
private var running = false
|
||||
private var thread: Thread? = null
|
||||
|
||||
fun start() {
|
||||
running = true
|
||||
thread = Thread {
|
||||
while (running) {
|
||||
try {
|
||||
dataSource.connection.use { conn ->
|
||||
conn.autoCommit = true
|
||||
val pgConn = conn.unwrap(PGConnection::class.java)
|
||||
|
||||
conn.createStatement().use { stmt ->
|
||||
stmt.execute("LISTEN cache_changed")
|
||||
}
|
||||
|
||||
logger.info { "Listening for cache changes via PG NOTIFY..." }
|
||||
|
||||
while (running && !conn.isClosed) {
|
||||
val notifications = pgConn.notifications
|
||||
if (notifications != null && notifications.isNotEmpty()) {
|
||||
logger.debug { "Cache invalidation received" }
|
||||
onInvalidated()
|
||||
}
|
||||
Thread.sleep(500)
|
||||
}
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
if (!running) break
|
||||
logger.warn(e) { "Cache listener error, reconnecting in 5s..." }
|
||||
Thread.sleep(5000)
|
||||
}
|
||||
}
|
||||
}.also { it.isDaemon = true }.apply {
|
||||
name = "pg-notify-listener"
|
||||
start()
|
||||
}
|
||||
}
|
||||
|
||||
fun stop() {
|
||||
running = false
|
||||
thread?.interrupt()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user