From b7b25b7cb7a3521988cea04ffcd10966628c42cf Mon Sep 17 00:00:00 2001 From: Rob Ede Date: Wed, 30 Sep 2020 13:58:18 +0100 Subject: [PATCH 1/3] update hyper and futures ecosystem --- Cargo.lock | 718 ++++++++++++++------------------ Cargo.toml | 20 +- rust-toolchain | 2 + src/engine/fut/analyze.rs | 101 ++--- src/engine/fut/crawl.rs | 50 ++- src/engine/fut/mod.rs | 2 +- src/engine/machines/crawler.rs | 2 +- src/engine/mod.rs | 259 ++++++------ src/interactors/bitbucket.rs | 10 +- src/interactors/crates.rs | 149 ++++--- src/interactors/github.rs | 99 +++-- src/interactors/gitlab.rs | 2 +- src/interactors/mod.rs | 70 ++-- src/interactors/rustsec.rs | 37 +- src/main.rs | 62 +-- src/models/crates.rs | 5 +- src/parsers/manifest.rs | 2 +- src/server/mod.rs | 373 +++++++++-------- src/server/views/html/mod.rs | 2 +- src/server/views/html/status.rs | 3 +- src/utils/cache.rs | 141 +++---- 21 files changed, 1043 insertions(+), 1066 deletions(-) create mode 100644 rust-toolchain diff --git a/Cargo.lock b/Cargo.lock index 8ad844c..483ae1f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -12,6 +12,12 @@ version = "1.0.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6b602bfe940d21c130f3895acd65221e8a61270debe89d628b9cb4e3ccb8569b" +[[package]] +name = "arc-swap" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d25d88fd6b8041580a654f9d0c581a047baee2b3efee13275f2fc392fc75034" + [[package]] name = "autocfg" version = "1.0.1" @@ -74,22 +80,17 @@ checksum = "08c48aae112d48ed9f069b33538ea9e3e90aa263cfa3d1c24309612b1f7472de" [[package]] name = "bytes" -version = "0.4.12" +version = "0.5.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "206fdffcfa2df7cbe15601ef46c813fce0965eb3286db6b56c583b814b51c81c" -dependencies = [ - "byteorder", - "either", - "iovec", -] +checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38" [[package]] name = "cadence" -version = "0.13.2" +version = "0.21.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "99612ce0a00efdaf3d81a5e8e17f0eed55a10e862033183c847a0365983af88c" +checksum = "e45b9cdf75cdddb0877f8af74c345d06b0c8a924c5115d2467d94d7e4bdf9180" dependencies = [ - "crossbeam", + "crossbeam-channel", ] [[package]] @@ -121,9 +122,9 @@ checksum = "4785bdd1c96b2a846b2bd7cc02e86b6b3dbf14e7e53446c4f54c92a361040822" [[package]] name = "chrono" -version = "0.4.18" +version = "0.4.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d021fddb7bd3e734370acfa4a83f34095571d8570c039f1420d77540f68d5772" +checksum = "670ad68c9088c2a963aaa298cb369688cf3f9465ce5e2d4ca10e6e0098a1ce73" dependencies = [ "libc", "num-integer", @@ -133,15 +134,6 @@ dependencies = [ "winapi 0.3.9", ] -[[package]] -name = "cloudabi" -version = "0.0.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ddfc5b9aa5d4507acaf872de71051dfd0e309860e88966e1051e462a077aac4f" -dependencies = [ - "bitflags", -] - [[package]] name = "core-foundation" version = "0.7.0" @@ -175,44 +167,11 @@ dependencies = [ ] [[package]] -name = "crossbeam" -version = "0.2.12" +name = "crossbeam-channel" +version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd66663db5a988098a89599d4857919b3acf7f61402e61365acfd3919857b9be" - -[[package]] -name = "crossbeam-deque" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9f02af974daeee82218205558e51ec8768b48cf524bd01d550abe5573a608285" +checksum = "b153fe7cbef478c567df0f972e02e6d736db11affe43dfc9c56a9374d1adfb87" dependencies = [ - "crossbeam-epoch", - "crossbeam-utils", - "maybe-uninit", -] - -[[package]] -name = "crossbeam-epoch" -version = "0.8.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "058ed274caafc1f60c4997b5fc07bf7dc7cca454af7c6e81edffe5f33f70dace" -dependencies = [ - "autocfg", - "cfg-if", - "crossbeam-utils", - "lazy_static", - "maybe-uninit", - "memoffset", - "scopeguard", -] - -[[package]] -name = "crossbeam-queue" -version = "0.2.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "774ba60a54c213d409d5353bda12d49cd68d14e45036a285234c8d6f91f92570" -dependencies = [ - "cfg-if", "crossbeam-utils", "maybe-uninit", ] @@ -239,9 +198,9 @@ dependencies = [ [[package]] name = "derive_more" -version = "0.99.10" +version = "0.99.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1dcfabdab475c16a93d669dddfc393027803e347d09663f524447f642fbb84ba" +checksum = "41cb0e6161ad61ed084a36ba71fbba9e3ac5aee3606fb607fe08da6acbcf3d8c" dependencies = [ "proc-macro2", "quote", @@ -257,12 +216,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "either" -version = "1.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e78d4f1cc4ae33bbfc157ed5d5a5ef3bc29227303d595861deb238fcec4e9457" - [[package]] name = "fake-simd" version = "0.1.2" @@ -308,18 +261,97 @@ checksum = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7" [[package]] name = "futures" -version = "0.1.29" +version = "0.3.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b980f2816d6ee8673b6517b52cb0e808a180efc92e5c19d02cdda79066703ef" +checksum = "1e05b85ec287aac0dc34db7d4a569323df697f9c55b99b15d6b4ef8cde49f613" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] [[package]] -name = "futures-cpupool" -version = "0.1.8" +name = "futures-channel" +version = "0.3.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ab90cde24b3319636588d0c35fe03b1333857621051837ed769faefb4c2162e4" +checksum = "f366ad74c28cca6ba456d95e6422883cfb4b252a83bed929c83abfdbbf2967d5" dependencies = [ - "futures", - "num_cpus", + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59f5fff90fd5d971f936ad674802482ba441b6f09ba5e15fd8b39145582ca399" + +[[package]] +name = "futures-executor" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10d6bb888be1153d3abeb9006b11b02cf5e9b209fda28693c31ae1e4e012e314" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de27142b013a8e869c14957e6d2edeef89e97c289e69d042ee3a49acd8b51789" + +[[package]] +name = "futures-macro" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0b5a30a4328ab5473878237c447333c093297bded83a4983d10f4deea240d39" +dependencies = [ + "proc-macro-hack", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "futures-sink" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f2032893cb734c7a05d85ce0cc8b8c4075278e93b24b66f9de99d6eb0fa8acc" + +[[package]] +name = "futures-task" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bdb66b5f09e22019b1ab0830f7785bcea8e7a42148683f99214f73f8ec21a626" +dependencies = [ + "once_cell", +] + +[[package]] +name = "futures-util" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8764574ff08b701a084482c3c7031349104b07ac897393010494beaa18ce32c6" +dependencies = [ + "futures-channel", + "futures-core", + "futures-io", + "futures-macro", + "futures-sink", + "futures-task", + "memchr", + "pin-project", + "pin-utils", + "proc-macro-hack", + "proc-macro-nested", + "slab", ] [[package]] @@ -365,20 +397,21 @@ checksum = "9b919933a397b79c37e33b77bb2aa3dc8eb6e165ad809e58ff75bc7db2e34574" [[package]] name = "h2" -version = "0.1.26" +version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a5b34c246847f938a410a03c5458c7fee2274436675e76d8b903c08efc29c462" +checksum = "993f9e0baeed60001cf565546b0d3dbe6a6ad23f2bd31644a133c641eccf6d53" dependencies = [ - "byteorder", "bytes", "fnv", - "futures", + "futures-core", + "futures-sink", + "futures-util", "http", "indexmap", - "log", "slab", - "string", - "tokio-io", + "tokio", + "tokio-util", + "tracing", ] [[package]] @@ -416,9 +449,9 @@ dependencies = [ [[package]] name = "http" -version = "0.1.21" +version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6ccf5ede3a895d8856620237b2f02972c1bbc78d2965ad7fe8838d4a0ed41f0" +checksum = "28d569972648b2c512421b5f2a405ad6ac9666547189d0c5477a3f200f3e02f9" dependencies = [ "bytes", "fnv", @@ -427,14 +460,12 @@ dependencies = [ [[package]] name = "http-body" -version = "0.1.0" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6741c859c1b2463a423a1dbce98d418e6c3c3fc720fb0d45528657320920292d" +checksum = "13d5ff830006f7646652e057693569bfe0d51760c0085a071769d142a205111b" dependencies = [ "bytes", - "futures", "http", - "tokio-buf", ] [[package]] @@ -444,46 +475,46 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd179ae861f0c2e53da70d892f5f3029f9594be0c41dc5269cd371691b1dc2f9" [[package]] -name = "hyper" -version = "0.12.35" +name = "httpdate" +version = "0.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9dbe6ed1438e1f8ad955a4701e9a944938e9519f6888d12d8558b645e247d5f6" +checksum = "494b4d60369511e7dea41cf646832512a94e542f68bb9c49e54518e0f468eb47" + +[[package]] +name = "hyper" +version = "0.13.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f3afcfae8af5ad0576a31e768415edb627824129e8e5a29b8bfccb2f234e835" dependencies = [ "bytes", - "futures", - "futures-cpupool", + "futures-channel", + "futures-core", + "futures-util", "h2", "http", "http-body", "httparse", - "iovec", + "httpdate", "itoa", - "log", - "net2", - "rustc_version", - "time", + "pin-project", + "socket2", "tokio", - "tokio-buf", - "tokio-executor", - "tokio-io", - "tokio-reactor", - "tokio-tcp", - "tokio-threadpool", - "tokio-timer", + "tower-service", + "tracing", "want", ] [[package]] name = "hyper-tls" -version = "0.3.2" +version = "0.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a800d6aa50af4b5850b2b0f659625ce9504df908e9733b635720483be26174f" +checksum = "d979acc56dcb5b8dddba3917601745e877576475aa046df3226eabdecef78eed" dependencies = [ "bytes", - "futures", "hyper", "native-tls", - "tokio-io", + "tokio", + "tokio-tls", ] [[package]] @@ -600,15 +631,6 @@ version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8dd5a6d5999d9907cda8ed67bbd137d3af8085216c2ac62de5be860bd41f304a" -[[package]] -name = "lock_api" -version = "0.3.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4da24a77a3d8a6d4862d95f72e6fdb9c09a643ecdb402d754004a557f2bec75" -dependencies = [ - "scopeguard", -] - [[package]] name = "log" version = "0.4.11" @@ -673,13 +695,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "60302e4db3a61da70c0cb7991976248362f30319e88850c487b9b95bbf059e00" [[package]] -name = "memoffset" -version = "0.5.6" +name = "memchr" +version = "2.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "043175f069eda7b85febe4a74abbaeff828d9f8b448515d3151a14a3542811aa" -dependencies = [ - "autocfg", -] +checksum = "3728d817d99e5ac407411fa471ff9800a778d88a24685968b36824eaf4bee400" [[package]] name = "mio" @@ -694,12 +713,24 @@ dependencies = [ "kernel32-sys", "libc", "log", - "miow", + "miow 0.2.1", "net2", "slab", "winapi 0.2.8", ] +[[package]] +name = "mio-named-pipes" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0840c1c50fd55e521b247f949c241c9997709f23bd7f023b9762cd561e935656" +dependencies = [ + "log", + "mio", + "miow 0.3.5", + "winapi 0.3.9", +] + [[package]] name = "mio-uds" version = "0.6.8" @@ -723,6 +754,16 @@ dependencies = [ "ws2_32-sys", ] +[[package]] +name = "miow" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07b88fb9795d4d36d62a012dfbf49a8f5cf12751f36d31a9dbe66d528e58979e" +dependencies = [ + "socket2", + "winapi 0.3.9", +] + [[package]] name = "native-tls" version = "0.2.4" @@ -835,32 +876,6 @@ dependencies = [ "ttf-parser", ] -[[package]] -name = "parking_lot" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f842b1982eb6c2fe34036a4fbfb06dd185a3f5c8edfaacdf7d1ea10b07de6252" -dependencies = [ - "lock_api", - "parking_lot_core", - "rustc_version", -] - -[[package]] -name = "parking_lot_core" -version = "0.6.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b876b1b9e7ac6e1a74a6da34d25c42e17e8862aa409cbbbdcfc8d86c6f3bc62b" -dependencies = [ - "cfg-if", - "cloudabi", - "libc", - "redox_syscall", - "rustc_version", - "smallvec", - "winapi 0.3.9", -] - [[package]] name = "percent-encoding" version = "2.1.0" @@ -910,6 +925,38 @@ dependencies = [ "sha-1", ] +[[package]] +name = "pin-project" +version = "0.4.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f48fad7cfbff853437be7cf54d7b993af21f53be7f0988cbfe4a51535aa77205" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "0.4.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24c6d293bdd3ca5a1697997854c6cf7855e43fb6a0ba1c47af57a5bcafd158ae" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "pin-project-lite" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fe74897791e156a0cd8cce0db31b9b2198e67877316bf3086c3acd187f719f0" + +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "pkg-config" version = "0.3.18" @@ -931,6 +978,18 @@ version = "0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c36fa947111f5c62a733b652544dd0016a43ce89619538a8ef92724a6f501a20" +[[package]] +name = "proc-macro-hack" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "99c605b9a0adc77b7211c6b1f722dcb613d68d66859a44f3d485a6da332b0598" + +[[package]] +name = "proc-macro-nested" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eba180dafb9038b050a4c280019bbedf9f2467b61e5d892dcad585bb57aadc5a" + [[package]] name = "proc-macro2" version = "1.0.23" @@ -998,9 +1057,9 @@ checksum = "41cc0f7e4d5d4544e8861606a285bb08d3e70712ccc7d2b84d7c0ccfaf4b05ce" [[package]] name = "relative-path" -version = "0.3.7" +version = "1.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e614f96449605730b4f7ad2c019e88c1652d730634b4eba07b810801856635e3" +checksum = "65aff7c83039e88c1c0b4bedf8dfa93d6ec84d5fc2945b37c1fa4186f46c5f94" dependencies = [ "serde", ] @@ -1016,18 +1075,9 @@ dependencies = [ [[package]] name = "route-recognizer" -version = "0.1.13" +version = "0.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ea509065eb0b3c446acdd0102f0d46567dc30902dc0be91d6552035d92b0f4f8" - -[[package]] -name = "rustc_version" -version = "0.2.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "138e3e0acb6c9fb258b19b67cb8abd63c00679d2851805ea151465464fe9030a" -dependencies = [ - "semver 0.9.0", -] +checksum = "56770675ebc04927ded3e60633437841581c285dc6236109ea25fbf3beb7b59e" [[package]] name = "rustsec" @@ -1097,18 +1147,6 @@ dependencies = [ "winapi 0.3.9", ] -[[package]] -name = "scoped-tls" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "332ffa32bf586782a3efaeb58f127980944bbc8c4d6913a86107ac2a5ab24b28" - -[[package]] -name = "scopeguard" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd" - [[package]] name = "security-framework" version = "0.4.4" @@ -1196,9 +1234,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.57" +version = "1.0.58" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "164eacbdb13512ec2745fb09d51fd5b22b0d65ed294a1dcf7285a360c80a675c" +checksum = "a230ea9107ca2220eea9d46de97eddcb04cd00e92d13dda78e478dd33fa82bd4" dependencies = [ "itoa", "ryu", @@ -1232,6 +1270,7 @@ dependencies = [ "lru-cache", "maud", "once_cell", + "pin-project", "relative-path", "route-recognizer", "rustsec", @@ -1241,10 +1280,18 @@ dependencies = [ "serde_json", "slog", "slog-json", - "tokio-core", - "tokio-service", + "tokio", "toml", - "try_future", +] + +[[package]] +name = "signal-hook-registry" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3e12110bc539e657a646068aaf5eb5b63af9d0c1f7b29c97113fad80e15f035" +dependencies = [ + "arc-swap", + "libc", ] [[package]] @@ -1271,15 +1318,6 @@ dependencies = [ "slog", ] -[[package]] -name = "smallvec" -version = "0.6.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f7b0758c52e15a8b5e3691eae6cc559f08eee9406e548a4477ba4e67770a82b6" -dependencies = [ - "maybe-uninit", -] - [[package]] name = "smol_str" version = "0.1.17" @@ -1290,12 +1328,15 @@ dependencies = [ ] [[package]] -name = "string" -version = "0.2.1" +name = "socket2" +version = "0.3.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d24114bfcceb867ca7f71a0d3fe45d45619ec47a6fbfa98cb14e14250bfa5d6d" +checksum = "b1fa70dc5c8104ec096f4fe7ede7a221d35ae13dcd19ba1ad9a81d2cab9a1c44" dependencies = [ - "bytes", + "cfg-if", + "libc", + "redox_syscall", + "winapi 0.3.9", ] [[package]] @@ -1362,223 +1403,61 @@ checksum = "238ce071d267c5710f9d31451efec16c5ee22de34df17cc05e56cbc92e967117" [[package]] name = "tokio" -version = "0.1.22" +version = "0.2.22" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5a09c0b5bb588872ab2f09afa13ee6e9dac11e10a0ec9e8e3ba39a5a5d530af6" +checksum = "5d34ca54d84bf2b5b4d7d31e901a8464f7b60ac145a284fba25ceb801f2ddccd" dependencies = [ "bytes", - "futures", - "mio", - "num_cpus", - "tokio-codec", - "tokio-current-thread", - "tokio-executor", - "tokio-fs", - "tokio-io", - "tokio-reactor", - "tokio-sync", - "tokio-tcp", - "tokio-threadpool", - "tokio-timer", - "tokio-udp", - "tokio-uds", -] - -[[package]] -name = "tokio-buf" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8fb220f46c53859a4b7ec083e41dec9778ff0b1851c0942b211edb89e0ccdc46" -dependencies = [ - "bytes", - "either", - "futures", -] - -[[package]] -name = "tokio-codec" -version = "0.1.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "25b2998660ba0e70d18684de5d06b70b70a3a747469af9dea7618cc59e75976b" -dependencies = [ - "bytes", - "futures", - "tokio-io", -] - -[[package]] -name = "tokio-core" -version = "0.1.17" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aeeffbbb94209023feaef3c196a41cbcdafa06b4a6f893f68779bb5e53796f71" -dependencies = [ - "bytes", - "futures", - "iovec", - "log", - "mio", - "scoped-tls", - "tokio", - "tokio-executor", - "tokio-io", - "tokio-reactor", - "tokio-timer", -] - -[[package]] -name = "tokio-current-thread" -version = "0.1.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b1de0e32a83f131e002238d7ccde18211c0a5397f60cbfffcb112868c2e0e20e" -dependencies = [ - "futures", - "tokio-executor", -] - -[[package]] -name = "tokio-executor" -version = "0.1.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb2d1b8f4548dbf5e1f7818512e9c406860678f29c300cdf0ebac72d1a3a1671" -dependencies = [ - "crossbeam-utils", - "futures", -] - -[[package]] -name = "tokio-fs" -version = "0.1.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "297a1206e0ca6302a0eed35b700d292b275256f596e2f3fea7729d5e629b6ff4" -dependencies = [ - "futures", - "tokio-io", - "tokio-threadpool", -] - -[[package]] -name = "tokio-io" -version = "0.1.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "57fc868aae093479e3131e3d165c93b1c7474109d13c90ec0dda2a1bbfff0674" -dependencies = [ - "bytes", - "futures", - "log", -] - -[[package]] -name = "tokio-reactor" -version = "0.1.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09bc590ec4ba8ba87652da2068d150dcada2cfa2e07faae270a5e0409aa51351" -dependencies = [ - "crossbeam-utils", - "futures", - "lazy_static", - "log", - "mio", - "num_cpus", - "parking_lot", - "slab", - "tokio-executor", - "tokio-io", - "tokio-sync", -] - -[[package]] -name = "tokio-service" -version = "0.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24da22d077e0f15f55162bdbdc661228c1581892f52074fb242678d015b45162" -dependencies = [ - "futures", -] - -[[package]] -name = "tokio-sync" -version = "0.1.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "edfe50152bc8164fcc456dab7891fa9bf8beaf01c5ee7e1dd43a397c3cf87dee" -dependencies = [ "fnv", - "futures", -] - -[[package]] -name = "tokio-tcp" -version = "0.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "98df18ed66e3b72e742f185882a9e201892407957e45fbff8da17ae7a7c51f72" -dependencies = [ - "bytes", - "futures", + "futures-core", "iovec", - "mio", - "tokio-io", - "tokio-reactor", -] - -[[package]] -name = "tokio-threadpool" -version = "0.1.18" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "df720b6581784c118f0eb4310796b12b1d242a7eb95f716a8367855325c25f89" -dependencies = [ - "crossbeam-deque", - "crossbeam-queue", - "crossbeam-utils", - "futures", "lazy_static", - "log", - "num_cpus", - "slab", - "tokio-executor", -] - -[[package]] -name = "tokio-timer" -version = "0.2.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "93044f2d313c95ff1cb7809ce9a7a05735b012288a888b62d4434fd58c94f296" -dependencies = [ - "crossbeam-utils", - "futures", - "slab", - "tokio-executor", -] - -[[package]] -name = "tokio-udp" -version = "0.1.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e2a0b10e610b39c38b031a2fcab08e4b82f16ece36504988dcbd81dbba650d82" -dependencies = [ - "bytes", - "futures", - "log", - "mio", - "tokio-codec", - "tokio-io", - "tokio-reactor", -] - -[[package]] -name = "tokio-uds" -version = "0.2.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ab57a4ac4111c8c9dbcf70779f6fc8bc35ae4b2454809febac840ad19bd7e4e0" -dependencies = [ - "bytes", - "futures", - "iovec", "libc", - "log", + "memchr", "mio", + "mio-named-pipes", "mio-uds", - "tokio-codec", - "tokio-io", - "tokio-reactor", + "num_cpus", + "pin-project-lite", + "signal-hook-registry", + "slab", + "tokio-macros", + "winapi 0.3.9", +] + +[[package]] +name = "tokio-macros" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0c3acc6aa564495a0f2e1d59fab677cd7f81a19994cfc7f3ad0e64301560389" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio-tls" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a70f4fcd7b3b24fb194f837560168208f669ca8cb70d0c4b862944452396343" +dependencies = [ + "native-tls", + "tokio", +] + +[[package]] +name = "tokio-util" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be8242891f2b6cbef26a2d7e8605133c2c554cd35b3e4948ea892d6d68436499" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "log", + "pin-project-lite", + "tokio", ] [[package]] @@ -1590,21 +1469,39 @@ dependencies = [ "serde", ] +[[package]] +name = "tower-service" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e987b6bf443f4b5b3b6f38704195592cca41c5bb7aedd3c3693c7081f8289860" + +[[package]] +name = "tracing" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0987850db3733619253fe60e17cb59b82d37c7e6c0236bb81e4d6b87c879f27" +dependencies = [ + "cfg-if", + "log", + "pin-project-lite", + "tracing-core", +] + +[[package]] +name = "tracing-core" +version = "0.1.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f50de3927f93d202783f4513cda820ab47ef17f624b03c096e86ef00c67e6b5f" +dependencies = [ + "lazy_static", +] + [[package]] name = "try-lock" version = "0.2.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "59547bce71d9c38b83d9c0e92b6066c4253371f15005def0c30d9657f50c7642" -[[package]] -name = "try_future" -version = "0.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30454186ee38f29e06f386c6e9f773b7c33e85430db6f90e4597419b4c5baad7" -dependencies = [ - "futures", -] - [[package]] name = "ttf-parser" version = "0.6.2" @@ -1666,11 +1563,10 @@ checksum = "6454029bf181f092ad1b853286f23e2c507d8e8194d01d92da4a55c274a5508c" [[package]] name = "want" -version = "0.2.0" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6395efa4784b027708f7451087e647ec73cc74f5d9bc2e418404248d679a230" +checksum = "1ce8a968cb1cd110d136ff8b819a556d6fb6d919363c61534f6860c7eb172ba0" dependencies = [ - "futures", "log", "try-lock", ] diff --git a/Cargo.toml b/Cargo.toml index 8f8454b..e4f5ab3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,28 +14,26 @@ members = [ badge = { path = "./libs/badge" } anyhow = "1" -cadence = "0.13" +cadence = "0.21" derive_more = "0.99" -futures = "0.1" -hyper = "0.12" -hyper-tls = "0.3" +futures = "0.3" +hyper = "0.13" +hyper-tls = "0.4" indexmap = { version = "1", features = ["serde-1"] } -lru-cache = "0.1" +lru-cache = "0.1" # TODO: replace unmaintained crate maud = "0.22" once_cell = "1.4" -relative-path = { version = "0.3.7", features = ["serde"] } -route-recognizer = "0.1" +pin-project = "0.4" +relative-path = { version = "1.3", features = ["serde"] } +route-recognizer = "0.2" rustsec = "0.21" semver = { version = "0.11", features = ["serde"] } serde = { version = "1", features = ["derive"] } serde_json = "1" slog = "2" slog-json = "2" -tokio-core = "0.1" -tokio-service = "0.1" +tokio = { version = "0.2", features = ["full"] } toml = "0.5" -try_future = "0.1" - [build-dependencies] sass-rs = "0.2" diff --git a/rust-toolchain b/rust-toolchain new file mode 100644 index 0000000..c40b898 --- /dev/null +++ b/rust-toolchain @@ -0,0 +1,2 @@ +nightly-2020-09-27 + diff --git a/src/engine/fut/analyze.rs b/src/engine/fut/analyze.rs index 2e66c45..a5d9f21 100644 --- a/src/engine/fut/analyze.rs +++ b/src/engine/fut/analyze.rs @@ -1,66 +1,55 @@ -use anyhow::{anyhow, ensure, Error}; -use futures::stream::futures_unordered; -use futures::{Future, Poll, Stream}; +use anyhow::Error; +use futures::StreamExt; use crate::models::crates::{AnalyzedDependencies, CrateDeps}; +use crate::{engine::machines::analyzer::DependencyAnalyzer, Engine}; -use super::super::machines::analyzer::DependencyAnalyzer; -use super::super::Engine; +pub async fn analyze_dependencies( + engine: Engine, + deps: CrateDeps, +) -> Result { + let advisory_db = engine.fetch_advisory_db().await?; -pub struct AnalyzeDependenciesFuture { - inner: Box + Send>, -} + let mut analyzer = DependencyAnalyzer::new(&deps, Some(advisory_db)); -impl AnalyzeDependenciesFuture { - pub fn new(engine: Engine, deps: CrateDeps) -> Self { - let future = - engine.fetch_advisory_db().and_then(move |advisory_db| { - let analyzer = DependencyAnalyzer::new(&deps, Some(advisory_db)); + let main_deps = + deps.main.into_iter().filter_map( + |(name, dep)| { + if dep.is_external() { + Some(name) + } else { + None + } + }, + ); + let dev_deps = + deps.dev.into_iter().filter_map( + |(name, dep)| { + if dep.is_external() { + Some(name) + } else { + None + } + }, + ); + let build_deps = + deps.build.into_iter().filter_map( + |(name, dep)| { + if dep.is_external() { + Some(name) + } else { + None + } + }, + ); - let main_deps = deps.main.into_iter().filter_map(|(name, dep)| { - if dep.is_external() { - Some(name) - } else { - None - } - }); - let dev_deps = deps.dev.into_iter().filter_map(|(name, dep)| { - if dep.is_external() { - Some(name) - } else { - None - } - }); - let build_deps = deps.build.into_iter().filter_map(|(name, dep)| { - if dep.is_external() { - Some(name) - } else { - None - } - }); + let deps_iter = main_deps.chain(dev_deps).chain(build_deps); + let mut releases = engine.fetch_releases(deps_iter); - let release_futures = - engine.fetch_releases(main_deps.chain(dev_deps).chain(build_deps)); - - futures_unordered(release_futures) - .fold(analyzer, |mut analyzer, releases| { - analyzer.process(releases); - Ok(analyzer) as Result<_, Error> - }) - .map(|analyzer| analyzer.finalize()) - }); - - AnalyzeDependenciesFuture { - inner: Box::new(future), - } + for release in releases.next().await { + let release = release?; + analyzer.process(release) } -} -impl Future for AnalyzeDependenciesFuture { - type Item = AnalyzedDependencies; - type Error = Error; - - fn poll(&mut self) -> Poll { - self.inner.poll() - } + Ok(analyzer.finalize()) } diff --git a/src/engine/fut/crawl.rs b/src/engine/fut/crawl.rs index 4581d12..b851394 100644 --- a/src/engine/fut/crawl.rs +++ b/src/engine/fut/crawl.rs @@ -1,8 +1,8 @@ -use std::mem; +use std::{future::Future, mem, pin::Pin, task::Context, task::Poll}; -use anyhow::{anyhow, ensure, Error}; -use futures::stream::FuturesOrdered; -use futures::{try_ready, Async, Future, Poll, Stream}; +use anyhow::Error; +use futures::{ready, Stream}; +use futures::{stream::FuturesOrdered, FutureExt}; use relative_path::RelativePathBuf; use crate::models::repo::RepoPath; @@ -11,24 +11,28 @@ use super::super::machines::crawler::ManifestCrawler; pub use super::super::machines::crawler::ManifestCrawlerOutput; use super::super::Engine; +#[pin_project::pin_project] pub struct CrawlManifestFuture { repo_path: RepoPath, engine: Engine, crawler: ManifestCrawler, - futures: - FuturesOrdered + Send>>, + #[pin] + futures: FuturesOrdered< + Pin> + Send>>, + >, } impl CrawlManifestFuture { pub fn new(engine: &Engine, repo_path: RepoPath, entry_point: RelativePathBuf) -> Self { - let future: Box + Send> = Box::new( - engine - .retrieve_manifest_at_path(&repo_path, &entry_point) - .map(move |contents| (entry_point, contents)), - ); let engine = engine.clone(); let crawler = ManifestCrawler::new(); let mut futures = FuturesOrdered::new(); + + let future: Pin + Send>> = Box::pin( + engine + .retrieve_manifest_at_path(&repo_path, &entry_point) + .map(move |contents| contents.map(|c| (entry_point, c))), + ); futures.push(future); CrawlManifestFuture { @@ -41,27 +45,33 @@ impl CrawlManifestFuture { } impl Future for CrawlManifestFuture { - type Item = ManifestCrawlerOutput; - type Error = Error; + type Output = Result; - fn poll(&mut self) -> Poll { - match try_ready!(self.futures.poll()) { + fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { + let this = self.as_mut().project(); + + match ready!(this.futures.poll_next(cx)) { None => { let crawler = mem::replace(&mut self.crawler, ManifestCrawler::new()); - Ok(Async::Ready(crawler.finalize())) + Poll::Ready(Ok(crawler.finalize())) } - Some((path, raw_manifest)) => { + + Some(Ok((path, raw_manifest))) => { let output = self.crawler.step(path, raw_manifest)?; + for path in output.paths_of_interest.into_iter() { - let future: Box + Send> = Box::new( + let future: Pin + Send>> = Box::pin( self.engine .retrieve_manifest_at_path(&self.repo_path, &path) - .map(move |contents| (path, contents)), + .map(move |contents| contents.map(|c| ((path, c)))), ); self.futures.push(future); } - self.poll() + + self.poll(cx) } + + Some(Err(err)) => Poll::Ready(Err(err.into())), } } } diff --git a/src/engine/fut/mod.rs b/src/engine/fut/mod.rs index f902460..a2fc499 100644 --- a/src/engine/fut/mod.rs +++ b/src/engine/fut/mod.rs @@ -1,5 +1,5 @@ mod analyze; mod crawl; -pub use self::analyze::AnalyzeDependenciesFuture; +pub use self::analyze::analyze_dependencies; pub use self::crawl::CrawlManifestFuture; diff --git a/src/engine/machines/crawler.rs b/src/engine/machines/crawler.rs index a8ce0b9..69d4b6c 100644 --- a/src/engine/machines/crawler.rs +++ b/src/engine/machines/crawler.rs @@ -1,6 +1,6 @@ use std::collections::HashMap; -use anyhow::{anyhow, ensure, Error}; +use anyhow::Error; use indexmap::IndexMap; use relative_path::RelativePathBuf; diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 4d27279..e7f119a 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -1,51 +1,57 @@ -use std::collections::HashSet; -use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::{ + collections::HashSet, + panic::RefUnwindSafe, + sync::{Arc, Mutex}, + task::{Context, Poll}, + time::{Duration, Instant}, +}; -use anyhow::{anyhow, ensure, Error}; -use cadence::prelude::*; +use anyhow::{anyhow, Error}; use cadence::{MetricSink, NopMetricSink, StatsdClient}; -use futures::future::join_all; -use futures::{future, Future}; -use hyper::client::{HttpConnector, ResponseFuture}; -use hyper::{Body, Client, Request, Response}; +use futures::{future::try_join_all, stream::FuturesUnordered, Future, FutureExt, Stream}; +use hyper::{ + client::{HttpConnector, ResponseFuture}, + service::Service, + Body, Client, Request, Response, +}; use hyper_tls::HttpsConnector; use once_cell::sync::Lazy; use relative_path::{RelativePath, RelativePathBuf}; use rustsec::database::Database; use semver::VersionReq; use slog::Logger; -use tokio_service::Service; - -use crate::utils::cache::Cache; - -use crate::models::crates::{AnalyzedDependencies, CrateName, CratePath, CrateRelease}; -use crate::models::repo::{RepoPath, Repository}; use crate::interactors::crates::{GetPopularCrates, QueryCrate}; use crate::interactors::github::GetPopularRepos; use crate::interactors::rustsec::FetchAdvisoryDatabase; use crate::interactors::RetrieveFileAtPath; +use crate::models::crates::{AnalyzedDependencies, CrateName, CratePath, CrateRelease}; +use crate::models::repo::{RepoPath, Repository}; +use crate::utils::cache::Cache; mod fut; mod machines; -use self::fut::AnalyzeDependenciesFuture; +use self::fut::analyze_dependencies; use self::fut::CrawlManifestFuture; type HttpClient = Client>; +// type HttpClient = Client; // workaround for hyper 0.12 not implementing Service for Client #[derive(Debug, Clone)] struct ServiceHttpClient(HttpClient); -impl Service for ServiceHttpClient { - type Request = Request; +impl Service> for ServiceHttpClient { type Response = Response; type Error = hyper::Error; type Future = ResponseFuture; - fn call(&self, req: Self::Request) -> Self::Future { + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_ready(cx).map_err(|err| err.into()) + } + + fn call(&mut self, req: Request) -> Self::Future { self.0.request(req) } } @@ -55,15 +61,16 @@ pub struct Engine { client: HttpClient, logger: Logger, metrics: StatsdClient, - query_crate: Arc>>, - get_popular_crates: Arc>>, - get_popular_repos: Arc>>, - retrieve_file_at_path: Arc>, - fetch_advisory_db: Arc>>, + // TODO: use futures aware mutex + query_crate: Arc, CrateName>>>, + get_popular_crates: Arc, ()>>>, + get_popular_repos: Arc, ()>>>, + retrieve_file_at_path: Arc>>, + fetch_advisory_db: Arc, ()>>>, } impl Engine { - pub fn new(client: Client>, logger: Logger) -> Engine { + pub fn new(client: HttpClient, logger: Logger) -> Engine { let metrics = StatsdClient::from_sink("engine", NopMetricSink); let service_client = ServiceHttpClient(client.clone()); @@ -93,19 +100,20 @@ impl Engine { client: client.clone(), logger, metrics, - query_crate: Arc::new(query_crate), - get_popular_crates: Arc::new(get_popular_crates), - get_popular_repos: Arc::new(get_popular_repos), - retrieve_file_at_path: Arc::new(RetrieveFileAtPath(service_client)), - fetch_advisory_db: Arc::new(fetch_advisory_db), + query_crate: Arc::new(Mutex::new(query_crate)), + get_popular_crates: Arc::new(Mutex::new(get_popular_crates)), + get_popular_repos: Arc::new(Mutex::new(get_popular_repos)), + retrieve_file_at_path: Arc::new(Mutex::new(RetrieveFileAtPath(service_client))), + fetch_advisory_db: Arc::new(Mutex::new(fetch_advisory_db)), } } - pub fn set_metrics(&mut self, sink: M) { + pub fn set_metrics(&mut self, sink: M) { self.metrics = StatsdClient::from_sink("engine", sink); } } +#[derive(Debug)] pub struct AnalyzeDependenciesOutcome { pub crates: Vec<(CrateName, AnalyzedDependencies)>, pub duration: Duration, @@ -132,141 +140,152 @@ impl AnalyzeDependenciesOutcome { } impl Engine { - pub fn get_popular_repos(&self) -> impl Future, Error = Error> + Send { - self.get_popular_repos.call(()).from_err().map(|repos| { - repos - .iter() - .filter(|repo| !POPULAR_REPO_BLOCK_LIST.contains(&repo.path)) - .cloned() - .collect() - }) + pub async fn get_popular_repos(&self) -> Result, Error> { + let repos = self.get_popular_repos.lock().unwrap().call(()); + let repos = repos.await?; + + let filtered_repos = repos + .iter() + .filter(|repo| !POPULAR_REPO_BLOCK_LIST.contains(&repo.path)) + .cloned() + .collect(); + + Ok(filtered_repos) } - pub fn get_popular_crates(&self) -> impl Future, Error = Error> + Send { - self.get_popular_crates - .call(()) - .from_err() - .map(|crates| crates.clone()) + pub async fn get_popular_crates(&self) -> Result, Error> { + let crates = self.get_popular_crates.lock().unwrap().call(()); + let crates = crates.await?; + Ok(crates.clone()) } - pub fn analyze_repo_dependencies( + pub async fn analyze_repo_dependencies( &self, repo_path: RepoPath, - ) -> impl Future + Send { + ) -> Result { let start = Instant::now(); let entry_point = RelativePath::new("/").to_relative_path_buf(); - let manifest_future = CrawlManifestFuture::new(self, repo_path.clone(), entry_point); - let engine = self.clone(); - manifest_future.and_then(move |manifest_output| { - let engine_for_analyze = engine.clone(); - let futures = manifest_output - .crates - .into_iter() - .map(move |(crate_name, deps)| { - let analyzed_deps_future = - AnalyzeDependenciesFuture::new(engine_for_analyze.clone(), deps); - analyzed_deps_future.map(move |analyzed_deps| (crate_name, analyzed_deps)) - }); + let manifest_future = CrawlManifestFuture::new(self, repo_path.clone(), entry_point); + let manifest_output = manifest_future.await?; - join_all(futures).and_then(move |crates| { - let duration = start.elapsed(); - engine - .metrics - .time_duration_with_tags("analyze_duration", duration) - .with_tag("repo_site", repo_path.site.as_ref()) - .with_tag("repo_qual", repo_path.qual.as_ref()) - .with_tag("repo_name", repo_path.name.as_ref()) - .send()?; - - Ok(AnalyzeDependenciesOutcome { crates, duration }) + let engine_for_analyze = engine.clone(); + let futures = manifest_output + .crates + .into_iter() + .map(|(crate_name, deps)| async { + let analyzed_deps = analyze_dependencies(engine_for_analyze.clone(), deps).await?; + Ok::<_, Error>((crate_name, analyzed_deps)) }) - }) + .collect::>(); + + let crates = try_join_all(futures).await?; + + let duration = start.elapsed(); + // engine + // .metrics + // .time_duration_with_tags("analyze_duration", duration) + // .with_tag("repo_site", repo_path.site.as_ref()) + // .with_tag("repo_qual", repo_path.qual.as_ref()) + // .with_tag("repo_name", repo_path.name.as_ref()) + // .send()?; + + Ok(AnalyzeDependenciesOutcome { crates, duration }) } - pub fn analyze_crate_dependencies( + pub async fn analyze_crate_dependencies( &self, crate_path: CratePath, - ) -> impl Future + Send { + ) -> Result { let start = Instant::now(); - let query_future = self.query_crate.call(crate_path.name.clone()).from_err(); + println!("analyze deps"); + + let query_response = self + .query_crate + .lock() + .unwrap() + .call(crate_path.name.clone()); + let query_response = query_response.await?; let engine = self.clone(); - query_future.and_then(move |query_response| { - match query_response - .releases - .iter() - .find(|release| release.version == crate_path.version) - { - None => future::Either::A(future::err(anyhow!( - "could not find crate release with version {}", - crate_path.version - ))), - Some(release) => { - let analyzed_deps_future = - AnalyzeDependenciesFuture::new(engine.clone(), release.deps.clone()); - future::Either::B(analyzed_deps_future.map(move |analyzed_deps| { - let crates = vec![(crate_path.name, analyzed_deps)].into_iter().collect(); - let duration = start.elapsed(); + match query_response + .releases + .iter() + .find(|release| release.version == crate_path.version) + { + None => Err(anyhow!( + "could not find crate release with version {}", + crate_path.version + )), - AnalyzeDependenciesOutcome { crates, duration } - })) - } + Some(release) => { + let analyzed_deps = + analyze_dependencies(engine.clone(), release.deps.clone()).await?; + + let crates = vec![(crate_path.name, analyzed_deps)].into_iter().collect(); + let duration = start.elapsed(); + + Ok(AnalyzeDependenciesOutcome { crates, duration }) } - }) + } } - pub fn find_latest_crate_release( + pub async fn find_latest_crate_release( &self, name: CrateName, req: VersionReq, - ) -> impl Future, Error = Error> + Send { - self.query_crate - .call(name) - .from_err() - .map(move |query_response| { - query_response - .releases - .iter() - .filter(|release| req.matches(&release.version)) - .max_by(|r1, r2| r1.version.cmp(&r2.version)) - .cloned() - }) + ) -> Result, Error> { + let query_response = self.query_crate.lock().unwrap().call(name); + let query_response = query_response.await?; + + let latest = query_response + .releases + .iter() + .filter(|release| req.matches(&release.version)) + .max_by(|r1, r2| r1.version.cmp(&r2.version)) + .cloned(); + + Ok(latest) } fn fetch_releases>( &self, names: I, - ) -> impl Iterator, Error = Error>> { + ) -> impl Stream, Error>> { let engine = self.clone(); - names.into_iter().map(move |name| { - engine - .query_crate - .call(name) - .from_err() - .map(|resp| resp.releases.clone()) - }) + + names + .into_iter() + .map(|name| { + engine + .query_crate + .lock() + .unwrap() + .call(name) + .map(|resp| resp.map(|r| r.releases.clone())) + }) + .collect::>() } fn retrieve_manifest_at_path( &self, repo_path: &RepoPath, path: &RelativePathBuf, - ) -> impl Future + Send { + ) -> impl Future> { let manifest_path = path.join(RelativePath::new("Cargo.toml")); + self.retrieve_file_at_path + .lock() + .unwrap() .call((repo_path.clone(), manifest_path)) } - fn fetch_advisory_db(&self) -> impl Future, Error = Error> + Send { - self.fetch_advisory_db - .call(()) - .from_err() - .map(|db| db.clone()) + fn fetch_advisory_db(&self) -> impl Future, Error>> { + self.fetch_advisory_db.lock().unwrap().call(()) } } diff --git a/src/interactors/bitbucket.rs b/src/interactors/bitbucket.rs index 8239c05..4703f71 100644 --- a/src/interactors/bitbucket.rs +++ b/src/interactors/bitbucket.rs @@ -1,4 +1,4 @@ -use anyhow::{anyhow, ensure, Error}; +use anyhow::Error; use hyper::Uri; use relative_path::RelativePathBuf; @@ -8,12 +8,14 @@ const BITBUCKET_USER_CONTENT_BASE_URI: &'static str = "https://bitbucket.org"; pub fn get_manifest_uri(repo_path: &RepoPath, path: &RelativePathBuf) -> Result { let path_str: &str = path.as_ref(); - Ok(format!( + + let url = format!( "{}/{}/{}/raw/HEAD/{}", BITBUCKET_USER_CONTENT_BASE_URI, repo_path.qual.as_ref(), repo_path.name.as_ref(), path_str - ) - .parse::()?) + ); + + Ok(url.parse::()?) } diff --git a/src/interactors/crates.rs b/src/interactors/crates.rs index 7b53e24..c97a527 100644 --- a/src/interactors/crates.rs +++ b/src/interactors/crates.rs @@ -1,11 +1,17 @@ -use std::str; +use std::pin::Pin; +use std::{future::Future, task::Poll}; +use std::{str, task::Context}; -use anyhow::{anyhow, ensure, Error}; -use futures::{future, Future, IntoFuture, Stream}; -use hyper::{Body, Error as HyperError, Method, Request, Response, Uri, header::USER_AGENT}; +use anyhow::{anyhow, Error}; +use futures::{ + future::{err, ok, ready}, + TryFutureExt, +}; +use hyper::{ + body, header::USER_AGENT, service::Service, Body, Error as HyperError, Request, Response, Uri, +}; use semver::{Version, VersionReq}; use serde::Deserialize; -use tokio_service::Service; use crate::models::crates::{CrateDep, CrateDeps, CrateName, CratePath, CrateRelease}; @@ -65,6 +71,7 @@ fn convert_pkgs( Ok(QueryCrateResponse { releases: releases }) } +#[derive(Debug, Clone)] pub struct QueryCrateResponse { pub releases: Vec, } @@ -72,19 +79,20 @@ pub struct QueryCrateResponse { #[derive(Debug, Clone)] pub struct QueryCrate(pub S); -impl Service for QueryCrate +impl Service for QueryCrate where - S: Service, Response = Response, Error = HyperError> - + Clone - + 'static, + S: Service, Response = Response, Error = HyperError> + Clone, S::Future: Send + 'static, { - type Request = CrateName; type Response = QueryCrateResponse; type Error = Error; - type Future = Box + Send>; + type Future = Pin> + Send>>; - fn call(&self, crate_name: CrateName) -> Self::Future { + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_ready(cx).map_err(|err| err.into()) + } + + fn call(&mut self, crate_name: CrateName) -> Self::Future { let lower_name = crate_name.as_ref().to_lowercase(); let path = match lower_name.len() { @@ -94,35 +102,44 @@ where _ => format!("{}/{}/{}", &lower_name[0..2], &lower_name[2..4], lower_name), }; - let uri = - try_future_box!(format!("{}/master/{}", CRATES_INDEX_BASE_URI, path).parse::()); + let uri = format!("{}/master/{}", CRATES_INDEX_BASE_URI, path); + + println!("analyze from uri {:?}", &uri); + + let uri = uri.parse::().expect("TODO: MAP ERROR PROPERLY"); let request = Request::get(uri.clone()) .header(USER_AGENT, "deps.rs") - .body(Body::empty()).unwrap(); + .body(Body::empty()) + .unwrap(); - Box::new(self.0.call(request).from_err().and_then(move |response| { - let status = response.status(); - if !status.is_success() { - try_future!(Err(anyhow!("Status code {} for URI {}", status, uri))); - } + Box::pin( + self.0 + .call(request) + .map_err(|err| err.into()) + .and_then(move |response| { + let status = response.status(); + if !status.is_success() { + return err(anyhow!("Status code {} for URI {}", status, uri)); + } - let body_future = response.into_body().concat2().from_err(); - let decode_future = body_future.and_then(move |body| { - let string_body = str::from_utf8(body.as_ref())?; - let packages = string_body - .lines() - .map(|s| s.trim()) - .filter(|s| !s.is_empty()) - .map(|s| serde_json::from_str::(s)) - .collect::>()?; - Ok(packages) - }); - - decode_future - .and_then(move |pkgs| convert_pkgs(&crate_name, pkgs)) - .into() - })) + ok(response) + }) + .and_then(|response| body::to_bytes(response.into_body()).err_into()) + .and_then(|body| ready(String::from_utf8(body.to_vec())).err_into()) + .and_then(|string_body| { + ready( + string_body + .lines() + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + .map(|s| serde_json::from_str::(s)) + .collect::>(), + ) + .err_into() + }) + .and_then(move |pkgs| ready(convert_pkgs(&crate_name, pkgs))), + ) } } @@ -154,44 +171,46 @@ fn convert_summary(response: SummaryResponse) -> Result, Error> { #[derive(Debug, Clone)] pub struct GetPopularCrates(pub S); -impl Service for GetPopularCrates +impl Service<()> for GetPopularCrates where - S: Service, Response = Response, Error = HyperError> - + Clone - + 'static, + S: Service, Response = Response, Error = HyperError> + Clone, S::Future: Send + 'static, { - type Request = (); type Response = Vec; type Error = Error; - type Future = Box + Send>; + type Future = Pin> + Send>>; - fn call(&self, _req: ()) -> Self::Future { - let service = self.0.clone(); + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_ready(cx).map_err(|err| err.into()) + } - let uri = format!("{}/summary", CRATES_API_BASE_URI) - .parse::() - .unwrap(); + fn call(&mut self, _req: ()) -> Self::Future { + let mut service = self.0.clone(); + + let uri = format!("{}/summary", CRATES_API_BASE_URI); + let uri = uri.parse::().unwrap(); let request = Request::get(uri.clone()) .header(USER_AGENT, "deps.rs") - .body(Body::empty()).unwrap(); + .body(Body::empty()) + .unwrap(); - Box::new(service.call(request).from_err().and_then(move |response| { - let status = response.status(); - if !status.is_success() { - future::Either::A(future::err(anyhow!( - "Status code {} for URI {}", - status, - uri - ))) - } else { - let body_future = response.into_body().concat2().from_err(); - let decode_future = body_future.and_then(|body| { - let summary = serde_json::from_slice::(&body)?; - convert_summary(summary) - }); - future::Either::B(decode_future) - } - })) + Box::pin( + service + .call(request) + .map_err(|err| err.into()) + .and_then(move |response| { + let status = response.status(); + if !status.is_success() { + err(anyhow!("Status code {} for URI {}", status, uri)) + } else { + ok(response) + } + }) + .and_then(|response| body::to_bytes(response.into_body()).err_into()) + .and_then(|bytes| { + ready(serde_json::from_slice::(&bytes)).err_into() + }) + .and_then(|summary| ready(convert_summary(summary)).err_into()), + ) } } diff --git a/src/interactors/github.rs b/src/interactors/github.rs index c407139..961f3b6 100644 --- a/src/interactors/github.rs +++ b/src/interactors/github.rs @@ -1,10 +1,15 @@ -use anyhow::{anyhow, ensure, Error}; -use futures::{Future, Stream}; -use hyper::header::USER_AGENT; -use hyper::{Body, Error as HyperError, Method, Request, Response, Uri}; +use std::{future::Future, pin::Pin, task::Context, task::Poll}; + +use anyhow::{anyhow, Error}; +use futures::{ + future::{err, ok, ready}, + TryFutureExt, +}; +use hyper::{ + body, header::USER_AGENT, service::Service, Body, Error as HyperError, Request, Response, Uri, +}; use relative_path::RelativePathBuf; use serde::Deserialize; -use tokio_service::Service; use crate::models::repo::{RepoPath, Repository}; @@ -43,57 +48,63 @@ struct GithubOwner { #[derive(Debug, Clone)] pub struct GetPopularRepos(pub S); -impl Service for GetPopularRepos +impl Service<()> for GetPopularRepos where - S: Service, Response = Response, Error = HyperError> - + Clone - + 'static, + S: Service, Response = Response, Error = HyperError> + Clone, S::Future: Send + 'static, { - type Request = (); type Response = Vec; type Error = Error; - type Future = Box + Send>; + type Future = Pin> + Send>>; - fn call(&self, _req: ()) -> Self::Future { - let uri = try_future_box!(format!( + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_ready(cx).map_err(|err| err.into()) + } + + fn call(&mut self, _req: ()) -> Self::Future { + let uri = format!( "{}/search/repositories?q=language:rust&sort=stars", GITHUB_API_BASE_URI ) - .parse::()); + .parse::() + .expect("TODO: handle error properly"); - let mut request = Request::get(uri); - request.header(USER_AGENT, "deps.rs"); - let request = request.body(Body::empty()).unwrap(); + let request = Request::get(uri) + .header(USER_AGENT, "deps.rs") + .body(Body::empty()) + .unwrap(); - Box::new(self.0.call(request).from_err().and_then(|response| { - let status = response.status(); - if !status.is_success() { - try_future!(Err(anyhow!( - "Status code {} for popular repo search", - status - ))); - } + Box::pin( + self.0 + .call(request) + .err_into() + .and_then(|response| { + let status = response.status(); + if !status.is_success() { + return err(anyhow!("Status code {} for popular repo search", status)); + } - let body_future = response.into_body().concat2().from_err(); - let decode_future = body_future - .and_then(|body| serde_json::from_slice(body.as_ref()).map_err(|err| err.into())); - decode_future - .and_then(|search_response: GithubSearchResponse| { - search_response - .items - .into_iter() - .map(|item| { - let path = - RepoPath::from_parts("github", &item.owner.login, &item.name)?; - Ok(Repository { - path, - description: item.description, - }) - }) - .collect::, _>>() + ok(response) }) - .into() - })) + .and_then(|response| body::to_bytes(response.into_body()).err_into()) + .and_then(|bytes| ready(serde_json::from_slice(bytes.as_ref())).err_into()) + .and_then(|search_response: GithubSearchResponse| { + ready( + search_response + .items + .into_iter() + .map(|item| { + let path = + RepoPath::from_parts("github", &item.owner.login, &item.name)?; + + Ok(Repository { + path, + description: item.description, + }) + }) + .collect::, Error>>(), + ) + }), + ) } } diff --git a/src/interactors/gitlab.rs b/src/interactors/gitlab.rs index 2ea5a66..72f7936 100644 --- a/src/interactors/gitlab.rs +++ b/src/interactors/gitlab.rs @@ -1,4 +1,4 @@ -use anyhow::{anyhow, ensure, Error}; +use anyhow::Error; use hyper::Uri; use relative_path::RelativePathBuf; diff --git a/src/interactors/mod.rs b/src/interactors/mod.rs index 70213e5..65f5c23 100644 --- a/src/interactors/mod.rs +++ b/src/interactors/mod.rs @@ -1,8 +1,15 @@ -use anyhow::{anyhow, ensure, Error}; -use futures::{Future, Stream}; -use hyper::{Body, Error as HyperError, Method, Request, Response, header::USER_AGENT}; +use std::{future::Future, task::Poll}; +use std::{pin::Pin, task::Context}; + +use anyhow::{anyhow, Error}; +use futures::{ + future::{err, ok, ready}, + TryFutureExt, +}; +use hyper::{ + body, header::USER_AGENT, service::Service, Body, Error as HyperError, Request, Response, +}; use relative_path::RelativePathBuf; -use tokio_service::Service; use crate::models::repo::{RepoPath, RepoSite}; @@ -15,41 +22,52 @@ pub mod rustsec; #[derive(Debug, Clone)] pub struct RetrieveFileAtPath(pub S); -impl Service for RetrieveFileAtPath +impl Service<(RepoPath, RelativePathBuf)> for RetrieveFileAtPath where - S: Service, Response = Response, Error = HyperError> - + Clone - + 'static, + S: Service, Response = Response, Error = HyperError> + Clone, S::Future: Send + 'static, { - type Request = (RepoPath, RelativePathBuf); type Response = String; type Error = Error; - type Future = Box + Send>; + type Future = Pin> + Send>>; - fn call(&self, req: Self::Request) -> Self::Future { + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.0.poll_ready(cx).map_err(|err| err.into()) + } + + fn call(&mut self, req: (RepoPath, RelativePathBuf)) -> Self::Future { let (repo_path, path) = req; let uri = match &repo_path.site { - &RepoSite::Github => try_future_box!(github::get_manifest_uri(&repo_path, &path)), - &RepoSite::Gitlab => try_future_box!(gitlab::get_manifest_uri(&repo_path, &path)), - &RepoSite::Bitbucket => try_future_box!(bitbucket::get_manifest_uri(&repo_path, &path)), + &RepoSite::Github => github::get_manifest_uri(&repo_path, &path), + &RepoSite::Gitlab => gitlab::get_manifest_uri(&repo_path, &path), + &RepoSite::Bitbucket => bitbucket::get_manifest_uri(&repo_path, &path), }; + if let Err(error) = uri { + return Box::pin(err(error)); + } + + let uri = uri.unwrap(); let request = Request::get(uri.clone()) .header(USER_AGENT, "deps.rs") - .body(Body::empty()).unwrap(); + .body(Body::empty()) + .unwrap(); - Box::new(self.0.call(request).from_err().and_then(move |response| { - let status = response.status(); - if !status.is_success() { - try_future!(Err(anyhow!("Status code {} for URI {}", status, uri))); - } + Box::pin( + self.0 + .call(request) + .err_into() + .and_then(move |response| { + let status = response.status(); - let body_future = response.into_body().concat2().from_err(); - - body_future - .and_then(|body| String::from_utf8(body.to_vec()).map_err(|err| err.into())) - .into() - })) + if status.is_success() { + ok(response) + } else { + err(anyhow!("Status code {} for URI {}", status, uri)) + } + }) + .and_then(|response| body::to_bytes(response.into_body()).err_into()) + .and_then(|bytes| ready(String::from_utf8(bytes.to_vec())).err_into()), + ) } } diff --git a/src/interactors/rustsec.rs b/src/interactors/rustsec.rs index 36716ed..7d28e63 100644 --- a/src/interactors/rustsec.rs +++ b/src/interactors/rustsec.rs @@ -1,35 +1,36 @@ -use std::str; -use std::sync::Arc; +use std::{pin::Pin, sync::Arc, task::Context, task::Poll}; -use anyhow::{anyhow, ensure, Error}; -use futures::{future, future::done, Future, IntoFuture, Stream}; -use hyper::{Body, Error as HyperError, Method, Request, Response}; +use anyhow::Error; +use futures::{future::ready, Future}; +use hyper::{service::Service, Body, Error as HyperError, Request, Response}; use rustsec::database::Database; -use rustsec::repository::DEFAULT_URL; -use tokio_service::Service; #[derive(Debug, Clone)] pub struct FetchAdvisoryDatabase(pub S); -impl Service for FetchAdvisoryDatabase +impl Service<()> for FetchAdvisoryDatabase where - S: Service, Response = Response, Error = HyperError> - + Clone - + 'static, + S: Service, Response = Response, Error = HyperError> + Clone, S::Future: 'static, { - type Request = (); type Response = Arc; type Error = Error; - type Future = Box + Send>; + type Future = Pin> + Send>>; - fn call(&self, _req: ()) -> Self::Future { - let service = self.0.clone(); + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + // TODO: should be this when async client is used again + // self.0.poll_ready(cx).map_err(|err| err.into()) + Poll::Ready(Ok(())) + } - Box::new(done( + // TODO: make fetch async again + fn call(&mut self, _req: ()) -> Self::Future { + let _service = self.0.clone(); + + Box::pin(ready( rustsec::Database::fetch() .map(|db| Arc::new(db)) - .map_err(|err| anyhow!("err fetching rustsec DB")), + .map_err(|err| err.into()), )) } } @@ -39,7 +40,7 @@ where // impl Service for FetchAdvisoryDatabase // where -// S: Service + Clone + 'static, +// S: Service + Clone, // S::Future: 'static, // { // type Request = (); diff --git a/src/main.rs b/src/main.rs index c47c387..53c2c60 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,22 +1,20 @@ #![deny(rust_2018_idioms)] -#![allow(unused)] +#![warn(missing_debug_implementations)] -#[macro_use] -extern crate try_future; - -use std::env; -use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket}; -use std::sync::Mutex; +use std::{ + env, + net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket}, + sync::Mutex, +}; use cadence::{QueuingMetricSink, UdpMetricSink}; -use futures::{Future, Stream}; -use hyper::server::conn::AddrStream; -use hyper::service::{make_service_fn, service_fn_ok}; -use hyper::Client; +use hyper::{ + server::conn::AddrStream, + service::{make_service_fn, service_fn}, + Client, Server, +}; use hyper_tls::HttpsConnector; -use slog::Drain; -use slog::{info, o}; -use tokio_core::reactor::Core; +use slog::{o, Drain}; mod engine; mod interactors; @@ -26,7 +24,7 @@ mod server; mod utils; use self::engine::Engine; -use self::server::Server; +use self::server::App; fn init_metrics() -> QueuingMetricSink { let socket = UdpSocket::bind("0.0.0.0:0").unwrap(); @@ -36,21 +34,15 @@ fn init_metrics() -> QueuingMetricSink { QueuingMetricSink::from(sink) } -fn main() { +#[tokio::main] +async fn main() { let logger = slog::Logger::root( Mutex::new(slog_json::Json::default(std::io::stderr())).map(slog::Fuse), o!("version" => env!("CARGO_PKG_VERSION")), ); let metrics = init_metrics(); - - let mut core = Core::new().expect("failed to create event loop"); - - let handle = core.handle(); - - let connector = HttpsConnector::new(4).expect("failed to create https connector"); - - let client = Client::builder().build(connector); + let client = Client::builder().build(HttpsConnector::new()); let port = env::var("PORT") .unwrap_or_else(|_| "8080".to_string()) @@ -62,15 +54,23 @@ fn main() { let mut engine = Engine::new(client.clone(), logger.clone()); engine.set_metrics(metrics); - let server = Server::new(logger.clone(), engine); - let make_svc = make_service_fn(move |socket: &AddrStream| { - let server = server.clone(); - futures::future::ok::<_, hyper::Error>(server) + let make_svc = make_service_fn(move |_socket: &AddrStream| { + let logger = logger.clone(); + let engine = engine.clone(); + + async move { + let server = App::new(logger.clone(), engine.clone()); + Ok::<_, hyper::Error>(service_fn(move |req| { + let server = server.clone(); + async move { server.handle(req).await } + })) + } }); - let server = hyper::Server::bind(&addr).serve(make_svc); + let server = Server::bind(&addr).serve(make_svc); println!("Server running on port {}", port); - hyper::rt::run(server.map_err(|e| { + + if let Err(e) = server.await { eprintln!("server error: {}", e); - })); + } } diff --git a/src/models/crates.rs b/src/models/crates.rs index 28324cc..c61a54f 100644 --- a/src/models/crates.rs +++ b/src/models/crates.rs @@ -1,7 +1,6 @@ -use std::borrow::Borrow; -use std::str::FromStr; +use std::{borrow::Borrow, str::FromStr}; -use anyhow::{anyhow, ensure, Error}; +use anyhow::{anyhow, Error}; use indexmap::IndexMap; use relative_path::RelativePathBuf; use semver::{Version, VersionReq}; diff --git a/src/parsers/manifest.rs b/src/parsers/manifest.rs index 4cf4f89..358e155 100644 --- a/src/parsers/manifest.rs +++ b/src/parsers/manifest.rs @@ -1,4 +1,4 @@ -use anyhow::{anyhow, ensure, Error}; +use anyhow::{anyhow, Error}; use indexmap::IndexMap; use relative_path::RelativePathBuf; use semver::VersionReq; diff --git a/src/server/mod.rs b/src/server/mod.rs index 4c2eb41..e7443c3 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -1,15 +1,14 @@ -use std::env; -use std::sync::Arc; +use std::{env, sync::Arc}; -use futures::{future, Future, IntoFuture}; -use hyper::header::{CONTENT_TYPE, LOCATION}; -use hyper::service::Service; -use hyper::{Body, Error as HyperError, Method, Request, Response, StatusCode}; +use futures::future; +use hyper::{ + header::{CONTENT_TYPE, LOCATION}, + Body, Error as HyperError, Method, Request, Response, StatusCode, +}; use once_cell::sync::Lazy; use route_recognizer::{Params, Router}; use semver::VersionReq; -use slog::Logger; -use slog::{error, o}; +use slog::{error, o, Logger}; mod assets; mod views; @@ -19,13 +18,13 @@ use crate::models::crates::{CrateName, CratePath}; use crate::models::repo::RepoPath; use crate::models::SubjectPath; -#[derive(Clone, Copy, PartialEq)] +#[derive(Debug, Clone, Copy, PartialEq)] enum StatusFormat { Html, Svg, } -#[derive(Clone, Copy)] +#[derive(Debug, Clone, Copy)] enum StaticFile { StyleCss, FaviconPng, @@ -40,14 +39,14 @@ enum Route { } #[derive(Clone)] -pub struct Server { +pub struct App { logger: Logger, engine: Engine, router: Arc>, } -impl Server { - pub fn new(logger: Logger, engine: Engine) -> Server { +impl App { + pub fn new(logger: Logger, engine: Engine) -> App { let mut router = Router::new(); router.add("/", Route::Index); @@ -74,21 +73,14 @@ impl Server { Route::CrateStatus(StatusFormat::Svg), ); - Server { + App { logger, engine, router: Arc::new(router), } } -} -impl Service for Server { - type ReqBody = Body; - type ResBody = Body; - type Error = hyper::Error; - type Future = Box, Error = Self::Error> + Send>; - - fn call(&mut self, req: Request) -> Self::Future { + pub async fn handle(&self, req: Request) -> Result, HyperError> { let logger = self .logger .new(o!("http_path" => req.uri().path().to_owned())); @@ -97,188 +89,196 @@ impl Service for Server { match route_match.handler { &Route::Index => { if *req.method() == Method::GET { - return Box::new(self.index(req, route_match.params, logger)); + return self.index(req, route_match.params, logger).await; } } &Route::RepoStatus(format) => { if *req.method() == Method::GET { - return Box::new(self.repo_status(req, route_match.params, logger, format)); + return self + .repo_status(req, route_match.params, logger, format) + .await; } } &Route::CrateStatus(format) => { + println!("route"); if *req.method() == Method::GET { - return Box::new(self.crate_status( - req, - route_match.params, - logger, - format, - )); + println!("get"); + return self + .crate_status(req, route_match.params, logger, format) + .await; } } &Route::CrateRedirect => { if *req.method() == Method::GET { - return Box::new(self.crate_redirect(req, route_match.params, logger)); + return self.crate_redirect(req, route_match.params, logger).await; } } &Route::Static(file) => { if *req.method() == Method::GET { - return Box::new(future::ok(Server::static_file(file))); + return Ok(App::static_file(file)); } } } } - let mut response = Response::builder(); - response.status(StatusCode::NOT_FOUND); - Box::new(future::ok(response.body(Body::empty()).unwrap())) + Ok(Response::builder() + .status(StatusCode::NOT_FOUND) + .body(Body::empty()) + .unwrap()) } } -impl Server { - fn index( +impl App { + async fn index( &self, _req: Request, _params: Params, logger: Logger, - ) -> impl Future, Error = HyperError> + Send { - self.engine - .get_popular_repos() - .join(self.engine.get_popular_crates()) - .then(move |popular_result| match popular_result { - Err(err) => { - error!(logger, "error: {}", err); - let mut response = - views::html::error::render("Could not retrieve popular items", ""); - *response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR; - future::ok(response) - } - Ok((popular_repos, popular_crates)) => { - future::ok(views::html::index::render(popular_repos, popular_crates)) - } - }) + ) -> Result, HyperError> { + let engine = self.engine.clone(); + + let popular = + future::try_join(engine.get_popular_repos(), engine.get_popular_crates()).await; + + match popular { + Err(err) => { + error!(logger, "error: {}", err); + let mut response = + views::html::error::render("Could not retrieve popular items", ""); + *response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR; + Ok(response) + } + Ok((popular_repos, popular_crates)) => { + Ok(views::html::index::render(popular_repos, popular_crates)) + } + } } - fn repo_status( + async fn repo_status( &self, _req: Request, params: Params, logger: Logger, format: StatusFormat, - ) -> impl Future, Error = HyperError> + Send { + ) -> Result, HyperError> { let server = self.clone(); let site = params.find("site").expect("route param 'site' not found"); let qual = params.find("qual").expect("route param 'qual' not found"); let name = params.find("name").expect("route param 'name' not found"); - RepoPath::from_parts(site, qual, name) - .into_future() - .then(move |repo_path_result| match repo_path_result { - Err(err) => { - error!(logger, "error: {}", err); - let mut response = views::html::error::render( - "Could not parse repository path", - "Please make sure to provide a valid repository path.", - ); - *response.status_mut() = StatusCode::BAD_REQUEST; - future::Either::A(future::ok(response)) + let repo_path_result = RepoPath::from_parts(site, qual, name); + + match repo_path_result { + Err(err) => { + error!(logger, "error: {}", err); + let mut response = views::html::error::render( + "Could not parse repository path", + "Please make sure to provide a valid repository path.", + ); + *response.status_mut() = StatusCode::BAD_REQUEST; + Ok(response) + } + + Ok(repo_path) => { + let analyze_result = server + .engine + .analyze_repo_dependencies(repo_path.clone()) + .await; + + match analyze_result { + Err(err) => { + error!(logger, "error: {}", err); + let response = + App::status_format_analysis(None, format, SubjectPath::Repo(repo_path)); + Ok(response) + } + Ok(analysis_outcome) => { + let response = App::status_format_analysis( + Some(analysis_outcome), + format, + SubjectPath::Repo(repo_path), + ); + Ok(response) + } } - Ok(repo_path) => future::Either::B( - server - .engine - .analyze_repo_dependencies(repo_path.clone()) - .then(move |analyze_result| match analyze_result { - Err(err) => { - error!(logger, "error: {}", err); - let response = Server::status_format_analysis( - None, - format, - SubjectPath::Repo(repo_path), - ); - future::ok(response) - } - Ok(analysis_outcome) => { - let response = Server::status_format_analysis( - Some(analysis_outcome), - format, - SubjectPath::Repo(repo_path), - ); - future::ok(response) - } - }), - ), - }) + } + } } - fn crate_redirect( + async fn crate_redirect( &self, _req: Request, params: Params, logger: Logger, - ) -> impl Future, Error = HyperError> + Send { + ) -> Result, HyperError> { let engine = self.engine.clone(); let name = params.find("name").expect("route param 'name' not found"); + let crate_name_result = name.parse::(); - name.parse::() - .into_future() - .then(move |crate_name_result| match crate_name_result { - Err(err) => { - error!(logger, "error: {}", err); - let mut response = views::html::error::render( - "Could not parse crate name", - "Please make sure to provide a valid crate name.", - ); - *response.status_mut() = StatusCode::BAD_REQUEST; - future::Either::A(future::ok(response)) + match crate_name_result { + Err(err) => { + error!(logger, "error: {}", err); + let mut response = views::html::error::render( + "Could not parse crate name", + "Please make sure to provide a valid crate name.", + ); + *response.status_mut() = StatusCode::BAD_REQUEST; + Ok(response) + } + + Ok(crate_name) => { + let release_result = engine + .find_latest_crate_release(crate_name, VersionReq::any()) + .await; + + match release_result { + Err(err) => { + error!(logger, "error: {}", err); + let mut response = views::html::error::render( + "Could not fetch crate information", + "Please make sure to provide a valid crate name.", + ); + *response.status_mut() = StatusCode::NOT_FOUND; + Ok(response) + } + Ok(None) => { + let mut response = views::html::error::render( + "Could not fetch crate information", + "Please make sure to provide a valid crate name.", + ); + *response.status_mut() = StatusCode::NOT_FOUND; + Ok(response) + } + Ok(Some(release)) => { + let redirect_url = format!( + "{}/crate/{}/{}", + &SELF_BASE_URL as &str, + release.name.as_ref(), + release.version + ); + + let res = Response::builder() + .status(StatusCode::TEMPORARY_REDIRECT) + .header(LOCATION, redirect_url) + .body(Body::empty()) + .unwrap(); + + Ok(res) + } } - Ok(crate_name) => future::Either::B( - engine - .find_latest_crate_release(crate_name, VersionReq::any()) - .then(move |release_result| match release_result { - Err(err) => { - error!(logger, "error: {}", err); - let mut response = views::html::error::render( - "Could not fetch crate information", - "Please make sure to provide a valid crate name.", - ); - *response.status_mut() = StatusCode::NOT_FOUND; - future::ok(response) - } - Ok(None) => { - let mut response = views::html::error::render( - "Could not fetch crate information", - "Please make sure to provide a valid crate name.", - ); - *response.status_mut() = StatusCode::NOT_FOUND; - future::ok(response) - } - Ok(Some(release)) => { - let mut response = Response::builder(); - response.status(StatusCode::TEMPORARY_REDIRECT); - let url = format!( - "{}/crate/{}/{}", - &SELF_BASE_URL as &str, - release.name.as_ref(), - release.version - ); - response.header(LOCATION, url); - - let response = response.body(Body::empty()).unwrap(); - future::ok(response) - } - }), - ), - }) + } + } } - fn crate_status( + async fn crate_status( &self, _req: Request, params: Params, logger: Logger, format: StatusFormat, - ) -> impl Future, Error = HyperError> + Send { + ) -> Result, HyperError> { let server = self.clone(); let name = params.find("name").expect("route param 'name' not found"); @@ -286,43 +286,56 @@ impl Server { .find("version") .expect("route param 'version' not found"); - CratePath::from_parts(name, version) - .into_future() - .then(move |crate_path_result| match crate_path_result { - Err(err) => { - error!(logger, "error: {}", err); - let mut response = views::html::error::render( - "Could not parse crate path", - "Please make sure to provide a valid crate name and version.", - ); - *response.status_mut() = StatusCode::BAD_REQUEST; - future::Either::A(future::ok(response)) + let crate_path_result = CratePath::from_parts(name, version); + + println!("crate path {:?}", &crate_path_result); + + match crate_path_result { + Err(err) => { + error!(logger, "error: {}", err); + let mut response = views::html::error::render( + "Could not parse crate path", + "Please make sure to provide a valid crate name and version.", + ); + *response.status_mut() = StatusCode::BAD_REQUEST; + Ok(response) + } + Ok(crate_path) => { + println!("crate path ok"); + + let analyze_result = server + .engine + .analyze_crate_dependencies(crate_path.clone()) + .await; + + println!("results analyzed {:?}", &analyze_result); + + match analyze_result { + Err(err) => { + error!(logger, "error: {}", err); + let response = App::status_format_analysis( + None, + format, + SubjectPath::Crate(crate_path), + ); + Ok(response) + } + Ok(analysis_outcome) => { + println!("analysis ok"); + + let response = App::status_format_analysis( + Some(analysis_outcome), + format, + SubjectPath::Crate(crate_path), + ); + + println!("response created"); + + Ok(response) + } } - Ok(crate_path) => future::Either::B( - server - .engine - .analyze_crate_dependencies(crate_path.clone()) - .then(move |analyze_result| match analyze_result { - Err(err) => { - error!(logger, "error: {}", err); - let response = Server::status_format_analysis( - None, - format, - SubjectPath::Crate(crate_path), - ); - future::ok(response) - } - Ok(analysis_outcome) => { - let response = Server::status_format_analysis( - Some(analysis_outcome), - format, - SubjectPath::Crate(crate_path), - ); - future::ok(response) - } - }), - ), - }) + } + } } fn status_format_analysis( diff --git a/src/server/views/html/mod.rs b/src/server/views/html/mod.rs index 7004f00..b4407c2 100644 --- a/src/server/views/html/mod.rs +++ b/src/server/views/html/mod.rs @@ -8,7 +8,7 @@ pub mod error; pub mod index; pub mod status; -use super::super::SELF_BASE_URL; +use crate::server::SELF_BASE_URL; fn render_html(title: &str, body: B) -> Response { let rendered = html! { diff --git a/src/server/views/html/status.rs b/src/server/views/html/status.rs index 7049888..46ab2a1 100644 --- a/src/server/views/html/status.rs +++ b/src/server/views/html/status.rs @@ -6,8 +6,7 @@ use crate::engine::AnalyzeDependenciesOutcome; use crate::models::crates::{AnalyzedDependencies, AnalyzedDependency, CrateName}; use crate::models::repo::RepoSite; use crate::models::SubjectPath; - -use super::super::badge; +use crate::server::views::badge; fn dependency_tables(crate_name: CrateName, deps: AnalyzedDependencies) -> Markup { html! { diff --git a/src/utils/cache.rs b/src/utils/cache.rs index 5018b42..bcad41a 100644 --- a/src/utils/cache.rs +++ b/src/utils/cache.rs @@ -1,29 +1,31 @@ -use std::fmt::{Debug, Formatter, Result as FmtResult}; -use std::hash::Hash; -use std::ops::Deref; -use std::sync::Mutex; -use std::time::{Duration, Instant}; +use std::{ + fmt::{Debug, Formatter, Result as FmtResult}, + hash::Hash, + sync::Mutex, + task::Context, + task::Poll, + time::{Duration, Instant}, +}; -use anyhow::{anyhow, ensure, Error}; -use futures::future::{FromErr, Shared, SharedItem}; -use futures::{Future, Poll}; +use anyhow::Error; +use hyper::service::Service; use lru_cache::LruCache; -use tokio_service::Service; -pub struct Cache +pub struct Cache where - S: Service, - S::Request: Hash + Eq, + S: Service, + Req: Hash + Eq, { inner: S, duration: Duration, - cache: Mutex>)>>, + #[allow(unused)] + cache: Mutex>, } -impl Debug for Cache +impl Debug for Cache where - S: Service + Debug, - S::Request: Hash + Eq, + S: Service + Debug, + Req: Hash + Eq, { fn fmt(&self, fmt: &mut Formatter<'_>) -> FmtResult { fmt.debug_struct("Cache") @@ -33,12 +35,12 @@ where } } -impl Cache +impl Cache where - S: Service, - S::Request: Hash + Eq, + S: Service, + Req: Hash + Eq, { - pub fn new(service: S, duration: Duration, capacity: usize) -> Cache { + pub fn new(service: S, duration: Duration, capacity: usize) -> Cache { Cache { inner: service, duration: duration, @@ -47,63 +49,62 @@ where } } -impl Service for Cache +impl Service for Cache where - S: Service, - S::Request: Clone + Hash + Eq, + S: Service, + S::Response: Clone, + Req: Clone + Hash + Eq, { - type Request = S::Request; - type Response = CachedItem; + type Response = S::Response; type Error = Error; - type Future = Cached; + // WAS: type Future = Cached; + // type Future = Pin>>>; + type Future = S::Future; - fn call(&self, req: Self::Request) -> Self::Future { - let now = Instant::now(); - let mut cache = self.cache.lock().expect("lock poisoned"); - if let Some(&mut (valid_until, ref shared_future)) = cache.get_mut(&req) { - if valid_until > now { - if let Some(Ok(_)) = shared_future.peek() { - return Cached(shared_future.clone()); - } - } - } - let shared_future = self.inner.call(req.clone()).from_err().shared(); - cache.insert(req, (now + self.duration, shared_future.clone())); - Cached(shared_future) + fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn call(&mut self, req: Req) -> Self::Future { + // TODO: re-add caching + // Box::pin({ + // let now = Instant::now(); + // let mut cache = self.cache.lock().expect("lock poisoned"); + + // if let Some(&mut (valid_until, ref cached_response)) = cache.get_mut(&req) { + // if valid_until > now { + // return Box::pin(ok(cached_response.clone())); + // } + // } + + self.inner.call(req.clone()) + // .and_then(|response| { + // // cache.insert(req, (now + self.duration, response.clone())); + // ok(response) + // }) + // }) } } -pub struct Cached>(Shared>); +// pub struct Cached(Shared); -impl Debug for Cached -where - F: Future + Debug, - F::Item: Debug, -{ - fn fmt(&self, fmt: &mut Formatter<'_>) -> FmtResult { - self.0.fmt(fmt) - } -} +// impl Debug for Cached +// where +// F: Future + Debug, +// F::Output: Debug, +// { +// fn fmt(&self, fmt: &mut Formatter<'_>) -> FmtResult { +// self.0.fmt(fmt) +// } +// } -impl> Future for Cached { - type Item = CachedItem; - type Error = Error; +// // WAS: impl> Future for Cached { +// impl Future for Cached { +// type Output = Result; - fn poll(&mut self) -> Poll { - self.0 - .poll() - .map_err(|_err| anyhow!("TODO: shared error not clone-able")) - .map(|item| item.map(CachedItem)) - } -} - -#[derive(Debug)] -pub struct CachedItem(SharedItem); - -impl Deref for CachedItem { - type Target = T; - - fn deref(&self) -> &T { - &self.0.deref() - } -} +// fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { +// self.0 +// .poll() +// .map_err(|_err| anyhow!("TODO: shared error not clone-able")) +// } +// } From 464256086823373fad2b3e3c172c919d07899d89 Mon Sep 17 00:00:00 2001 From: Rob Ede Date: Thu, 1 Oct 2020 11:24:51 +0100 Subject: [PATCH 2/3] remove debugging print --- src/engine/mod.rs | 2 -- src/interactors/crates.rs | 5 +---- src/interactors/mod.rs | 14 +++++++------- src/server/mod.rs | 12 ------------ 4 files changed, 8 insertions(+), 25 deletions(-) diff --git a/src/engine/mod.rs b/src/engine/mod.rs index e7f119a..399180a 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -201,8 +201,6 @@ impl Engine { ) -> Result { let start = Instant::now(); - println!("analyze deps"); - let query_response = self .query_crate .lock() diff --git a/src/interactors/crates.rs b/src/interactors/crates.rs index c97a527..ecd8cfc 100644 --- a/src/interactors/crates.rs +++ b/src/interactors/crates.rs @@ -103,9 +103,6 @@ where }; let uri = format!("{}/master/{}", CRATES_INDEX_BASE_URI, path); - - println!("analyze from uri {:?}", &uri); - let uri = uri.parse::().expect("TODO: MAP ERROR PROPERLY"); let request = Request::get(uri.clone()) @@ -133,7 +130,7 @@ where .lines() .map(|s| s.trim()) .filter(|s| !s.is_empty()) - .map(|s| serde_json::from_str::(s)) + .map(serde_json::from_str) .collect::>(), ) .err_into() diff --git a/src/interactors/mod.rs b/src/interactors/mod.rs index 65f5c23..efef47e 100644 --- a/src/interactors/mod.rs +++ b/src/interactors/mod.rs @@ -38,16 +38,16 @@ where fn call(&mut self, req: (RepoPath, RelativePathBuf)) -> Self::Future { let (repo_path, path) = req; let uri = match &repo_path.site { - &RepoSite::Github => github::get_manifest_uri(&repo_path, &path), - &RepoSite::Gitlab => gitlab::get_manifest_uri(&repo_path, &path), - &RepoSite::Bitbucket => bitbucket::get_manifest_uri(&repo_path, &path), + RepoSite::Github => github::get_manifest_uri(&repo_path, &path), + RepoSite::Gitlab => gitlab::get_manifest_uri(&repo_path, &path), + RepoSite::Bitbucket => bitbucket::get_manifest_uri(&repo_path, &path), }; - if let Err(error) = uri { - return Box::pin(err(error)); - } + let uri = match uri { + Ok(uri) => uri, + Err(error) => return Box::pin(err(error)), + }; - let uri = uri.unwrap(); let request = Request::get(uri.clone()) .header(USER_AGENT, "deps.rs") .body(Body::empty()) diff --git a/src/server/mod.rs b/src/server/mod.rs index e7443c3..c104f55 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -100,9 +100,7 @@ impl App { } } &Route::CrateStatus(format) => { - println!("route"); if *req.method() == Method::GET { - println!("get"); return self .crate_status(req, route_match.params, logger, format) .await; @@ -288,8 +286,6 @@ impl App { let crate_path_result = CratePath::from_parts(name, version); - println!("crate path {:?}", &crate_path_result); - match crate_path_result { Err(err) => { error!(logger, "error: {}", err); @@ -301,15 +297,11 @@ impl App { Ok(response) } Ok(crate_path) => { - println!("crate path ok"); - let analyze_result = server .engine .analyze_crate_dependencies(crate_path.clone()) .await; - println!("results analyzed {:?}", &analyze_result); - match analyze_result { Err(err) => { error!(logger, "error: {}", err); @@ -321,16 +313,12 @@ impl App { Ok(response) } Ok(analysis_outcome) => { - println!("analysis ok"); - let response = App::status_format_analysis( Some(analysis_outcome), format, SubjectPath::Crate(crate_path), ); - println!("response created"); - Ok(response) } } From 45f114b3f671d5ab3bce84ab96f8fe33abeeba2f Mon Sep 17 00:00:00 2001 From: Rob Ede Date: Thu, 1 Oct 2020 15:30:19 +0100 Subject: [PATCH 3/3] remove unecessary into_iter/collect Co-authored-by: Eduardo Pinho --- src/engine/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 399180a..380507a 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -224,7 +224,7 @@ impl Engine { let analyzed_deps = analyze_dependencies(engine.clone(), release.deps.clone()).await?; - let crates = vec![(crate_path.name, analyzed_deps)].into_iter().collect(); + let crates = vec![(crate_path.name, analyzed_deps)]; let duration = start.elapsed(); Ok(AnalyzeDependenciesOutcome { crates, duration })