libeufin

Integration and sandbox testing for FinTech APIs and data formats
Log | Files | Refs | Submodules | README | LICENSE

notifications.kt (3479B)

      1 /*
      2  * This file is part of LibEuFin.
      3  * Copyright (C) 2024 Taler Systems S.A.
      4  *
      5  * LibEuFin is free software; you can redistribute it and/or modify
      6  * it under the terms of the GNU Affero General Public License as
      7  * published by the Free Software Foundation; either version 3, or
      8  * (at your option) any later version.
      9  *
     10  * LibEuFin is distributed in the hope that it will be useful, but
     11  * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
     12  * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Affero General
     13  * Public License for more details.
     14  *
     15  * You should have received a copy of the GNU Affero General Public
     16  * License along with LibEuFin; see the file COPYING.  If not, see
     17  * <http://www.gnu.org/licenses/>
     18  */
     19 
     20 package tech.libeufin.common.db
     21 
     22 import kotlinx.coroutines.*
     23 import kotlinx.coroutines.flow.Flow
     24 import kotlinx.coroutines.flow.MutableSharedFlow
     25 import org.postgresql.ds.PGSimpleDataSource
     26 import org.slf4j.Logger
     27 import tech.libeufin.common.ExpoBackoffDecorr
     28 import tech.libeufin.common.fmtLog
     29 import java.io.Closeable
     30 import java.util.concurrent.ConcurrentHashMap
     31 
     32 // SharedFlow that are manually counted for manual garbage collection
     33 class CountedSharedFlow<T> {
     34     val flow: MutableSharedFlow<T> = MutableSharedFlow()
     35     var count: Int = 0
     36 }
     37 
     38 fun watchNotifications(
     39     pgSource: PGSimpleDataSource, 
     40     schema: String,
     41     logger: Logger,
     42     listeners: Map<String, (suspend (String) -> Unit)>
     43 ): Closeable {
     44     val backoff = ExpoBackoffDecorr()
     45     // JDBC notification reads block, so keep them off the caller's dispatcher.
     46     val job = CoroutineScope(Dispatchers.IO).launch {
     47         while (isActive) {
     48             try {
     49                 pgSource.pgConnection(schema).use { conn ->
     50 
     51                     // Listen to all notifications channels
     52                     for (channel in listeners.keys) {
     53                         conn.execSQLUpdate("LISTEN $channel")
     54                     }
     55 
     56                     backoff.reset()
     57 
     58                     while (isActive) {
     59                         // Bound the blocking read so close() can cancel an idle watcher.
     60                         conn.getNotifications(1000)
     61                             .forEach {
     62                             // Dispatch
     63                             try {
     64                                 listeners[it.name]!!(it.parameter)
     65                             } catch (e: Exception) {
     66                                 ensureActive()
     67                                 throw Exception("channel ${it.name} with input '${it.parameter}'", e)
     68                             }
     69                         }
     70                     }
     71                 }
     72             } catch (e: Exception) {
     73                 ensureActive()
     74                 e.fmtLog(logger)
     75                 delay(backoff.next())
     76             }
     77         }
     78     }
     79     return Closeable { runBlocking { job.cancelAndJoin() } }
     80 }
     81 
     82 /** Listen to flow from [map] for [key] using [lambda]*/
     83 suspend fun <R, K, V> listen(map: ConcurrentHashMap<K, CountedSharedFlow<V>>, key: K, lambda: suspend (Flow<V>) -> R): R {
     84     // Register listener, create a new flow if missing
     85     val flow = map.compute(key) { _, v ->
     86         val tmp = v ?: CountedSharedFlow()
     87         tmp.count++
     88         tmp
     89     }!!.flow
     90 
     91     try {
     92         return lambda(flow)
     93     } finally {
     94         // Unregister listener, removing unused flow
     95         map.compute(key) { _, v ->
     96             v!!
     97             v.count--
     98             if (v.count > 0) v else null
     99         }
    100     }
    101 }