HMC/handmade-revolt-backend
6
0
Fork 0
forked from jmug/stoatchat
Code Pull Requests Actions Packages Releases 2 Activity

merge: pull request #301 from TheBobBobs/bonfire-redis

Browse Source 
fix(bonfire): random disconnects from redis
This commit is contained in:
Paul Makles
2024-02-06 13:18:55 +00:00
committed by GitHub
parent 09a848f5a6 308193377d
commit c451e431cd
4 changed files with 517 additions and 331 deletions
Download Patch File Download Diff File Expand all files Collapse all files

372
Cargo.lock generated
View File

@@ -104,6 +104,12 @@ version = "1.0.68"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2cb2f989d18dd141ab8ae82f64d1a8cdd37e0840f73a406896cf5e99502fab61"
[[package]]
name = "arc-swap"
version = "1.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bddcadddf5e9015d310179a59bb28c4d4b9920ad0f11e8e14dbadf654890c9a6"
[[package]]
name = "arrayref"
version = "0.3.6"
@@ -165,7 +171,7 @@ dependencies = [
"futures-lite",
"once_cell",
"tokio 0.2.25",
"tokio 1.18.2",
"tokio 1.35.1",
]
[[package]]
@@ -246,7 +252,7 @@ dependencies = [
"log",
"memchr",
"once_cell",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"pin-utils",
"slab",
"wasm-bindgen-futures",
@@ -315,7 +321,7 @@ dependencies = [
"futures-io",
"futures-util",
"log",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"tungstenite",
]
@@ -588,9 +594,19 @@ checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38"
[[package]]
name = "bytes"
version = "1.1.0"
version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c4872d67bab6358e59559027aa3b9157c53d9358c51423c17554809a8858e0f8"
checksum = "a2bd12c1caf447e69cd4528f47f94d203fd2582878ecb9e9465484c4148a8223"
[[package]]
name = "bytes-utils"
version = "0.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7dafe3a8757b027e2be6e4e5601ed563c55989fcf1546e933c66c8eb3a058d35"
dependencies = [
"bytes 1.5.0",
"either",
]
[[package]]
name = "cache-padded"
@@ -612,7 +628,7 @@ dependencies = [
"instant",
"once_cell",
"thiserror",
"tokio 1.18.2",
"tokio 1.35.1",
]
[[package]]
@@ -712,11 +728,11 @@ version = "4.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2a604e93b79d1808327a6fca85a6f2d69de66461e7620f5a4cbf5fb4d1d7c948"
dependencies = [
"bytes 1.1.0",
"bytes 1.5.0",
"futures-core",
"memchr",
"pin-project-lite 0.2.9",
"tokio 1.18.2",
"pin-project-lite 0.2.13",
"tokio 1.35.1",
"tokio-util 0.7.2",
]
@@ -784,6 +800,12 @@ dependencies = [
"version_check",
]
[[package]]
name = "cookie-factory"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "396de984970346b0d9e93d1415082923c679e5ae5c3ee3dcbd104f5610af126b"
[[package]]
name = "core-foundation"
version = "0.9.3"
@@ -809,6 +831,12 @@ dependencies = [
"libc",
]
[[package]]
name = "crc16"
version = "0.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "338089f42c427b86394a5ee60ff321da23a5c89c9d89514c829687b26359fcff"
[[package]]
name = "crossbeam-queue"
version = "0.3.5"
@@ -1002,7 +1030,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "16a2561fd313df162315935989dceb8c99db4ee1933358270a57a3cfb8c957f3"
dependencies = [
"crossbeam-queue",
"tokio 1.18.2",
"tokio 1.35.1",
]
[[package]]
@@ -1348,6 +1376,15 @@ dependencies = [
"winapi 0.3.9",
]
[[package]]
name = "float-cmp"
version = "0.9.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "98de4bbd547a563b716d8dfa9aad1cb19bfab00f4fa09a6a4ed21dbcf44ce9c4"
dependencies = [
"num-traits",
]
[[package]]
name = "fnv"
version = "1.0.7"
@@ -1371,14 +1408,40 @@ checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b"
[[package]]
name = "form_urlencoded"
version = "1.0.1"
version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5fc25a87fa4fd2094bffb06925852034d90a17f0d1e05197d4956d3555752191"
checksum = "e13624c2627564efccf4934284bdd98cbaa14e79b0b5a141218e507b3a823456"
dependencies = [
"matches",
"percent-encoding",
]
[[package]]
name = "fred"
version = "8.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3b2a2ac060e3266004c552235c241b481e438e2b1ea75715ea1176914ef2868"
dependencies = [
"arc-swap",
"async-trait",
"bytes 1.5.0",
"bytes-utils",
"crossbeam-queue",
"float-cmp",
"futures",
"lazy_static",
"log",
"parking_lot",
"rand 0.8.5",
"redis-protocol",
"semver 1.0.9",
"socket2 0.5.5",
"tokio 1.35.1",
"tokio-stream",
"tokio-util 0.7.2",
"url",
"urlencoding",
]
[[package]]
name = "fuchsia-cprng"
version = "0.1.1"
@@ -1460,7 +1523,7 @@ dependencies = [
"futures-io",
"memchr",
"parking",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"waker-fn",
]
@@ -1472,7 +1535,7 @@ checksum = "45ec6fe3675af967e67c5536c0b9d44e34e6c52f86bedc4ea49c5317b8e94d06"
dependencies = [
"futures-channel",
"futures-task",
"tokio 1.18.2",
"tokio 1.35.1",
]
[[package]]
@@ -1517,7 +1580,7 @@ dependencies = [
"futures-sink",
"futures-task",
"memchr",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"pin-utils",
"slab",
]
@@ -1661,7 +1724,7 @@ version = "0.3.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "37a82c6d637fc9515a4694bbf1cb2457b79d81ce52b3108bdeea58b07dd34a57"
dependencies = [
"bytes 1.1.0",
"bytes 1.5.0",
"fnv",
"futures-core",
"futures-sink",
@@ -1669,7 +1732,7 @@ dependencies = [
"http",
"indexmap",
"slab",
"tokio 1.18.2",
"tokio 1.35.1",
"tokio-util 0.7.2",
"tracing",
]
@@ -1802,7 +1865,7 @@ version = "0.2.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ff8670570af52249509a86f5e3e18a08c60b177071826898fde8997cf5f6bfbb"
dependencies = [
"bytes 1.1.0",
"bytes 1.5.0",
"fnv",
"itoa 1.0.2",
]
@@ -1823,9 +1886,9 @@ version = "0.4.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d5f38f16d184e36f2408a55281cd658ecbd3ca05cce6d6510a176eca393e26d1"
dependencies = [
"bytes 1.1.0",
"bytes 1.5.0",
"http",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
]
[[package]]
@@ -1885,7 +1948,7 @@ version = "0.14.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42dc3c131584288d375f2d07f822b0cb012d8c6fb899a5b9fdb3cb7eb9b6004f"
dependencies = [
"bytes 1.1.0",
"bytes 1.5.0",
"futures-channel",
"futures-core",
"futures-util",
@@ -1895,9 +1958,9 @@ dependencies = [
"httparse",
"httpdate 1.0.2",
"itoa 1.0.2",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"socket2 0.4.4",
"tokio 1.18.2",
"tokio 1.35.1",
"tower-service",
"tracing",
"want",
@@ -1922,10 +1985,10 @@ version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905"
dependencies = [
"bytes 1.1.0",
"bytes 1.5.0",
"hyper 0.14.19",
"native-tls",
"tokio 1.18.2",
"tokio 1.35.1",
"tokio-native-tls",
]
@@ -1946,6 +2009,16 @@ dependencies = [
"unicode-normalization",
]
[[package]]
name = "idna"
version = "0.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "634d9b1461af396cad843f47fdba5597a4f9e6ddd4bfb6ff5d85028c25cb12f6"
dependencies = [
"unicode-bidi",
"unicode-normalization",
]
[[package]]
name = "if_chain"
version = "1.0.2"
@@ -2183,7 +2256,7 @@ dependencies = [
"futures-util",
"hostname",
"httpdate 1.0.2",
"idna",
"idna 0.2.3",
"mime",
"native-tls",
"nom",
@@ -2194,9 +2267,9 @@ dependencies = [
[[package]]
name = "libc"
version = "0.2.126"
version = "0.2.153"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "349d5a591cd28b49e1d1037471617a32ddcda5731b99419008085f72d5a53836"
checksum = "9c198f91728a82281a64e1f4f9eeb25d82cb32a5de251c6bd1b5154d63a8e7bd"
[[package]]
name = "libgit2-sys"
@@ -2429,14 +2502,13 @@ dependencies = [
[[package]]
name = "mio"
version = "0.8.3"
version = "0.8.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "713d550d9b44d89174e066b7a6217ae06234c10cb47819a88290d2b353c31799"
checksum = "8f3d0b296e374a4e6f3c7b0a1f5a51d748a0d34c85e7dc48fc3fa9a87657fe09"
dependencies = [
"libc",
"log",
"wasi 0.11.0+wasi-snapshot-preview1",
"windows-sys",
"windows-sys 0.48.0",
]
[[package]]
@@ -2466,7 +2538,7 @@ dependencies = [
"log",
"metrics",
"thiserror",
"tokio 1.18.2",
"tokio 1.35.1",
"tracing",
"tracing-subscriber",
]
@@ -2518,7 +2590,7 @@ dependencies = [
"strsim",
"take_mut",
"thiserror",
"tokio 1.18.2",
"tokio 1.35.1",
"tokio-rustls",
"tokio-util 0.7.2",
"trust-dns-proto",
@@ -2534,7 +2606,7 @@ version = "2.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5f8f35e687561d5c1667590911e6698a8cb714a134a7505718a182e7bc9d3836"
dependencies = [
"bytes 1.1.0",
"bytes 1.5.0",
"encoding_rs",
"futures-util",
"http",
@@ -2543,7 +2615,7 @@ dependencies = [
"memchr",
"mime",
"spin 0.9.3",
"tokio 1.18.2",
"tokio 1.35.1",
"tokio-util 0.6.10",
"version_check",
]
@@ -2846,7 +2918,7 @@ dependencies = [
"libc",
"redox_syscall",
"smallvec",
"windows-sys",
"windows-sys 0.36.1",
]
[[package]]
@@ -2924,9 +2996,9 @@ dependencies = [
[[package]]
name = "percent-encoding"
version = "2.1.0"
version = "2.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e"
checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
[[package]]
name = "pest"
@@ -2999,9 +3071,9 @@ checksum = "257b64915a082f7811703966789728173279bdebb956b143dbcd23f6f970a777"
[[package]]
name = "pin-project-lite"
version = "0.2.9"
version = "0.2.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e0a7ae3ac2f1173085d398531c705756c94a4c56843785df85a60c1a0afac116"
checksum = "8afb450f006bf6385ca15ef45d71d2288452bc3683ce2e2cacc0d18e4be60b58"
[[package]]
name = "pin-utils"
@@ -3114,9 +3186,9 @@ dependencies = [
[[package]]
name = "proc-macro2"
version = "1.0.56"
version = "1.0.78"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2b63bdb0cd06f1f4dedf69b254734f9b45af66e4a031e42a7480257d9898b435"
checksum = "e2422ad645d89c99f8f3e6b88a9fdeca7fabeac836b1002371c4367c8f984aae"
dependencies = [
"unicode-ident",
]
@@ -3334,15 +3406,15 @@ source = "git+https://github.com/revoltchat/redis-rs?rev=1a41faf356fd21aebba71ce
dependencies = [
"async-std",
"async-trait",
"bytes 1.1.0",
"bytes 1.5.0",
"combine",
"futures-util",
"itoa 1.0.2",
"percent-encoding",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"ryu",
"sha1_smol",
"tokio 1.18.2",
"tokio 1.35.1",
"tokio-util 0.7.2",
"url",
]
@@ -3354,14 +3426,14 @@ source = "git+https://github.com/revoltchat/redis-rs?rev=f8ca28ab85da59d2ccde526
dependencies = [
"async-std",
"async-trait",
"bytes 1.1.0",
"bytes 1.5.0",
"combine",
"futures-util",
"itoa 1.0.2",
"percent-encoding",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"ryu",
"tokio 1.18.2",
"tokio 1.35.1",
"tokio-util 0.7.2",
"url",
]
@@ -3382,6 +3454,20 @@ dependencies = [
"serde_json",
]
[[package]]
name = "redis-protocol"
version = "4.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9c31deddf734dc0a39d3112e73490e88b61a05e83e074d211f348404cee4d2c6"
dependencies = [
"bytes 1.5.0",
"bytes-utils",
"cookie-factory",
"crc16",
"log",
"nom",
]
[[package]]
name = "redox_syscall"
version = "0.2.13"
@@ -3453,7 +3539,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "46a1f7aa4f35e5e8b4160449f51afc758f0ce6454315a9fa7d0d113e958c41eb"
dependencies = [
"base64 0.13.0",
"bytes 1.1.0",
"bytes 1.5.0",
"encoding_rs",
"futures-core",
"futures-util",
@@ -3469,11 +3555,11 @@ dependencies = [
"mime",
"native-tls",
"percent-encoding",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"serde",
"serde_json",
"serde_urlencoded",
"tokio 1.18.2",
"tokio 1.35.1",
"tokio-native-tls",
"url",
"wasm-bindgen",
@@ -3498,6 +3584,8 @@ version = "0.6.9"
dependencies = [
"async-std",
"async-tungstenite",
"bincode",
"fred",
"futures",
"log",
"once_cell",
@@ -3830,7 +3918,7 @@ dependencies = [
"atomic",
"atty",
"binascii",
"bytes 1.1.0",
"bytes 1.5.0",
"either",
"figment",
"futures",
@@ -3840,7 +3928,7 @@ dependencies = [
"multer",
"num_cpus",
"parking_lot",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"rand 0.8.5",
"ref-cast",
"rocket_codegen",
@@ -3850,7 +3938,7 @@ dependencies = [
"state",
"tempfile",
"time 0.3.17",
"tokio 1.18.2",
"tokio 1.35.1",
"tokio-stream",
"tokio-util 0.7.2",
"ubyte",
@@ -3933,14 +4021,14 @@ dependencies = [
"memchr",
"pear",
"percent-encoding",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"ref-cast",
"serde",
"smallvec",
"stable-pattern",
"state",
"time 0.3.17",
"tokio 1.18.2",
"tokio 1.35.1",
"uncased",
]
@@ -4088,7 +4176,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "88d6731146462ea25d9244b2ed5fd1d716d25c52e4d54aa4fb0f3c4e9854dbe2"
dependencies = [
"lazy_static",
"windows-sys",
"windows-sys 0.36.1",
]
[[package]]
@@ -4222,7 +4310,7 @@ dependencies = [
"sentry-debug-images",
"sentry-panic",
"sentry-tracing",
"tokio 1.18.2",
"tokio 1.35.1",
"ureq",
]
@@ -4523,6 +4611,16 @@ dependencies = [
"winapi 0.3.9",
]
[[package]]
name = "socket2"
version = "0.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7b5fac59a5cb5dd637972e5fca70daf0523c9067fcdc4842f053dae04a18f8e9"
dependencies = [
"libc",
"windows-sys 0.48.0",
]
[[package]]
name = "spin"
version = "0.5.2"
@@ -4783,33 +4881,32 @@ dependencies = [
[[package]]
name = "tokio"
version = "1.18.2"
version = "1.35.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4903bf0427cf68dddd5aa6a93220756f8be0c34fcfa9f5e6191e103e15a31395"
checksum = "c89b4efa943be685f629b149f53829423f8f5531ea21249408e8e2f8671ec104"
dependencies = [
"bytes 1.1.0",
"backtrace",
"bytes 1.5.0",
"libc",
"memchr",
"mio 0.8.3",
"mio 0.8.10",
"num_cpus",
"once_cell",
"parking_lot",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"signal-hook-registry",
"socket2 0.4.4",
"socket2 0.5.5",
"tokio-macros",
"winapi 0.3.9",
"windows-sys 0.48.0",
]
[[package]]
name = "tokio-macros"
version = "1.7.0"
version = "2.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b557f72f448c511a979e2564e55d74e6c4432fc96ff4f6241bc6bded342643b7"
checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
dependencies = [
"proc-macro2",
"quote 1.0.26",
"syn 1.0.107",
"syn 2.0.15",
]
[[package]]
@@ -4819,7 +4916,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f7d995660bd2b7f8c1568414c1126076c13fbb725c40112dc0120b78eb9b717b"
dependencies = [
"native-tls",
"tokio 1.18.2",
"tokio 1.35.1",
]
[[package]]
@@ -4829,7 +4926,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c43ee83903113e03984cb9e5cebe6c04a5116269e900e3ddba8f068a62adda59"
dependencies = [
"rustls",
"tokio 1.18.2",
"tokio 1.35.1",
"webpki",
]
@@ -4840,8 +4937,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "50145484efff8818b5ccd256697f36863f587da82cf8b409c53adf1e840798e3"
dependencies = [
"futures-core",
"pin-project-lite 0.2.9",
"tokio 1.18.2",
"pin-project-lite 0.2.13",
"tokio 1.35.1",
]
[[package]]
@@ -4874,12 +4971,12 @@ version = "0.6.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "36943ee01a6d67977dd3f84a5a1d2efeb4ada3a1ae771cadfaa535d9d9fc6507"
dependencies = [
"bytes 1.1.0",
"bytes 1.5.0",
"futures-core",
"futures-sink",
"log",
"pin-project-lite 0.2.9",
"tokio 1.18.2",
"pin-project-lite 0.2.13",
"tokio 1.35.1",
]
[[package]]
@@ -4888,12 +4985,12 @@ version = "0.7.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f988a1a1adc2fb21f9c12aa96441da33a1728193ae0b95d2be22dbd17fcb4e5c"
dependencies = [
"bytes 1.1.0",
"bytes 1.5.0",
"futures-core",
"futures-io",
"futures-sink",
"pin-project-lite 0.2.9",
"tokio 1.18.2",
"pin-project-lite 0.2.13",
"tokio 1.35.1",
"tracing",
]
@@ -4932,7 +5029,7 @@ checksum = "5d0ecdcb44a79f0fe9844f0c4f33a342cbcbb5117de8001e6ba0dc2351327d09"
dependencies = [
"cfg-if 1.0.0",
"log",
"pin-project-lite 0.2.9",
"pin-project-lite 0.2.13",
"tracing-attributes",
"tracing-core",
]
@@ -5010,7 +5107,7 @@ dependencies = [
"futures-channel",
"futures-io",
"futures-util",
"idna",
"idna 0.2.3",
"ipnet",
"lazy_static",
"log",
@@ -5018,7 +5115,7 @@ dependencies = [
"smallvec",
"thiserror",
"tinyvec",
"tokio 1.18.2",
"tokio 1.35.1",
"url",
]
@@ -5038,7 +5135,7 @@ dependencies = [
"resolv-conf",
"smallvec",
"thiserror",
"tokio 1.18.2",
"tokio 1.35.1",
"trust-dns-proto",
]
@@ -5056,7 +5153,7 @@ checksum = "d96a2dea40e7570482f28eb57afbe42d97551905da6a9400acc5c328d24004f5"
dependencies = [
"base64 0.13.0",
"byteorder",
"bytes 1.1.0",
"bytes 1.5.0",
"http",
"httparse",
"log",
@@ -5169,9 +5266,9 @@ dependencies = [
[[package]]
name = "unicode-bidi"
version = "0.3.8"
version = "0.3.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "099b7128301d285f79ddd55b9a83d5e6b9e97c92e0ea0daebee7263e932de992"
checksum = "08f95100a766bf4f8f28f90d77e0a5461bbdb219042e7679bebe79004fed8d75"
[[package]]
name = "unicode-ident"
@@ -5181,9 +5278,9 @@ checksum = "d22af068fba1eb5edcb4aea19d382b2a3deb4c8f9d475c589b6ada9e0fd493ee"
[[package]]
name = "unicode-normalization"
version = "0.1.19"
version = "0.1.22"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d54590932941a9e9266f0832deed84ebe1bf2e4c9e4a3554d393d18f5e854bf9"
checksum = "5c5713f0fc4b5db668a2ac63cdb7bb4469d8c9fed047b1d0292cc7b0ce2ba921"
dependencies = [
"tinyvec",
]
@@ -5243,13 +5340,12 @@ dependencies = [
[[package]]
name = "url"
version = "2.2.2"
version = "2.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a507c383b2d33b5fc35d1861e77e6b383d158b2da5e14fe51b83dfedf6fd578c"
checksum = "31e6302e3bb753d46e83516cae55ae196fc0c309407cf11ab35cc51a4c2a4633"
dependencies = [
"form_urlencoded",
"idna",
"matches",
"idna 0.5.0",
"percent-encoding",
"serde",
]
@@ -5263,6 +5359,12 @@ dependencies = [
"percent-encoding",
]
[[package]]
name = "urlencoding"
version = "2.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
[[package]]
name = "utf-8"
version = "0.7.6"
@@ -5295,7 +5397,7 @@ version = "0.15.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f07b0a1390e01c0fc35ebb26b28ced33c9a3808f7f9fbe94d3cc01e233bfeed5"
dependencies = [
"idna",
"idna 0.2.3",
"lazy_static",
"regex",
"serde",
@@ -5310,7 +5412,7 @@ version = "0.16.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32ad5bf234c7d3ad1042e5252b7eddb2c4669ee23f32c7dd0e9b7705f07ef591"
dependencies = [
"idna",
"idna 0.2.3",
"lazy_static",
"regex",
"serde",
@@ -5626,43 +5728,109 @@ version = "0.36.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ea04155a16a59f9eab786fe12a4a450e75cdb175f9e0d80da1e17db09f55b8d2"
dependencies = [
"windows_aarch64_msvc",
"windows_i686_gnu",
"windows_i686_msvc",
"windows_x86_64_gnu",
"windows_x86_64_msvc",
"windows_aarch64_msvc 0.36.1",
"windows_i686_gnu 0.36.1",
"windows_i686_msvc 0.36.1",
"windows_x86_64_gnu 0.36.1",
"windows_x86_64_msvc 0.36.1",
]
[[package]]
name = "windows-sys"
version = "0.48.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9"
dependencies = [
"windows-targets",
]
[[package]]
name = "windows-targets"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c"
dependencies = [
"windows_aarch64_gnullvm",
"windows_aarch64_msvc 0.48.5",
"windows_i686_gnu 0.48.5",
"windows_i686_msvc 0.48.5",
"windows_x86_64_gnu 0.48.5",
"windows_x86_64_gnullvm",
"windows_x86_64_msvc 0.48.5",
]
[[package]]
name = "windows_aarch64_gnullvm"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8"
[[package]]
name = "windows_aarch64_msvc"
version = "0.36.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9bb8c3fd39ade2d67e9874ac4f3db21f0d710bee00fe7cab16949ec184eeaa47"
[[package]]
name = "windows_aarch64_msvc"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc"
[[package]]
name = "windows_i686_gnu"
version = "0.36.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "180e6ccf01daf4c426b846dfc66db1fc518f074baa793aa7d9b9aaeffad6a3b6"
[[package]]
name = "windows_i686_gnu"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e"
[[package]]
name = "windows_i686_msvc"
version = "0.36.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e2e7917148b2812d1eeafaeb22a97e4813dfa60a3f8f78ebe204bcc88f12f024"
[[package]]
name = "windows_i686_msvc"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406"
[[package]]
name = "windows_x86_64_gnu"
version = "0.36.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4dcd171b8776c41b97521e5da127a2d86ad280114807d0b2ab1e462bc764d9e1"
[[package]]
name = "windows_x86_64_gnu"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e"
[[package]]
name = "windows_x86_64_gnullvm"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc"
[[package]]
name = "windows_x86_64_msvc"
version = "0.36.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c811ca4a8c853ef420abd8592ba53ddbbac90410fab6903b3e79972a631f7680"
[[package]]
name = "windows_x86_64_msvc"
version = "0.48.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538"
[[package]]
name = "winreg"
version = "0.7.0"

4
crates/bonfire/Cargo.toml
View File

@@ -18,6 +18,7 @@ querystring = "1.1.0"
revolt-quark = { path = "../quark" }
# serde
bincode = "1.3.3"
serde_json = "1.0.79"
rmp-serde = "1.0.0"
serde = "1.0.136"
@@ -34,3 +35,6 @@ async-std = { version = "1.8.0", features = [
# core
revolt-presence = { path = "../core/presence", features = ["redis-is-patched"] }
sentry = "0.31.5"
# redis
fred = { version = "8.0.1", features = ["subscriber-client"] }

6
crates/bonfire/src/main.rs
View File

@@ -29,6 +29,10 @@ async fn main() {
// Start accepting new connections and spawn a client for each connection.
while let Ok((stream, addr)) = listener.accept().await {
websocket::spawn_client(database::get_db(), stream, addr);
async_std::task::spawn(async move {
info!("User connected from {addr:?}");
websocket::client(database::get_db(), stream, addr).await;
info!("User disconnected from {addr:?}");
});
}
}

466
crates/bonfire/src/websocket.rs
View File

@@ -1,6 +1,16 @@
use std::net::SocketAddr;
use futures::{channel::oneshot, pin_mut, select, FutureExt, SinkExt, StreamExt, TryStreamExt};
use async_tungstenite::WebSocketStream;
use fred::{
interfaces::{ClientLike, EventInterface, PubsubInterface},
types::RedisConfig,
};
use futures::{
channel::oneshot,
pin_mut, select,
stream::{SplitSink, SplitStream},
FutureExt, SinkExt, StreamExt, TryStreamExt,
};
use revolt_presence::{create_session, delete_session};
use revolt_quark::{
events::{
@@ -9,249 +19,249 @@ use revolt_quark::{
state::{State, SubscriptionStateChange},
},
models::{user::UserHint, User},
redis_kiss, Database,
redis_kiss::{PayloadType, REDIS_PAYLOAD_TYPE, REDIS_URI},
Database,
};
use async_std::{net::TcpStream, sync::Mutex, task};
use async_std::{net::TcpStream, sync::Mutex};
use crate::config::WebsocketHandshakeCallback;
use crate::config::{ProtocolConfiguration, WebsocketHandshakeCallback};
/// Spawn a new WebSocket client worker given access to the database,
type WsReader = SplitStream<WebSocketStream<TcpStream>>;
type WsWriter = SplitSink<WebSocketStream<TcpStream>, async_tungstenite::tungstenite::Message>;
/// Start a new WebSocket client worker given access to the database,
/// the relevant TCP stream and the remote address of the client.
pub fn spawn_client(db: &'static Database, stream: TcpStream, addr: SocketAddr) {
// Spawn a new Async task to work on.
task::spawn(async move {
info!("User connected from {addr:?}");
pub async fn client(db: &'static Database, stream: TcpStream, addr: SocketAddr) {
// Upgrade the TCP connection to a WebSocket connection.
// In this process, we also parse any additional parameters given.
// e.g. wss://example.com?format=json&version=1
let (sender, receiver) = oneshot::channel();
let Ok(ws) = async_tungstenite::accept_hdr_async_with_config(
stream,
WebsocketHandshakeCallback::from(sender),
None,
)
.await
else {
return;
};
// Verify we've received a valid config, otherwise we should just drop the connection.
let Ok(mut config) = receiver.await else {
return;
};
info!(
"User {addr:?} provided protocol configuration (version = {}, format = {:?})",
config.get_protocol_version(),
config.get_protocol_format()
);
// Upgrade the TCP connection to a WebSocket connection.
// In this process, we also parse any additional parameters given.
// e.g. wss://example.com?format=json&version=1
let (sender, receiver) = oneshot::channel();
if let Ok(ws) = async_tungstenite::accept_hdr_async_with_config(
stream,
WebsocketHandshakeCallback::from(sender),
None,
)
// Split the socket for simultaneously read and write.
let (mut write, mut read) = ws.split();
// If the user has not provided authentication, request information.
if config.get_session_token().is_none() {
while let Ok(message) = read.try_next().await {
if let Ok(ClientMessage::Authenticate { token }) =
config.decode(message.as_ref().unwrap())
{
config.set_session_token(token);
break;
}
}
}
// Try to authenticate the user.
let Some(token) = config.get_session_token().as_ref() else {
return;
};
let user = match User::from_token(db, token, UserHint::Any).await {
Ok(user) => user,
Err(err) => {
write.send(config.encode(&err)).await.ok();
return;
}
};
info!("User {addr:?} authenticated as @{}", user.username);
// Create local state.
let mut state = State::from(user);
let user_id = state.cache.user_id.clone();
// Notify socket we have authenticated.
if write
.send(config.encode(&EventV1::Authenticated))
.await
{
// Verify we've received a valid config, otherwise we should just drop the connection.
if let Ok(mut config) = receiver.await {
info!(
"User {addr:?} provided protocol configuration (version = {}, format = {:?})",
config.get_protocol_version(),
config.get_protocol_format()
);
.is_err()
{
return;
}
// Split the socket for simultaneously read and write.
let (write, mut read) = ws.split();
let write = Mutex::new(write);
// Download required data to local cache and send Ready payload.
let Ok(ready_payload) = state.generate_ready_payload(db).await else {
return;
};
if write.send(config.encode(&ready_payload)).await.is_err() {
return;
}
// If the user has not provided authentication, request information.
if config.get_session_token().is_none() {
'outer: while let Ok(message) = read.try_next().await {
if let Ok(ClientMessage::Authenticate { token }) =
config.decode(message.as_ref().unwrap())
{
config.set_session_token(token);
break 'outer;
}
}
// Create presence session.
let (first_session, session_id) = create_session(&user_id, 0).await;
// If this was the first session, notify other users that we just went online.
if first_session {
state.broadcast_presence_change(true).await;
}
{
let write = Mutex::new(write);
// Create a PubSub connection to poll on.
let listener = listener(db, &mut state, addr, &config, &write).fuse();
// Read from WebSocket stream.
let worker = worker(user_id.clone(), &config, read, &write).fuse();
// Pin both tasks.
pin_mut!(listener, worker);
// Wait for either disconnect or for listener to die.
select!(
() = listener => {},
() = worker => {}
);
}
// Clean up presence session.
let last_session = delete_session(&user_id, session_id).await;
// If this was the last session, notify other users that we just went offline.
if last_session {
state.broadcast_presence_change(false).await;
}
}
async fn listener(
db: &'static Database,
state: &mut State,
addr: SocketAddr,
config: &ProtocolConfiguration,
write: &Mutex<WsWriter>,
) {
let redis_config = RedisConfig::from_url(&REDIS_URI).unwrap();
let Ok(subscriber) = fred::types::Builder::from_config(redis_config).build_subscriber_client()
else {
return;
};
if subscriber.init().await.is_err() {
return;
};
let mut message_rx = subscriber.message_rx();
loop {
// Check for state changes for subscriptions.
match state.apply_state() {
SubscriptionStateChange::Reset => {
subscriber.unsubscribe_all().await.unwrap();
for id in state.iter_subscriptions() {
subscriber.subscribe(id).await.unwrap();
}
// Try to authenticate the user.
if let Some(token) = config.get_session_token().as_ref() {
match User::from_token(db, token, UserHint::Any).await {
Ok(user) => {
info!("User {addr:?} authenticated as @{}", user.username);
#[cfg(debug_assertions)]
info!("{addr:?} has reset their subscriptions");
}
SubscriptionStateChange::Change { add, remove } => {
for id in remove {
#[cfg(debug_assertions)]
info!("{addr:?} unsubscribing from {id}");
// Create local state.
let mut state = State::from(user);
let user_id = state.cache.user_id.clone();
subscriber.unsubscribe(id).await.unwrap();
}
// Create presence session.
let (first_session, session_id) = create_session(&user_id, 0).await;
for id in add {
#[cfg(debug_assertions)]
info!("{addr:?} subscribing to {id}");
// Notify socket we have authenticated.
write
.lock()
.await
.send(config.encode(&EventV1::Authenticated))
.await
.ok();
// Download required data to local cache and send Ready payload.
if let Ok(ready_payload) = state.generate_ready_payload(db).await {
write
.lock()
.await
.send(config.encode(&ready_payload))
.await
.ok();
// If this was the first session, notify other users that we just went online.
if first_session {
state.broadcast_presence_change(true).await;
}
// Create a PubSub connection to poll on.
let listener = async {
if let Ok(mut conn) = redis_kiss::open_pubsub_connection().await
{
loop {
// Check for state changes for subscriptions.
match state.apply_state() {
SubscriptionStateChange::Reset => {
for id in state.iter_subscriptions() {
conn.subscribe(id).await.unwrap();
}
#[cfg(debug_assertions)]
info!("{addr:?} has reset their subscriptions");
}
SubscriptionStateChange::Change { add, remove } => {
for id in remove {
#[cfg(debug_assertions)]
info!("{addr:?} unsubscribing from {id}");
conn.unsubscribe(id).await.unwrap();
}
for id in add {
#[cfg(debug_assertions)]
info!("{addr:?} subscribing to {id}");
conn.subscribe(id).await.unwrap();
}
}
SubscriptionStateChange::None => {}
}
// * Debug logging of current subscriptions.
/*#[cfg(debug_assertions)]
info!(
"User {addr:?} is subscribed to {:?}",
state
.iter_subscriptions()
.collect::<Vec<&String>>()
);*/
// Handle incoming events.
match conn.on_message().next().await.map(|res| {
res.map(|item|(
item.get_channel_name().to_string(),
redis_kiss::decode_payload::<EventV1>(&item),
))
}) {
Some(Ok((channel, item))) => {
if let Ok(mut event) = item {
if state
.handle_incoming_event_v1(
db, &mut event,
)
.await
&& write.lock().await
.send(config.encode(&event))
.await
.is_err()
{
break;
}
} else {
warn!("Failed to deserialise an event for {channel}!");
}
}
Some(Err(e)) => {
info!("Error while consuming pub/sub messages: {e:?}");
sentry::capture_error(&e);
break
}
// No more data, assume we disconnected or otherwise
// something bad occurred, so disconnect user.
None => break,
}
}
}
}
.fuse();
// Read from WebSocket stream.
let worker =
async {
while let Ok(Some(msg)) = read.try_next().await {
if let Ok(payload) = config.decode(&msg) {
match payload {
ClientMessage::BeginTyping { channel } => {
EventV1::ChannelStartTyping {
id: channel.clone(),
user: user_id.clone(),
}
.p(channel.clone())
.await;
}
ClientMessage::EndTyping { channel } => {
EventV1::ChannelStopTyping {
id: channel.clone(),
user: user_id.clone(),
}
.p(channel.clone())
.await;
}
ClientMessage::Ping { data, responded } => {
if responded.is_none() {
write
.lock()
.await
.send(config.encode(
&EventV1::Pong { data },
))
.await
.ok();
}
}
_ => {}
}
}
}
}
.fuse();
// Pin both tasks.
pin_mut!(listener, worker);
// Wait for either disconnect or for listener to die.
select!(
() = listener => {},
() = worker => {}
);
// * Combine the streams back once we are ready to disconnect.
/* ws = read.reunite(write).unwrap(); */
}
// Clean up presence session.
let last_session = delete_session(&user_id, session_id).await;
// If this was the last session, notify other users that we just went offline.
if last_session {
state.broadcast_presence_change(false).await;
}
}
Err(err) => {
write.lock().await.send(config.encode(&err)).await.ok();
}
}
subscriber.subscribe(id).await.unwrap();
}
}
// * Disconnect the WebSocket if it isn't already.
/*ws.close(Some(CloseFrame {
code: CloseCode::Normal,
reason: std::borrow::Cow::from(""),
}))
.await
.unwrap();*/
SubscriptionStateChange::None => {}
}
info!("User disconnected from {addr:?}");
});
// Handle incoming events.
let Ok(message) = message_rx.recv().await.map_err(|e| {
info!("Error while consuming pub/sub messages: {e:?}");
sentry::capture_error(&e);
}) else {
return;
};
let event = match *REDIS_PAYLOAD_TYPE {
PayloadType::Json => message
.value
.as_str()
.and_then(|s| serde_json::from_str::<EventV1>(s.as_ref()).ok()),
PayloadType::Msgpack => message
.value
.as_bytes()
.and_then(|b| rmp_serde::from_slice::<EventV1>(b).ok()),
PayloadType::Bincode => message
.value
.as_bytes()
.and_then(|b| bincode::deserialize::<EventV1>(b).ok()),
};
let Some(mut event) = event else {
warn!("Failed to deserialise an event for {}!", message.channel);
return;
};
let should_send = state.handle_incoming_event_v1(db, &mut event).await;
if should_send
&& write
.lock()
.await
.send(config.encode(&event))
.await
.is_err()
{
return;
}
}
}
async fn worker(
user_id: String,
config: &ProtocolConfiguration,
mut read: WsReader,
write: &Mutex<WsWriter>,
) {
while let Ok(Some(msg)) = read.try_next().await {
let Ok(payload) = config.decode(&msg) else {
continue;
};
match payload {
ClientMessage::BeginTyping { channel } => {
EventV1::ChannelStartTyping {
id: channel.clone(),
user: user_id.clone(),
}
.p(channel.clone())
.await;
}
ClientMessage::EndTyping { channel } => {
EventV1::ChannelStopTyping {
id: channel.clone(),
user: user_id.clone(),
}
.p(channel.clone())
.await;
}
ClientMessage::Ping { data, responded } => {
if responded.is_none() {
write
.lock()
.await
.send(config.encode(&EventV1::Pong { data }))
.await
.ok();
}
}
_ => {}
}
}
}