Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 1f3fec22eb | |||
8135256ff4
|
|||
|
722eb9b025
|
|||
| e0dd9a085f | |||
| f7c93132b9 | |||
|
da5a5a5dff
|
|||
| 4eb2f23013 | |||
| 257ebefe0c |
14 changed files with 1336 additions and 351 deletions
258
Cargo.lock
generated
258
Cargo.lock
generated
|
|
@ -14,18 +14,18 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "aho-corasick"
|
||||
version = "1.1.4"
|
||||
version = "1.1.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
|
||||
checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba"
|
||||
dependencies = [
|
||||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "android_system_properties"
|
||||
version = "0.1.5"
|
||||
version = "0.1.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311"
|
||||
checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc"
|
||||
dependencies = [
|
||||
"libc",
|
||||
]
|
||||
|
|
@ -109,9 +109,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
|
|||
|
||||
[[package]]
|
||||
name = "aws-lc-rs"
|
||||
version = "1.17.3"
|
||||
version = "1.18.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1"
|
||||
checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e"
|
||||
dependencies = [
|
||||
"aws-lc-sys",
|
||||
"untrusted 0.7.1",
|
||||
|
|
@ -120,9 +120,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "aws-lc-sys"
|
||||
version = "0.43.0"
|
||||
version = "0.44.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c"
|
||||
checksum = "f09fae7be8bb3174e05c6afdb34199e6dc0c7c04ba9fa237b1967adfbde27483"
|
||||
dependencies = [
|
||||
"cc",
|
||||
"cmake",
|
||||
|
|
@ -143,6 +143,12 @@ version = "0.22.1"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
|
||||
|
||||
[[package]]
|
||||
name = "base64"
|
||||
version = "0.23.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5"
|
||||
|
||||
[[package]]
|
||||
name = "base64ct"
|
||||
version = "1.8.3"
|
||||
|
|
@ -211,9 +217,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
|
|||
|
||||
[[package]]
|
||||
name = "cc"
|
||||
version = "1.4.0"
|
||||
version = "1.4.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9"
|
||||
checksum = "9066c49992464636f92905fa096ec58baaa4d57ec19a5c096c68d3e25ef3d136"
|
||||
dependencies = [
|
||||
"find-msvc-tools",
|
||||
"jobserver",
|
||||
|
|
@ -319,12 +325,6 @@ dependencies = [
|
|||
"memchr",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "const-oid"
|
||||
version = "0.9.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
|
||||
|
||||
[[package]]
|
||||
name = "const-oid"
|
||||
version = "0.10.2"
|
||||
|
|
@ -411,8 +411,23 @@ dependencies = [
|
|||
"cfg-if",
|
||||
"cpufeatures 0.2.17",
|
||||
"curve25519-dalek-derive",
|
||||
"digest 0.10.7",
|
||||
"fiat-crypto",
|
||||
"fiat-crypto 0.2.9",
|
||||
"rustc_version",
|
||||
"subtle",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "curve25519-dalek"
|
||||
version = "5.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b5eed333089e2e1c1ac8c6c0398e5e2497b4c9926ca6d0365ed1e099afa5bc23"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures 0.3.0",
|
||||
"curve25519-dalek-derive",
|
||||
"digest 0.11.3",
|
||||
"fiat-crypto 0.3.0",
|
||||
"rustc_version",
|
||||
"subtle",
|
||||
"zeroize",
|
||||
|
|
@ -445,20 +460,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "data-encoding"
|
||||
version = "2.11.0"
|
||||
version = "2.11.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8"
|
||||
|
||||
[[package]]
|
||||
name = "der"
|
||||
version = "0.7.10"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb"
|
||||
dependencies = [
|
||||
"const-oid 0.9.6",
|
||||
"pem-rfc7468",
|
||||
"zeroize",
|
||||
]
|
||||
checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06"
|
||||
|
||||
[[package]]
|
||||
name = "der"
|
||||
|
|
@ -466,7 +470,8 @@ version = "0.8.1"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a69dedd701da44b0536442edf09c81a64b0ab97a7a4a5e3d1971f00027cbc63d"
|
||||
dependencies = [
|
||||
"const-oid 0.10.2",
|
||||
"const-oid",
|
||||
"pem-rfc7468",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
|
|
@ -523,7 +528,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2"
|
||||
dependencies = [
|
||||
"block-buffer 0.12.1",
|
||||
"const-oid 0.10.2",
|
||||
"const-oid",
|
||||
"crypto-common 0.2.2",
|
||||
"ctutils",
|
||||
]
|
||||
|
|
@ -563,24 +568,25 @@ checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
|
|||
|
||||
[[package]]
|
||||
name = "ed25519"
|
||||
version = "2.2.3"
|
||||
version = "3.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53"
|
||||
checksum = "29fcf32e6c73d1079f83ab4d782de2d81620346a5f38c6237a86a22f8368980a"
|
||||
dependencies = [
|
||||
"pkcs8 0.10.2",
|
||||
"signature 2.2.0",
|
||||
"pkcs8",
|
||||
"signature 3.0.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ed25519-dalek"
|
||||
version = "2.2.0"
|
||||
version = "3.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9"
|
||||
checksum = "6ebaa1a2bf1290ab3bfe5a7b771d050ebffab2711c19a81691c683a5144a25de"
|
||||
dependencies = [
|
||||
"curve25519-dalek",
|
||||
"curve25519-dalek 5.0.0",
|
||||
"ed25519",
|
||||
"serde",
|
||||
"sha2 0.10.9",
|
||||
"sha2 0.11.0",
|
||||
"signature 3.0.0",
|
||||
"subtle",
|
||||
"zeroize",
|
||||
]
|
||||
|
|
@ -616,7 +622,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
|
||||
dependencies = [
|
||||
"libc",
|
||||
"windows-sys 0.59.0",
|
||||
<<<<<<< HEAD
|
||||
"windows-sys 0.61.2",
|
||||
=======
|
||||
"windows-sys 0.52.0",
|
||||
>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -644,10 +654,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d"
|
||||
|
||||
[[package]]
|
||||
name = "find-msvc-tools"
|
||||
version = "0.1.9"
|
||||
name = "fiat-crypto"
|
||||
version = "0.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
|
||||
checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24"
|
||||
|
||||
[[package]]
|
||||
name = "find-msvc-tools"
|
||||
version = "0.1.10"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de"
|
||||
|
||||
[[package]]
|
||||
name = "fixedbitset"
|
||||
|
|
@ -953,9 +969,9 @@ checksum = "1a9fcbcc408c5526c3ab80d534e5c86e7967c1fb7aa0a8c76abd1edc27deb877"
|
|||
|
||||
[[package]]
|
||||
name = "http"
|
||||
version = "1.4.2"
|
||||
version = "1.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425"
|
||||
checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"itoa",
|
||||
|
|
@ -998,9 +1014,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
|
|||
|
||||
[[package]]
|
||||
name = "hybrid-array"
|
||||
version = "0.4.13"
|
||||
version = "0.4.14"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c"
|
||||
checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b"
|
||||
dependencies = [
|
||||
"ctutils",
|
||||
"typenum",
|
||||
|
|
@ -1214,9 +1230,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "ipnet"
|
||||
version = "2.12.0"
|
||||
version = "2.12.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2"
|
||||
checksum = "6a756c3fac73139e83f14c2d742155dd2b78d3ee56597b419a0579b7bdd6dd78"
|
||||
|
||||
[[package]]
|
||||
name = "itertools"
|
||||
|
|
@ -1363,9 +1379,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "keccak"
|
||||
version = "0.2.0"
|
||||
version = "0.2.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "9e24a010dd405bd7ed803e5253182815b41bf2e6a80cc3bfc066658e03a198aa"
|
||||
checksum = "ffd9697dc4a9a62e2da93389f34400b77a28f0287711263cabb203b3ccb9c0e4"
|
||||
dependencies = [
|
||||
"cfg-if",
|
||||
"cpufeatures 0.3.0",
|
||||
|
|
@ -1452,9 +1468,15 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "livekit-protocol"
|
||||
version = "0.7.11"
|
||||
<<<<<<< HEAD
|
||||
version = "0.7.10"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "735a237deeff60124edd7831fcabce32b4b9f667d5c4c5688c9d18a2808422c4"
|
||||
checksum = "4d26880e94e2f9bab298445e7d86a3794453d211a12ddbd051bd9991a343f9ff"
|
||||
=======
|
||||
version = "0.7.12"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "526f22bddf409e5f15449d55cf341647d7c16a1f43df9db9b05be106e4208e1c"
|
||||
>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
|
||||
dependencies = [
|
||||
"pbjson",
|
||||
"pbjson-types",
|
||||
|
|
@ -1531,12 +1553,12 @@ version = "0.1.1"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "add6b9d92e496f16f4526d68ff29da1483aba4b119baeab8bed3b9e3544a6f3d"
|
||||
dependencies = [
|
||||
"const-oid 0.10.2",
|
||||
"const-oid",
|
||||
"crypto-common 0.2.2",
|
||||
"ctutils",
|
||||
"hybrid-array",
|
||||
"module-lattice",
|
||||
"pkcs8 0.11.0",
|
||||
"pkcs8",
|
||||
"shake",
|
||||
"signature 3.0.0",
|
||||
]
|
||||
|
|
@ -1581,7 +1603,7 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "mtp"
|
||||
version = "0.2.0"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
|
||||
dependencies = [
|
||||
"mtp-client",
|
||||
"mtp-codec",
|
||||
|
|
@ -1597,7 +1619,7 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "mtp-client"
|
||||
version = "0.2.0"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
|
||||
dependencies = [
|
||||
"mtp-codec",
|
||||
"mtp-common",
|
||||
|
|
@ -1610,9 +1632,9 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "mtp-codec"
|
||||
version = "0.2.0"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"base64 0.23.1",
|
||||
"byteorder",
|
||||
"mtp-common",
|
||||
"mtp-crypto",
|
||||
|
|
@ -1623,7 +1645,7 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "mtp-common"
|
||||
version = "0.2.0"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
|
||||
dependencies = [
|
||||
"quinn",
|
||||
"rustls",
|
||||
|
|
@ -1634,9 +1656,9 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "mtp-crypto"
|
||||
version = "0.2.0"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
|
||||
dependencies = [
|
||||
"base64 0.22.1",
|
||||
"base64 0.23.1",
|
||||
"chacha20poly1305",
|
||||
"ed25519-dalek",
|
||||
"getrandom 0.4.3",
|
||||
|
|
@ -1648,7 +1670,7 @@ dependencies = [
|
|||
"rustls",
|
||||
"serde",
|
||||
"sha2 0.11.0",
|
||||
"thiserror 1.0.69",
|
||||
"thiserror 2.0.19",
|
||||
"tokio",
|
||||
"zeroize",
|
||||
]
|
||||
|
|
@ -1656,24 +1678,24 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "mtp-files"
|
||||
version = "0.2.0"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
|
||||
dependencies = [
|
||||
"mtp-crypto",
|
||||
"rand 0.10.2",
|
||||
"thiserror 1.0.69",
|
||||
"thiserror 2.0.19",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "mtp-host"
|
||||
version = "0.2.0"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
|
||||
dependencies = [
|
||||
"mtp-codec",
|
||||
"mtp-common",
|
||||
"mtp-crypto",
|
||||
"mtp-transport",
|
||||
"rand 0.8.7",
|
||||
"rand 0.10.2",
|
||||
"tokio",
|
||||
"tracing",
|
||||
"wtransport",
|
||||
|
|
@ -1682,7 +1704,7 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "mtp-transport"
|
||||
version = "0.2.0"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"mtp-codec",
|
||||
|
|
@ -1700,7 +1722,7 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "mtp-type-map"
|
||||
version = "0.2.0"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
|
||||
dependencies = [
|
||||
"serde",
|
||||
"serde_yaml",
|
||||
|
|
@ -1709,7 +1731,7 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "mtp-webserver"
|
||||
version = "0.2.0"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
|
||||
source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
|
||||
dependencies = [
|
||||
"async-trait",
|
||||
"bytes",
|
||||
|
|
@ -2102,9 +2124,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "pem-rfc7468"
|
||||
version = "0.7.0"
|
||||
version = "1.0.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412"
|
||||
checksum = "a6305423e0e7738146434843d1694d621cce767262b2a86910beab705e4493d9"
|
||||
dependencies = [
|
||||
"base64ct",
|
||||
]
|
||||
|
|
@ -2131,24 +2153,14 @@ version = "0.2.17"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
|
||||
|
||||
[[package]]
|
||||
name = "pkcs8"
|
||||
version = "0.10.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7"
|
||||
dependencies = [
|
||||
"der 0.7.10",
|
||||
"spki 0.7.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pkcs8"
|
||||
version = "0.11.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "451913da69c775a56034ea8d9003d27ee8948e12443eae7c038ba100a4f21cb7"
|
||||
dependencies = [
|
||||
"der 0.8.1",
|
||||
"spki 0.8.0",
|
||||
"der",
|
||||
"spki",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -2327,7 +2339,11 @@ dependencies = [
|
|||
"once_cell",
|
||||
"socket2",
|
||||
"tracing",
|
||||
"windows-sys 0.59.0",
|
||||
<<<<<<< HEAD
|
||||
"windows-sys 0.61.2",
|
||||
=======
|
||||
"windows-sys 0.52.0",
|
||||
>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -2474,9 +2490,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "regex-automata"
|
||||
version = "0.4.16"
|
||||
version = "0.4.18"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad"
|
||||
checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2"
|
||||
dependencies = [
|
||||
"aho-corasick",
|
||||
"memchr",
|
||||
|
|
@ -2581,14 +2597,18 @@ dependencies = [
|
|||
"errno",
|
||||
"libc",
|
||||
"linux-raw-sys",
|
||||
"windows-sys 0.59.0",
|
||||
<<<<<<< HEAD
|
||||
"windows-sys 0.61.2",
|
||||
=======
|
||||
"windows-sys 0.52.0",
|
||||
>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rustls"
|
||||
version = "0.23.42"
|
||||
version = "0.23.43"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138"
|
||||
checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06"
|
||||
dependencies = [
|
||||
"aws-lc-rs",
|
||||
"log",
|
||||
|
|
@ -2640,7 +2660,11 @@ dependencies = [
|
|||
"security-framework",
|
||||
"security-framework-sys",
|
||||
"webpki-root-certs",
|
||||
"windows-sys 0.59.0",
|
||||
<<<<<<< HEAD
|
||||
"windows-sys 0.61.2",
|
||||
=======
|
||||
"windows-sys 0.52.0",
|
||||
>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -2844,7 +2868,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "09057cb2149ad4cbd2da1e26b351f9a4c354219421229c69c3063e6f61947c4a"
|
||||
dependencies = [
|
||||
"digest 0.11.3",
|
||||
"keccak 0.2.0",
|
||||
"keccak 0.2.1",
|
||||
"sponge-cursor",
|
||||
]
|
||||
|
||||
|
|
@ -2927,16 +2951,6 @@ dependencies = [
|
|||
"windows-sys 0.61.2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "spki"
|
||||
version = "0.7.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d"
|
||||
dependencies = [
|
||||
"base64ct",
|
||||
"der 0.7.10",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "spki"
|
||||
version = "0.8.0"
|
||||
|
|
@ -2944,7 +2958,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "1d9efca8738c78ee9484207732f728b1ef517bbb1833d6fc0879ca898a522f6f"
|
||||
dependencies = [
|
||||
"base64ct",
|
||||
"der 0.8.1",
|
||||
"der",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -3032,10 +3046,14 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
|
||||
dependencies = [
|
||||
"fastrand",
|
||||
"getrandom 0.3.4",
|
||||
"getrandom 0.4.3",
|
||||
"once_cell",
|
||||
"rustix",
|
||||
"windows-sys 0.59.0",
|
||||
<<<<<<< HEAD
|
||||
"windows-sys 0.61.2",
|
||||
=======
|
||||
"windows-sys 0.52.0",
|
||||
>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -3080,9 +3098,9 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "time"
|
||||
version = "0.3.54"
|
||||
version = "0.3.55"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3e1d5e639ff6bab73cb6885cc7e7b1de96c3f32c68ec55f3952614bec1092244"
|
||||
checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134"
|
||||
dependencies = [
|
||||
"deranged",
|
||||
"num-conv",
|
||||
|
|
@ -3152,13 +3170,13 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "tokio-macros"
|
||||
version = "2.7.1"
|
||||
version = "2.7.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6328af13490e73a9b4694030fafd93f8c8c6a9dede33e821c3fc63eddf8042ba"
|
||||
checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
"syn 2.0.119",
|
||||
"syn 3.0.3",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -3596,7 +3614,11 @@ version = "0.1.11"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
|
||||
dependencies = [
|
||||
"windows-sys 0.59.0",
|
||||
<<<<<<< HEAD
|
||||
"windows-sys 0.61.2",
|
||||
=======
|
||||
"windows-sys 0.52.0",
|
||||
>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -3851,7 +3873,7 @@ version = "2.0.1"
|
|||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "c7e468321c81fb07fa7f4c636c3972b9100f0346e5b6a9f2bd0603a52f7ed277"
|
||||
dependencies = [
|
||||
"curve25519-dalek",
|
||||
"curve25519-dalek 4.1.3",
|
||||
"rand_core 0.6.4",
|
||||
"serde",
|
||||
"zeroize",
|
||||
|
|
@ -3911,18 +3933,18 @@ dependencies = [
|
|||
|
||||
[[package]]
|
||||
name = "zerocopy"
|
||||
version = "0.8.55"
|
||||
version = "0.8.56"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b5a105cd7b140f6eeec8acff2ea38135d3cab283ada58540f629fe51e46696eb"
|
||||
checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb"
|
||||
dependencies = [
|
||||
"zerocopy-derive",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "zerocopy-derive"
|
||||
version = "0.8.55"
|
||||
version = "0.8.56"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0fe976fb70c78cd64cccfe3a6fc142244e8a77b70959b30faf9d0ac37ee228eb"
|
||||
checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@ dotenv = "0.15.0"
|
|||
strum = "0.28.0"
|
||||
strum_macros = "0.28.0"
|
||||
livekit-api = { version = "0.6.0", features = ["rustls-tls-native-roots"] }
|
||||
livekit-protocol = "0.7.10"
|
||||
livekit-protocol = "=0.7.10"
|
||||
thiserror = "2.0.19"
|
||||
trust-dns-resolver = "0.23.2"
|
||||
serde_json = "1.0.151"
|
||||
|
|
|
|||
|
|
@ -1 +1 @@
|
|||
Subproject commit f753cfd406fc0050a8ccd76a90897205efceacd8
|
||||
Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e
|
||||
|
|
@ -8,7 +8,6 @@ use uuid::Uuid;
|
|||
use crate::anonymous_clients::anonymous_manager::{self, generate_username};
|
||||
use crate::app_state::AppState;
|
||||
use crate::calls::call_group::call_invite_secret_from_cv;
|
||||
use crate::data::user::UserStatus;
|
||||
use crate::rho::connection::{GeneralConnection, MtpReceiver, MtpSender};
|
||||
use crate::util::data_type_id;
|
||||
use crate::util::logger::PrintType;
|
||||
|
|
@ -222,9 +221,13 @@ impl AnonymousClientConnection {
|
|||
.await;
|
||||
}
|
||||
|
||||
// Handle client status changes
|
||||
if cv.is_type(CommunicationType::ClientChanged) {
|
||||
self.handle_client_changed(cv).await;
|
||||
// Presence is account-scoped and anonymous sessions have no
|
||||
// persisted account preference to change.
|
||||
if cv.is_type(CommunicationType::SetUserState)
|
||||
|| cv.is_type(CommunicationType::ClientChanged)
|
||||
{
|
||||
self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -299,7 +302,6 @@ impl AnonymousClientConnection {
|
|||
|
||||
if cv.is_type(CommunicationType::GetUserData)
|
||||
|| cv.is_type(CommunicationType::GetIotaData)
|
||||
|| cv.is_type(CommunicationType::DeleteUser)
|
||||
{
|
||||
self.handle_omega_forward(cv).await;
|
||||
return;
|
||||
|
|
@ -321,22 +323,6 @@ impl AnonymousClientConnection {
|
|||
});
|
||||
}
|
||||
|
||||
/// Handle client status change
|
||||
async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) {
|
||||
if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) {
|
||||
let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online);
|
||||
self.state
|
||||
.omega
|
||||
.client_changed(
|
||||
self.user_id as i64,
|
||||
self.user_id as i64,
|
||||
self.user_id as i64,
|
||||
user_status,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle call invite
|
||||
async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
|
||||
let receiver_id: i64 = cv.get_data(DataType::ReceiverId).as_number().unwrap_or(0) as i64;
|
||||
|
|
|
|||
|
|
@ -1,10 +1,12 @@
|
|||
use std::env;
|
||||
use std::{env, time::Duration};
|
||||
|
||||
use thiserror::Error;
|
||||
|
||||
const DEFAULT_RHO_PORT: u16 = 443;
|
||||
const DEFAULT_OMEGA_HOST: &str = "tensamin.net";
|
||||
const DEFAULT_OMEGA_PORT: u16 = 9187;
|
||||
const DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS: u64 = 20;
|
||||
const DEFAULT_OMEGA_SYNC_RETRIES: u32 = 3;
|
||||
|
||||
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||
pub struct LiveKitConfig {
|
||||
|
|
@ -19,6 +21,11 @@ pub struct Config {
|
|||
pub omega_host: String,
|
||||
pub omega_port: u16,
|
||||
pub omikron_id: u64,
|
||||
/// Maximum duration of one route or subscription synchronization request.
|
||||
pub omega_sync_timeout: Duration,
|
||||
/// Number of synchronization requests before the transport is closed and
|
||||
/// the normal reconnect loop starts.
|
||||
pub omega_sync_retries: u32,
|
||||
pub livekit: Option<LiveKitConfig>,
|
||||
}
|
||||
|
||||
|
|
@ -38,6 +45,15 @@ impl Config {
|
|||
let rho_port = parse_or_default("RHO_PORT", DEFAULT_RHO_PORT)?;
|
||||
let omega_port = parse_or_default("OMEGA_PORT", DEFAULT_OMEGA_PORT)?;
|
||||
let omikron_id = parse_or_default("ID", 0_u64)?;
|
||||
// Each synchronization request is bounded by this timeout. After
|
||||
// omega_sync_retries attempts, Omikron closes the authenticated
|
||||
// transport so the normal reconnect loop can establish a clean state.
|
||||
let omega_sync_timeout = Duration::from_secs(parse_or_default(
|
||||
"OMEGA_SYNC_TIMEOUT_SECONDS",
|
||||
DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS,
|
||||
)?);
|
||||
let omega_sync_retries =
|
||||
parse_or_default("OMEGA_SYNC_RETRIES", DEFAULT_OMEGA_SYNC_RETRIES)?.max(1);
|
||||
let omega_host = env::var("OMEGA_HOST")
|
||||
.unwrap_or_else(|_| DEFAULT_OMEGA_HOST.to_string())
|
||||
.trim()
|
||||
|
|
@ -55,6 +71,8 @@ impl Config {
|
|||
omega_host,
|
||||
omega_port,
|
||||
omikron_id,
|
||||
omega_sync_timeout,
|
||||
omega_sync_retries,
|
||||
livekit: livekit_from_environment()?,
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,4 @@
|
|||
use strum::IntoEnumIterator;
|
||||
use strum_macros::EnumIter;
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, EnumIter, Eq)]
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
#[allow(unused, non_camel_case_types)]
|
||||
pub enum UserStatus {
|
||||
user_offline,
|
||||
|
|
@ -20,12 +17,40 @@ impl UserStatus {
|
|||
pub fn to_string(&self) -> String {
|
||||
format!("{:?}", self)
|
||||
}
|
||||
pub fn from_str(s: &str) -> Option<UserStatus> {
|
||||
for sel in UserStatus::iter() {
|
||||
if &sel.to_string() == s {
|
||||
return Some(sel);
|
||||
}
|
||||
pub fn from_client_preference(s: &str) -> Option<Self> {
|
||||
match s {
|
||||
"user_online" => Some(Self::user_online),
|
||||
"user_idle" => Some(Self::user_idle),
|
||||
"user_dnd" => Some(Self::user_dnd),
|
||||
"user_wc" => Some(Self::user_wc),
|
||||
"user_invisible" => Some(Self::user_invisible),
|
||||
_ => None,
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
pub fn public_value(&self) -> Self {
|
||||
match self {
|
||||
Self::user_invisible => Self::user_offline,
|
||||
value => value.clone(),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn from_str(s: &str) -> Option<Self> {
|
||||
Self::from_client_preference(s)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::UserStatus;
|
||||
|
||||
#[test]
|
||||
fn network_parser_rejects_derived_states() {
|
||||
assert_eq!(UserStatus::from_str("user_offline"), None);
|
||||
assert_eq!(UserStatus::from_str("iota_offline"), None);
|
||||
assert_eq!(
|
||||
UserStatus::from_str("user_online"),
|
||||
Some(UserStatus::user_online)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
145
src/omega/capabilities.rs
Normal file
145
src/omega/capabilities.rs
Normal file
|
|
@ -0,0 +1,145 @@
|
|||
use std::collections::BTreeSet;
|
||||
|
||||
const OMIKRON_PREFIX: &str = "omikron;caps=";
|
||||
const OMEGA_PREFIX: &str = "omega;caps=";
|
||||
const SET_USER_STATE: &str = "set_user_state_v1";
|
||||
const STATE_SUBSCRIBE: &str = "state_subscribe_v1";
|
||||
const SESSION_SNAPSHOT: &str = "session_snapshot_v1";
|
||||
const CLIENT_STATE_PUSH: &str = "client_state_push_v1";
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq, Eq)]
|
||||
pub struct PeerCapabilities {
|
||||
pub set_user_state_v1: bool,
|
||||
pub state_subscribe_v1: bool,
|
||||
pub session_snapshot_v1: bool,
|
||||
pub client_state_push_v1: bool,
|
||||
}
|
||||
|
||||
impl PeerCapabilities {
|
||||
pub fn current() -> Self {
|
||||
Self {
|
||||
set_user_state_v1: true,
|
||||
state_subscribe_v1: true,
|
||||
session_snapshot_v1: true,
|
||||
client_state_push_v1: true,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn identification_description(&self) -> String {
|
||||
let mut names = Vec::new();
|
||||
if self.set_user_state_v1 {
|
||||
names.push(SET_USER_STATE);
|
||||
}
|
||||
if self.state_subscribe_v1 {
|
||||
names.push(STATE_SUBSCRIBE);
|
||||
}
|
||||
if self.session_snapshot_v1 {
|
||||
names.push(SESSION_SNAPSHOT);
|
||||
}
|
||||
if self.client_state_push_v1 {
|
||||
names.push(CLIENT_STATE_PUSH);
|
||||
}
|
||||
format!("{OMIKRON_PREFIX}{}", names.join(","))
|
||||
}
|
||||
|
||||
pub fn from_identification_description(description: Option<&str>) -> Result<Self, ()> {
|
||||
parse_capabilities(description, OMIKRON_PREFIX)
|
||||
}
|
||||
|
||||
pub fn from_omega_identification_description(description: Option<&str>) -> Result<Self, ()> {
|
||||
// A missing Omega descriptor selects the documented legacy wire
|
||||
// contracts: tuple route snapshots, GetStates-only subscriptions,
|
||||
// and ClientChanged pushes.
|
||||
parse_capabilities(description, OMEGA_PREFIX)
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_capabilities(description: Option<&str>, prefix: &str) -> Result<PeerCapabilities, ()> {
|
||||
let Some(description) = description else {
|
||||
return Ok(PeerCapabilities::default());
|
||||
};
|
||||
if description == "omega" {
|
||||
return Ok(PeerCapabilities::default());
|
||||
}
|
||||
let Some(capabilities) = description.strip_prefix(prefix) else {
|
||||
return Err(());
|
||||
};
|
||||
let mut seen = BTreeSet::new();
|
||||
for capability in capabilities.split(',') {
|
||||
if capability.is_empty() || !seen.insert(capability) {
|
||||
return Err(());
|
||||
}
|
||||
}
|
||||
if seen.iter().any(|capability| {
|
||||
!matches!(
|
||||
*capability,
|
||||
SET_USER_STATE | STATE_SUBSCRIBE | SESSION_SNAPSHOT | CLIENT_STATE_PUSH
|
||||
)
|
||||
}) {
|
||||
return Err(());
|
||||
}
|
||||
Ok(PeerCapabilities {
|
||||
set_user_state_v1: seen.contains(SET_USER_STATE),
|
||||
state_subscribe_v1: seen.contains(STATE_SUBSCRIBE),
|
||||
session_snapshot_v1: seen.contains(SESSION_SNAPSHOT),
|
||||
client_state_push_v1: seen.contains(CLIENT_STATE_PUSH),
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn capability_description_round_trips() {
|
||||
let capabilities = PeerCapabilities::current();
|
||||
assert_eq!(
|
||||
PeerCapabilities::from_identification_description(Some(
|
||||
&capabilities.identification_description()
|
||||
)),
|
||||
Ok(capabilities)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn malformed_capabilities_are_rejected() {
|
||||
assert!(
|
||||
PeerCapabilities::from_identification_description(Some("omikron;caps=unknown_v9"))
|
||||
.is_err()
|
||||
);
|
||||
assert!(
|
||||
PeerCapabilities::from_identification_description(Some(
|
||||
"omikron;caps=state_subscribe_v1,state_subscribe_v1"
|
||||
))
|
||||
.is_err()
|
||||
);
|
||||
assert!(PeerCapabilities::from_identification_description(Some("omikron;caps=")).is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn missing_capabilities_mean_legacy_peer() {
|
||||
assert_eq!(
|
||||
PeerCapabilities::from_identification_description(None),
|
||||
Ok(PeerCapabilities::default())
|
||||
);
|
||||
assert_eq!(
|
||||
PeerCapabilities::from_omega_identification_description(Some("omega")),
|
||||
Ok(PeerCapabilities::default())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn omega_capabilities_use_the_omega_namespace() {
|
||||
assert!(
|
||||
PeerCapabilities::from_omega_identification_description(Some(
|
||||
"omega;caps=set_user_state_v1"
|
||||
))
|
||||
.unwrap()
|
||||
.set_user_state_v1
|
||||
);
|
||||
assert!(
|
||||
PeerCapabilities::from_identification_description(Some("omega;caps=set_user_state_v1"))
|
||||
.is_err()
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
@ -1 +1,2 @@
|
|||
pub mod capabilities;
|
||||
pub mod omega_connection;
|
||||
|
|
|
|||
|
|
@ -1,6 +1,9 @@
|
|||
use super::capabilities::PeerCapabilities;
|
||||
use crate::{
|
||||
config::Config, data::user::UserStatus, log_cv_in, log_cv_out, log_err, log_in, log_out,
|
||||
rho::rho_manager::RhoManager, util::logger::PrintType,
|
||||
config::Config,
|
||||
log_cv_in, log_cv_out, log_err, log_in, log_out,
|
||||
rho::rho_manager::RhoManager,
|
||||
util::{data_type_id, logger::PrintType},
|
||||
};
|
||||
use dashmap::DashMap;
|
||||
use mtp::client::{Client, MTPConnection, Sender};
|
||||
|
|
@ -13,7 +16,7 @@ use mtp::{
|
|||
use mtp_transport::ConnectionHandle;
|
||||
use std::{sync::Arc, time::Duration};
|
||||
use tokio::{
|
||||
sync::{Mutex, RwLock, mpsc, watch},
|
||||
sync::{Mutex, RwLock, mpsc, oneshot, watch},
|
||||
task::JoinHandle,
|
||||
time::{Instant, sleep},
|
||||
};
|
||||
|
|
@ -23,6 +26,7 @@ use uuid::Uuid;
|
|||
const RECONNECT_DELAY: Duration = Duration::from_secs(5);
|
||||
const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(300);
|
||||
const CONNECTION_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
const CAPABILITY_NEGOTIATION_TIMEOUT: Duration = Duration::from_secs(1);
|
||||
const PING_INTERVAL: Duration = Duration::from_secs(5);
|
||||
const TASK_CLEANUP_INTERVAL: Duration = Duration::from_secs(60);
|
||||
const TASK_MAX_AGE: Duration = Duration::from_secs(60);
|
||||
|
|
@ -30,6 +34,37 @@ const MAX_CONCURRENT_REQUESTS: usize = 128;
|
|||
const CIRCUIT_BREAKER_FAILURE_THRESHOLD: u32 = 3;
|
||||
const CIRCUIT_BREAKER_COOLDOWN: Duration = Duration::from_secs(30);
|
||||
|
||||
fn client_changed_target(value: &CommunicationValue) -> Option<(i64, i64)> {
|
||||
if !value.is_type(CommunicationType::ClientChanged) {
|
||||
return None;
|
||||
}
|
||||
let receiver = i64::try_from(value.get_receiver())
|
||||
.ok()
|
||||
.filter(|id| *id > 0)?;
|
||||
let session_id = value
|
||||
.get_data(DataType::SessionId)
|
||||
.as_signed_number()
|
||||
.and_then(|id| i64::try_from(id).ok())
|
||||
.filter(|id| *id > 0)?;
|
||||
value
|
||||
.get_data(DataType::UserId)
|
||||
.as_signed_number()
|
||||
.and_then(|id| i64::try_from(id).ok())
|
||||
.filter(|id| *id > 0)?;
|
||||
value.get_data(DataType::UserState).as_str()?;
|
||||
Some((receiver, session_id))
|
||||
}
|
||||
|
||||
fn parse_omega_capability_response(value: &CommunicationValue) -> Result<PeerCapabilities, String> {
|
||||
if !value.is_type(CommunicationType::IdentificationResponse) {
|
||||
return Err("not an identification response".to_string());
|
||||
}
|
||||
PeerCapabilities::from_omega_identification_description(
|
||||
value.get_data(DataType::Description).as_str(),
|
||||
)
|
||||
.map_err(|_| "Omega advertised an invalid capability set".to_string())
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Waiting Task System
|
||||
// ============================================================================
|
||||
|
|
@ -85,24 +120,33 @@ pub fn start_task_cleanup_loop(omega: Arc<OmegaConnection>) {
|
|||
pub enum ConnectionState {
|
||||
Disconnected,
|
||||
Connecting,
|
||||
Connected { identified: bool },
|
||||
Identifying,
|
||||
SynchronizingRoutes,
|
||||
SynchronizingSubscriptions,
|
||||
Ready,
|
||||
}
|
||||
|
||||
#[allow(unused_variables)]
|
||||
impl ConnectionState {
|
||||
pub fn is_connected(&self) -> bool {
|
||||
match self {
|
||||
ConnectionState::Connected { identified } => true,
|
||||
_ => false,
|
||||
}
|
||||
!matches!(
|
||||
self,
|
||||
ConnectionState::Disconnected | ConnectionState::Connecting
|
||||
)
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn is_identified(&self) -> bool {
|
||||
match self {
|
||||
ConnectionState::Connected { identified: true } => true,
|
||||
_ => false,
|
||||
}
|
||||
matches!(
|
||||
self,
|
||||
ConnectionState::SynchronizingRoutes
|
||||
| ConnectionState::SynchronizingSubscriptions
|
||||
| ConnectionState::Ready
|
||||
)
|
||||
}
|
||||
|
||||
pub fn is_ready(&self) -> bool {
|
||||
matches!(self, ConnectionState::Ready)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -117,6 +161,8 @@ pub struct OmegaConnection {
|
|||
host: String,
|
||||
port: u16,
|
||||
omikron_id: u64,
|
||||
sync_timeout: Duration,
|
||||
sync_retries: u32,
|
||||
pub connection_id: Uuid,
|
||||
shutdown_tx: Arc<Mutex<Option<watch::Sender<bool>>>>,
|
||||
// Track if we should reconnect on close
|
||||
|
|
@ -126,6 +172,14 @@ pub struct OmegaConnection {
|
|||
keyring: mtp::crypto::Keyring,
|
||||
rho: Arc<RhoManager>,
|
||||
waiting_tasks: DashMap<u32, WaitingTask>,
|
||||
session_presence: DashMap<(i64, i64), SessionPresenceState>,
|
||||
peer_capabilities: Arc<RwLock<PeerCapabilities>>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Default, PartialEq, Eq)]
|
||||
pub struct SessionPresenceState {
|
||||
pub subscribed_user_ids: Vec<i64>,
|
||||
pub pending: bool,
|
||||
}
|
||||
|
||||
impl OmegaConnection {
|
||||
|
|
@ -143,6 +197,8 @@ impl OmegaConnection {
|
|||
host: config.omega_host.clone(),
|
||||
port: config.omega_port,
|
||||
omikron_id: config.omikron_id,
|
||||
sync_timeout: config.omega_sync_timeout,
|
||||
sync_retries: config.omega_sync_retries,
|
||||
connection_id: Uuid::new_v4(),
|
||||
shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))),
|
||||
reconnect_on_close: Arc::new(RwLock::new(true)),
|
||||
|
|
@ -154,6 +210,8 @@ impl OmegaConnection {
|
|||
keyring,
|
||||
rho,
|
||||
waiting_tasks: DashMap::new(),
|
||||
session_presence: DashMap::new(),
|
||||
peer_capabilities: Arc::new(RwLock::new(PeerCapabilities::default())),
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -267,6 +325,7 @@ impl OmegaConnection {
|
|||
|
||||
let client_config = ClientConfig::new(format!("https://{}:{}", self.host, self.port))
|
||||
.with_client_id(self.omikron_id)
|
||||
.with_description(PeerCapabilities::current().identification_description())
|
||||
.with_policy(
|
||||
Policy::default()
|
||||
.with_send_mode(SendMode::SingleStreamPerMessage)
|
||||
|
|
@ -302,18 +361,43 @@ impl OmegaConnection {
|
|||
// Store sender
|
||||
let sender_arc = Arc::new(connection.sender.clone());
|
||||
*self.sender.write().await = Some(sender_arc.clone());
|
||||
*self.state.write().await = ConnectionState::Connected { identified: false };
|
||||
// A new authenticated transport gets a fresh request circuit. A
|
||||
// previous transport can have failed while synchronizing and must not
|
||||
// prevent this connection from making its initial requests.
|
||||
self.circuit_breaker.lock().await.record_success();
|
||||
*self.state.write().await = ConnectionState::Identifying;
|
||||
|
||||
// Get handle for close monitoring
|
||||
let sender_handle = sender_arc.handle().clone();
|
||||
let connection = Arc::new(connection);
|
||||
let (capability_tx, capability_rx) = oneshot::channel();
|
||||
|
||||
// Start read loop
|
||||
let read_self = self.clone();
|
||||
let read_handle = tokio::spawn(async move {
|
||||
read_self.read_loop(connection, sender_handle).await;
|
||||
read_self
|
||||
.read_loop(connection, sender_handle, Some(capability_tx))
|
||||
.await;
|
||||
});
|
||||
|
||||
let negotiated_capabilities =
|
||||
match tokio::time::timeout(CAPABILITY_NEGOTIATION_TIMEOUT, capability_rx).await {
|
||||
Ok(Ok(Ok(capabilities))) => capabilities,
|
||||
Ok(Ok(Err(error))) => {
|
||||
self.fail_synchronization("capability negotiation", error)
|
||||
.await;
|
||||
let _ = read_handle.await;
|
||||
return Err("Omega capability negotiation failed".to_string());
|
||||
}
|
||||
Ok(Err(_)) | Err(_) => {
|
||||
// An Omega from before capability negotiation sends no second
|
||||
// identification response. Fall back to its documented
|
||||
// legacy contracts rather than guessing modern support.
|
||||
PeerCapabilities::default()
|
||||
}
|
||||
};
|
||||
*self.peer_capabilities.write().await = negotiated_capabilities;
|
||||
|
||||
// Tell omega our current state now that we're actually connected -
|
||||
// doing this after teardown (as before) sent into a sender that had
|
||||
// already been cleared, silently dropping the sync every time.
|
||||
|
|
@ -354,8 +438,10 @@ impl OmegaConnection {
|
|||
}
|
||||
}
|
||||
|
||||
async fn sync_client_iota_status(self: Arc<Self>) {
|
||||
async fn sync_client_iota_status(&self) {
|
||||
*self.state.write().await = ConnectionState::SynchronizingRoutes;
|
||||
let mut connected_iota_ids: Vec<DataValue> = Vec::new();
|
||||
let mut connected_sessions: Vec<DataValue> = Vec::new();
|
||||
let mut connected_user_ids: Vec<DataValue> = Vec::new();
|
||||
|
||||
let iota_ids = self.rho.iota_ids().await;
|
||||
|
|
@ -367,21 +453,120 @@ impl OmegaConnection {
|
|||
|
||||
for rho in rho_connections {
|
||||
for client_conn in rho.get_client_connections().await {
|
||||
connected_user_ids.push(DataValue::SignedNumber(
|
||||
client_conn.get_user_id().await.into(),
|
||||
));
|
||||
let type_map = mtp::codec::TypeMap::latest();
|
||||
let user_id = DataValue::SignedNumber(client_conn.get_user_id().await.into());
|
||||
let session_id = DataValue::SignedNumber((client_conn.session_id as i128).into());
|
||||
let iota_id = DataValue::SignedNumber((rho.get_iota_id().await as i128).into());
|
||||
let peer_capabilities = self.peer_capabilities.read().await.clone();
|
||||
if peer_capabilities.session_snapshot_v1 {
|
||||
connected_sessions.push(DataValue::Container(vec![
|
||||
(data_type_id(DataType::UserId, &type_map), user_id),
|
||||
(data_type_id(DataType::SessionId, &type_map), session_id),
|
||||
(data_type_id(DataType::IotaId, &type_map), iota_id),
|
||||
]));
|
||||
} else {
|
||||
connected_user_ids.push(user_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let sync_msg = CommunicationValue::new(CommunicationType::SyncClientIotaStatus)
|
||||
.add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids))
|
||||
.add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids))
|
||||
.add_typed_default(
|
||||
DataType::RhoConnections,
|
||||
DataValue::SignedNumber(self.rho.connection_count().await as i128),
|
||||
);
|
||||
let peer_capabilities = self.peer_capabilities.read().await.clone();
|
||||
let sync_msg = if peer_capabilities.session_snapshot_v1 {
|
||||
CommunicationValue::new(CommunicationType::SyncClientIotaStatus)
|
||||
.add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids))
|
||||
.add_typed_default(DataType::UserStates, DataValue::Array(connected_sessions))
|
||||
.add_typed_default(
|
||||
DataType::RhoConnections,
|
||||
DataValue::SignedNumber(self.rho.connection_count().await as i128),
|
||||
)
|
||||
} else {
|
||||
// Legacy Omega has no request/response contract for this message.
|
||||
// Its exact snapshot is IotaIds + UserIds + RhoConnections.
|
||||
CommunicationValue::new(CommunicationType::SyncClientIotaStatus)
|
||||
.add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids))
|
||||
.add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids))
|
||||
.add_typed_default(
|
||||
DataType::RhoConnections,
|
||||
DataValue::SignedNumber(self.rho.connection_count().await as i128),
|
||||
)
|
||||
};
|
||||
|
||||
self.send_message(&sync_msg).await;
|
||||
if !peer_capabilities.session_snapshot_v1 {
|
||||
if let Err(error) = self.send_message_result(&sync_msg).await {
|
||||
self.fail_synchronization("legacy route synchronization", error)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
*self.state.write().await = ConnectionState::SynchronizingSubscriptions;
|
||||
if let Err(error) = self.restore_presence_subscriptions().await {
|
||||
self.fail_synchronization("subscription restoration", error)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
*self.state.write().await = ConnectionState::Ready;
|
||||
return;
|
||||
}
|
||||
|
||||
let mut response = Err("route synchronization did not start".to_string());
|
||||
for attempt in 0..self.sync_retries {
|
||||
response = self
|
||||
.await_response(&sync_msg, Some(self.sync_timeout))
|
||||
.await;
|
||||
match &response {
|
||||
Ok(value) if value.is_type(CommunicationType::Success) => break,
|
||||
Ok(value) if value.is_type(CommunicationType::ErrorInvalidData) => break,
|
||||
_ if attempt + 1 < self.sync_retries => {
|
||||
sleep(Duration::from_millis(250 * 2u64.pow(attempt))).await
|
||||
}
|
||||
_ => break,
|
||||
}
|
||||
}
|
||||
match response {
|
||||
Ok(response) if response.is_type(CommunicationType::Success) => {
|
||||
*self.state.write().await = ConnectionState::SynchronizingSubscriptions;
|
||||
if let Err(error) = self.restore_presence_subscriptions().await {
|
||||
log_err!(
|
||||
0,
|
||||
PrintType::Omega,
|
||||
"Presence subscription restoration failed: {}",
|
||||
error
|
||||
);
|
||||
self.fail_synchronization("subscription restoration", error)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
*self.state.write().await = ConnectionState::Ready;
|
||||
}
|
||||
Ok(response) => {
|
||||
let error = format!(
|
||||
"Omega rejected presence snapshot with {}",
|
||||
response.get_type()
|
||||
);
|
||||
self.fail_synchronization("route synchronization", error)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
Err(error) => {
|
||||
self.fail_synchronization("route synchronization", error)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn fail_synchronization(&self, phase: &str, error: String) {
|
||||
log_err!(
|
||||
0,
|
||||
PrintType::Omega,
|
||||
"Omega {} failed after {} attempts: {}; closing transport for reconnect",
|
||||
phase,
|
||||
self.sync_retries,
|
||||
error
|
||||
);
|
||||
*self.state.write().await = ConnectionState::Disconnected;
|
||||
if let Some(sender) = self.sender.read().await.as_ref() {
|
||||
sender.close().await;
|
||||
}
|
||||
}
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
|
|
@ -392,6 +577,7 @@ impl OmegaConnection {
|
|||
self: Arc<Self>,
|
||||
connection: Arc<MTPConnection>,
|
||||
sender_handle: Arc<ConnectionHandle>,
|
||||
mut capability_tx: Option<oneshot::Sender<Result<PeerCapabilities, String>>>,
|
||||
) {
|
||||
// Monitor both receiver and sender handle for close
|
||||
let mut close_rx = sender_handle.subscribe_close();
|
||||
|
|
@ -403,6 +589,14 @@ impl OmegaConnection {
|
|||
Ok(cv) => {
|
||||
log_cv_in!(PrintType::Omega, &cv);
|
||||
|
||||
if cv.is_type(CommunicationType::IdentificationResponse) {
|
||||
if let Some(tx) = capability_tx.take() {
|
||||
let result = parse_omega_capability_response(&cv);
|
||||
let _ = tx.send(result);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
let msg_id = cv.get_id();
|
||||
if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) {
|
||||
if (task.task)(self.clone(), cv.clone()) {
|
||||
|
|
@ -430,6 +624,36 @@ impl OmegaConnection {
|
|||
}
|
||||
}
|
||||
}
|
||||
if cv.is_type(CommunicationType::EraseHostedUserData) {
|
||||
let Some(iota_id) = cv
|
||||
.get_data(DataType::IotaId)
|
||||
.as_number()
|
||||
.and_then(|id| i64::try_from(id).ok())
|
||||
else {
|
||||
log_err!(0, PrintType::Omega, "Discarded hosted-data erasure without IotaId");
|
||||
continue;
|
||||
};
|
||||
if let Some(rho) = self.rho.get_by_iota(iota_id).await {
|
||||
rho.get_iota_connection().send_message(&cv).await;
|
||||
} else {
|
||||
log_err!(iota_id, PrintType::Omega, "Hosted-data erasure is pending because the Iota is offline");
|
||||
}
|
||||
continue;
|
||||
}
|
||||
if cv.is_type(CommunicationType::ClientChanged) {
|
||||
let Some((receiver, session_id)) = client_changed_target(&cv)
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
if let Some(client) = self
|
||||
.rho
|
||||
.get_client_connection(receiver, session_id)
|
||||
.await
|
||||
{
|
||||
client.send_message(&cv).await;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
log_err!(0, PrintType::Omega, "Receive error: {}", e);
|
||||
|
|
@ -488,6 +712,14 @@ impl OmegaConnection {
|
|||
let _ = self.try_send_message(cv).await;
|
||||
}
|
||||
|
||||
pub async fn send_message_result(&self, cv: &CommunicationValue) -> Result<(), String> {
|
||||
self.try_send_message(cv).await
|
||||
}
|
||||
|
||||
pub async fn supports_set_user_state(&self) -> bool {
|
||||
self.peer_capabilities.read().await.set_user_state_v1
|
||||
}
|
||||
|
||||
pub async fn await_connection(&self, timeout_duration: Option<Duration>) -> Result<(), String> {
|
||||
if self.state.read().await.is_connected() {
|
||||
return Ok(());
|
||||
|
|
@ -607,72 +839,371 @@ impl OmegaConnection {
|
|||
self.state.read().await.is_identified()
|
||||
}
|
||||
|
||||
pub async fn is_ready(&self) -> bool {
|
||||
self.state.read().await.is_ready()
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub async fn close_iota(&self, iota_id: i64) {
|
||||
pub async fn close_iota(&self, iota_id: i64) -> Result<(), String> {
|
||||
let cv = CommunicationValue::new(CommunicationType::IotaDisconnected)
|
||||
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
|
||||
self.send_message(&cv).await;
|
||||
let request_id = cv.get_id();
|
||||
let result = self.lifecycle_request(&cv).await;
|
||||
if let Err(error) = &result {
|
||||
self.log_lifecycle_failure("IotaDisconnected", iota_id, None, request_id, error);
|
||||
self.sync_client_iota_status().await;
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
pub async fn client_changed(
|
||||
&self,
|
||||
_iota_id: i64,
|
||||
user_id: i64,
|
||||
session_id: i64,
|
||||
state: UserStatus,
|
||||
) {
|
||||
let cv = CommunicationValue::new(CommunicationType::ClientChanged)
|
||||
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
|
||||
.add_typed_default(
|
||||
DataType::SessionId,
|
||||
DataValue::SignedNumber(session_id.into()),
|
||||
)
|
||||
.add_typed_default(DataType::UserState, DataValue::Str(state.to_string()));
|
||||
self.send_message(&cv).await;
|
||||
}
|
||||
|
||||
pub async fn client_disconnected(&self, user_id: i64, session_id: i64) {
|
||||
pub async fn client_disconnected(&self, user_id: i64, session_id: i64) -> Result<(), String> {
|
||||
self.session_presence.remove(&(user_id, session_id));
|
||||
let cv = CommunicationValue::new(CommunicationType::UserDisconnected)
|
||||
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
|
||||
.add_typed_default(
|
||||
DataType::SessionId,
|
||||
DataValue::SignedNumber(session_id.into()),
|
||||
);
|
||||
self.send_message(&cv).await;
|
||||
let request_id = cv.get_id();
|
||||
let result = self.lifecycle_request(&cv).await;
|
||||
if let Err(error) = &result {
|
||||
self.log_lifecycle_failure(
|
||||
"UserDisconnected",
|
||||
user_id,
|
||||
Some(session_id),
|
||||
request_id,
|
||||
error,
|
||||
);
|
||||
self.sync_client_iota_status().await;
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
pub async fn user_states(&self, user_id: i64, user_ids: Vec<i64>) {
|
||||
let user_ids = user_ids
|
||||
.iter()
|
||||
.map(|v| DataValue::SignedNumber((*v).into()))
|
||||
.collect();
|
||||
|
||||
let cv = CommunicationValue::new(CommunicationType::GetStates)
|
||||
.add_typed_default(DataType::UserIds, DataValue::Array(user_ids));
|
||||
let msg_id = cv.get_id();
|
||||
|
||||
let rho_manager = self.rho.clone();
|
||||
self.waiting_tasks.insert(
|
||||
msg_id,
|
||||
WaitingTask {
|
||||
task: Box::new(
|
||||
move |_: Arc<OmegaConnection>, response: CommunicationValue| {
|
||||
let rho_manager = rho_manager.clone();
|
||||
tokio::spawn(async move {
|
||||
let rho = rho_manager.get_for_user(user_id).await;
|
||||
if let Some(rho) = rho {
|
||||
for client in rho.get_client_connections_for_user(user_id).await {
|
||||
client.send_message(&response).await;
|
||||
}
|
||||
}
|
||||
});
|
||||
true
|
||||
},
|
||||
pub async fn user_connected(
|
||||
&self,
|
||||
user_id: i64,
|
||||
session_id: i64,
|
||||
iota_id: i64,
|
||||
) -> Result<(), String> {
|
||||
let request = CommunicationValue::new(CommunicationType::UserConnected)
|
||||
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
|
||||
.add_typed_default(
|
||||
DataType::SessionId,
|
||||
DataValue::SignedNumber(session_id.into()),
|
||||
)
|
||||
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()))
|
||||
.add_typed_default(
|
||||
DataType::UserState,
|
||||
DataValue::Str("user_online".to_string()),
|
||||
)
|
||||
.add_typed_default(
|
||||
DataType::UpdatedAt,
|
||||
DataValue::SignedNumber(
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_millis() as i128,
|
||||
),
|
||||
inserted_at: Instant::now(),
|
||||
);
|
||||
let request_id = request.get_id();
|
||||
let result = self.lifecycle_request(&request).await;
|
||||
if let Err(error) = &result {
|
||||
self.log_lifecycle_failure(
|
||||
"UserConnected",
|
||||
user_id,
|
||||
Some(session_id),
|
||||
request_id,
|
||||
error,
|
||||
);
|
||||
self.sync_client_iota_status().await;
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
pub async fn iota_connected(&self, iota_id: i64) -> Result<(), String> {
|
||||
let request = CommunicationValue::new(CommunicationType::IotaConnected)
|
||||
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
|
||||
let request_id = request.get_id();
|
||||
let result = self.lifecycle_request(&request).await;
|
||||
if let Err(error) = &result {
|
||||
self.log_lifecycle_failure("IotaConnected", iota_id, None, request_id, error);
|
||||
self.sync_client_iota_status().await;
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
pub async fn reconcile_routes(&self) {
|
||||
self.sync_client_iota_status().await;
|
||||
}
|
||||
|
||||
async fn lifecycle_request(&self, request: &CommunicationValue) -> Result<(), String> {
|
||||
if !self.is_ready().await {
|
||||
return Err("Omega is not synchronized".to_string());
|
||||
}
|
||||
if !self.peer_capabilities.read().await.session_snapshot_v1 {
|
||||
// Legacy Omega handles lifecycle messages as notifications and
|
||||
// never returns Success. A transport-level send result is the
|
||||
// only acknowledgement available under that contract.
|
||||
return self.send_message_result(request).await;
|
||||
}
|
||||
let response = self
|
||||
.await_response(request, Some(Duration::from_secs(20)))
|
||||
.await?;
|
||||
if response.is_type(CommunicationType::Success) {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(format!("Omega rejected {}", response.get_type()))
|
||||
}
|
||||
}
|
||||
|
||||
fn log_lifecycle_failure(
|
||||
&self,
|
||||
operation: &str,
|
||||
entity_id: i64,
|
||||
session_id: Option<i64>,
|
||||
request_id: u32,
|
||||
error: &str,
|
||||
) {
|
||||
log_err!(
|
||||
0,
|
||||
PrintType::Omega,
|
||||
"Lifecycle operation={} entity_id={} session_id={:?} request_id={} failed: {}",
|
||||
operation,
|
||||
entity_id,
|
||||
session_id,
|
||||
request_id,
|
||||
error
|
||||
);
|
||||
}
|
||||
|
||||
pub async fn replace_state_subscription(
|
||||
&self,
|
||||
user_id: i64,
|
||||
session_id: i64,
|
||||
mut user_ids: Vec<i64>,
|
||||
) -> Result<(), String> {
|
||||
if user_id <= 0 || session_id <= 0 {
|
||||
return Err("invalid user or session id".to_string());
|
||||
}
|
||||
|
||||
user_ids.sort_unstable();
|
||||
user_ids.dedup();
|
||||
|
||||
let key = (user_id, session_id);
|
||||
// Store the latest authoritative snapshot before checking readiness.
|
||||
// A client can synchronize while the transport is reconnecting.
|
||||
self.session_presence.insert(
|
||||
key,
|
||||
SessionPresenceState {
|
||||
subscribed_user_ids: user_ids.clone(),
|
||||
pending: true,
|
||||
},
|
||||
);
|
||||
|
||||
self.send_message(&cv).await;
|
||||
if !self.is_ready().await {
|
||||
return Err("Omega is not synchronized".to_string());
|
||||
}
|
||||
|
||||
self.deliver_state_subscription(key, user_ids).await
|
||||
}
|
||||
|
||||
async fn deliver_state_subscription(
|
||||
&self,
|
||||
key: (i64, i64),
|
||||
user_ids: Vec<i64>,
|
||||
) -> Result<(), String> {
|
||||
let (user_id, session_id) = key;
|
||||
|
||||
if !self.peer_capabilities.read().await.state_subscribe_v1 {
|
||||
let delivered_ids = user_ids.clone();
|
||||
let result = self
|
||||
.request_user_states(user_id, session_id, user_ids)
|
||||
.await;
|
||||
if result.is_ok() {
|
||||
self.mark_presence_delivered(key, &delivered_ids).await;
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
let subscription = CommunicationValue::new(CommunicationType::StateSubscribe)
|
||||
.with_sender(user_id as u64)
|
||||
.add_typed_default(
|
||||
DataType::SessionId,
|
||||
DataValue::SignedNumber(session_id.into()),
|
||||
)
|
||||
.add_typed_default(
|
||||
DataType::UserIds,
|
||||
DataValue::Array(
|
||||
user_ids
|
||||
.iter()
|
||||
.map(|user_id| DataValue::SignedNumber((*user_id).into()))
|
||||
.collect(),
|
||||
),
|
||||
);
|
||||
let response = self
|
||||
.await_response(&subscription, Some(Duration::from_secs(20)))
|
||||
.await?;
|
||||
if !response.is_type(CommunicationType::Success) {
|
||||
return Err(format!(
|
||||
"StateSubscribe rejected with {}",
|
||||
response.get_type()
|
||||
));
|
||||
}
|
||||
|
||||
let delivered_ids = user_ids.clone();
|
||||
self.request_user_states(user_id, session_id, user_ids)
|
||||
.await?;
|
||||
self.mark_presence_delivered(key, &delivered_ids).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn restore_presence_subscriptions(&self) -> Result<(), String> {
|
||||
let mut active = Vec::new();
|
||||
for rho in self.rho.connections().await {
|
||||
active.extend(rho.get_client_connections().await);
|
||||
}
|
||||
let active_keys: std::collections::HashSet<_> = active
|
||||
.iter()
|
||||
.map(|client| (client.user_id as i64, client.session_id as i64))
|
||||
.collect();
|
||||
|
||||
// Do not replay subscriptions for sessions that disappeared while the
|
||||
// transport was down.
|
||||
self.session_presence
|
||||
.retain(|key, _| active_keys.contains(key));
|
||||
|
||||
for key in active_keys {
|
||||
let Some(snapshot) = self.session_presence.get(&key).map(|entry| entry.clone()) else {
|
||||
continue;
|
||||
};
|
||||
if !snapshot.pending {
|
||||
continue;
|
||||
}
|
||||
self.deliver_state_subscription(key, snapshot.subscribed_user_ids)
|
||||
.await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn mark_presence_delivered(&self, key: (i64, i64), delivered_ids: &[i64]) {
|
||||
if let Some(mut entry) = self.session_presence.get_mut(&key)
|
||||
&& entry.subscribed_user_ids == delivered_ids
|
||||
{
|
||||
entry.pending = false;
|
||||
}
|
||||
}
|
||||
|
||||
async fn request_user_states(
|
||||
&self,
|
||||
user_id: i64,
|
||||
session_id: i64,
|
||||
user_ids: Vec<i64>,
|
||||
) -> Result<(), String> {
|
||||
let request = CommunicationValue::new(CommunicationType::GetStates)
|
||||
.add_typed_default(
|
||||
DataType::SessionId,
|
||||
DataValue::SignedNumber(session_id.into()),
|
||||
)
|
||||
.add_typed_default(
|
||||
DataType::UserIds,
|
||||
DataValue::Array(
|
||||
user_ids
|
||||
.iter()
|
||||
.map(|user_id| DataValue::SignedNumber((*user_id).into()))
|
||||
.collect(),
|
||||
),
|
||||
);
|
||||
let response = self
|
||||
.await_response(&request, Some(Duration::from_secs(20)))
|
||||
.await?;
|
||||
let legacy_peer = !self.peer_capabilities.read().await.client_state_push_v1;
|
||||
let response =
|
||||
if legacy_peer && response.get_data(DataType::SessionId).as_number().is_none() {
|
||||
response.add_typed_default(
|
||||
DataType::SessionId,
|
||||
DataValue::SignedNumber(session_id.into()),
|
||||
)
|
||||
} else {
|
||||
response
|
||||
};
|
||||
if !legacy_peer
|
||||
&& response.get_data(DataType::SessionId).as_number() != Some(session_id as i128)
|
||||
{
|
||||
return Err("GetStates response had the wrong session".to_string());
|
||||
}
|
||||
|
||||
if let Some(client) = self.rho.get_client_connection(user_id, session_id).await {
|
||||
client.send_message(&response).await;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{PeerCapabilities, client_changed_target, parse_omega_capability_response};
|
||||
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
|
||||
|
||||
fn notification() -> CommunicationValue {
|
||||
CommunicationValue::new(CommunicationType::ClientChanged)
|
||||
.with_receiver(7)
|
||||
.add_typed_default(DataType::SessionId, DataValue::SignedNumber(11))
|
||||
.add_typed_default(DataType::UserId, DataValue::SignedNumber(20))
|
||||
.add_typed_default(DataType::UserState, DataValue::Str("user_online".into()))
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn client_changed_target_requires_all_routing_fields() {
|
||||
assert_eq!(client_changed_target(¬ification()), Some((7, 11)));
|
||||
|
||||
let mut missing_session = notification();
|
||||
missing_session.remove_data(DataType::SessionId);
|
||||
assert_eq!(client_changed_target(&missing_session), None);
|
||||
|
||||
let mut missing_state = notification();
|
||||
missing_state.remove_data(DataType::UserState);
|
||||
assert_eq!(client_changed_target(&missing_state), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn client_changed_target_rejects_non_positive_ids() {
|
||||
let invalid = CommunicationValue::new(CommunicationType::ClientChanged)
|
||||
.with_receiver(0)
|
||||
.add_typed_default(DataType::SessionId, DataValue::SignedNumber(11))
|
||||
.add_typed_default(DataType::UserId, DataValue::SignedNumber(20))
|
||||
.add_typed_default(DataType::UserState, DataValue::Str("user_online".into()));
|
||||
assert_eq!(client_changed_target(&invalid), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn capability_response_negotiates_new_omega() {
|
||||
let response = CommunicationValue::new(CommunicationType::IdentificationResponse)
|
||||
.add_typed_default(
|
||||
DataType::Description,
|
||||
DataValue::Str(
|
||||
"omega;caps=set_user_state_v1,state_subscribe_v1,session_snapshot_v1,client_state_push_v1"
|
||||
.into(),
|
||||
),
|
||||
);
|
||||
let capabilities = parse_omega_capability_response(&response).unwrap();
|
||||
assert!(capabilities.state_subscribe_v1);
|
||||
assert!(capabilities.client_state_push_v1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn missing_capability_response_is_legacy() {
|
||||
let response = CommunicationValue::new(CommunicationType::IdentificationResponse);
|
||||
let capabilities = parse_omega_capability_response(&response).unwrap();
|
||||
assert_eq!(capabilities, PeerCapabilities::default());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unsupported_capability_response_is_rejected() {
|
||||
let response = CommunicationValue::new(CommunicationType::IdentificationResponse)
|
||||
.add_typed_default(
|
||||
DataType::Description,
|
||||
DataValue::Str("omega;caps=unsupported_v9".into()),
|
||||
);
|
||||
assert!(parse_omega_capability_response(&response).is_err());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -24,7 +24,6 @@ pub struct ClientConnection {
|
|||
pub receiver: Arc<MtpReceiver>,
|
||||
pub_key: Arc<RwLock<Option<Vec<u8>>>>,
|
||||
pub rho_connection: Arc<RwLock<Option<Arc<RhoConnection>>>>,
|
||||
pub interested_users: Arc<RwLock<Vec<i64>>>,
|
||||
is_open: Arc<RwLock<bool>>,
|
||||
message_slots: Arc<tokio::sync::Semaphore>,
|
||||
}
|
||||
|
|
@ -35,7 +34,6 @@ impl ClientConnection {
|
|||
state: general.state.clone(),
|
||||
pub_key: Arc::new(RwLock::new(None)),
|
||||
rho_connection: general.rho_connection.clone(),
|
||||
interested_users: Arc::new(RwLock::new(Vec::new())),
|
||||
is_open: Arc::new(RwLock::new(true)),
|
||||
sender: general.sender.clone(),
|
||||
receiver: general.receiver.clone(),
|
||||
|
|
@ -79,6 +77,12 @@ impl ClientConnection {
|
|||
*self.rho_connection.write().await = Some(rho_connection);
|
||||
}
|
||||
|
||||
/// A membership snapshot can remove an account from an Iota while keeping
|
||||
/// its authenticated Omega connection alive.
|
||||
pub async fn clear_rho_connection(&self) {
|
||||
*self.rho_connection.write().await = None;
|
||||
}
|
||||
|
||||
/// Send a CommunicationValue to the client
|
||||
pub async fn send_message(self: Arc<Self>, cv: &CommunicationValue) {
|
||||
if !*self.is_open.read().await {
|
||||
|
|
@ -104,9 +108,26 @@ impl ClientConnection {
|
|||
|
||||
let mut cv = cv;
|
||||
|
||||
// Handle client status changes
|
||||
if cv.is_type(CommunicationType::ClientChanged) {
|
||||
self.handle_client_changed(cv).await;
|
||||
if cv.is_type(CommunicationType::SetUserState) {
|
||||
self.handle_set_user_state(cv).await;
|
||||
return;
|
||||
}
|
||||
|
||||
// Compatibility for clients predating SetUserState. The target
|
||||
// user fields, if present, are deliberately ignored: an
|
||||
// authenticated connection may only change its own state.
|
||||
if cv.is_type(CommunicationType::ClientChanged)
|
||||
&& cv.get_data_opt(DataType::UserState).is_some()
|
||||
{
|
||||
self.handle_set_user_state(
|
||||
CommunicationValue::new(CommunicationType::SetUserState)
|
||||
.with_id(cv.get_id())
|
||||
.add_typed_default(
|
||||
DataType::UserState,
|
||||
cv.get_data(DataType::UserState).clone(),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
|
||||
|
|
@ -191,6 +212,46 @@ impl ClientConnection {
|
|||
|| cv.is_type(CommunicationType::GetIotaData)
|
||||
|| cv.is_type(CommunicationType::DeleteUser)
|
||||
{
|
||||
if cv.is_type(CommunicationType::ChangeUserData)
|
||||
&& cv.get_data_opt(DataType::OnlineStatus).is_some()
|
||||
{
|
||||
let mut profile_request = cv.clone();
|
||||
let preference = profile_request.remove_data(DataType::OnlineStatus);
|
||||
let state_request = CommunicationValue::new(CommunicationType::SetUserState)
|
||||
.with_id(cv.get_id())
|
||||
.add_typed_default(DataType::UserState, preference.unwrap());
|
||||
let state_response = match self.request_set_user_state(state_request).await {
|
||||
Ok(response) => response,
|
||||
Err(error_type) => {
|
||||
self.send_error_response(cv.get_id(), error_type).await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
if !state_response.is_type(CommunicationType::Success) {
|
||||
self.send_message(&state_response).await;
|
||||
return;
|
||||
}
|
||||
|
||||
if profile_request.data_len() == 0 {
|
||||
self.send_message(&state_response).await;
|
||||
return;
|
||||
}
|
||||
|
||||
match self
|
||||
.await_omega_response(profile_request.with_sender(self.user_id))
|
||||
.await
|
||||
{
|
||||
Ok(response) if response.is_type(CommunicationType::Success) => {
|
||||
self.send_message(&response).await;
|
||||
}
|
||||
Ok(response) => self.send_message(&response).await,
|
||||
Err(_) => {
|
||||
self.send_error_response(cv.get_id(), CommunicationType::ErrorInternal)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
let sender = self.get_user_id().await;
|
||||
self.handle_omega_forward(cv.with_sender(sender as u64))
|
||||
.await;
|
||||
|
|
@ -290,34 +351,91 @@ impl ClientConnection {
|
|||
});
|
||||
}
|
||||
async fn handle_omega_forward(self: Arc<Self>, cv: CommunicationValue) {
|
||||
let client_for_closure = self.clone();
|
||||
let response_cv = self
|
||||
.state
|
||||
.omega
|
||||
.clone()
|
||||
.await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20)))
|
||||
.await;
|
||||
if let Ok(response_cv) = response_cv {
|
||||
client_for_closure.send_message(&response_cv).await;
|
||||
let request_id = cv.get_id();
|
||||
match self.await_omega_response(cv).await {
|
||||
Ok(response_cv) => self.send_message(&response_cv).await,
|
||||
Err(_) => {
|
||||
self.send_error_response(request_id, CommunicationType::ErrorInternal)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle client status change
|
||||
async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) {
|
||||
let user_id = self.get_user_id().await;
|
||||
if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) {
|
||||
let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online);
|
||||
if let Some(rho_conn) = self.get_rho_connection().await {
|
||||
self.state
|
||||
.omega
|
||||
.client_changed(
|
||||
rho_conn.get_iota_id().await as i64,
|
||||
user_id as i64,
|
||||
self.session_id as i64,
|
||||
user_status,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
async fn await_omega_response(&self, cv: CommunicationValue) -> Result<CommunicationValue, ()> {
|
||||
self.state
|
||||
.omega
|
||||
.clone()
|
||||
.await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20)))
|
||||
.await
|
||||
.map_err(|_| ())
|
||||
}
|
||||
|
||||
async fn request_set_user_state(
|
||||
&self,
|
||||
cv: CommunicationValue,
|
||||
) -> Result<CommunicationValue, CommunicationType> {
|
||||
if !self.state.omega.is_ready().await {
|
||||
return Err(CommunicationType::ErrorInternal);
|
||||
}
|
||||
let Some(state) = cv
|
||||
.get_data(DataType::UserState)
|
||||
.as_str()
|
||||
.and_then(UserStatus::from_client_preference)
|
||||
else {
|
||||
return Err(CommunicationType::ErrorInvalidData);
|
||||
};
|
||||
let Some(rho) = self.get_rho_connection().await else {
|
||||
return Err(CommunicationType::ErrorNoIota);
|
||||
};
|
||||
let request = if self.state.omega.supports_set_user_state().await {
|
||||
CommunicationValue::new(CommunicationType::SetUserState)
|
||||
.with_id(cv.get_id())
|
||||
.with_sender(self.user_id)
|
||||
.add_typed_default(
|
||||
DataType::UserId,
|
||||
DataValue::SignedNumber(self.user_id as i128),
|
||||
)
|
||||
.add_typed_default(
|
||||
DataType::SessionId,
|
||||
DataValue::SignedNumber(self.session_id as i128),
|
||||
)
|
||||
.add_typed_default(
|
||||
DataType::IotaId,
|
||||
DataValue::SignedNumber(rho.get_iota_id().await as i128),
|
||||
)
|
||||
.add_typed_default(DataType::UserState, DataValue::Str(state.to_string()))
|
||||
} else {
|
||||
// Legacy Omega accepts the original ClientChanged request. It is
|
||||
// intentionally limited to the authenticated user's ID and a
|
||||
// writable preference, with no derived connectivity state.
|
||||
CommunicationValue::new(CommunicationType::ClientChanged)
|
||||
.with_id(cv.get_id())
|
||||
.with_sender(self.user_id)
|
||||
.add_typed_default(
|
||||
DataType::UserId,
|
||||
DataValue::SignedNumber(self.user_id as i128),
|
||||
)
|
||||
.add_typed_default(DataType::UserState, DataValue::Str(state.to_string()))
|
||||
};
|
||||
if !self.state.omega.supports_set_user_state().await {
|
||||
self.state
|
||||
.omega
|
||||
.send_message_result(&request)
|
||||
.await
|
||||
.map_err(|_| CommunicationType::ErrorInternal)?;
|
||||
return Ok(CommunicationValue::new(CommunicationType::Success)
|
||||
.with_id(cv.get_id())
|
||||
.add_typed_default(DataType::UserState, DataValue::Str(state.to_string())));
|
||||
}
|
||||
self.await_omega_response(request)
|
||||
.await
|
||||
.map_err(|_| CommunicationType::ErrorInternal)
|
||||
}
|
||||
|
||||
async fn handle_set_user_state(self: Arc<Self>, cv: CommunicationValue) {
|
||||
match self.request_set_user_state(cv.clone()).await {
|
||||
Ok(response) => self.send_message(&response).await,
|
||||
Err(error_type) => self.send_error_response(cv.get_id(), error_type).await,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -830,35 +948,6 @@ impl ClientConnection {
|
|||
let _ = self.sender.close();
|
||||
}
|
||||
|
||||
/// Set interested users list
|
||||
pub async fn set_interested_users(self: Arc<Self>, interested_ids: Vec<i64>) {
|
||||
let mut interested_guard = self.interested_users.write().await;
|
||||
*interested_guard = interested_ids;
|
||||
}
|
||||
#[allow(dead_code)]
|
||||
pub async fn get_interested_users(self: Arc<Self>) -> Vec<i64> {
|
||||
let interested_guard = self.interested_users.read().await;
|
||||
interested_guard.clone()
|
||||
}
|
||||
|
||||
/// Check if interested in a user and send notification
|
||||
#[allow(dead_code)]
|
||||
pub async fn are_you_interested(self: Arc<Self>, user_id: i64, user_status: &str) {
|
||||
let interested_guard = self.clone().get_interested_users().await;
|
||||
if interested_guard.contains(&user_id) {
|
||||
let status = if user_status == "user_invisible" {
|
||||
"user_offline"
|
||||
} else {
|
||||
user_status
|
||||
};
|
||||
let notification = CommunicationValue::new(CommunicationType::ClientChanged)
|
||||
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
|
||||
.add_typed_default(DataType::UserState, DataValue::Str(status.to_string()));
|
||||
|
||||
self.send_message(¬ification).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle connection close
|
||||
pub async fn handle_close(self: Arc<Self>) {
|
||||
let user_id = self.get_user_id().await;
|
||||
|
|
@ -880,7 +969,6 @@ impl Clone for ClientConnection {
|
|||
client_version: self.client_version.clone(),
|
||||
pub_key: Arc::clone(&self.pub_key),
|
||||
rho_connection: Arc::clone(&self.rho_connection),
|
||||
interested_users: Arc::clone(&self.interested_users),
|
||||
is_open: Arc::clone(&self.is_open),
|
||||
message_slots: Arc::clone(&self.message_slots),
|
||||
}
|
||||
|
|
|
|||
|
|
@ -136,10 +136,23 @@ impl GeneralConnection {
|
|||
if let Some(rho_conn) = rho {
|
||||
rho_conn.bind_user_id(user_id).await;
|
||||
rho_conn.add_client_connection(client.clone()).await;
|
||||
self.notify_user_connected(user_id, rho_conn.get_iota_id().await as i64)
|
||||
.await;
|
||||
self.send_initial_client_state_request(&rho_conn, user_id)
|
||||
.await;
|
||||
match self
|
||||
.notify_user_connected(user_id, rho_conn.get_iota_id().await as i64)
|
||||
.await
|
||||
{
|
||||
Ok(()) => {
|
||||
self.send_initial_client_state_request(&rho_conn, user_id)
|
||||
.await;
|
||||
}
|
||||
Err(error) => {
|
||||
log_err!(
|
||||
user_id,
|
||||
PrintType::Client,
|
||||
"UserConnected acknowledgement failed: {}",
|
||||
error
|
||||
);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
log_err!(
|
||||
user_id,
|
||||
|
|
@ -155,29 +168,13 @@ impl GeneralConnection {
|
|||
true
|
||||
}
|
||||
|
||||
async fn notify_user_connected(&self, user_id: i64, iota_id: i64) {
|
||||
async fn notify_user_connected(&self, user_id: i64, iota_id: i64) -> Result<(), String> {
|
||||
let session_id = *self.session_id.read().await as i64;
|
||||
let notify = CommunicationValue::new(CommunicationType::UserConnected)
|
||||
.add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
|
||||
.add_typed_default(
|
||||
DataType::SessionId,
|
||||
DataValue::SignedNumber(session_id.into()),
|
||||
)
|
||||
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()))
|
||||
.add_typed_default(
|
||||
DataType::UserState,
|
||||
DataValue::Str("user_online".to_string()),
|
||||
)
|
||||
.add_typed_default(
|
||||
DataType::UpdatedAt,
|
||||
DataValue::SignedNumber(
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_millis() as i128,
|
||||
),
|
||||
);
|
||||
self.state.omega.clone().send_message(¬ify).await;
|
||||
self.state
|
||||
.omega
|
||||
.clone()
|
||||
.user_connected(user_id, session_id, iota_id)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn find_user_rho(&self, user_id: i64) -> Option<Arc<RhoConnection>> {
|
||||
|
|
@ -227,9 +224,6 @@ impl GeneralConnection {
|
|||
async fn migrate_iota(self: &Arc<Self>) {
|
||||
let id = self.id;
|
||||
log_in!(id as i64, PrintType::Iota, "Iota {} connected", id);
|
||||
let notify = CommunicationValue::new(CommunicationType::IotaConnected)
|
||||
.add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.into()));
|
||||
self.state.omega.clone().send_message(¬ify).await;
|
||||
|
||||
let iota = IotaConnection::from_general(self.clone(), id).await;
|
||||
let previous = self.state.rho.get_by_iota(id as i64).await;
|
||||
|
|
@ -246,6 +240,14 @@ impl GeneralConnection {
|
|||
}
|
||||
}
|
||||
self.state.rho.add(rho).await;
|
||||
if let Err(error) = self.state.omega.clone().iota_connected(id as i64).await {
|
||||
log_err!(
|
||||
id as i64,
|
||||
PrintType::Iota,
|
||||
"IotaConnected acknowledgement failed: {}",
|
||||
error
|
||||
);
|
||||
}
|
||||
self.load_iota_users(id).await;
|
||||
iota.start();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,6 +23,47 @@ use tokio::sync::mpsc;
|
|||
|
||||
use super::rho_connection::RhoConnection;
|
||||
|
||||
fn contact_snapshot(value: &CommunicationValue) -> Option<(i64, i64, Vec<i64>)> {
|
||||
if !value.is_type(CommunicationType::ClientStateSync)
|
||||
&& !value.is_type(CommunicationType::AddConversation)
|
||||
{
|
||||
return None;
|
||||
}
|
||||
|
||||
let user_id = i64::try_from(value.get_receiver())
|
||||
.ok()
|
||||
.filter(|id| *id > 0)?;
|
||||
let session_id = value
|
||||
.get_data(DataType::SessionId)
|
||||
.as_signed_number()
|
||||
.and_then(|id| i64::try_from(id).ok())
|
||||
.filter(|id| *id > 0)?;
|
||||
let DataValue::Array(values) = value.get_data(DataType::UserIds) else {
|
||||
return None;
|
||||
};
|
||||
|
||||
let user_id_type = data_type_id(DataType::UserId, &TypeMap::latest());
|
||||
let mut contact_ids = Vec::with_capacity(values.len());
|
||||
for value in values {
|
||||
let contact_id = match value {
|
||||
DataValue::SignedNumber(id) => i64::try_from(*id).ok(),
|
||||
DataValue::Container(entries) => entries.iter().find_map(|(key, value)| {
|
||||
(*key == user_id_type)
|
||||
.then(|| value.as_signed_number())
|
||||
.flatten()
|
||||
.and_then(|id| i64::try_from(id).ok())
|
||||
}),
|
||||
_ => None,
|
||||
}
|
||||
.filter(|id| *id > 0)?;
|
||||
contact_ids.push(contact_id);
|
||||
}
|
||||
contact_ids.sort_unstable();
|
||||
contact_ids.dedup();
|
||||
|
||||
Some((user_id, session_id, contact_ids))
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub struct IotaConnection {
|
||||
pub state: Arc<AppState>,
|
||||
|
|
@ -172,6 +213,29 @@ impl IotaConnection {
|
|||
}
|
||||
}
|
||||
|
||||
async fn replace_state_subscription(
|
||||
&self,
|
||||
user_id: i64,
|
||||
session_id: i64,
|
||||
contact_ids: Vec<i64>,
|
||||
) -> Result<(), String> {
|
||||
let Some(rho_connection) = self.get_rho_connection().await else {
|
||||
return Err("Rho connection is unavailable".to_string());
|
||||
};
|
||||
if rho_connection
|
||||
.get_client_connection(user_id, session_id)
|
||||
.await
|
||||
.is_none()
|
||||
{
|
||||
return Err("target client session is no longer connected".to_string());
|
||||
}
|
||||
|
||||
self.state
|
||||
.omega
|
||||
.replace_state_subscription(user_id, session_id, contact_ids)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Send a CommunicationValue to the Iota
|
||||
pub async fn send_message(&self, cv: &CommunicationValue) {
|
||||
log_cv_out!(PrintType::Iota, cv);
|
||||
|
|
@ -206,6 +270,34 @@ impl IotaConnection {
|
|||
cv
|
||||
};
|
||||
|
||||
// Iota's UserIds field is an authoritative replacement snapshot for
|
||||
// the requesting client session, including contact mutations.
|
||||
if let Some((user_id, session_id, contact_ids)) = contact_snapshot(&cv) {
|
||||
if let Err(error) = self
|
||||
.replace_state_subscription(user_id, session_id, contact_ids)
|
||||
.await
|
||||
{
|
||||
log_err!(
|
||||
self.iota_id as i64,
|
||||
PrintType::Omega,
|
||||
"Failed to replace presence subscription for user {} session {}: {}",
|
||||
user_id,
|
||||
session_id,
|
||||
error
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if cv.is_type(CommunicationType::StateSubscribe) {
|
||||
self.send_error_response(
|
||||
cv.get_id(),
|
||||
CommunicationType::ErrorInvalidData,
|
||||
Some("StateSubscribe must come from an authoritative contact snapshot"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
|
||||
// Handle GET_CHATS
|
||||
if cv.is_type(CommunicationType::GetChats) {
|
||||
self.handle_get_chats(cv).await;
|
||||
|
|
@ -350,6 +442,12 @@ impl IotaConnection {
|
|||
|| cv.is_type(CommunicationType::GetUserData)
|
||||
|| cv.is_type(CommunicationType::GetIotaData)
|
||||
|| cv.is_type(CommunicationType::DeleteIota)
|
||||
|| cv.is_type(CommunicationType::AttachUserBegin)
|
||||
|| cv.is_type(CommunicationType::AttachUserComplete)
|
||||
|| cv.is_type(CommunicationType::ReleaseUserFromIota)
|
||||
|| cv.is_type(CommunicationType::DeleteUserCredentialBegin)
|
||||
|| cv.is_type(CommunicationType::DeleteUserCredentialComplete)
|
||||
|| cv.is_type(CommunicationType::EraseHostedUserDataAck)
|
||||
{
|
||||
let sender = self.get_iota_id().await;
|
||||
|
||||
|
|
@ -517,7 +615,20 @@ impl IotaConnection {
|
|||
}
|
||||
|
||||
let mut interested_ids: Vec<i64> = Vec::new();
|
||||
let session_id = cv.get_data(DataType::SessionId).as_signed_number();
|
||||
let Some(session_id) = cv
|
||||
.get_data(DataType::SessionId)
|
||||
.as_signed_number()
|
||||
.and_then(|id| i64::try_from(id).ok())
|
||||
.filter(|id| *id > 0)
|
||||
else {
|
||||
self.forward_to_client(
|
||||
CommunicationValue::new(CommunicationType::ErrorInvalidData)
|
||||
.with_id(cv.get_id())
|
||||
.with_receiver(user_id),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
};
|
||||
let tm = TypeMap::latest();
|
||||
|
||||
// Presence interest is the complete contact set, independent of
|
||||
|
|
@ -658,20 +769,18 @@ impl IotaConnection {
|
|||
// ============================
|
||||
// Notify Omega
|
||||
// ============================
|
||||
self.state
|
||||
.omega
|
||||
.user_states(user_id as i64, interested_ids.clone())
|
||||
.await;
|
||||
|
||||
// ============================
|
||||
// Notify Rho
|
||||
// ============================
|
||||
if let Some(rho_conn) = self.get_rho_connection().await {
|
||||
if let Some(session_id) = session_id.and_then(|id| i64::try_from(id).ok()) {
|
||||
rho_conn
|
||||
.set_interested(user_id as i64, session_id, interested_ids)
|
||||
.await;
|
||||
}
|
||||
if let Err(error) = self
|
||||
.replace_state_subscription(user_id as i64, session_id, interested_ids.clone())
|
||||
.await
|
||||
{
|
||||
log_err!(
|
||||
self.iota_id as i64,
|
||||
PrintType::Omega,
|
||||
"Failed to replace presence subscription for user {} session {}: {}",
|
||||
user_id,
|
||||
session_id,
|
||||
error
|
||||
);
|
||||
}
|
||||
|
||||
// ============================
|
||||
|
|
@ -779,3 +888,48 @@ impl std::fmt::Debug for IotaConnection {
|
|||
.finish()
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::contact_snapshot;
|
||||
use crate::util::data_type_id;
|
||||
use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue, TypeMap};
|
||||
|
||||
fn snapshot(user_ids: DataValue) -> CommunicationValue {
|
||||
CommunicationValue::new(CommunicationType::ClientStateSync)
|
||||
.with_receiver(7)
|
||||
.add_typed_default(DataType::SessionId, DataValue::SignedNumber(11))
|
||||
.add_typed_default(DataType::UserIds, user_ids)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn contact_snapshot_deduplicates_flat_and_typed_ids() {
|
||||
let user_id_type = data_type_id(DataType::UserId, &TypeMap::latest());
|
||||
let parsed = contact_snapshot(&snapshot(DataValue::Array(vec![
|
||||
DataValue::SignedNumber(20),
|
||||
DataValue::Container(vec![(user_id_type, DataValue::SignedNumber(21))]),
|
||||
DataValue::SignedNumber(20),
|
||||
])))
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(parsed, (7, 11, vec![20, 21]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn contact_snapshot_accepts_empty_contact_sets() {
|
||||
assert_eq!(
|
||||
contact_snapshot(&snapshot(DataValue::Array(Vec::new()))),
|
||||
Some((7, 11, Vec::new()))
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn contact_snapshot_rejects_malformed_contact_sets() {
|
||||
assert_eq!(
|
||||
contact_snapshot(&snapshot(DataValue::Array(vec![DataValue::Str(
|
||||
"bad".into(),
|
||||
)]))),
|
||||
None
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
use super::{client_connection::ClientConnection, iota_connection::IotaConnection};
|
||||
|
||||
use crate::{data::user::UserStatus, rho::app_connection::AppConnection};
|
||||
use crate::{log_err, rho::app_connection::AppConnection};
|
||||
use dashmap::DashMap;
|
||||
use mtp::codec::{CommunicationValue, DataType};
|
||||
use std::sync::Arc;
|
||||
|
|
@ -86,6 +86,14 @@ impl RhoConnection {
|
|||
.collect()
|
||||
}
|
||||
|
||||
pub async fn detach_user_clients(&self, user_id: i64) {
|
||||
let clients = self.get_client_connections_for_user(user_id).await;
|
||||
for client in clients {
|
||||
client.clear_rho_connection().await;
|
||||
}
|
||||
self.client_connections.retain(|(id, _), _| *id != user_id as u64);
|
||||
}
|
||||
|
||||
#[allow(unused)]
|
||||
pub async fn get_app_connections(
|
||||
&self,
|
||||
|
|
@ -178,11 +186,21 @@ impl RhoConnection {
|
|||
return;
|
||||
}
|
||||
self.client_connections.remove(&key);
|
||||
self.iota_connection
|
||||
if let Err(error) = self
|
||||
.iota_connection
|
||||
.state
|
||||
.omega
|
||||
.client_disconnected(target_user_id, target_session_id)
|
||||
.await;
|
||||
.await
|
||||
{
|
||||
log_err!(
|
||||
target_user_id,
|
||||
crate::util::logger::PrintType::Client,
|
||||
"UserDisconnected acknowledgement failed for session {}: {}",
|
||||
target_session_id,
|
||||
error
|
||||
);
|
||||
}
|
||||
let remaining_for_user = self
|
||||
.client_connections
|
||||
.iter()
|
||||
|
|
@ -194,16 +212,6 @@ impl RhoConnection {
|
|||
.rho
|
||||
.remove_user_binding(target_user_id, self.get_iota_id().await as i64)
|
||||
.await;
|
||||
self.iota_connection
|
||||
.state
|
||||
.omega
|
||||
.client_changed(
|
||||
self.get_iota_id().await as i64,
|
||||
target_user_id,
|
||||
target_session_id,
|
||||
UserStatus::user_offline,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -228,7 +236,14 @@ impl RhoConnection {
|
|||
}
|
||||
|
||||
// Notify OmegaConnection
|
||||
self.iota_connection.state.omega.close_iota(iota_id).await;
|
||||
if let Err(error) = self.iota_connection.state.omega.close_iota(iota_id).await {
|
||||
log_err!(
|
||||
iota_id,
|
||||
crate::util::logger::PrintType::Iota,
|
||||
"IotaDisconnected acknowledgement failed: {}",
|
||||
error
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Send message from Iota to specific client
|
||||
|
|
@ -255,25 +270,6 @@ impl RhoConnection {
|
|||
self.iota_connection.send_message(&cv).await;
|
||||
}
|
||||
|
||||
/// Set interested users for a specific client
|
||||
pub async fn set_interested(&self, user_id: i64, session_id: i64, interested_ids: Vec<i64>) {
|
||||
if let Some(connection) = self.get_client_connection(user_id, session_id).await {
|
||||
connection.set_interested_users(interested_ids).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Check if clients are interested in a user
|
||||
#[allow(dead_code)]
|
||||
pub async fn are_they_interested(&self, user_id: i64, user_status: &str) {
|
||||
let connections = self.get_client_connections().await;
|
||||
for connection in connections.iter() {
|
||||
connection
|
||||
.clone()
|
||||
.are_you_interested(user_id, user_status)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Check if this RhoConnection contains a specific user ID
|
||||
#[allow(dead_code)]
|
||||
pub async fn contains_user(&self, user_id: &i64) -> bool {
|
||||
|
|
|
|||
|
|
@ -66,7 +66,11 @@ impl RhoManager {
|
|||
)
|
||||
.await;
|
||||
|
||||
let replacement = user_ids.iter().copied().collect::<std::collections::HashSet<_>>();
|
||||
for user_id in previous_users {
|
||||
if !replacement.contains(&user_id) {
|
||||
rho.detach_user_clients(user_id).await;
|
||||
}
|
||||
if self
|
||||
.users
|
||||
.get(&user_id)
|
||||
|
|
@ -135,4 +139,17 @@ impl RhoManager {
|
|||
.map(|entry| entry.value().clone())
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub async fn get_client_connection(
|
||||
&self,
|
||||
user_id: i64,
|
||||
session_id: i64,
|
||||
) -> Option<Arc<super::client_connection::ClientConnection>> {
|
||||
for rho in self.connections().await {
|
||||
if let Some(client) = rho.get_client_connection(user_id, session_id).await {
|
||||
return Some(client);
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue