+303 -20

Cargo.lock

··· 39 39 ] 40 40 41 41 [[package]] 42 42 + name = "allocator-api2" 43 43 + version = "0.2.21" 44 44 + source = "registry+https://github.com/rust-lang/crates.io-index" 45 45 + checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" 46 46 + 47 47 + [[package]] 42 48 name = "android-tzdata" 43 49 version = "0.1.1" 44 50 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 162 168 ] 163 169 164 170 [[package]] 171 171 + name = "async-lock" 172 172 + version = "3.4.0" 173 173 + source = "registry+https://github.com/rust-lang/crates.io-index" 174 174 + checksum = "ff6e472cdea888a4bd64f342f09b3f50e1886d32afe8df3d663c01140b811b18" 175 175 + dependencies = [ 176 176 + "event-listener", 177 177 + "event-listener-strategy", 178 178 + "pin-project-lite", 179 179 + ] 180 180 + 181 181 + [[package]] 165 182 name = "async-trait" 166 183 version = "0.1.87" 167 184 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 180 197 181 198 [[package]] 182 199 name = "atrium-api" 183 183 - version = "0.24.10" 200 200 + version = "0.25.0" 184 201 source = "registry+https://github.com/rust-lang/crates.io-index" 185 185 - checksum = "9c5d74937642f6b21814e82d80f54d55ebd985b681bffbe27c8a76e726c3c4db" 202 202 + checksum = "ea3ea578c768ec91082e424a8d139517b2cb5c75149bf3cec04371a1e74f00f2" 186 203 dependencies = [ 204 204 + "atrium-common", 187 205 "atrium-xrpc", 188 206 "chrono", 189 207 "http", ··· 195 213 "serde_json", 196 214 "thiserror 1.0.69", 197 215 "trait-variant", 216 216 + ] 217 217 + 218 218 + [[package]] 219 219 + name = "atrium-common" 220 220 + version = "0.1.0" 221 221 + source = "registry+https://github.com/rust-lang/crates.io-index" 222 222 + checksum = "168e558408847bfed69df1033a32fd051f7a037ebc90ea46e588ccb2bfbd7233" 223 223 + dependencies = [ 224 224 + "dashmap", 225 225 + "lru", 226 226 + "moka", 227 227 + "thiserror 1.0.69", 228 228 + "tokio", 229 229 + "trait-variant", 230 230 + "web-time", 198 231 ] 199 232 200 233 [[package]] ··· 577 610 checksum = "5b63caa9aa9397e2d9480a9b13673856c78d8ac123288526c37d7839f2a86990" 578 611 579 612 [[package]] 613 613 + name = "concurrent-queue" 614 614 + version = "2.5.0" 615 615 + source = "registry+https://github.com/rust-lang/crates.io-index" 616 616 + checksum = "4ca0197aee26d1ae37445ee532fefce43251d24cc7c166799f4d46817f1d3973" 617 617 + dependencies = [ 618 618 + "crossbeam-utils", 619 619 + ] 620 620 + 621 621 + [[package]] 580 622 name = "constellation" 581 623 version = "0.1.0" 582 624 dependencies = [ ··· 646 688 ] 647 689 648 690 [[package]] 691 691 + name = "crossbeam-channel" 692 692 + version = "0.5.14" 693 693 + source = "registry+https://github.com/rust-lang/crates.io-index" 694 694 + checksum = "06ba6d68e24814cb8de6bb986db8222d3a027d15872cabc0d18817bc3c0e4471" 695 695 + dependencies = [ 696 696 + "crossbeam-utils", 697 697 + ] 698 698 + 699 699 + [[package]] 649 700 name = "crossbeam-epoch" 650 701 version = "0.9.18" 651 702 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 716 767 ] 717 768 718 769 [[package]] 770 770 + name = "dashmap" 771 771 + version = "6.1.0" 772 772 + source = "registry+https://github.com/rust-lang/crates.io-index" 773 773 + checksum = "5041cc499144891f3790297212f32a74fb938e5136a14943f338ef9e0ae276cf" 774 774 + dependencies = [ 775 775 + "cfg-if", 776 776 + "crossbeam-utils", 777 777 + "hashbrown 0.14.5", 778 778 + "lock_api", 779 779 + "once_cell", 780 780 + "parking_lot_core", 781 781 + ] 782 782 + 783 783 + [[package]] 719 784 name = "data-encoding" 720 785 version = "2.7.0" 721 786 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 795 860 ] 796 861 797 862 [[package]] 863 863 + name = "event-listener" 864 864 + version = "5.4.0" 865 865 + source = "registry+https://github.com/rust-lang/crates.io-index" 866 866 + checksum = "3492acde4c3fc54c845eaab3eed8bd00c7a7d881f78bfc801e43a93dec1331ae" 867 867 + dependencies = [ 868 868 + "concurrent-queue", 869 869 + "parking", 870 870 + "pin-project-lite", 871 871 + ] 872 872 + 873 873 + [[package]] 874 874 + name = "event-listener-strategy" 875 875 + version = "0.5.3" 876 876 + source = "registry+https://github.com/rust-lang/crates.io-index" 877 877 + checksum = "3c3e4e0dd3673c1139bf041f3008816d9cf2946bbfac2945c09e523b8d7b05b2" 878 878 + dependencies = [ 879 879 + "event-listener", 880 880 + "pin-project-lite", 881 881 + ] 882 882 + 883 883 + [[package]] 798 884 name = "fastrand" 799 885 version = "2.3.0" 800 886 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 816 902 source = "registry+https://github.com/rust-lang/crates.io-index" 817 903 checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" 818 904 dependencies = [ 819 819 - "futures-core", 820 820 - "futures-sink", 821 821 - "nanorand", 822 905 "spin", 823 906 ] 824 907 ··· 922 1005 ] 923 1006 924 1007 [[package]] 1008 1008 + name = "generator" 1009 1009 + version = "0.8.4" 1010 1010 + source = "registry+https://github.com/rust-lang/crates.io-index" 1011 1011 + checksum = "cc6bd114ceda131d3b1d665eba35788690ad37f5916457286b32ab6fd3c438dd" 1012 1012 + dependencies = [ 1013 1013 + "cfg-if", 1014 1014 + "libc", 1015 1015 + "log", 1016 1016 + "rustversion", 1017 1017 + "windows", 1018 1018 + ] 1019 1019 + 1020 1020 + [[package]] 925 1021 name = "generic-array" 926 1022 version = "0.14.7" 927 1023 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 938 1034 checksum = "c4567c8db10ae91089c99af84c68c38da3ec2f087c3f82960bcdbf3656b6f4d7" 939 1035 dependencies = [ 940 1036 "cfg-if", 941 941 - "js-sys", 942 1037 "libc", 943 1038 "wasi 0.11.0+wasi-snapshot-preview1", 944 944 - "wasm-bindgen", 945 1039 ] 946 1040 947 1041 [[package]] ··· 995 1089 996 1090 [[package]] 997 1091 name = "hashbrown" 1092 1092 + version = "0.14.5" 1093 1093 + source = "registry+https://github.com/rust-lang/crates.io-index" 1094 1094 + checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" 1095 1095 + 1096 1096 + [[package]] 1097 1097 + name = "hashbrown" 998 1098 version = "0.15.2" 999 1099 source = "registry+https://github.com/rust-lang/crates.io-index" 1000 1100 checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289" 1001 1101 dependencies = [ 1102 1102 + "allocator-api2", 1103 1103 + "equivalent", 1002 1104 "foldhash", 1003 1105 ] 1004 1106 ··· 1390 1492 "atrium-api", 1391 1493 "chrono", 1392 1494 "clap", 1393 1393 - "flume", 1394 1495 "futures-util", 1395 1496 "log", 1396 1497 "serde", ··· 1542 1643 checksum = "04cbf5b083de1c7e0222a7a51dbfdba1cbe1c6ab0b15e29fff3f6c077fd9cd9f" 1543 1644 1544 1645 [[package]] 1646 1646 + name = "loom" 1647 1647 + version = "0.7.2" 1648 1648 + source = "registry+https://github.com/rust-lang/crates.io-index" 1649 1649 + checksum = "419e0dc8046cb947daa77eb95ae174acfbddb7673b4151f56d1eed8e93fbfaca" 1650 1650 + dependencies = [ 1651 1651 + "cfg-if", 1652 1652 + "generator", 1653 1653 + "scoped-tls", 1654 1654 + "tracing", 1655 1655 + "tracing-subscriber", 1656 1656 + ] 1657 1657 + 1658 1658 + [[package]] 1659 1659 + name = "lru" 1660 1660 + version = "0.12.5" 1661 1661 + source = "registry+https://github.com/rust-lang/crates.io-index" 1662 1662 + checksum = "234cf4f4a04dc1f57e24b96cc0cd600cf2af460d4161ac5ecdd0af8e1f3b2a38" 1663 1663 + dependencies = [ 1664 1664 + "hashbrown 0.15.2", 1665 1665 + ] 1666 1666 + 1667 1667 + [[package]] 1545 1668 name = "lz4-sys" 1546 1669 version = "1.11.1+lz4-1.10.0" 1547 1670 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1558 1681 checksum = "19b955cdeb2a02b9117f121ce63aa52d08ade45de53e48fe6a38b39c10f6f709" 1559 1682 dependencies = [ 1560 1683 "libc", 1684 1684 + ] 1685 1685 + 1686 1686 + [[package]] 1687 1687 + name = "matchers" 1688 1688 + version = "0.1.0" 1689 1689 + source = "registry+https://github.com/rust-lang/crates.io-index" 1690 1690 + checksum = "8263075bb86c5a1b1427b5ae862e8889656f126e9f77c484496e8b47cf5c5558" 1691 1691 + dependencies = [ 1692 1692 + "regex-automata 0.1.10", 1561 1693 ] 1562 1694 1563 1695 [[package]] ··· 1683 1815 ] 1684 1816 1685 1817 [[package]] 1818 1818 + name = "moka" 1819 1819 + version = "0.12.10" 1820 1820 + source = "registry+https://github.com/rust-lang/crates.io-index" 1821 1821 + checksum = "a9321642ca94a4282428e6ea4af8cc2ca4eac48ac7a6a4ea8f33f76d0ce70926" 1822 1822 + dependencies = [ 1823 1823 + "async-lock", 1824 1824 + "crossbeam-channel", 1825 1825 + "crossbeam-epoch", 1826 1826 + "crossbeam-utils", 1827 1827 + "event-listener", 1828 1828 + "futures-util", 1829 1829 + "loom", 1830 1830 + "parking_lot", 1831 1831 + "portable-atomic", 1832 1832 + "rustc_version", 1833 1833 + "smallvec", 1834 1834 + "tagptr", 1835 1835 + "thiserror 1.0.69", 1836 1836 + "uuid", 1837 1837 + ] 1838 1838 + 1839 1839 + [[package]] 1686 1840 name = "multibase" 1687 1841 version = "0.9.1" 1688 1842 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1702 1856 "core2", 1703 1857 "serde", 1704 1858 "unsigned-varint", 1705 1705 - ] 1706 1706 - 1707 1707 - [[package]] 1708 1708 - name = "nanorand" 1709 1709 - version = "0.7.0" 1710 1710 - source = "registry+https://github.com/rust-lang/crates.io-index" 1711 1711 - checksum = "6a51313c5820b0b02bd422f4b44776fbf47961755c74ce64afc73bfad10226c3" 1712 1712 - dependencies = [ 1713 1713 - "getrandom 0.2.15", 1714 1859 ] 1715 1860 1716 1861 [[package]] ··· 1753 1898 ] 1754 1899 1755 1900 [[package]] 1901 1901 + name = "nu-ansi-term" 1902 1902 + version = "0.46.0" 1903 1903 + source = "registry+https://github.com/rust-lang/crates.io-index" 1904 1904 + checksum = "77a8165726e8236064dbb45459242600304b42a5ea24ee2948e18e023bf7ba84" 1905 1905 + dependencies = [ 1906 1906 + "overload", 1907 1907 + "winapi", 1908 1908 + ] 1909 1909 + 1910 1910 + [[package]] 1756 1911 name = "num-conv" 1757 1912 version = "0.1.0" 1758 1913 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1835 1990 "pkg-config", 1836 1991 "vcpkg", 1837 1992 ] 1993 1993 + 1994 1994 + [[package]] 1995 1995 + name = "overload" 1996 1996 + version = "0.1.1" 1997 1997 + source = "registry+https://github.com/rust-lang/crates.io-index" 1998 1998 + checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39" 1999 1999 + 2000 2000 + [[package]] 2001 2001 + name = "parking" 2002 2002 + version = "2.2.1" 2003 2003 + source = "registry+https://github.com/rust-lang/crates.io-index" 2004 2004 + checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" 1838 2005 1839 2006 [[package]] 1840 2007 name = "parking_lot" ··· 2055 2222 dependencies = [ 2056 2223 "aho-corasick", 2057 2224 "memchr", 2058 2058 - "regex-automata", 2059 2059 - "regex-syntax", 2225 2225 + "regex-automata 0.4.9", 2226 2226 + "regex-syntax 0.8.5", 2227 2227 + ] 2228 2228 + 2229 2229 + [[package]] 2230 2230 + name = "regex-automata" 2231 2231 + version = "0.1.10" 2232 2232 + source = "registry+https://github.com/rust-lang/crates.io-index" 2233 2233 + checksum = "6c230d73fb8d8c1b9c0b3135c5142a8acee3a0558fb8db5cf1cb65f8d7862132" 2234 2234 + dependencies = [ 2235 2235 + "regex-syntax 0.6.29", 2060 2236 ] 2061 2237 2062 2238 [[package]] ··· 2067 2243 dependencies = [ 2068 2244 "aho-corasick", 2069 2245 "memchr", 2070 2070 - "regex-syntax", 2246 2246 + "regex-syntax 0.8.5", 2071 2247 ] 2248 2248 + 2249 2249 + [[package]] 2250 2250 + name = "regex-syntax" 2251 2251 + version = "0.6.29" 2252 2252 + source = "registry+https://github.com/rust-lang/crates.io-index" 2253 2253 + checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1" 2072 2254 2073 2255 [[package]] 2074 2256 name = "regex-syntax" ··· 2108 2290 checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2" 2109 2291 2110 2292 [[package]] 2293 2293 + name = "rustc_version" 2294 2294 + version = "0.4.1" 2295 2295 + source = "registry+https://github.com/rust-lang/crates.io-index" 2296 2296 + checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" 2297 2297 + dependencies = [ 2298 2298 + "semver", 2299 2299 + ] 2300 2300 + 2301 2301 + [[package]] 2111 2302 name = "rustix" 2112 2303 version = "0.38.44" 2113 2304 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2142 2333 ] 2143 2334 2144 2335 [[package]] 2336 2336 + name = "scoped-tls" 2337 2337 + version = "1.0.1" 2338 2338 + source = "registry+https://github.com/rust-lang/crates.io-index" 2339 2339 + checksum = "e1cf6437eb19a8f4a6cc0f7dca544973b0b78843adbfeb3683d1a94a0024a294" 2340 2340 + 2341 2341 + [[package]] 2145 2342 name = "scopeguard" 2146 2343 version = "1.2.0" 2147 2344 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2169 2366 "core-foundation-sys", 2170 2367 "libc", 2171 2368 ] 2369 2369 + 2370 2370 + [[package]] 2371 2371 + name = "semver" 2372 2372 + version = "1.0.26" 2373 2373 + source = "registry+https://github.com/rust-lang/crates.io-index" 2374 2374 + checksum = "56e6fa9c48d24d85fb3de5ad847117517440f6beceb7798af16b4a87d616b8d0" 2172 2375 2173 2376 [[package]] 2174 2377 name = "serde" ··· 2288 2491 ] 2289 2492 2290 2493 [[package]] 2494 2494 + name = "sharded-slab" 2495 2495 + version = "0.1.7" 2496 2496 + source = "registry+https://github.com/rust-lang/crates.io-index" 2497 2497 + checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" 2498 2498 + dependencies = [ 2499 2499 + "lazy_static", 2500 2500 + ] 2501 2501 + 2502 2502 + [[package]] 2291 2503 name = "shlex" 2292 2504 version = "1.3.0" 2293 2505 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2383 2595 ] 2384 2596 2385 2597 [[package]] 2598 2598 + name = "tagptr" 2599 2599 + version = "0.2.0" 2600 2600 + source = "registry+https://github.com/rust-lang/crates.io-index" 2601 2601 + checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" 2602 2602 + 2603 2603 + [[package]] 2386 2604 name = "tempfile" 2387 2605 version = "3.16.0" 2388 2606 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2437 2655 ] 2438 2656 2439 2657 [[package]] 2658 2658 + name = "thread_local" 2659 2659 + version = "1.1.8" 2660 2660 + source = "registry+https://github.com/rust-lang/crates.io-index" 2661 2661 + checksum = "8b9ef9bad013ada3808854ceac7b46812a6465ba368859a37e2100283d2d719c" 2662 2662 + dependencies = [ 2663 2663 + "cfg-if", 2664 2664 + "once_cell", 2665 2665 + ] 2666 2666 + 2667 2667 + [[package]] 2440 2668 name = "time" 2441 2669 version = "0.3.37" 2442 2670 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2609 2837 checksum = "e672c95779cf947c5311f83787af4fa8fffd12fb27e4993211a84bdfd9610f9c" 2610 2838 dependencies = [ 2611 2839 "once_cell", 2840 2840 + "valuable", 2841 2841 + ] 2842 2842 + 2843 2843 + [[package]] 2844 2844 + name = "tracing-log" 2845 2845 + version = "0.2.0" 2846 2846 + source = "registry+https://github.com/rust-lang/crates.io-index" 2847 2847 + checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" 2848 2848 + dependencies = [ 2849 2849 + "log", 2850 2850 + "once_cell", 2851 2851 + "tracing-core", 2852 2852 + ] 2853 2853 + 2854 2854 + [[package]] 2855 2855 + name = "tracing-subscriber" 2856 2856 + version = "0.3.19" 2857 2857 + source = "registry+https://github.com/rust-lang/crates.io-index" 2858 2858 + checksum = "e8189decb5ac0fa7bc8b96b7cb9b2701d60d48805aca84a238004d665fcc4008" 2859 2859 + dependencies = [ 2860 2860 + "matchers", 2861 2861 + "nu-ansi-term", 2862 2862 + "once_cell", 2863 2863 + "regex", 2864 2864 + "sharded-slab", 2865 2865 + "smallvec", 2866 2866 + "thread_local", 2867 2867 + "tracing", 2868 2868 + "tracing-core", 2869 2869 + "tracing-log", 2612 2870 ] 2613 2871 2614 2872 [[package]] ··· 2740 2998 checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" 2741 2999 2742 3000 [[package]] 3001 3001 + name = "uuid" 3002 3002 + version = "1.15.1" 3003 3003 + source = "registry+https://github.com/rust-lang/crates.io-index" 3004 3004 + checksum = "e0f540e3240398cce6128b64ba83fdbdd86129c16a3aa1a3a252efd66eb3d587" 3005 3005 + dependencies = [ 3006 3006 + "getrandom 0.3.1", 3007 3007 + ] 3008 3008 + 3009 3009 + [[package]] 3010 3010 + name = "valuable" 3011 3011 + version = "0.1.1" 3012 3012 + source = "registry+https://github.com/rust-lang/crates.io-index" 3013 3013 + checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" 3014 3014 + 3015 3015 + [[package]] 2743 3016 name = "vcpkg" 2744 3017 version = "0.2.15" 2745 3018 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2838 3111 version = "0.3.77" 2839 3112 source = "registry+https://github.com/rust-lang/crates.io-index" 2840 3113 checksum = "33b6dd2ef9186f1f2072e409e99cd22a975331a6b3591b12c764e0e55c60d5d2" 3114 3114 + dependencies = [ 3115 3115 + "js-sys", 3116 3116 + "wasm-bindgen", 3117 3117 + ] 3118 3118 + 3119 3119 + [[package]] 3120 3120 + name = "web-time" 3121 3121 + version = "1.1.0" 3122 3122 + source = "registry+https://github.com/rust-lang/crates.io-index" 3123 3123 + checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" 2841 3124 dependencies = [ 2842 3125 "js-sys", 2843 3126 "wasm-bindgen",

+1 -2

jetstream/Cargo.toml

··· 10 10 11 11 [dependencies] 12 12 async-trait = "0.1.83" 13 13 - atrium-api = { version = "0.24.7", default-features = false, features = [ 13 13 + atrium-api = { version = "0.25", default-features = false, features = [ 14 14 "namespace-appbsky", 15 15 ] } 16 16 tokio = { version = "1.41.1", features = ["full", "sync", "time"] } ··· 26 26 chrono = "0.4.38" 27 27 zstd = "0.13.2" 28 28 thiserror = "2.0.3" 29 29 - flume = "0.11.1" 30 29 log = "0.4.22" 31 30 tokio-util = "0.7.13" 32 31

+8 -4

jetstream/examples/arbitrary_record.rs

··· 38 38 ..Default::default() 39 39 }; 40 40 41 41 - let jetstream: JetstreamConnector<serde_json::Value> = JetstreamConnector::new(config)?; 42 42 - let receiver = jetstream.connect().await?; 41 41 + let jetstream = JetstreamConnector::new(config)?; 42 42 + let mut receiver = jetstream.connect().await?; 43 43 44 44 - println!("Listening for '{}' events on DIDs: {:?}", &*args.nsid, dids); 44 44 + println!( 45 45 + "Listening for '{}' events on DIDs: {:?}", 46 46 + args.nsid.as_str(), 47 47 + dids 48 48 + ); 45 49 46 46 - while let Ok(event) = receiver.recv_async().await { 50 50 + while let Some(event) = receiver.recv().await { 47 51 if let Commit(CommitEvent::Create { commit, .. }) = event { 48 52 println!("got record: {:?}", commit.record); 49 53 }

+10 -5

jetstream/examples/basic.rs

··· 41 41 }; 42 42 43 43 let jetstream = JetstreamConnector::new(config)?; 44 44 - let receiver = jetstream.connect().await?; 44 44 + let mut receiver = jetstream.connect().await?; 45 45 46 46 - println!("Listening for '{}' events on DIDs: {:?}", &*args.nsid, dids); 46 46 + println!( 47 47 + "Listening for '{}' events on DIDs: {:?}", 48 48 + args.nsid.as_str(), 49 49 + dids 50 50 + ); 47 51 48 48 - while let Ok(event) = receiver.recv_async().await { 52 52 + while let Some(event) = receiver.recv().await { 49 53 if let Commit(commit) = event { 50 54 match commit { 51 55 CommitEvent::Create { info: _, commit } => { 52 56 if let AppBskyFeedPost(record) = commit.record { 53 57 println!( 54 58 "New post created! ({})\n\n'{}'", 55 55 - commit.info.rkey, record.text 59 59 + commit.info.rkey.as_str(), 60 60 + record.text 56 61 ); 57 62 } 58 63 } 59 64 CommitEvent::Delete { info: _, commit } => { 60 60 - println!("A post has been deleted. ({})", commit.rkey); 65 65 + println!("A post has been deleted. ({})", commit.rkey.as_str()); 61 66 } 62 67 _ => {} 63 68 }

+2

jetstream/src/error.rs

··· 40 40 CompressionDecoderError(io::Error), 41 41 #[error("all receivers were dropped but the websocket connection failed to close cleanly")] 42 42 WebSocketCloseFailure, 43 43 + #[error("failed to send ping or pong: {0}")] 44 44 + PingPongError(#[from] tokio_tungstenite::tungstenite::Error), 43 45 }

+1 -1

jetstream/src/events/commit.rs

··· 42 42 /// The type of commit operation that was performed. 43 43 pub operation: CommitType, 44 44 pub rev: String, 45 45 - pub rkey: String, 45 45 + pub rkey: exports::RecordKey, 46 46 /// The NSID of the record type that this commit is associated with. 47 47 pub collection: exports::Nsid, 48 48 }

+1

jetstream/src/exports.rs

··· 5 5 Did, 6 6 Handle, 7 7 Nsid, 8 8 + RecordKey, 8 9 };

+23 -11

jetstream/src/lib.rs

··· 24 24 use serde::de::DeserializeOwned; 25 25 use tokio::{ 26 26 net::TcpStream, 27 27 - sync::Mutex, 27 27 + sync::{ 28 28 + mpsc::{ 29 29 + channel, 30 30 + Receiver, 31 31 + Sender, 32 32 + }, 33 33 + Mutex, 34 34 + }, 28 35 }; 29 36 use tokio_tungstenite::{ 30 37 connect_async, ··· 113 120 const JETSTREAM_ZSTD_DICTIONARY: &[u8] = include_bytes!("../zstd/dictionary"); 114 121 115 122 /// A receiver channel for consuming Jetstream events. 116 116 - pub type JetstreamReceiver<R> = flume::Receiver<JetstreamEvent<R>>; 123 123 + pub type JetstreamReceiver<R> = Receiver<JetstreamEvent<R>>; 117 124 118 125 /// An internal sender channel for sending Jetstream events to [JetstreamReceiver]'s. 119 119 - type JetstreamSender<R> = flume::Sender<JetstreamEvent<R>>; 126 126 + type JetstreamSender<R> = Sender<JetstreamEvent<R>>; 120 127 121 128 /// A wrapper connector type for working with a WebSocket connection to a Jetstream instance to 122 129 /// receive and consume events. See [JetstreamConnector::connect] for more info. ··· 191 198 wanted_dids: Vec::new(), 192 199 compression: JetstreamCompression::None, 193 200 cursor: None, 194 194 - channel_size: 1024, 201 201 + channel_size: 4096, // a few seconds of firehose buffer 195 202 record_type: PhantomData, 196 203 } 197 204 } ··· 274 281 .validate() 275 282 .map_err(ConnectionError::InvalidConfig)?; 276 283 277 277 - let (send_channel, receive_channel) = flume::bounded(self.config.channel_size); 284 284 + let (send_channel, receive_channel) = channel(self.config.channel_size); 278 285 279 286 let configured_endpoint = self 280 287 .config ··· 306 313 } 307 314 308 315 if retry_attempt >= max_retries { 316 316 + eprintln!("max retries, bye"); 309 317 break; 310 318 } 319 319 + 320 320 + eprintln!("will try to reconnect"); 311 321 312 322 // Exponential backoff 313 323 let delay_ms = base_delay_ms * (2_u64.pow(retry_attempt)); ··· 344 354 let false = ping_cancelled.is_cancelled() else { 345 355 break; 346 356 }; 347 347 - log::trace!("Sending ping"); 348 357 match ping_shared_socket_write 349 358 .lock() 350 359 .await 351 351 - .send(Message::Ping("ping".as_bytes().to_vec())) 360 360 + .send(Message::Ping("ping!!!!".as_bytes().to_vec())) 352 361 .await 353 362 { 354 363 Ok(_) => (), 355 364 Err(error) => { 365 365 + eprintln!("ping send failed"); 356 366 log::error!("Ping failed: {error}"); 357 367 break; 358 368 } 359 369 } 360 370 } 371 371 + eprintln!("oh this is bad news."); 361 372 }); 362 373 363 374 let mut closing_connection = false; ··· 369 380 let event = serde_json::from_str(&json) 370 381 .map_err(JetstreamEventError::ReceivedMalformedJSON)?; 371 382 372 372 - if send_channel.send(event).is_err() { 383 383 + if send_channel.send(event).await.is_err() { 373 384 // We can assume that all receivers have been dropped, so we can close 374 385 // the connection and exit the task. 375 386 log::info!( ··· 394 405 let event = serde_json::from_str(&json) 395 406 .map_err(JetstreamEventError::ReceivedMalformedJSON)?; 396 407 397 397 - if send_channel.send(event).is_err() { 408 408 + if send_channel.send(event).await.is_err() { 398 409 // We can assume that all receivers have been dropped, so we can close 399 410 // the connection and exit the task. 400 411 log::info!( ··· 405 416 } 406 417 Message::Ping(vec) => { 407 418 log::trace!("Ping recieved, responding"); 408 408 - _ = shared_socket_write 419 419 + shared_socket_write 409 420 .lock() 410 421 .await 411 422 .send(Message::Pong(vec)) 412 412 - .await; 423 423 + .await 424 424 + .map_err(JetstreamEventError::PingPongError)?; 413 425 } 414 426 Message::Close(close_frame) => { 415 427 if let Some(close_frame) = close_frame {

+148

ufos/src/consumer.rs

··· 1 1 + use jetstream::exports::Did; 2 2 + use jetstream::{ 3 3 + events::{ 4 4 + account::AccountEvent, 5 5 + commit::{CommitData, CommitEvent, CommitInfo}, 6 6 + EventInfo, JetstreamEvent, 7 7 + }, 8 8 + DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector, 9 9 + JetstreamReceiver, 10 10 + }; 11 11 + use std::mem; 12 12 + use std::time::Duration; 13 13 + use tokio::sync::mpsc::{channel, error::TrySendError, Receiver, Sender}; 14 14 + 15 15 + use crate::{DeleteRecord, EventBatch, SetRecord}; 16 16 + 17 17 + const MAX_BATCHED_RECORDS: usize = 32; // non-blocking limit. drops oldest batched record per collection. 18 18 + const MAX_BATCHED_COLLECTIONS: usize = 256; // block at this point. pretty arbitrary, limit unbounded growth since during replay it could grow a lot. 19 19 + const MAX_BATCHED_DELETES: usize = 1024; // block at this point. fairly arbitrary, limit unbounded. 20 20 + const MAX_ACCOUNT_REMOVES: usize = 1; // block at this point. these can be heavy so hold at each one. 21 21 + 22 22 + const SEND_TIMEOUT_S: f64 = 3.; 23 23 + 24 24 + #[derive(Debug)] 25 25 + struct Batcher { 26 26 + jetstream_receiver: JetstreamReceiver<serde_json::Value>, 27 27 + batch_sender: Sender<EventBatch>, 28 28 + current_batch: EventBatch, 29 29 + } 30 30 + 31 31 + pub async fn consume(jetstream_endpoint: &str) -> anyhow::Result<Receiver<EventBatch>> { 32 32 + let config: JetstreamConfig<serde_json::Value> = JetstreamConfig { 33 33 + endpoint: DefaultJetstreamEndpoints::endpoint_or_shortcut(jetstream_endpoint), 34 34 + compression: JetstreamCompression::Zstd, 35 35 + ..Default::default() 36 36 + }; 37 37 + let jetstream_receiver = JetstreamConnector::new(config)?.connect().await?; 38 38 + let (batch_sender, batch_reciever) = channel::<EventBatch>(1); // *almost* rendezvous: one message in the middle 39 39 + let mut batcher = Batcher::new(jetstream_receiver, batch_sender); 40 40 + tokio::task::spawn(async move { batcher.run().await }); 41 41 + Ok(batch_reciever) 42 42 + } 43 43 + 44 44 + impl Batcher { 45 45 + fn new( 46 46 + jetstream_receiver: JetstreamReceiver<serde_json::Value>, 47 47 + batch_sender: Sender<EventBatch>, 48 48 + ) -> Self { 49 49 + Self { 50 50 + jetstream_receiver, 51 51 + batch_sender, 52 52 + current_batch: Default::default(), 53 53 + } 54 54 + } 55 55 + 56 56 + async fn run(&mut self) -> anyhow::Result<()> { 57 57 + loop { 58 58 + if let Some(event) = self.jetstream_receiver.recv().await { 59 59 + self.handle_event(event).await? 60 60 + } else { 61 61 + anyhow::bail!("channel closed"); 62 62 + } 63 63 + } 64 64 + } 65 65 + 66 66 + async fn handle_event( 67 67 + &mut self, 68 68 + event: JetstreamEvent<serde_json::Value>, 69 69 + ) -> anyhow::Result<()> { 70 70 + let batch_full = match event { 71 71 + JetstreamEvent::Commit(CommitEvent::Create { commit, info }) => { 72 72 + self.handle_set_record(true, commit, info) 73 73 + } 74 74 + JetstreamEvent::Commit(CommitEvent::Update { commit, info }) => { 75 75 + self.handle_set_record(false, commit, info) 76 76 + } 77 77 + JetstreamEvent::Commit(CommitEvent::Delete { commit, info }) => { 78 78 + self.handle_delete_record(commit, info) 79 79 + } 80 80 + JetstreamEvent::Identity(_) => false, // identity events are noops for us 81 81 + JetstreamEvent::Account(AccountEvent { info, account }) if !account.active => { 82 82 + self.handle_remove_account(info.did) 83 83 + } 84 84 + JetstreamEvent::Account(_) => false, // ignore account *activations* 85 85 + }; 86 86 + if batch_full { 87 87 + self.batch_sender 88 88 + .send_timeout( 89 89 + mem::take(&mut self.current_batch), 90 90 + Duration::from_secs_f64(SEND_TIMEOUT_S), 91 91 + ) 92 92 + .await?; 93 93 + } else { 94 94 + match self.batch_sender.try_reserve() { 95 95 + Ok(permit) => permit.send(mem::take(&mut self.current_batch)), 96 96 + Err(TrySendError::Full(())) => {} // no worries if not, keep batching while waiting for capacity 97 97 + Err(TrySendError::Closed(())) => anyhow::bail!("batch channel closed"), 98 98 + } 99 99 + } 100 100 + Ok(()) 101 101 + } 102 102 + 103 103 + fn handle_set_record( 104 104 + &mut self, 105 105 + new: bool, 106 106 + commit: CommitData<serde_json::Value>, 107 107 + info: EventInfo, 108 108 + ) -> bool { 109 109 + let record = SetRecord { 110 110 + new, 111 111 + did: info.did, 112 112 + rkey: commit.info.rkey, 113 113 + record: commit.record, 114 114 + }; 115 115 + let mut created_collection = false; 116 116 + let collection = self 117 117 + .current_batch 118 118 + .records 119 119 + .entry(commit.info.collection) 120 120 + .or_insert_with(|| { 121 121 + created_collection = true; 122 122 + Default::default() 123 123 + }); 124 124 + collection.push_front(record); 125 125 + collection.truncate(MAX_BATCHED_RECORDS); 126 126 + 127 127 + if created_collection { 128 128 + self.current_batch.records.len() >= MAX_BATCHED_COLLECTIONS // full if we have collections to the max 129 129 + } else { 130 130 + false 131 131 + } 132 132 + } 133 133 + 134 134 + fn handle_delete_record(&mut self, commit_info: CommitInfo, info: EventInfo) -> bool { 135 135 + let rm = DeleteRecord { 136 136 + did: info.did, 137 137 + collection: commit_info.collection, 138 138 + rkey: commit_info.rkey, 139 139 + }; 140 140 + self.current_batch.record_deletes.push(rm); 141 141 + self.current_batch.record_deletes.len() >= MAX_BATCHED_DELETES 142 142 + } 143 143 + 144 144 + fn handle_remove_account(&mut self, did: Did) -> bool { 145 145 + self.current_batch.account_removes.push(did); 146 146 + self.current_batch.account_removes.len() >= MAX_ACCOUNT_REMOVES 147 147 + } 148 148 + }

+27

ufos/src/lib.rs

··· 1 1 + pub mod consumer; 2 2 + pub mod store; 3 3 + 4 4 + use jetstream::exports::{Did, Nsid, RecordKey}; 5 5 + use std::collections::{HashMap, VecDeque}; 6 6 + 7 7 + #[derive(Debug)] 8 8 + pub struct SetRecord { 9 9 + pub new: bool, 10 10 + pub did: Did, 11 11 + pub rkey: RecordKey, 12 12 + pub record: serde_json::Value, 13 13 + } 14 14 + 15 15 + #[derive(Debug)] 16 16 + pub struct DeleteRecord { 17 17 + pub did: Did, 18 18 + pub collection: Nsid, 19 19 + pub rkey: RecordKey, 20 20 + } 21 21 + 22 22 + #[derive(Debug, Default)] 23 23 + pub struct EventBatch { 24 24 + pub records: HashMap<Nsid, VecDeque<SetRecord>>, 25 25 + pub record_deletes: Vec<DeleteRecord>, 26 26 + pub account_removes: Vec<Did>, 27 27 + }

+4 -70

ufos/src/main.rs

··· 1 1 use clap::Parser; 2 2 use std::path::PathBuf; 3 3 - use std::time::{Duration, Instant}; 4 4 - 5 5 - use tokio::select; 6 6 - use tokio_util::sync::CancellationToken; 7 7 - 8 8 - use jetstream::{ 9 9 - events::{commit::CommitEvent, JetstreamEvent::Commit}, 10 10 - DefaultJetstreamEndpoints, JetstreamCompression, JetstreamConfig, JetstreamConnector, 11 11 - }; 3 3 + use ufos::consumer; 4 4 + use ufos::store; 12 5 13 6 /// Aggregate links in the at-mosphere 14 7 #[derive(Parser, Debug)] ··· 26 19 #[tokio::main] 27 20 async fn main() -> anyhow::Result<()> { 28 21 let args = Args::parse(); 29 29 - 30 30 - let config: JetstreamConfig<serde_json::Value> = JetstreamConfig { 31 31 - endpoint: DefaultJetstreamEndpoints::endpoint_or_shortcut(&args.jetstream), 32 32 - compression: JetstreamCompression::Zstd, 33 33 - ..Default::default() 34 34 - }; 35 35 - 36 36 - let stay_alive = CancellationToken::new(); 37 37 - 38 38 - ctrlc::set_handler({ 39 39 - let mut desperation: u8 = 0; 40 40 - let stay_alive = stay_alive.clone(); 41 41 - move || match desperation { 42 42 - 0 => { 43 43 - println!("ok, signalling shutdown..."); 44 44 - stay_alive.cancel(); 45 45 - desperation += 1; 46 46 - } 47 47 - 1.. => panic!("fine, panicking!"), 48 48 - } 49 49 - })?; 50 50 - 51 51 - let jetstream: JetstreamConnector<serde_json::Value> = JetstreamConnector::new(config)?; 52 52 - let receiver = jetstream.connect().await?; 53 53 - 54 54 - println!("Jetstream ready"); 55 55 - 56 56 - let print_throttle = Duration::from_millis(400); 57 57 - let mut last = Instant::now(); 58 58 - loop { 59 59 - select! { 60 60 - _ = stay_alive.cancelled() => { 61 61 - eprintln!("byeeee"); 62 62 - break 63 63 - } 64 64 - ev = receiver.recv_async() => { 65 65 - match ev { 66 66 - Ok(event) => { 67 67 - if let Commit(CommitEvent::Create { commit, .. }) = event { 68 68 - let now = Instant::now(); 69 69 - let since = now - last; 70 70 - if since >= print_throttle { 71 71 - let overshoot = since - print_throttle; // adjust to keep the rate on average 72 72 - last = now - overshoot; 73 73 - println!( 74 74 - "{}: {}", 75 75 - &*commit.info.collection, 76 76 - serde_json::to_string(&commit.record)? 77 77 - ); 78 78 - } 79 79 - } 80 80 - }, 81 81 - Err(e) => { 82 82 - eprintln!("jetstream event error: {e:?}"); 83 83 - break 84 84 - } 85 85 - } 86 86 - } 87 87 - } 88 88 - } 89 89 - 22 22 + let batches = consumer::consume(&args.jetstream).await?; 23 23 + store::receive(batches).await?; 90 24 Ok(()) 91 25 }

+32

ufos/src/store.rs

··· 1 1 + use crate::EventBatch; 2 2 + use std::time::Duration; 3 3 + use tokio::sync::mpsc::Receiver; 4 4 + use tokio::time::sleep; 5 5 + 6 6 + pub async fn receive(mut receiver: Receiver<EventBatch>) -> anyhow::Result<()> { 7 7 + loop { 8 8 + eprintln!("receive loop, sleeping:"); 9 9 + sleep(Duration::from_secs_f64(0.5)).await; 10 10 + eprintln!("slept."); 11 11 + if let Some(batch) = receiver.recv().await { 12 12 + eprintln!("got batch"); 13 13 + summarize(batch) 14 14 + } else { 15 15 + anyhow::bail!("receive channel closed") 16 16 + } 17 17 + } 18 18 + } 19 19 + 20 20 + fn summarize(batch: EventBatch) { 21 21 + let EventBatch { 22 22 + records, 23 23 + record_deletes, 24 24 + account_removes, 25 25 + } = batch; 26 26 + println!( 27 27 + "got batch with {} collections, {} record deletes, {} account removes", 28 28 + records.len(), 29 29 + record_deletes.len(), 30 30 + account_removes.len() 31 31 + ); 32 32 + }