///|
priv struct Handle[C] {
connection : C
mut broken : Bool
}
///|
/// Add lifetime management to an existing AsyncDriver and its connection source.
/// acquire transfers an exclusive connection; release receives discard=true when
/// it must not be reused. Both callbacks must be cancellation-safe. A reusable
/// pool's release callback must reset transaction/session state before reuse.
/// close stops that source from accepting new work; active leases finish normally.
/// No Value, Row or ExecResult conversions occur here.
pub fn[C : @moondb.AsyncDriver] database(
acquire~ : async () -> C,
release~ : async (C, Bool) -> Unit,
close~ : () -> Unit,
max_leases~ : Int,
max_waiters? : Int = 128,
checkout_timeout_ms? : Int = 5000,
) -> @sql.Database[@moondb.Value, @moondb.Row, @moondb.ExecResult, Unit] raise {
database_with_options(
acquire~,
release~,
close~,
begin=async fn(conn, _) { @moondb.AsyncDriver::begin(conn) },
max_leases~,
max_waiters~,
checkout_timeout_ms~,
)
}
///|
/// Keep backend transaction options with the backend. The supplied begin callback
/// opens a transaction with those options; commit/rollback use AsyncDriver.
pub fn[C : @moondb.AsyncDriver, O] database_with_options(
acquire~ : async () -> C,
release~ : async (C, Bool) -> Unit,
close~ : () -> Unit,
begin~ : async (C, O) -> Unit,
max_leases~ : Int,
max_waiters? : Int = 128,
checkout_timeout_ms? : Int = 5000,
) -> @sql.Database[@moondb.Value, @moondb.Row, @moondb.ExecResult, O] raise {
guard max_leases > 0 && max_leases <= 1000000 else {
raise @sql.InvalidConfig("Invalid admission limits")
}
let permits : @async.Queue[Unit] = @async.Queue(kind=Unbounded)
for _ in 0.. false })
{ connection: acquire(), broken: false, }
},
release: async fn(handle, discard) {
defer ignore(permits.try_put(()) catch { _ => false })
release(handle.connection, discard || handle.broken)
},
close: fn() {
permits.close(error=@sql.Closed, clear=true)
close()
},
executor: @sql.Separate(
query=async fn(handle, sql, params) {
@async.protect_from_cancel(async fn() {
errdefer {
handle.broken = true
@moondb.AsyncDriver::close(handle.connection)
}
@moondb.AsyncDriver::query(handle.connection, sql, params)
})
},
execute=async fn(handle, sql, params) {
@async.protect_from_cancel(async fn() {
errdefer {
handle.broken = true
@moondb.AsyncDriver::close(handle.connection)
}
@moondb.AsyncDriver::execute(handle.connection, sql, params)
})
},
),
begin: async fn(handle, options) { begin(handle.connection, options) },
commit: async fn(handle) {
@moondb.AsyncDriver::commit(handle.connection)
},
rollback: async fn(handle) {
// A failed operation closes the connection, which aborts its transaction.
// Do not issue ROLLBACK into possibly unfinished protocol state.
if !handle.broken {
@moondb.AsyncDriver::rollback(handle.connection)
}
},
},
max_leases~,
max_waiters~,
checkout_timeout_ms~,
)
}