cuprate_txpool/
service.rs

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
//! [`tower::Service`] integeration + thread-pool.
//!
//! ## `service`
//! The `service` module implements the [`tower`] integration,
//! along with the reader/writer thread-pool system.
//!
//! The thread-pool allows outside crates to communicate with it by
//! sending database [`Request`][req_r]s and receiving [`Response`][resp]s `async`hronously -
//! without having to actually worry and handle the database themselves.
//!
//! The system is managed by this crate, and only requires [`init`] by the user.
//!
//! ## Handles
//! The 2 handles to the database are:
//! - [`TxpoolReadHandle`]
//! - [`TxpoolWriteHandle`]
//!
//! The 1st allows any caller to send [`ReadRequest`][req_r]s.
//!
//! The 2nd allows any caller to send [`WriteRequest`][req_w]s.
//!
//! Both the handles are cheaply [`Clone`]able.
//!
//! ## Initialization
//! The database & thread-pool system can be initialized with [`init()`].
//!
//! This causes the underlying database/threads to be setup
//! and returns a read/write handle to that database.
//!
//! ## Shutdown
//! Upon the above handles being dropped, the corresponding thread(s) will automatically exit, i.e:
//! - The last [`TxpoolReadHandle`] is dropped => reader thread-pool exits
//! - The last [`TxpoolWriteHandle`] is dropped => writer thread exits
//!
//! Upon dropping the [`cuprate_database::Env`]:
//! - All un-processed database transactions are completed
//! - All data gets flushed to disk (caused by [`Drop::drop`] impl on `Env`)
//!
//! ## Request and Response
//! To interact with the database (whether reading or writing data),
//! a `Request` can be sent using one of the above handles.
//!
//! Both the handles implement [`tower::Service`], so they can be [`tower::Service::call`]ed.
//!
//! An `async`hronous channel will be returned from the call.
//! This channel can be `.await`ed upon to (eventually) receive
//! the corresponding `Response` to your `Request`.
//!
//! [req_r]: interface::TxpoolReadRequest
//!
//! [req_w]: interface::TxpoolWriteRequest
//!
//! // TODO: we have 2 responses
//!
//! [resp]: interface::TxpoolWriteResponse
//!
//! # Example
//! Simple usage of `service`.
//!
//! ```rust
//! use std::sync::Arc;
//!
//! use hex_literal::hex;
//! use tower::{Service, ServiceExt};
//!
//! use cuprate_test_utils::data::TX_V1_SIG2;
//!
//! use cuprate_txpool::{
//!     cuprate_database::Env,
//!     config::ConfigBuilder,
//!     service::interface::{
//!         TxpoolWriteRequest,
//!         TxpoolWriteResponse,
//!         TxpoolReadRequest,
//!         TxpoolReadResponse
//!     }
//! };
//!
//! # #[tokio::main]
//! # async fn main() -> Result<(), Box<dyn std::error::Error>> {
//! // Create a configuration for the database environment.
//! use cuprate_test_utils::data::TX_V1_SIG2;
//! let tmp_dir = tempfile::tempdir()?;
//! let db_dir = tmp_dir.path().to_owned();
//! let config = ConfigBuilder::new()
//!     .db_directory(db_dir.into())
//!     .build();
//!
//! // Initialize the database thread-pool.
//! let (mut read_handle, mut write_handle, _) = cuprate_txpool::service::init(config)?;
//!
//! // Prepare a request to write block.
//! let tx = TX_V1_SIG2.clone();
//! let request = TxpoolWriteRequest::AddTransaction {
//!     tx: Arc::new(tx.try_into().unwrap()),
//!     state_stem: false,
//! };
//!
//! // Send the request.
//! // We receive back an `async` channel that will
//! // eventually yield the result when `service`
//! // is done writing the tx.
//! let response_channel = write_handle.ready().await?.call(request);
//!
//! // Block write was OK.
//! let TxpoolWriteResponse::AddTransaction(double_spent) = response_channel.await? else {
//!     panic!("tx-pool returned wrong response!");
//! };
//! assert!(double_spent.is_none());
//!
//! // Now, let's try getting the block hash
//! // of the block we just wrote.
//! let request = TxpoolReadRequest::TxBlob(TX_V1_SIG2.tx_hash);
//! let response_channel = read_handle.ready().await?.call(request);
//! let response = response_channel.await?;
//!
//! // This causes the writer thread on the
//! // other side of this handle to exit...
//! drop(write_handle);
//! // ...and this causes the reader thread-pool to exit.
//! drop(read_handle);
//! # Ok(()) }
//! ```

mod free;
pub mod interface;
mod read;
mod types;
mod write;

pub use free::init;
pub use types::{TxpoolReadHandle, TxpoolWriteHandle};