diff --git a/Cargo.lock b/Cargo.lock index 24b483e..4b5f8fd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -41,131 +41,34 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" [[package]] -name = "async-channel" -version = "1.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "81953c529336010edd6d8e358f886d9581267795c61b19475b71314bffa46d35" -dependencies = [ - "concurrent-queue", - "event-listener 2.5.3", - "futures-core", -] - -[[package]] -name = "async-channel" -version = "2.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "89b47800b0be77592da0afd425cc03468052844aff33b84e33cc696f64e77b6a" -dependencies = [ - "concurrent-queue", - "event-listener-strategy", - "futures-core", - "pin-project-lite", -] - -[[package]] -name = "async-executor" -version = "1.13.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bb812ffb58524bdd10860d7d974e2f01cc0950c2438a74ee5ec2e2280c6c4ffa" -dependencies = [ - "async-task", - "concurrent-queue", - "fastrand", - "futures-lite", - "pin-project-lite", - "slab", -] - -[[package]] -name = "async-global-executor" -version = "2.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "05b1b633a2115cd122d73b955eadd9916c18c8f510ec9cd1686404c60ad1c29c" -dependencies = [ - "async-channel 2.3.1", - "async-executor", - "async-io", - "async-lock", - "blocking", - "futures-lite", - "once_cell", -] - -[[package]] -name = "async-io" -version = "2.4.1" +name = "autocfg" +version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1237c0ae75a0f3765f58910ff9cdd0a12eeb39ab2f4c7de23262f337f0aacbb3" -dependencies = [ - "async-lock", - "cfg-if", - "concurrent-queue", - "futures-io", - "futures-lite", - "parking", - "polling", - "rustix 1.0.7", - "slab", - "tracing", - "windows-sys 0.59.0", -] +checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26" [[package]] -name = "async-lock" -version = "3.4.0" +name = "aws-lc-rs" +version = "1.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ff6e472cdea888a4bd64f342f09b3f50e1886d32afe8df3d663c01140b811b18" +checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e" dependencies = [ - "event-listener 5.4.0", - "event-listener-strategy", - "pin-project-lite", + "aws-lc-sys", + "zeroize", ] [[package]] -name = "async-std" -version = "1.13.1" +name = "aws-lc-sys" +version = "0.44.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "730294c1c08c2e0f85759590518f6333f0d5a0a766a27d519c1b244c3dfd8a24" +checksum = "f09fae7be8bb3174e05c6afdb34199e6dc0c7c04ba9fa237b1967adfbde27483" dependencies = [ - "async-channel 1.9.0", - "async-global-executor", - "async-io", - "async-lock", - "crossbeam-utils", - "futures-channel", - "futures-core", - "futures-io", - "futures-lite", - "gloo-timers", - "kv-log-macro", - "log", - "memchr", - "once_cell", - "pin-project-lite", - "pin-utils", - "slab", - "wasm-bindgen-futures", + "cc", + "cmake", + "dunce", + "fs_extra", + "pkg-config", ] -[[package]] -name = "async-task" -version = "4.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b75356056920673b02621b35afd0f7dda9306d03c79a30f5c56c44cf256e3de" - -[[package]] -name = "atomic-waker" -version = "1.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" - -[[package]] -name = "autocfg" -version = "1.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26" - [[package]] name = "backtrace" version = "0.3.75" @@ -356,25 +259,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "blocking" -version = "1.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "703f41c54fc768e63e091340b424302bb1c29ef4aa0c7f10fe849dfb114d29ea" -dependencies = [ - "async-channel 2.3.1", - "async-task", - "futures-io", - "futures-lite", - "piper", -] - -[[package]] -name = "bumpalo" -version = "3.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1628fb46dfa0b37568d12e5edd512553eccf6a22a78e8bde00bb4aed84d5bdbf" - [[package]] name = "byteorder" version = "1.5.0" @@ -409,10 +293,11 @@ dependencies = [ [[package]] name = "cc" -version = "1.2.25" +version = "1.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d0fc897dc1e865cc67c0e05a836d9d3f1df3cbe442aa4a9473b18e12624a4951" +checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d" dependencies = [ + "find-msvc-tools", "jobserver", "libc", "shlex", @@ -435,12 +320,12 @@ dependencies = [ ] [[package]] -name = "concurrent-queue" -version = "2.5.0" +name = "cmake" +version = "0.1.58" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" +checksum = "c0f78a02292a74a88ac736019ab962ece0bc380e3f977bf72e376c5d78ff0678" dependencies = [ - "crossbeam-utils", + "cc", ] [[package]] @@ -503,6 +388,12 @@ dependencies = [ "subtle", ] +[[package]] +name = "dunce" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" + [[package]] name = "either" version = "1.15.0" @@ -542,14 +433,16 @@ name = "electrum_streaming_client" version = "0.4.0" dependencies = [ "anyhow", - "async-std", "bdk_testenv", "bitcoin", "futures", + "rustls 0.23.43", "serde", "serde_json", "tokio", + "tokio-rustls", "tokio-util", + "webpki-roots 1.0.9", ] [[package]] @@ -562,33 +455,6 @@ dependencies = [ "windows-sys 0.59.0", ] -[[package]] -name = "event-listener" -version = "2.5.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0206175f82b8d6bf6652ff7d71a1e27fd2e4efde587fd368662814d6ec1d9ce0" - -[[package]] -name = "event-listener" -version = "5.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3492acde4c3fc54c845eaab3eed8bd00c7a7d881f78bfc801e43a93dec1331ae" -dependencies = [ - "concurrent-queue", - "parking", - "pin-project-lite", -] - -[[package]] -name = "event-listener-strategy" -version = "0.5.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" -dependencies = [ - "event-listener 5.4.0", - "pin-project-lite", -] - [[package]] name = "fastrand" version = "2.3.0" @@ -607,6 +473,12 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "find-msvc-tools" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890" + [[package]] name = "flate2" version = "1.1.1" @@ -617,6 +489,12 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "fs_extra" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" + [[package]] name = "futures" version = "0.3.31" @@ -657,7 +535,6 @@ dependencies = [ "futures-core", "futures-task", "futures-util", - "num_cpus", ] [[package]] @@ -666,19 +543,6 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" -[[package]] -name = "futures-lite" -version = "2.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f5edaec856126859abb19ed65f39e90fea3a9574b9707f13539acf4abf7eb532" -dependencies = [ - "fastrand", - "futures-core", - "futures-io", - "parking", - "pin-project-lite", -] - [[package]] name = "futures-macro" version = "0.3.31" @@ -687,7 +551,7 @@ checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.101", ] [[package]] @@ -759,24 +623,6 @@ version = "0.31.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07e28edb80900c19c28f1072f2e8aeca7fa06b23cd4169cefe1af5aa3260783f" -[[package]] -name = "gloo-timers" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbb143cf96099802033e0d4f4963b19fd2e0b728bcf076cd9cf7f6634f092994" -dependencies = [ - "futures-channel", - "futures-core", - "js-sys", - "wasm-bindgen", -] - -[[package]] -name = "hermit-abi" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f154ce46856750ed433c8649605bf7ed2de3bc35fd9d2a9f30cddd873c80cb08" - [[package]] name = "hex-conservative" version = "0.2.1" @@ -835,16 +681,6 @@ dependencies = [ "libc", ] -[[package]] -name = "js-sys" -version = "0.3.77" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1cfaf33c695fc6e08064efbc1f72ec937429614f25eef83af942d0e227c3a28f" -dependencies = [ - "once_cell", - "wasm-bindgen", -] - [[package]] name = "jsonrpc" version = "0.18.0" @@ -857,15 +693,6 @@ dependencies = [ "serde_json", ] -[[package]] -name = "kv-log-macro" -version = "1.0.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0de8b303297635ad57c9f5059fd9cee7a47f8e8daa09df0fcd07dd39fb22977f" -dependencies = [ - "log", -] - [[package]] name = "libc" version = "0.2.172" @@ -910,9 +737,6 @@ name = "log" version = "0.4.27" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13dc2df351e3202783a1fe0d44375f7295ffb4049267b0f3018346dc122a1d94" -dependencies = [ - "value-bag", -] [[package]] name = "memchr" @@ -955,11 +779,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0d2aaba477837b46ec1289588180fabfccf0c3b1d1a0c6b1866240cd6cd5ce9" dependencies = [ "log", - "rustls", - "rustls-webpki", + "rustls 0.21.12", + "rustls-webpki 0.101.7", "serde", "serde_json", - "webpki-roots", + "webpki-roots 0.25.4", ] [[package]] @@ -993,16 +817,6 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9" -[[package]] -name = "num_cpus" -version = "1.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" -dependencies = [ - "hermit-abi", - "libc", -] - [[package]] name = "object" version = "0.36.7" @@ -1018,12 +832,6 @@ version = "1.21.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" -[[package]] -name = "parking" -version = "2.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" - [[package]] name = "parking_lot" version = "0.12.4" @@ -1082,38 +890,12 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" -[[package]] -name = "piper" -version = "0.2.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96c8c490f422ef9a4efd2cb5b42b76c8613d7e7dfc1caf667b8a3350a5acc066" -dependencies = [ - "atomic-waker", - "fastrand", - "futures-io", -] - [[package]] name = "pkg-config" version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" -[[package]] -name = "polling" -version = "3.8.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b53a684391ad002dd6a596ceb6c74fd004fdce75f4be2e3f615068abbea5fd50" -dependencies = [ - "cfg-if", - "concurrent-queue", - "hermit-abi", - "pin-project-lite", - "rustix 1.0.7", - "tracing", - "windows-sys 0.59.0", -] - [[package]] name = "powerfmt" version = "0.2.0" @@ -1246,10 +1028,34 @@ checksum = "3f56a14d1f48b391359b22f731fd4bd7e43c97f3c50eee276f3aa09c94784d3e" dependencies = [ "log", "ring", - "rustls-webpki", + "rustls-webpki 0.101.7", "sct", ] +[[package]] +name = "rustls" +version = "0.23.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" +dependencies = [ + "aws-lc-rs", + "log", + "once_cell", + "rustls-pki-types", + "rustls-webpki 0.103.14", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + [[package]] name = "rustls-webpki" version = "0.101.7" @@ -1261,10 +1067,16 @@ dependencies = [ ] [[package]] -name = "rustversion" -version = "1.0.21" +name = "rustls-webpki" +version = "0.103.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8a0d197bd2c9dc6e53b84da9556a69ba4cdfab8619eb41a8bd1cc2027a0f6b1d" +checksum = "0527518605e68109d875e248ea259b6758801cf165e4b2c2733ae3b51f12535a" +dependencies = [ + "aws-lc-rs", + "ring", + "rustls-pki-types", + "untrusted", +] [[package]] name = "ryu" @@ -1311,22 +1123,32 @@ dependencies = [ [[package]] name = "serde" -version = "1.0.219" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f0e2c6ed6606019b4e29e69dbaba95b11854410e5347d525002456dbbb786b6" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" dependencies = [ "serde_derive", ] [[package]] name = "serde_derive" -version = "1.0.219" +version = "1.0.229" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b0276cf7f2c73365f7157c8123c21cd9a50fbbd844757af28ca1f5925fc2a00" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.3", ] [[package]] @@ -1365,9 +1187,9 @@ dependencies = [ [[package]] name = "shlex" -version = "1.3.0" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" [[package]] name = "signal-hook-registry" @@ -1420,6 +1242,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "tar" version = "0.4.44" @@ -1489,7 +1322,17 @@ checksum = "6e06d43f1345a3bcd39f6a56dbb7dcab2ba47e68e8ac134855e7e2bdbaf8cab8" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.101", +] + +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls 0.23.43", + "tokio", ] [[package]] @@ -1506,22 +1349,6 @@ dependencies = [ "tokio", ] -[[package]] -name = "tracing" -version = "0.1.41" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0" -dependencies = [ - "pin-project-lite", - "tracing-core", -] - -[[package]] -name = "tracing-core" -version = "0.1.33" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e672c95779cf947c5311f83787af4fa8fffd12fb27e4993211a84bdfd9610f9c" - [[package]] name = "typenum" version = "1.18.0" @@ -1540,12 +1367,6 @@ version = "0.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" -[[package]] -name = "value-bag" -version = "1.11.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "943ce29a8a743eb10d6082545d861b24f9d1b160b7d741e0f2cdf726bec909c5" - [[package]] name = "version_check" version = "0.9.5" @@ -1568,92 +1389,20 @@ dependencies = [ ] [[package]] -name = "wasm-bindgen" -version = "0.2.100" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1edc8929d7499fc4e8f0be2262a241556cfc54a0bea223790e71446f2aab1ef5" -dependencies = [ - "cfg-if", - "once_cell", - "rustversion", - "wasm-bindgen-macro", -] - -[[package]] -name = "wasm-bindgen-backend" -version = "0.2.100" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2f0a0651a5c2bc21487bde11ee802ccaf4c51935d0d3d42a6101f98161700bc6" -dependencies = [ - "bumpalo", - "log", - "proc-macro2", - "quote", - "syn", - "wasm-bindgen-shared", -] - -[[package]] -name = "wasm-bindgen-futures" -version = "0.4.50" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "555d470ec0bc3bb57890405e5d4322cc9ea83cebb085523ced7be4144dac1e61" -dependencies = [ - "cfg-if", - "js-sys", - "once_cell", - "wasm-bindgen", - "web-sys", -] - -[[package]] -name = "wasm-bindgen-macro" -version = "0.2.100" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7fe63fc6d09ed3792bd0897b314f53de8e16568c2b3f7982f468c0bf9bd0b407" -dependencies = [ - "quote", - "wasm-bindgen-macro-support", -] - -[[package]] -name = "wasm-bindgen-macro-support" -version = "0.2.100" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de" -dependencies = [ - "proc-macro2", - "quote", - "syn", - "wasm-bindgen-backend", - "wasm-bindgen-shared", -] - -[[package]] -name = "wasm-bindgen-shared" -version = "0.2.100" +name = "webpki-roots" +version = "0.25.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a05d73b933a847d6cccdda8f838a22ff101ad9bf93e33684f39c1f5f0eece3d" -dependencies = [ - "unicode-ident", -] +checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" [[package]] -name = "web-sys" -version = "0.3.77" +name = "webpki-roots" +version = "1.0.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33b6dd2ef9186f1f2072e409e99cd22a975331a6b3591b12c764e0e55c60d5d2" +checksum = "7dcd9d09a39985f5344844e66b0c530a33843579125f23e21e9f0f220850f22a" dependencies = [ - "js-sys", - "wasm-bindgen", + "rustls-pki-types", ] -[[package]] -name = "webpki-roots" -version = "0.25.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1" - [[package]] name = "which" version = "4.4.2" @@ -1784,9 +1533,15 @@ checksum = "28a6e20d751156648aa063f3800b706ee209a32c0b4d9f24be3d980b01be55ef" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.101", ] +[[package]] +name = "zeroize" +version = "1.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ced3678a2879b30306d323f4542626697a464a97c0a07c9aebf7ebca65cd4dde" + [[package]] name = "zip" version = "0.6.6" diff --git a/Cargo.toml b/Cargo.toml index 5e3c99a..6189a5a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,27 +4,32 @@ version = "0.4.0" description = "Experimental but sane electrum client by @evanlinjin." license = "MIT OR Apache-2.0" edition = "2021" -rust-version = "1.70" +rust-version = "1.71" repository = "https://github.com/bitcoindevkit/electrum_streaming_client" documentation = "https://docs.rs/electrum_streaming_client" readme = "README.md" +[package.metadata.docs.rs] +all-features = true + [dependencies] futures = "0.3" serde = { version = "1", features = ["derive"] } serde_json = "1" bitcoin = { version = "0.32", features = ["serde"] } -tokio = { version = "1.44.2", features = ["io-util"], optional = true } +tokio = { version = "1.44.2", features = ["io-util", "net", "time"], optional = true } tokio-util = { version = "0.7.15", features = ["compat"], optional = true } +tokio-rustls = { version = "0.26", optional = true } +rustls = { version = "0.23", optional = true } +webpki-roots = { version = "1", optional = true } [features] default = ["tokio"] tokio = ["dep:tokio", "tokio-util"] +ssl = ["dep:rustls", "dep:webpki-roots", "dep:tokio-rustls"] [dev-dependencies] -async-std = "1.13.0" bdk_testenv = "0.11" -futures = { version = "0.3", features = ["thread-pool"] } tokio = { version = "1.44.2", features = ["full"] } anyhow = "1.0.98" diff --git a/README.md b/README.md index 77bac7c..59bb836 100644 --- a/README.md +++ b/README.md @@ -16,24 +16,24 @@ models. ## Example (async with Tokio) ```rust,no_run -use electrum_streaming_client::{AsyncClient, Event}; -use tokio::net::TcpStream; +use electrum_streaming_client::{request, AsyncClient, ConnectConfig, ServerUrl}; use futures::StreamExt; #[tokio::main] async fn main() -> anyhow::Result<()> { - let stream = TcpStream::connect("127.0.0.1:50001").await?; - let (reader, writer) = stream.into_split(); - let (client, mut events, worker) = AsyncClient::new_tokio(reader, writer); + let url: ServerUrl = "ssl://electrum.blockstream.info:50002".parse()?; + let (client, mut events, worker) = + AsyncClient::connect(&url, &ConnectConfig::default()).await?; + let worker = tokio::spawn(worker); - tokio::spawn(worker); // spawn the client worker task - - let relay_fee = client.send_request(electrum_streaming_client::request::RelayFee).await?; + let relay_fee = client.send_request(request::RelayFee).await?; println!("Relay fee: {relay_fee:?}"); - while let Some(event) = events.next().await { - println!("Event: {event:?}"); - } + client.send_event_request(request::HeadersSubscribe)?; + println!("Event: {:?}", events.next().await); + + drop(client); + worker.await??; Ok(()) } @@ -41,9 +41,9 @@ async fn main() -> anyhow::Result<()> { ## Optional Features -- `tokio`: Enables [`AsyncClient::new_tokio`] for use with Tokio-compatible streams. +- `tokio` (default): Enables Tokio transport support. +- `ssl`: Enables TLS via rustls. Async TLS additionally requires `tokio`. ## License MIT - diff --git a/src/client.rs b/src/client.rs index 2c55200..a7a7cd1 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,3 +1,5 @@ +use std::{future::Future, io, thread::JoinHandle, time::Duration}; + use crate::pending_request::{PendingRequest, RequestExt}; use crate::*; @@ -94,7 +96,7 @@ impl AsyncClient { ) -> ( Self, AsyncEventReceiver, - impl std::future::Future> + Send, + impl Future> + Send, ) where R: futures::AsyncRead + Send + Unpin, @@ -123,7 +125,7 @@ impl AsyncClient { Some(incoming_res) => { let event_opt = state .handle_incoming(incoming_res?) - .map_err(|error| std::io::Error::new(std::io::ErrorKind::Other, error))?; + .map_err(|error| io::Error::new(io::ErrorKind::Other, error))?; if let Some(event) = event_opt { if let Err(_err) = event_tx.unbounded_send(event) { break; @@ -134,7 +136,7 @@ impl AsyncClient { } } } - std::io::Result::<()>::Ok(()) + io::Result::<()>::Ok(()) }; (Self { tx: req_tx }, event_recv, fut) @@ -164,7 +166,7 @@ impl AsyncClient { ) -> ( Self, AsyncEventReceiver, - impl std::future::Future> + Send, + impl Future> + Send, ) where R: tokio::io::AsyncRead + Send + Unpin, @@ -179,6 +181,85 @@ impl AsyncClient { self.tx.close_channel(); } + /// Creates a new [`AsyncClient`] connected to `addr` over plaintext TCP via Tokio. + #[cfg(feature = "tokio")] + pub async fn connect_tcp( + addr: &crate::transport::ServerAddr, + timeout: Option, + ) -> io::Result<( + Self, + AsyncEventReceiver, + impl Future> + Send, + )> { + let stream = crate::transport::tokio::connect_tcp(addr, timeout).await?; + let (reader, writer) = tokio::io::split(stream); + Ok(Self::new_tokio(reader, writer)) + } + + /// Creates a new [`AsyncClient`] connected to `addr` over TLS via Tokio. + /// + /// `timeout` bounds DNS, TCP connect, and the TLS handshake. + /// `validate_domain` requires a domain host. + #[cfg(all(feature = "ssl", feature = "tokio"))] + pub async fn connect_ssl( + addr: &crate::transport::ServerAddr, + validate_domain: bool, + timeout: Option, + ) -> Result< + ( + Self, + AsyncEventReceiver, + impl Future> + Send, + ), + crate::ConnectError, + > { + let stream = crate::transport::tokio::connect_ssl(addr, validate_domain, timeout).await?; + let (reader, writer) = tokio::io::split(stream); + Ok(Self::new_tokio(reader, writer)) + } + + /// Connects to `url` using its scheme. + /// + /// A missing scheme defaults to plaintext TCP; `ssl://` requires the `ssl` feature. + #[cfg(feature = "tokio")] + pub async fn connect( + url: &crate::transport::ServerUrl, + config: &crate::transport::ConnectConfig, + ) -> Result< + ( + Self, + AsyncEventReceiver, + impl Future> + Send, + ), + crate::ConnectError, + > { + fn box_worker( + worker: F, + ) -> std::pin::Pin> + Send>> + where + F: Future> + Send + 'static, + { + Box::pin(worker) + } + + match url.scheme() { + crate::transport::Scheme::Tcp => { + let (client, events, worker) = + Self::connect_tcp(url.addr(), config.timeout()).await?; + Ok((client, events, box_worker(worker))) + } + #[cfg(feature = "ssl")] + crate::transport::Scheme::Ssl => { + let (client, events, worker) = + Self::connect_ssl(url.addr(), config.validate_domain(), config.timeout()) + .await?; + Ok((client, events, box_worker(worker))) + } + #[cfg(not(feature = "ssl"))] + crate::transport::Scheme::Ssl => Err(crate::ConnectError::SslUnsupported), + } + } + /// Sends a single tracked request to the Electrum server and awaits the response. /// /// This method is for request–response style interactions where only a single result is @@ -304,27 +385,27 @@ impl BlockingClient { ) -> ( Self, BlockingEventReceiver, - std::thread::JoinHandle>, - std::thread::JoinHandle>, + JoinHandle>, + JoinHandle>, ) where - R: std::io::Read + Send + 'static, - W: std::io::Write + Send + 'static, + R: io::Read + Send + 'static, + W: io::Write + Send + 'static, { use std::sync::mpsc::*; let (event_tx, event_recv) = channel::(); let (req_tx, req_recv) = channel::>(); - let incoming_stream = crate::io::ReadStreamer::new(std::io::BufReader::new(reader)); + let incoming_stream = crate::io::ReadStreamer::new(io::BufReader::new(reader)); let read_state = std::sync::Arc::new(std::sync::Mutex::new(RequestTracker::new())); let write_state = std::sync::Arc::clone(&read_state); - let read_join = std::thread::spawn(move || -> std::io::Result<()> { + let read_join = std::thread::spawn(move || -> io::Result<()> { for incoming_res in incoming_stream { let event_opt = read_state .lock() .unwrap() .handle_incoming(incoming_res?) - .map_err(|error| std::io::Error::new(std::io::ErrorKind::Other, error))?; + .map_err(|error| io::Error::new(io::ErrorKind::Other, error))?; if let Some(event) = event_opt { if let Err(_err) = event_tx.send(event) { break; @@ -333,7 +414,7 @@ impl BlockingClient { } Ok(()) }); - let write_join = std::thread::spawn(move || -> std::io::Result<()> { + let write_join = std::thread::spawn(move || -> io::Result<()> { let mut next_id = 0_u32; for req in req_recv { let raw_req = write_state.lock().unwrap().track_request(&mut next_id, req); @@ -344,6 +425,75 @@ impl BlockingClient { (Self { tx: req_tx }, event_recv, read_join, write_join) } + /// Connects to `url` using its scheme. + /// + /// A missing scheme defaults to plaintext TCP; `ssl://` requires the `ssl` feature. + #[allow(clippy::type_complexity)] + pub fn connect( + url: &crate::transport::ServerUrl, + config: &crate::transport::ConnectConfig, + ) -> Result< + ( + Self, + BlockingEventReceiver, + JoinHandle>, + JoinHandle>, + ), + crate::ConnectError, + > { + match url.scheme() { + crate::transport::Scheme::Tcp => { + Self::connect_tcp(url.addr(), config.timeout()).map_err(Into::into) + } + #[cfg(feature = "ssl")] + crate::transport::Scheme::Ssl => { + Self::connect_ssl(url.addr(), config.validate_domain(), config.timeout()) + } + #[cfg(not(feature = "ssl"))] + crate::transport::Scheme::Ssl => Err(crate::ConnectError::SslUnsupported), + } + } + + /// Creates a new [`BlockingClient`] connected to `addr` over plaintext TCP. + #[allow(clippy::type_complexity)] + pub fn connect_tcp( + addr: &crate::transport::ServerAddr, + timeout: Option, + ) -> io::Result<( + Self, + BlockingEventReceiver, + JoinHandle>, + JoinHandle>, + )> { + let writer = crate::transport::blocking::connect_tcp(addr, timeout)?; + let reader = writer.try_clone()?; + Ok(Self::new(reader, writer)) + } + + /// Creates a new [`BlockingClient`] connected to `addr` over TLS. + /// + /// `timeout` bounds TCP connect and the TLS handshake. + /// `validate_domain` requires a domain host. + #[cfg(feature = "ssl")] + #[allow(clippy::type_complexity)] + pub fn connect_ssl( + addr: &crate::transport::ServerAddr, + validate_domain: bool, + timeout: Option, + ) -> Result< + ( + Self, + BlockingEventReceiver, + JoinHandle>, + JoinHandle>, + ), + crate::ConnectError, + > { + let stream = crate::transport::blocking::connect_ssl(addr, validate_domain, timeout)?; + let (reader, writer) = stream.into_split(); + Ok(Self::new(reader, writer)) + } + /// Sends a single tracked request to the Electrum server and waits for its response. /// /// This method blocks the current thread until the server replies. It is intended for diff --git a/src/io.rs b/src/io.rs index 58a83ec..3518870 100644 --- a/src/io.rs +++ b/src/io.rs @@ -178,7 +178,8 @@ where { let mut b = serde_json::to_vec(&msg.into()).expect("must serialize"); b.push(b'\n'); - writer.write_all(&b) + writer.write_all(&b)?; + writer.flush() } /// Asynchronously writes a JSON-RPC request or batch to an async writer, followed by a newline. @@ -200,7 +201,8 @@ where use futures::AsyncWriteExt; let mut b = serde_json::to_vec(&msg.into()).expect("must serialize"); b.push(b'\n'); - writer.write_all(&b).await + writer.write_all(&b).await?; + writer.flush().await } /// Asynchronously writes a JSON-RPC request or batch to a tokio async writer, followed by a newline. @@ -215,5 +217,6 @@ where use tokio::io::AsyncWriteExt; let mut b = serde_json::to_vec(&msg.into()).expect("must serialize"); b.push(b'\n'); - writer.write_all(&b).await + writer.write_all(&b).await?; + writer.flush().await } diff --git a/src/lib.rs b/src/lib.rs index c48897e..b4a2b62 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -13,12 +13,20 @@ pub mod protocol; pub mod request; mod request_tracker; pub mod response; +pub mod transport; pub use hash_types::*; pub use pending_request::*; pub use protocol::*; pub use request::Request; pub use request_tracker::*; pub use serde_json; +pub use transport::{ + ConnectConfig, ConnectConfigBuilder, ConnectError, Host, ParseServerAddrError, Scheme, + ServerAddr, ServerUrl, +}; + +#[cfg(feature = "ssl")] +pub use transport::TlsError; /// An owned or borrowed static string. pub type CowStr = std::borrow::Cow<'static, str>; diff --git a/src/transport/mod.rs b/src/transport/mod.rs new file mode 100644 index 0000000..77584a0 --- /dev/null +++ b/src/transport/mod.rs @@ -0,0 +1,810 @@ +//! Address types and transport constructors for Electrum connections. +//! +//! Use [`crate::BlockingClient::connect`] or [`crate::AsyncClient::connect`] for built-in +//! TCP/TLS, or [`crate::BlockingClient::new`] / [`crate::AsyncClient::new`] for custom I/O. + +use std::fmt; +use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; +use std::str::FromStr; +#[cfg(feature = "ssl")] +use std::sync::Arc; + +#[cfg(feature = "ssl")] +mod tls; + +/// The connection scheme of an Electrum server, taken from the URL prefix. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Scheme { + /// Plaintext TCP (`tcp://`). + Tcp, + /// SSL/TLS encrypted TCP (`ssl://`). + Ssl, +} + +impl Scheme { + /// The URL prefix of this scheme (e.g. `tcp` for `tcp://`). + pub fn as_str(&self) -> &'static str { + match self { + Scheme::Tcp => "tcp", + Scheme::Ssl => "ssl", + } + } +} + +impl fmt::Display for Scheme { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +impl FromStr for Scheme { + type Err = ParseServerAddrError; + + /// Parses a scheme prefix. Lowercase only: `"tcp"` and `"ssl"` are accepted. + fn from_str(s: &str) -> Result { + match s { + "tcp" => Ok(Scheme::Tcp), + "ssl" => Ok(Scheme::Ssl), + other => Err(ParseServerAddrError(other.to_string())), + } + } +} + +/// The host portion of a [`ServerAddr`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Host { + /// A domain name, e.g. `electrum.example.com`. + Domain(String), + /// An IP literal. + Ip(IpAddr), +} + +impl fmt::Display for Host { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Host::Domain(domain) => f.write_str(domain), + Host::Ip(IpAddr::V4(ip)) => write!(f, "{}", ip), + Host::Ip(IpAddr::V6(ip)) => write!(f, "[{}]", ip), + } + } +} + +/// An Electrum server address: a [`Host`] and a port, without a connection scheme. +/// +/// Parses from `"host:port"`. IPv6 literals must be bracketed (`"[::1]:50001"`). +/// Use [`ServerUrl`] to parse scheme-prefixed addresses. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ServerAddr { + host: Host, + port: u16, +} + +impl ServerAddr { + /// Creates a new `ServerAddr` from a [`Host`] and port. + pub fn new(host: Host, port: u16) -> Self { + Self { host, port } + } + + /// The host portion of this address. + pub fn host(&self) -> &Host { + &self.host + } + + /// The port of this address. + pub fn port(&self) -> u16 { + self.port + } +} + +impl fmt::Display for ServerAddr { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}:{}", self.host, self.port) + } +} + +impl FromStr for ServerAddr { + type Err = ParseServerAddrError; + + fn from_str(s: &str) -> Result { + let invalid = || ParseServerAddrError(s.to_string()); + + if s.contains("://") { + return Err(invalid()); + } + + // Bracketed IP literal: "[]:". + if let Some(rest) = s.strip_prefix('[') { + let (ip_str, port_str) = rest.split_once("]:").ok_or_else(invalid)?; + return Ok(Self { + host: Host::Ip(ip_str.parse().map_err(|_| invalid())?), + port: port_str.parse().map_err(|_| invalid())?, + }); + } + + let (host_str, port_str) = s.rsplit_once(':').ok_or_else(invalid)?; + if host_str.is_empty() || host_str.contains(':') { + // Empty host, or an unbracketed IPv6 literal. + return Err(invalid()); + } + Ok(Self { + host: match host_str.parse::() { + Ok(ip) => Host::Ip(ip), + Err(_) => Host::Domain(host_str.to_string()), + }, + port: port_str.parse().map_err(|_| invalid())?, + }) + } +} + +impl ToSocketAddrs for ServerAddr { + type Iter = std::vec::IntoIter; + + /// Resolves this address via **local DNS**. + /// + /// Do not use this for `.onion` hosts — they are not resolvable via local DNS. + fn to_socket_addrs(&self) -> std::io::Result { + match &self.host { + Host::Ip(ip) => Ok(vec![SocketAddr::new(*ip, self.port)].into_iter()), + Host::Domain(domain) => (domain.as_str(), self.port).to_socket_addrs(), + } + } +} + +/// A full Electrum server URL: a connection [`Scheme`] and a [`ServerAddr`]. +/// +/// Parses from `"[scheme://]host:port"`. A missing scheme defaults to [`Scheme::Tcp`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ServerUrl { + scheme: Scheme, + addr: ServerAddr, +} + +impl ServerUrl { + /// Creates a new `ServerUrl` from a [`Scheme`] and [`ServerAddr`]. + pub fn new(scheme: Scheme, addr: ServerAddr) -> Self { + Self { scheme, addr } + } + + /// The connection scheme of this URL. + pub fn scheme(&self) -> Scheme { + self.scheme + } + + /// The server address (host and port) of this URL. + pub fn addr(&self) -> &ServerAddr { + &self.addr + } +} + +impl fmt::Display for ServerUrl { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}://{}", self.scheme, self.addr) + } +} + +impl FromStr for ServerUrl { + type Err = ParseServerAddrError; + + fn from_str(s: &str) -> Result { + let invalid = || ParseServerAddrError(s.to_string()); + let (scheme, addr_str) = match s.split_once("://") { + Some((scheme_str, addr_str)) => (scheme_str.parse().map_err(|_| invalid())?, addr_str), + None => (Scheme::Tcp, s), + }; + Ok(Self { + scheme, + addr: addr_str.parse().map_err(|_| invalid())?, + }) + } +} + +/// An error parsing a [`Scheme`], [`ServerAddr`] or [`ServerUrl`] from a string. +/// +/// The payload is the full input string that failed to parse. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ParseServerAddrError(pub String); + +impl fmt::Display for ParseServerAddrError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "invalid server address '{}'", self.0) + } +} + +impl std::error::Error for ParseServerAddrError {} + +/// Configuration for [`crate::client::BlockingClient::connect`] / +/// [`crate::client::AsyncClient::connect`]. +/// +/// Use [`ConnectConfig::builder`] to construct. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ConnectConfig { + /// Timeout for establishing the connection (`None` = no limit). + timeout: Option, + /// Whether to validate the server's TLS certificate against the domain (TLS only). + validate_domain: bool, +} + +impl ConnectConfig { + /// Returns a [`ConnectConfigBuilder`] with default values. + pub fn builder() -> ConnectConfigBuilder { + ConnectConfigBuilder::default() + } + + /// Timeout for establishing the connection. + /// + /// `None` means no limit. + pub fn timeout(&self) -> Option { + self.timeout + } + + /// Whether to validate the server's TLS certificate against the domain. + /// + /// This only applies to TLS connections and is ignored for plain TCP. Defaults to `true`. + pub fn validate_domain(&self) -> bool { + self.validate_domain + } +} + +impl Default for ConnectConfig { + fn default() -> Self { + Self { + timeout: None, + validate_domain: true, + } + } +} + +/// A builder for [`ConnectConfig`], obtained via [`ConnectConfig::builder`]. +#[derive(Debug, Clone, Default)] +pub struct ConnectConfigBuilder { + config: ConnectConfig, +} + +impl ConnectConfigBuilder { + /// Sets the connection timeout. See [`ConnectConfig::timeout`]. + pub fn timeout(mut self, timeout: Option) -> Self { + self.config.timeout = timeout; + self + } + + /// Sets whether to validate the server's TLS certificate against the domain. See + /// [`ConnectConfig::validate_domain`]. + pub fn validate_domain(mut self, validate_domain: bool) -> Self { + self.config.validate_domain = validate_domain; + self + } + + /// Builds the [`ConnectConfig`]. + pub fn build(self) -> ConnectConfig { + self.config + } +} + +/// Error establishing an Electrum connection. +#[non_exhaustive] +#[derive(Debug)] +pub enum ConnectError { + /// Transport or socket failure, including DNS, TCP, timeout, and handshake EOF/reset. + Io(std::io::Error), + /// TLS configuration, protocol, or certificate-validation failure. + #[cfg(feature = "ssl")] + Tls(TlsError), + /// The URL uses `ssl://` but the crate was built without the `ssl` feature. + SslUnsupported, +} + +#[cfg(feature = "ssl")] +impl ConnectError { + /// Returns the certificate-validation error, if present. + pub fn certificate_error(&self) -> Option<&rustls::CertificateError> { + match self { + ConnectError::Tls(e) => e.certificate_error(), + _ => None, + } + } + + /// Classifies embedded rustls errors as TLS failures and preserves other I/O errors. + fn from_handshake_io(err: std::io::Error) -> Self { + match err + .get_ref() + .and_then(|inner| inner.downcast_ref::()) + { + Some(tls) => ConnectError::Tls(TlsError::Rustls(tls.clone())), + None => ConnectError::Io(err), + } + } +} + +impl fmt::Display for ConnectError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + ConnectError::Io(e) => write!(f, "connection I/O error: {e}"), + #[cfg(feature = "ssl")] + ConnectError::Tls(e) => write!(f, "{e}"), + ConnectError::SslUnsupported => { + write!(f, "the 'ssl://' scheme requires the 'ssl' feature") + } + } + } +} + +impl std::error::Error for ConnectError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + ConnectError::Io(e) => Some(e), + #[cfg(feature = "ssl")] + ConnectError::Tls(e) => Some(e), + ConnectError::SslUnsupported => None, + } + } +} + +impl From for ConnectError { + fn from(e: std::io::Error) -> Self { + ConnectError::Io(e) + } +} + +#[cfg(feature = "ssl")] +impl From for ConnectError { + fn from(e: TlsError) -> Self { + ConnectError::Tls(e) + } +} + +/// Error in the TLS layer of a connection: configuration, protocol, or +/// certificate-validation failures. Transport failures are [`ConnectError::Io`]. +#[cfg(feature = "ssl")] +#[non_exhaustive] +#[derive(Debug)] +pub enum TlsError { + /// TLS protocol or certificate-validation failure. + Rustls(rustls::Error), + /// `validate_domain` is true and the host is [`Host::Ip`]. + MissingDomain, + /// Host string is not a valid TLS [`rustls::pki_types::ServerName`]. + InvalidServerName(String), +} + +#[cfg(feature = "ssl")] +impl TlsError { + /// Server certificate rejected by the local verifier. + pub fn certificate_error(&self) -> Option<&rustls::CertificateError> { + match self { + Self::Rustls(rustls::Error::InvalidCertificate(error)) => Some(error), + _ => None, + } + } +} + +#[cfg(feature = "ssl")] +impl fmt::Display for TlsError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + TlsError::MissingDomain => { + write!(f, "TLS certificate validation requires a domain name") + } + TlsError::InvalidServerName(name) => { + write!(f, "invalid TLS server name '{name}'") + } + TlsError::Rustls(e) => write!(f, "TLS error: {e}"), + } + } +} + +#[cfg(feature = "ssl")] +impl std::error::Error for TlsError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + TlsError::Rustls(e) => Some(e), + TlsError::MissingDomain | TlsError::InvalidServerName(_) => None, + } + } +} + +#[cfg(feature = "ssl")] +impl From for TlsError { + fn from(e: rustls::Error) -> Self { + TlsError::Rustls(e) + } +} + +#[cfg(feature = "ssl")] +impl From for ConnectError { + fn from(e: rustls::Error) -> Self { + ConnectError::Tls(TlsError::Rustls(e)) + } +} + +/// Process-default provider if the application installed one; otherwise aws-lc-rs. +#[cfg(feature = "ssl")] +fn crypto_provider() -> Arc { + rustls::crypto::CryptoProvider::get_default() + .cloned() + .unwrap_or_else(|| Arc::new(rustls::crypto::aws_lc_rs::default_provider())) +} + +#[cfg(feature = "ssl")] +fn client_config(validate_domain: bool) -> Result, rustls::Error> { + let provider = crypto_provider(); + let builder = rustls::ClientConfig::builder_with_provider(Arc::clone(&provider)) + .with_safe_default_protocol_versions()?; + let config = if validate_domain { + let roots = rustls::RootCertStore { + roots: webpki_roots::TLS_SERVER_ROOTS.to_vec(), + }; + builder.with_root_certificates(roots).with_no_client_auth() + } else { + builder + .dangerous() + .with_custom_certificate_verifier(Arc::new(danger::NoCertificateVerification::new( + (*provider).clone(), + ))) + .with_no_client_auth() + }; + Ok(Arc::new(config)) +} + +#[cfg(feature = "ssl")] +fn server_name( + addr: &ServerAddr, + validate_domain: bool, +) -> Result, TlsError> { + match addr.host() { + Host::Domain(domain) => rustls::pki_types::ServerName::try_from(domain.clone()) + .map_err(|_| TlsError::InvalidServerName(domain.clone())), + Host::Ip(ip) => { + if validate_domain { + Err(TlsError::MissingDomain) + } else { + Ok(rustls::pki_types::ServerName::from(*ip)) + } + } + } +} + +#[cfg(feature = "ssl")] +mod danger { + use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified}; + use rustls::crypto::CryptoProvider; + use rustls::pki_types::{CertificateDer, ServerName, UnixTime}; + use rustls::DigitallySignedStruct; + + #[derive(Debug)] + pub struct NoCertificateVerification(CryptoProvider); + + impl NoCertificateVerification { + pub fn new(provider: CryptoProvider) -> Self { + Self(provider) + } + } + + impl rustls::client::danger::ServerCertVerifier for NoCertificateVerification { + fn verify_server_cert( + &self, + _end_entity: &CertificateDer<'_>, + _intermediates: &[CertificateDer<'_>], + _server_name: &ServerName<'_>, + _ocsp: &[u8], + _now: UnixTime, + ) -> Result { + Ok(ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + _message: &[u8], + _cert: &CertificateDer<'_>, + _dss: &DigitallySignedStruct, + ) -> Result { + Ok(HandshakeSignatureValid::assertion()) + } + + fn verify_tls13_signature( + &self, + _message: &[u8], + _cert: &CertificateDer<'_>, + _dss: &DigitallySignedStruct, + ) -> Result { + Ok(HandshakeSignatureValid::assertion()) + } + + fn supported_verify_schemes(&self) -> Vec { + self.0.signature_verification_algorithms.supported_schemes() + } + } +} + +/// Blocking (std I/O) transport constructors. +pub mod blocking { + use std::io; + use std::net::{SocketAddr, TcpStream, ToSocketAddrs}; + use std::time::Duration; + #[cfg(feature = "ssl")] + use std::time::Instant; + + use super::ServerAddr; + + /// Connects to `addr` over plaintext TCP using blocking I/O. + /// + /// `timeout` bounds TCP connection, not DNS. No read/write timeout + /// on the returned stream. + pub fn connect_tcp(addr: &ServerAddr, timeout: Option) -> io::Result { + let addrs: Vec<_> = addr.to_socket_addrs()?.collect(); + match timeout { + Some(timeout) => connect_with_total_timeout(&addrs, timeout), + None => TcpStream::connect(addrs.as_slice()), + } + } + + #[cfg(feature = "ssl")] + pub use super::tls::{TlsReadHalf, TlsStream, TlsWriteHalf}; + + /// Connects to `addr` over TLS using blocking I/O. + /// + /// `timeout` is a single deadline covering TCP connect and the TLS handshake. + /// The stream can be used as a single `Read + Write`, or split with + /// [`TlsStream::into_split`] so a reader and writer can run on separate threads. + #[cfg(feature = "ssl")] + pub fn connect_ssl( + addr: &ServerAddr, + validate_domain: bool, + timeout: Option, + ) -> Result { + let deadline = timeout.map(|timeout| Instant::now() + timeout); + let server_name = super::server_name(addr, validate_domain)?; + let mut tcp = connect_tcp(addr, timeout).map_err(super::ConnectError::Io)?; + + let mut conn = + rustls::ClientConnection::new(super::client_config(validate_domain)?, server_name)?; + conn.complete_io(&mut HandshakeIo { + tcp: &mut tcp, + deadline, + }) + .map_err(super::ConnectError::from_handshake_io)?; + + tcp.set_read_timeout(None) + .map_err(super::ConnectError::Io)?; + tcp.set_write_timeout(None) + .map_err(super::ConnectError::Io)?; + TlsStream::new(conn, tcp).map_err(super::ConnectError::Io) + } + + /// Remaining time until `deadline`, or `TimedOut` if it has passed. + #[cfg(feature = "ssl")] + fn remaining_timeout(deadline: Instant) -> io::Result { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + Err(io::Error::new( + io::ErrorKind::TimedOut, + "TLS handshake timed out", + )) + } else { + Ok(remaining) + } + } + + /// `Read + Write` adapter for rustls `complete_io`, enforcing a single + /// deadline for the whole handshake. + /// + /// Without it, socket timeouts would apply per operation, so the handshake + /// could take several times longer than the intended deadline. + #[cfg(feature = "ssl")] + struct HandshakeIo<'a> { + tcp: &'a mut TcpStream, + deadline: Option, + } + + /// Make socket timeouts readable as `TimedOut` rather than rustls's + /// bare `WouldBlock` propagation. + #[cfg(feature = "ssl")] + fn map_timeout(deadline: Option, result: io::Result) -> io::Result { + match result { + Err(e) if e.kind() == io::ErrorKind::WouldBlock && deadline.is_some() => { + Err(io::Error::new(io::ErrorKind::TimedOut, e)) + } + other => other, + } + } + + #[cfg(feature = "ssl")] + impl HandshakeIo<'_> { + fn apply_deadline(&mut self) -> io::Result<()> { + let Some(deadline) = self.deadline else { + return Ok(()); + }; + let remaining = remaining_timeout(deadline)?; + self.tcp.set_read_timeout(Some(remaining))?; + self.tcp.set_write_timeout(Some(remaining))?; + Ok(()) + } + } + + #[cfg(feature = "ssl")] + impl io::Read for HandshakeIo<'_> { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.apply_deadline()?; + let result = self.tcp.read(buf); + map_timeout(self.deadline, result) + } + } + + #[cfg(feature = "ssl")] + impl io::Write for HandshakeIo<'_> { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.apply_deadline()?; + let result = self.tcp.write(buf); + map_timeout(self.deadline, result) + } + + fn flush(&mut self) -> io::Result<()> { + self.apply_deadline()?; + let result = self.tcp.flush(); + map_timeout(self.deadline, result) + } + } + + /// Tries each addr, splitting `timeout` across attempts. + fn connect_with_total_timeout( + addrs: &[SocketAddr], + mut timeout: Duration, + ) -> io::Result { + // Use the same algorithm as curl: 1/2 of the timeout on the first address, 1/4 on the + // second one, etc. https://curl.se/mail/lib-2014-11/0164.html + let mut last_err = None; + for (index, addr) in addrs.iter().enumerate() { + if index < addrs.len() - 1 { + timeout = timeout.div_f32(2.0); + } + match TcpStream::connect_timeout(addr, timeout) { + Ok(stream) => return Ok(stream), + Err(err) => last_err = Some(err), + } + } + Err(last_err.unwrap_or_else(|| { + io::Error::new( + io::ErrorKind::InvalidInput, + "could not resolve to any addresses", + ) + })) + } +} + +/// Tokio-based async transport constructors. +#[cfg(feature = "tokio")] +pub mod tokio { + use std::io; + use std::net::SocketAddr; + use std::time::Duration; + + use tokio::net::TcpStream; + + use super::{Host, ServerAddr}; + + /// Connects to `addr` over plaintext TCP using the Tokio runtime. + /// + /// `timeout` bounds DNS and TCP connection. + pub async fn connect_tcp( + addr: &ServerAddr, + timeout: Option, + ) -> io::Result { + let connect_fut = async { + match addr.host() { + Host::Domain(domain) => TcpStream::connect((domain.as_str(), addr.port())).await, + Host::Ip(ip) => TcpStream::connect(SocketAddr::new(*ip, addr.port())).await, + } + }; + match timeout { + Some(timeout) => match tokio::time::timeout(timeout, connect_fut).await { + Ok(res) => res, + Err(_elapsed) => Err(io::Error::new( + io::ErrorKind::TimedOut, + format!("connection to '{}' timed out", addr), + )), + }, + None => connect_fut.await, + } + } + + /// Connects to `addr` over TLS using the Tokio runtime. + /// + /// `timeout` bounds DNS, TCP connect, and the TLS handshake. + /// `validate_domain` requires a [`super::Host::Domain`]. + #[cfg(feature = "ssl")] + pub async fn connect_ssl( + addr: &ServerAddr, + validate_domain: bool, + timeout: Option, + ) -> Result, super::ConnectError> { + let server_name = super::server_name(addr, validate_domain)?; + let connector = tokio_rustls::TlsConnector::from(super::client_config(validate_domain)?); + let handshake = async { + let tcp = connect_tcp(addr, None) + .await + .map_err(super::ConnectError::Io)?; + connector + .connect(server_name, tcp) + .await + .map_err(super::ConnectError::from_handshake_io) + }; + match timeout { + Some(timeout) => match tokio::time::timeout(timeout, handshake).await { + Ok(res) => res, + Err(_elapsed) => Err(super::ConnectError::Io(io::Error::new( + io::ErrorKind::TimedOut, + format!("connection to '{}' timed out", addr), + ))), + }, + None => handshake.await, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; + + fn addr(s: &str) -> ServerAddr { + s.parse().unwrap_or_else(|e| panic!("{s:?}: {e}")) + } + + fn url(s: &str) -> ServerUrl { + s.parse().unwrap_or_else(|e| panic!("{s:?}: {e}")) + } + + #[test] + fn server_addr_parse() { + let a = addr("127.0.0.1:50001"); + assert_eq!(a.host(), &Host::Ip(IpAddr::V4(Ipv4Addr::LOCALHOST))); + assert_eq!(a.port(), 50001); + assert_eq!(a.to_string(), "127.0.0.1:50001"); + + let a = addr("localhost:50001"); + assert_eq!(a.host(), &Host::Domain("localhost".into())); + assert_eq!(a.port(), 50001); + + let a = addr("[::1]:50001"); + assert_eq!(a.host(), &Host::Ip(IpAddr::V6(Ipv6Addr::LOCALHOST))); + assert_eq!(a.port(), 50001); + assert_eq!(a.to_string(), "[::1]:50001"); + + for bad in [ + "tcp://127.0.0.1:50001", + "ssl://host:50002", + "::1:50001", + ":50001", + "host", + "host:", + "host:99999", + "[::1]50001", + "[example.com]:50001", + ] { + assert!(bad.parse::().is_err(), "{bad}"); + } + } + + #[test] + fn server_url_parse() { + let u = url("127.0.0.1:50001"); + assert_eq!(u.scheme(), Scheme::Tcp); + assert_eq!(u.addr(), &addr("127.0.0.1:50001")); + assert_eq!(u.to_string(), "tcp://127.0.0.1:50001"); + + let u = url("tcp://electrum.example.com:50001"); + assert_eq!(u.scheme(), Scheme::Tcp); + assert_eq!( + u.addr().host(), + &Host::Domain("electrum.example.com".into()) + ); + + let u = url("ssl://[::1]:50002"); + assert_eq!(u.scheme(), Scheme::Ssl); + assert_eq!(u.addr().port(), 50002); + assert_eq!(u.to_string(), "ssl://[::1]:50002"); + + for bad in ["http://host:80", "TCP://host:1", "ssl://", "tcp://host"] { + assert!(bad.parse::().is_err(), "{bad}"); + } + } +} diff --git a/src/transport/tls.rs b/src/transport/tls.rs new file mode 100644 index 0000000..4a5a422 --- /dev/null +++ b/src/transport/tls.rs @@ -0,0 +1,308 @@ +//! Split-capable blocking TLS stream built on rustls. +//! +//! `rustls::StreamOwned` cannot provide independent read and write halves. This adapter shares +//! the rustls connection state but releases its lock before blocking on socket I/O, allowing +//! reads and writes to progress independently. +//! +//! When both locks are needed, acquire TLS output before the rustls connection state. + +use std::io::{self, Read, Write}; +use std::net::{Shutdown, TcpStream}; +use std::sync::{Arc, Mutex, MutexGuard}; + +const CIPHERTEXT_CHUNK: usize = 16 * 1024; + +#[derive(Debug)] +struct TlsOutput { + /// Destination for queued TLS ciphertext. + writer: W, + /// TLS ciphertext from rustls before sending it to `writer`. + /// `offset` marks the already-written prefix. + pending_ciphertext: Vec, + /// Number of leading bytes in `pending_ciphertext` already accepted by `writer`. + offset: usize, +} + +impl TlsOutput { + fn new(writer: W) -> Self { + Self { + writer, + pending_ciphertext: Vec::new(), + offset: 0, + } + } + + /// Queues new ciphertext and writes all pending ciphertext. + fn queue_and_write_ciphertext(&mut self, ciphertext: &[u8]) -> io::Result<()> { + self.pending_ciphertext.extend_from_slice(ciphertext); + while self.offset < self.pending_ciphertext.len() { + match self.writer.write(&self.pending_ciphertext[self.offset..]) { + Ok(0) => return Err(io::Error::from(io::ErrorKind::WriteZero)), + Ok(n) => self.offset += n, + Err(err) if err.kind() == io::ErrorKind::Interrupted => continue, + Err(err) => return Err(err), + } + } + self.pending_ciphertext.clear(); + self.offset = 0; + Ok(()) + } + + fn flush(&mut self) -> io::Result<()> { + self.queue_and_write_ciphertext(&[])?; + self.writer.flush() + } +} + +#[derive(Debug)] +struct Shared { + /// TLS protocol state shared by the read and write halves. + tls_connection: Mutex, + /// Retains and sends TLS ciphertext produced by `tls_connection`. + /// The read half also writes here via `emit` for TLS alerts and KeyUpdate responses. + tls_output: Mutex>, + /// Extra handle used by either half's `Drop` to interrupt blocking socket I/O. + /// It avoids waiting on `tls_output`, which a blocked writer may hold locked. + shutdown_socket: TcpStream, +} + +impl Shared { + fn tls_connection(&self) -> io::Result> { + self.tls_connection.lock().map_err(|_poison| { + io::Error::new(io::ErrorKind::Other, "TLS connection state mutex poisoned") + }) + } + + fn tls_output(&self) -> io::Result>> { + self.tls_output + .lock() + .map_err(|_poison| io::Error::new(io::ErrorKind::Other, "TLS output mutex poisoned")) + } + + /// Sends queued TLS records, releasing TLS state first so reads can continue. + fn emit(&self) -> io::Result<()> { + let mut output = self.tls_output()?; + let ciphertext = { + let mut conn = self.tls_connection()?; + let mut ciphertext = Vec::new(); + drain_tls(&mut conn, &mut ciphertext)?; + ciphertext + }; + output.queue_and_write_ciphertext(&ciphertext)?; + output.flush() + } +} + +/// Blocking TLS stream over a TCP socket. +/// +/// Use [`into_split`](Self::into_split) to give the read and write threads independent halves. +/// A successful write may still have ciphertext queued; use [`Write::flush`] to report +/// pending socket errors. +#[derive(Debug)] +pub struct TlsStream { + reader: TlsReadHalf, + writer: TlsWriteHalf, +} + +/// Read half of a split [`TlsStream`]. +/// +/// Dropping this half shuts down the TCP socket so the writer unblocks. +#[derive(Debug)] +pub struct TlsReadHalf { + shared: Arc, + read_socket: TcpStream, + /// Inbound TCP bytes not yet consumed by `read_tls`. Incomplete records live in rustls. + pending_incoming: Vec, +} + +/// Write half of a split [`TlsStream`]. +/// +/// Dropping this half shuts down the TCP socket so the reader unblocks. +/// A successful write may still have ciphertext queued; use [`Write::flush`] to report +/// pending socket errors. +#[derive(Debug)] +pub struct TlsWriteHalf { + shared: Arc, +} + +impl TlsStream { + /// `conn` must already have completed the handshake. + pub(super) fn new(conn: rustls::ClientConnection, tcp: TcpStream) -> io::Result { + let read_socket = tcp.try_clone()?; + let write_socket = tcp.try_clone()?; + let shared = Arc::new(Shared { + tls_connection: Mutex::new(conn), + tls_output: Mutex::new(TlsOutput::new(write_socket)), + shutdown_socket: tcp, + }); + Ok(Self { + reader: TlsReadHalf { + shared: Arc::clone(&shared), + read_socket, + pending_incoming: Vec::new(), + }, + writer: TlsWriteHalf { shared }, + }) + } + + /// Splits this stream into independent read and write halves. + pub fn into_split(self) -> (TlsReadHalf, TlsWriteHalf) { + (self.reader, self.writer) + } +} + +impl Read for TlsStream { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.reader.read(buf) + } +} + +impl Write for TlsStream { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.writer.write(buf) + } + + fn flush(&mut self) -> io::Result<()> { + self.writer.flush() + } +} + +impl Read for TlsReadHalf { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + if buf.is_empty() { + return Ok(0); + } + loop { + if let Some(n) = self.try_read_plaintext(buf)? { + return Ok(n); + } + if self.pending_incoming.is_empty() && !self.read_ciphertext()? { + self.process_pending_incoming()?; + return match self.try_read_plaintext(buf)? { + Some(n) => Ok(n), + None => Err(io::Error::from(io::ErrorKind::UnexpectedEof)), + }; + } + self.process_pending_incoming()?; + } + } +} + +impl TlsReadHalf { + /// Reads buffered plaintext without touching the socket. + /// `None` needs more ciphertext; `Some(0)` is clean TLS EOF. + fn try_read_plaintext(&mut self, buf: &mut [u8]) -> io::Result> { + let mut conn = self.shared.tls_connection()?; + match conn.reader().read(buf) { + Err(err) if err.kind() == io::ErrorKind::WouldBlock => Ok(None), + other => other.map(Some), + } + } + + /// Reads TLS ciphertext into `pending_incoming`; returns `false` on TCP EOF. + fn read_ciphertext(&mut self) -> io::Result { + let mut buf = [0u8; CIPHERTEXT_CHUNK]; + loop { + match self.read_socket.read(&mut buf) { + Ok(0) => return Ok(false), + Ok(n) => { + self.pending_incoming.extend_from_slice(&buf[..n]); + return Ok(true); + } + Err(err) if err.kind() == io::ErrorKind::Interrupted => continue, + Err(err) => return Err(err), + } + } + } + + /// Feeds `pending_incoming` (or TCP EOF) into rustls and sends generated TLS responses. + fn process_pending_incoming(&mut self) -> io::Result<()> { + let mut protocol_err = None; + let wants_write = { + let mut conn = self.shared.tls_connection()?; + let n = { + let mut input = self.pending_incoming.as_slice(); + conn.read_tls(&mut input)? + }; + if n == 0 { + self.pending_incoming.clear(); + } else { + self.pending_incoming.drain(..n); + } + match conn.process_new_packets() { + Ok(_) => {} + Err(err) => protocol_err = Some(err), + } + // Make any queued TLS 1.3 KeyUpdate response available to `write_tls` + // without sending application data. + let _ = conn.writer().write(&[])?; + conn.wants_write() + }; + if wants_write { + let emitted = self.shared.emit(); + if protocol_err.is_none() { + emitted?; + } + } + match protocol_err { + Some(err) => Err(io::Error::new(io::ErrorKind::InvalidData, err)), + None => Ok(()), + } + } +} + +impl Drop for TlsReadHalf { + fn drop(&mut self) { + let _ = self.shared.shutdown_socket.shutdown(Shutdown::Both); + } +} + +impl Write for TlsWriteHalf { + fn write(&mut self, buf: &[u8]) -> io::Result { + if buf.is_empty() { + return Ok(0); + } + let mut output = self.shared.tls_output()?; + let prior_ciphertext = { + let mut conn = self.shared.tls_connection()?; + let mut ciphertext = Vec::new(); + drain_tls(&mut conn, &mut ciphertext)?; + ciphertext + }; + output.queue_and_write_ciphertext(&prior_ciphertext)?; + + let (ciphertext, n) = { + let mut conn = self.shared.tls_connection()?; + let mut ciphertext = Vec::new(); + let n = conn.writer().write(buf)?; + drain_tls(&mut conn, &mut ciphertext)?; + (ciphertext, n) + }; + // Plaintext accepted by rustls must be reported as written. + match output.queue_and_write_ciphertext(&ciphertext) { + Ok(()) => Ok(n), + Err(_err) if n > 0 => Ok(n), + Err(err) => Err(err), + } + } + + fn flush(&mut self) -> io::Result<()> { + self.shared.emit() + } +} + +impl Drop for TlsWriteHalf { + fn drop(&mut self) { + let _ = self.shared.shutdown_socket.shutdown(Shutdown::Both); + } +} + +/// Drains queued ciphertext so it can be written after releasing TLS state. +fn drain_tls(conn: &mut rustls::ClientConnection, out: &mut Vec) -> io::Result<()> { + while conn.wants_write() { + if conn.write_tls(out)? == 0 { + break; + } + } + Ok(()) +} diff --git a/tests/synopsis.rs b/tests/synopsis.rs index 129a605..a1d7b0b 100644 --- a/tests/synopsis.rs +++ b/tests/synopsis.rs @@ -1,206 +1,200 @@ +#![cfg(feature = "tokio")] + use std::time::Duration; -use async_std::{net::TcpStream, stream::StreamExt}; use bdk_testenv::{anyhow, bitcoincore_rpc::RpcApi, TestEnv}; use bitcoin::Amount; use electrum_streaming_client::{ - notification::Notification, request, AsyncClient, CompletedRequest, Event, -}; -use futures::{ - executor::{block_on, ThreadPool}, - task::SpawnExt, - AsyncReadExt, FutureExt, + notification::Notification, request, AsyncClient, CompletedRequest, ConnectConfig, Event, + ServerUrl, }; +use futures::{FutureExt, StreamExt}; -#[test] -fn synopsis() -> anyhow::Result<()> { +#[tokio::test] +async fn synopsis() -> anyhow::Result<()> { let env = TestEnv::new()?; - let electrum_addr = env.electrsd.electrum_url.clone(); - println!("URL: {}", electrum_addr); + let url: ServerUrl = format!("tcp://{}", env.electrsd.electrum_url).parse()?; + println!("URL: {url}"); let wallet_addr = env .rpc_client() .get_new_address(None, None)? .assume_checked(); - let pool = ThreadPool::new()?; - block_on(async { - let stream = TcpStream::connect(electrum_addr.as_str()).await?; - let (read_stream, write_strean) = stream.split(); - let (client, mut event_rx, run_fut) = AsyncClient::new(read_stream, write_strean); - let run_handle = pool.spawn_with_handle(run_fut)?; - - client.send_event_request(request::HeadersSubscribe)?; - client.send_event_request(request::ScriptHashSubscribe::from_script( - wallet_addr.script_pubkey(), - ))?; - assert!(matches!( - event_rx.next().await, - Some(Event::Response(CompletedRequest::HeadersSubscribe { .. })) - )); - assert!(matches!( - event_rx.next().await, - Some(Event::Response( - CompletedRequest::ScriptHashSubscribe { .. } - )) - )); - - const TO_MINE: usize = 3; - let blockhashes = env.mine_blocks(TO_MINE, Some(wallet_addr.clone()))?; - println!("MINED: {:?}", blockhashes); - while let Some(event) = event_rx.next().await { - if let Event::Notification(Notification::Header(n)) = event { - if n.height() > TO_MINE as u32 { - break; - } + let (client, mut event_rx, run_fut) = + AsyncClient::connect(&url, &ConnectConfig::default()).await?; + let run_handle = tokio::spawn(run_fut); + + client.send_event_request(request::HeadersSubscribe)?; + client.send_event_request(request::ScriptHashSubscribe::from_script( + wallet_addr.script_pubkey(), + ))?; + assert!(matches!( + event_rx.next().await, + Some(Event::Response(CompletedRequest::HeadersSubscribe { .. })) + )); + assert!(matches!( + event_rx.next().await, + Some(Event::Response( + CompletedRequest::ScriptHashSubscribe { .. } + )) + )); + + const TO_MINE: usize = 3; + let blockhashes = env.mine_blocks(TO_MINE, Some(wallet_addr.clone()))?; + println!("MINED: {:?}", blockhashes); + while let Some(event) = event_rx.next().await { + if let Event::Notification(Notification::Header(n)) = event { + if n.height() > TO_MINE as u32 { + break; } } + } - assert_eq!( - client - .send_request(request::HeaderWithProof { - height: 3, - cp_height: 3 - }) - .await? - .header, - { - let blockhash = env.rpc_client().get_block_hash(3)?; - env.rpc_client().get_block_header(&blockhash)? - }, - "header at height must match" - ); - - println!( - "HEADERS: {:?}", - client - .send_request(request::Headers { - start_height: 1, - count: 2, - }) - .await? - ); - - // Make unconfirmed balance. - env.mine_blocks(101, Some(wallet_addr.clone()))?; // create spendable balance - let txid = env.rpc_client().send_to_address( - &wallet_addr, - Amount::from_btc(1.0).unwrap(), - None, - None, - None, - None, - None, - None, - )?; - env.wait_until_electrum_sees_txid(txid, Duration::from_secs(10))?; - - let tx_resp = client.send_request(request::GetTx { txid }).await?; - println!("GOT TX: {:?}", tx_resp); - println!( - "BROADCAST RESULT: {}", - client - .send_request(request::BroadcastTx(tx_resp.tx)) - .await? - ); - - println!( - "GET BALANCE RESP: {:?}", - client - .send_request(request::GetBalance::from_script( - wallet_addr.script_pubkey(), - )) - .await? - ); - - let history_resp = client - .send_request(request::GetHistory::from_script( - wallet_addr.script_pubkey(), - )) - .await?; - println!( - "GET HISTORY RESP: first = {:?} last = {:?}", - history_resp.first().unwrap(), - history_resp.last().unwrap() - ); - - let block_hash = env.mine_blocks(1, None)?.first().copied().unwrap(); - let block_height = env.rpc_client().get_block_info(&block_hash)?.height as u32; - env.wait_until_electrum_sees_block(Duration::from_secs(5))?; - - let tx_merkle = client - .send_request(request::GetTxMerkle { - txid, - height: block_height, + assert_eq!( + client + .send_request(request::HeaderWithProof { + height: 3, + cp_height: 3 }) - .await?; - println!("GET MERKLE: {:?}", tx_merkle); - - let from_pos = client - .send_request(request::GetTxidFromPos { - height: block_height, - tx_pos: tx_merkle.pos, + .await? + .header, + { + let blockhash = env.rpc_client().get_block_hash(3)?; + env.rpc_client().get_block_header(&blockhash)? + }, + "header at height must match" + ); + + println!( + "HEADERS: {:?}", + client + .send_request(request::Headers { + start_height: 1, + count: 2, }) - .await?; - println!("TXID FROM POS: {}", from_pos.txid); - assert_eq!(txid, from_pos.txid); - - // NOTE: This does not work with `electrs` - // let mempool_history = request::GetMempool::from_script(addr.script_pubkey()) - // .send(&req_tx)? - // .await??; - // println!("GET MEMPOOL RESP: {:?}", mempool_history); - - let utxos = client - .send_request(request::ListUnspent::from_script( + .await? + ); + + // Make unconfirmed balance. + env.mine_blocks(101, Some(wallet_addr.clone()))?; // create spendable balance + let txid = env.rpc_client().send_to_address( + &wallet_addr, + Amount::from_btc(1.0).unwrap(), + None, + None, + None, + None, + None, + None, + )?; + env.wait_until_electrum_sees_txid(txid, Duration::from_secs(10))?; + + let tx_resp = client.send_request(request::GetTx { txid }).await?; + println!("GOT TX: {:?}", tx_resp); + println!( + "BROADCAST RESULT: {}", + client + .send_request(request::BroadcastTx(tx_resp.tx)) + .await? + ); + + println!( + "GET BALANCE RESP: {:?}", + client + .send_request(request::GetBalance::from_script( wallet_addr.script_pubkey(), )) - .await?; - println!( - "GET UTXOs: first = {:?} last = {:?}", - utxos.first().unwrap(), - utxos.last().unwrap() - ); - - // NOTE: This does not work with our version of `electrs` - // let unsub_resp = request::ScriptHashUnsubscribe::from_script(addr.script_pubkey()) - // .send(&req_tx)? - // .await??; - // println!("UNSUB RESP: {:?}", unsub_resp); - - let fee_histogram = client.send_request(request::GetFeeHistogram).await?; - println!("FEE HISTOGRAM: {:?}", fee_histogram); - - let server_banner = client.send_request(request::Banner).await?; - println!("SERVER BANNER: {}", server_banner); - - client.send_request(request::Ping).await?; - println!("PING SUCCESS!"); - - // // NOTE: Batching does not work until https://github.com/Blockstream/electrs/pull/108 is - // // merged. - // - // let txid1 = env.send(&wallet_addr, Amount::from_btc(0.1)?)?; - // let txid2 = env.send(&wallet_addr, Amount::from_btc(0.1)?)?; - // env.mine_blocks(1, None)?; - // env.wait_until_electrum_sees_txid(txid1, Duration::from_secs(10))?; - // env.wait_until_electrum_sees_txid(txid2, Duration::from_secs(10))?; - // let mut batch = client.batch(); - // let tx1_fut = batch.request(request::GetTx(txid1)); - // let tx2_fut = batch.request(request::GetTx(txid2)); - // batch.send()?; - // let (tx1_res, tx2_res) = futures::join!(tx1_fut, tx2_fut); - // let (tx1, tx2) = (tx1_res?, tx2_res?); - // println!("Got tx1: {:?}", tx1); - // println!("Got tx2: {:?}", tx2); - - // read remaining events. - while let Some(event) = event_rx.next().now_or_never() { - println!("EVENT: {:?}", event); - } + .await? + ); - drop(client); - run_handle.await?; - Ok(()) - }) + let history_resp = client + .send_request(request::GetHistory::from_script( + wallet_addr.script_pubkey(), + )) + .await?; + println!( + "GET HISTORY RESP: first = {:?} last = {:?}", + history_resp.first().unwrap(), + history_resp.last().unwrap() + ); + + let block_hash = env.mine_blocks(1, None)?.first().copied().unwrap(); + let block_height = env.rpc_client().get_block_info(&block_hash)?.height as u32; + env.wait_until_electrum_sees_block(Duration::from_secs(5))?; + + let tx_merkle = client + .send_request(request::GetTxMerkle { + txid, + height: block_height, + }) + .await?; + println!("GET MERKLE: {:?}", tx_merkle); + + let from_pos = client + .send_request(request::GetTxidFromPos { + height: block_height, + tx_pos: tx_merkle.pos, + }) + .await?; + println!("TXID FROM POS: {}", from_pos.txid); + assert_eq!(txid, from_pos.txid); + + // NOTE: This does not work with `electrs` + // let mempool_history = request::GetMempool::from_script(addr.script_pubkey()) + // .send(&req_tx)? + // .await??; + // println!("GET MEMPOOL RESP: {:?}", mempool_history); + + let utxos = client + .send_request(request::ListUnspent::from_script( + wallet_addr.script_pubkey(), + )) + .await?; + println!( + "GET UTXOs: first = {:?} last = {:?}", + utxos.first().unwrap(), + utxos.last().unwrap() + ); + + // NOTE: This does not work with our version of `electrs` + // let unsub_resp = request::ScriptHashUnsubscribe::from_script(addr.script_pubkey()) + // .send(&req_tx)? + // .await??; + // println!("UNSUB RESP: {:?}", unsub_resp); + + let fee_histogram = client.send_request(request::GetFeeHistogram).await?; + println!("FEE HISTOGRAM: {:?}", fee_histogram); + + let server_banner = client.send_request(request::Banner).await?; + println!("SERVER BANNER: {}", server_banner); + + client.send_request(request::Ping).await?; + println!("PING SUCCESS!"); + + // // NOTE: Batching does not work until https://github.com/Blockstream/electrs/pull/108 is + // // merged. + // + // let txid1 = env.send(&wallet_addr, Amount::from_btc(0.1)?)?; + // let txid2 = env.send(&wallet_addr, Amount::from_btc(0.1)?)?; + // env.mine_blocks(1, None)?; + // env.wait_until_electrum_sees_txid(txid1, Duration::from_secs(10))?; + // env.wait_until_electrum_sees_txid(txid2, Duration::from_secs(10))?; + // let mut batch = client.batch(); + // let tx1_fut = batch.request(request::GetTx(txid1)); + // let tx2_fut = batch.request(request::GetTx(txid2)); + // batch.send()?; + // let (tx1_res, tx2_res) = futures::join!(tx1_fut, tx2_fut); + // let (tx1, tx2) = (tx1_res?, tx2_res?); + // println!("Got tx1: {:?}", tx1); + // println!("Got tx2: {:?}", tx2); + + // read remaining events. + while let Some(event) = event_rx.next().now_or_never() { + println!("EVENT: {:?}", event); + } + + drop(client); + run_handle.await??; + Ok(()) }