umpx 0.6.0

Collection of ump extensions.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

//! Integration utilities for creating sqlsrv handlers.

use super::{
  r2d2,
  rusqlite::{params, Connection},
  ConnPool, WrConn
};

use threadpool::ThreadPool;

use ump::ReplyContext;

// Re-export rusqlite from sqlsrv
pub use sqlsrv::rusqlite;

/// Given an [`ConnPool`] and a [`ReplyContext`], run a closure (presumably
/// performing read-only database operations).
///
/// Translates the closure's `Result<T, E>` into a `ReplyContext::reply()` on
/// `Ok` and `ReplyContext::fail()` on `Err`.
///
/// # Errors
/// [`r2d2::Error`] is returned if a read-only connection could not be
/// acquired from the connection pool.
pub fn proc_rodb_req<R, E, F>(
  cpool: &ConnPool,
  rctx: ReplyContext<R, E>,
  f: F
) -> Result<(), r2d2::Error>
where
  R: Send + 'static,
  E: std::error::Error + Send + 'static,
  F: FnOnce(&Connection) -> Result<R, E> + Send + 'static
{
  let roconn = cpool.reader()?;

  let res = match f(&roconn) {
    Ok(reply) => rctx.reply(reply),
    Err(e) => rctx.fail(e)
  };
  if let Err(e) = res {
    eprintln!("Reply message pass failed; {e}");
  }

  Ok(())
}

/// Given an [`ConnPool`], a [`ThreadPool`] and a [`ReplyContext`], run a
/// closure (presumably performing read-only database operations) on a thread.
///
/// Translates the closure's `Result<T, E>` into a `ReplyContext::reply()` on
/// `Ok` and `ReplyContext::fail()` on `Err`.
///
/// # Errors
/// [`r2d2::Error`] is returned if a read-only connection could not be
/// acquired from the connection pool.
pub fn proc_rodb_req_thrd<R, E, F>(
  cpool: &ConnPool,
  tpool: &ThreadPool,
  rctx: ReplyContext<R, E>,
  f: F
) -> Result<(), r2d2::Error>
where
  R: Send + 'static,
  E: std::error::Error + Send + 'static,
  F: FnOnce(&Connection) -> Result<R, E> + Send + 'static
{
  let roconn = cpool.reader()?;

  tpool.execute(move || {
    let res = match f(&roconn) {
      Ok(reply) => rctx.reply(reply),
      Err(e) => rctx.fail(e)
    };
    if let Err(e) = res {
      eprintln!("Reply message pass failed; {e}");
    }
  });

  Ok(())
}


/// Given an [`ConnPool`] and a [`ReplyContext`], run a
/// closure (presumably performing read/write database operations) on a thread.
///
/// Translates the closure's `Result<T, E>` into a `ReplyContext::reply()` on
/// `Ok` and `ReplyContext::fail()` on `Err`.
pub fn proc_rwdb_req<R, E, F>(cpool: &ConnPool, rctx: ReplyContext<R, E>, f: F)
where
  R: Send + 'static,
  E: std::error::Error + Send + 'static,
  F: FnOnce(&mut WrConn) -> Result<R, E> + Send + 'static
{
  let mut rwconn = cpool.writer();

  let res = match f(&mut rwconn) {
    Ok(reply) => rctx.reply(reply),
    Err(e) => rctx.fail(e)
  };
  if let Err(e) = res {
    eprintln!("Reply message pass failed; {e}");
  }
}


/// Given an [`ConnPool`], a [`ThreadPool`] and a [`ReplyContext`], run a
/// closure (presumably performing read/write database operations) on a thread.
///
/// Translates the closure's `Result<T, E>` into a `ReplyContext::reply()` on
/// `Ok` and `ReplyContext::fail()` on `Err`.
pub fn proc_rwdb_req_thrd<R, E, F>(
  cpool: &ConnPool,
  tpool: &ThreadPool,
  rctx: ReplyContext<R, E>,
  f: F
) where
  R: Send + 'static,
  E: std::error::Error + Send + 'static,
  F: FnOnce(&mut WrConn) -> Result<R, E> + Send + 'static
{
  let mut rwconn = cpool.writer();

  tpool.execute(move || {
    let res = match f(&mut rwconn) {
      Ok(reply) => rctx.reply(reply),
      Err(e) => rctx.fail(e)
    };
    if let Err(e) = res {
      eprintln!("Reply message pass failed; {e}");
    }
  });
}


/// Perform an incremental vacuum.
///
/// `n` is the number of freelist nodes to reclaim.  If `None` all nodes will
/// be reclaimed.
#[must_use]
pub fn incremental_vacuum(
  cpool: &ConnPool,
  tpool: &ThreadPool,
  n: Option<usize>
) -> swctx::WaitCtx<(), (), rusqlite::Error> {
  let (sctx, wctx) = swctx::mkpair();

  let conn = cpool.writer();

  tpool.execute(move || {
    #[allow(clippy::option_if_let_else)]
    let res = if let Some(n) = n {
      conn.execute("PRAGMA incremental_vacuum(?);", params![n])
    } else {
      conn.execute("PRAGMA incremental_vacuum;", params![])
    };
    let res = match res {
      Ok(_) => sctx.set(()),
      Err(e) => sctx.fail(e)
    };
    if let Err(e) = res {
      eprintln!("Reply message pass failed; {e}");
    }
  });

  wctx
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :