Jonathan Bernard
3e19b3628d
The previous implementation expected to work with a context object that
contained a field `conn` that represented a valid database connection.
However, this required the caller to manage the connection's lifecycle
(close, renew, etc.). Now we expect to receive a context object that
provides a `withConn` procedure or template that accepts a statement
block and provides a `conn` variable to that code block. For example:
createRecord(db: DbContext): Record =
# withConn must be defined for DbContext
db.withConn:
# conn must be injected into the statement block context
conn.exec(sql("INSERT INTO..."))
In addition, this change provides a connection pooling mechanism
(`DbConnPool`) as a default implementation for this hypothetical
DbContext. There is also a new function `initPool` that will create an
DbConnPool instance.
Callers of this library can modify their DbContext objects to extend
DbConnPool or simply be a type alias of DbConnPool.
98 lines
2.7 KiB
Nim
98 lines
2.7 KiB
Nim
import std/db_postgres, std/sequtils, std/strutils, std/sugar
|
|
|
|
import namespaced_logging
|
|
|
|
|
|
type
|
|
DbConnPoolConfig* = object
|
|
connect*: () -> DbConn
|
|
poolSize*: int
|
|
hardCap*: bool
|
|
healthCheckQuery*: string
|
|
|
|
PooledDbConn = ref object
|
|
conn: DbConn
|
|
id: int
|
|
free: bool
|
|
|
|
DbConnPool* = ref object
|
|
conns: seq[PooledDbConn]
|
|
cfg: DbConnPoolConfig
|
|
lastId: int
|
|
|
|
var logNs {.threadvar.}: LoggingNamespace
|
|
|
|
template log(): untyped =
|
|
if logNs.isNil: logNs = initLoggingNamespace(name = "fiber_orm/pool", level = lvlNotice)
|
|
logNs
|
|
|
|
proc initDbConnPool*(cfg: DbConnPoolConfig): DbConnPool =
|
|
log().debug("Initializing new pool (size: " & $cfg.poolSize)
|
|
result = DbConnPool(
|
|
conns: @[],
|
|
cfg: cfg)
|
|
|
|
proc newConn(pool: DbConnPool): PooledDbConn =
|
|
log().debug("Creating a new connection to add to the pool.")
|
|
pool.lastId += 1
|
|
let conn = pool.cfg.connect()
|
|
result = PooledDbConn(
|
|
conn: conn,
|
|
id: pool.lastId,
|
|
free: true)
|
|
pool.conns.add(result)
|
|
|
|
proc maintain(pool: DbConnPool): void =
|
|
log().debug("Maintaining pool. $# connections." % [$pool.conns.len])
|
|
pool.conns.keepIf(proc (pc: PooledDbConn): bool =
|
|
if not pc.free: return true
|
|
|
|
try:
|
|
discard getRow(pc.conn, sql(pool.cfg.healthCheckQuery), [])
|
|
return true
|
|
except:
|
|
try: pc.conn.close() # try to close the connection
|
|
except: discard ""
|
|
return false
|
|
)
|
|
log().debug(
|
|
"Pruned dead connections. $# connections remaining." %
|
|
[$pool.conns.len])
|
|
|
|
let freeConns = pool.conns.filterIt(it.free)
|
|
if pool.conns.len > pool.cfg.poolSize and freeConns.len > 0:
|
|
let numToCull = min(freeConns.len, pool.conns.len - pool.cfg.poolSize)
|
|
let toCull = freeConns[0..numToCull]
|
|
pool.conns.keepIf((pc) => toCull.allIt(it.id != pc.id))
|
|
for culled in toCull:
|
|
try: culled.conn.close()
|
|
except: discard ""
|
|
log().debug(
|
|
"Trimming pool size. Culled $# free connections. $# connections remaining." %
|
|
[$toCull.len, $pool.conns.len])
|
|
|
|
proc take*(pool: DbConnPool): tuple[id: int, conn: DbConn] =
|
|
pool.maintain
|
|
let freeConns = pool.conns.filterIt(it.free)
|
|
|
|
log().debug(
|
|
"Providing a new connection ($# currently free)." % [$freeConns.len])
|
|
|
|
let reserved =
|
|
if freeConns.len > 0: freeConns[0]
|
|
else: pool.newConn()
|
|
|
|
reserved.free = false
|
|
log().debug("Reserve connection $#" % [$reserved.id])
|
|
return (id: reserved.id, conn: reserved.conn)
|
|
|
|
proc release*(pool: DbConnPool, connId: int): void =
|
|
log().debug("Reclaiming released connaction $#" % [$connId])
|
|
let foundConn = pool.conns.filterIt(it.id == connId)
|
|
if foundConn.len > 0: foundConn[0].free = true
|
|
|
|
template withConn*(pool: DbConnPool, stmt: untyped): untyped =
|
|
let (connId, conn {.inject.}) = take(pool)
|
|
stmt
|
|
release(pool, connId)
|