Robust proxying with retry

Also includes upgraded hyper-reverse-proxy, hyper, and futures.

This is looking and feeling way more ergonomic.

Cargo.lock

@@ -15,6 +15,12 @@ version = "1.0.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "85bb70cc08ec97ca5450e6eba421deeea5f172c0fc61f78b5357b2a8e8be195f"
 
+[[package]]
+name = "arc-swap"
+version = "0.4.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4d25d88fd6b8041580a654f9d0c581a047baee2b3efee13275f2fc392fc75034"
+
 [[package]]
 name = "atty"
 version = "0.2.14"
@@ -38,22 +44,11 @@ version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693"
 
-[[package]]
-name = "byteorder"
-version = "1.3.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-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"
+checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38"
-dependencies = [
- "byteorder",
- "either",
- "iovec",
-]
 
 [[package]]
 name = "cfg-if"
@@ -61,69 +56,6 @@ version = "0.1.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4785bdd1c96b2a846b2bd7cc02e86b6b3dbf14e7e53446c4f54c92a361040822"
 
-[[package]]
-name = "cloudabi"
-version = "0.0.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ddfc5b9aa5d4507acaf872de71051dfd0e309860e88966e1051e462a077aac4f"
-dependencies = [
- "bitflags",
-]
-
-[[package]]
-name = "crossbeam-deque"
-version = "0.7.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9f02af974daeee82218205558e51ec8768b48cf524bd01d550abe5573a608285"
-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",
-]
-
-[[package]]
-name = "crossbeam-utils"
-version = "0.7.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c3c7c73a2d1e9fc0886a08b93e98eb643461230d5f1925e4036204d5f2e261a8"
-dependencies = [
- "autocfg",
- "cfg-if",
- "lazy_static",
-]
-
-[[package]]
-name = "either"
-version = "1.5.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bb1f6b1ce1c140482ea30ddd3335fc0024ac7ee112895426e0a629a6c20adfe3"
-
 [[package]]
 name = "env_logger"
 version = "0.7.1"
@@ -161,18 +93,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"
+name = "futures-channel"
-version = "0.1.8"
+version = "0.3.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ab90cde24b3319636588d0c35fe03b1333857621051837ed769faefb4c2162e4"
+checksum = "f366ad74c28cca6ba456d95e6422883cfb4b252a83bed929c83abfdbbf2967d5"
 dependencies = [
- "futures",
+ "futures-core",
- "num_cpus",
+ "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]]
@@ -188,20 +199,21 @@ dependencies = [
 
 [[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",
- "tokio-io",
+ "tokio-util",
+ "tracing",
 ]
 
 [[package]]
@@ -215,9 +227,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",
@@ -226,14 +238,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]]
@@ -253,41 +263,33 @@ dependencies = [
 
 [[package]]
 name = "hyper"
-version = "0.12.35"
+version = "0.13.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9dbe6ed1438e1f8ad955a4701e9a944938e9519f6888d12d8558b645e247d5f6"
+checksum = "3e68a8dd9716185d9e64ea473ea6ef63529252e3e27623295a0378a19665d5eb"
 dependencies = [
  "bytes",
- "futures",
+ "futures-channel",
- "futures-cpupool",
+ "futures-core",
+ "futures-util",
  "h2",
  "http",
  "http-body",
  "httparse",
- "iovec",
  "itoa",
- "log",
+ "pin-project",
- "net2",
+ "socket2",
- "rustc_version",
  "time",
  "tokio",
- "tokio-buf",
+ "tower-service",
- "tokio-executor",
+ "tracing",
- "tokio-io",
- "tokio-reactor",
- "tokio-tcp",
- "tokio-threadpool",
- "tokio-timer",
  "want",
 ]
 
 [[package]]
 name = "hyper-reverse-proxy"
-version = "0.4.0"
+version = "0.5.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
+source = "git+https://github.com/felipenoris/hyper-reverse-proxy.git#d109fcbd629c62eeb5c34bec1074d737e7b58529"
-checksum = "474f19c624c13005d914fc4a73aa2a2921b0371f5f7b748146f5130685492ce8"
 dependencies = [
- "futures",
  "hyper",
  "lazy_static",
  "unicase",
@@ -339,15 +341,6 @@ version = "0.2.71"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9457b06509d27052635f90d6466700c65095fdf75409b3fbdd903e988b886f49"
 
-[[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.8"
@@ -357,27 +350,12 @@ dependencies = [
  "cfg-if",
 ]
 
-[[package]]
-name = "maybe-uninit"
-version = "2.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "60302e4db3a61da70c0cb7991976248362f30319e88850c487b9b95bbf059e00"
-
 [[package]]
 name = "memchr"
 version = "2.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3728d817d99e5ac407411fa471ff9800a778d88a24685968b36824eaf4bee400"
 
-[[package]]
-name = "memoffset"
-version = "0.5.4"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b4fc2c02a7e374099d4ee95a193111f72d2110197fe200272371758f6c3643d8"
-dependencies = [
- "autocfg",
-]
-
 [[package]]
 name = "mio"
 version = "0.6.22"
@@ -391,12 +369,35 @@ 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"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "afcb699eb26d4332647cc848492bbc15eafb26f08d0304550d5aa1f612e066f0"
+dependencies = [
+ "iovec",
+ "libc",
+ "mio",
+]
+
 [[package]]
 name = "miow"
 version = "0.2.1"
@@ -409,6 +410,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 = "net2"
 version = "0.2.34"
@@ -447,37 +458,61 @@ dependencies = [
 ]
 
 [[package]]
-name = "parking_lot"
+name = "once_cell"
-version = "0.9.0"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f842b1982eb6c2fe34036a4fbfb06dd185a3f5c8edfaacdf7d1ea10b07de6252"
+checksum = "260e51e7efe62b592207e9e13a68e43692a7a279171d6ba57abd208bf23645ad"
+
+[[package]]
+name = "pin-project"
+version = "0.4.23"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ca4433fff2ae79342e497d9f8ee990d174071408f28f726d6d83af93e58e48aa"
 dependencies = [
- "lock_api",
+ "pin-project-internal",
- "parking_lot_core",
- "rustc_version",
 ]
 
 [[package]]
-name = "parking_lot_core"
+name = "pin-project-internal"
-version = "0.6.2"
+version = "0.4.23"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b876b1b9e7ac6e1a74a6da34d25c42e17e8862aa409cbbbdcfc8d86c6f3bc62b"
+checksum = "2c0e815c3ee9a031fdf5af21c10aa17c573c9c6a566328d99e3936c34e36461f"
 dependencies = [
- "cfg-if",
+ "proc-macro2",
- "cloudabi",
+ "quote",
- "libc",
+ "syn",
- "redox_syscall",
- "rustc_version",
- "smallvec",
- "winapi 0.3.9",
 ]
 
+[[package]]
+name = "pin-project-lite"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "282adbf10f2698a7a77f8e983a74b2d18176c19a7fd32a45446139ae7b02b715"
+
+[[package]]
+name = "pin-utils"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
+
 [[package]]
 name = "ppv-lite86"
 version = "0.2.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "237a5ed80e274dbc66f86bd59c1e25edc039660be53194b5fe0a482e0f2612ea"
 
+[[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.18"
@@ -567,42 +602,12 @@ version = "0.6.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "26412eb97c6b088a6997e05f69403a802a92d520de2f8e63c2b65f9e0f47c4e8"
 
-[[package]]
-name = "rustc_version"
-version = "0.2.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "138e3e0acb6c9fb258b19b67cb8abd63c00679d2851805ea151465464fe9030a"
-dependencies = [
- "semver",
-]
-
 [[package]]
 name = "ryu"
 version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "71d301d4193d031abdd79ff7e3dd721168a9572ef3fe51a1517aba235bd8f86e"
 
-[[package]]
-name = "scopeguard"
-version = "1.1.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d29ab0c6d3fc0ee92fe66e2d99f700eab17a8d57d1c1d3b748380fb20baa78cd"
-
-[[package]]
-name = "semver"
-version = "0.9.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1d7eb9ef2c18661902cc47e535f9bc51b78acd254da71d375c2f6720d9a40403"
-dependencies = [
- "semver-parser",
-]
-
-[[package]]
-name = "semver-parser"
-version = "0.7.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3"
-
 [[package]]
 name = "serde"
 version = "1.0.114"
@@ -631,6 +636,16 @@ dependencies = [
  "serde",
 ]
 
+[[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]]
 name = "slab"
 version = "0.4.2"
@@ -638,21 +653,15 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c111b5bd5695e56cffe5129854aa230b39c93a305372fdbb2668ca2394eea9f8"
 
 [[package]]
-name = "smallvec"
+name = "socket2"
-version = "0.6.13"
+version = "0.3.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f7b0758c52e15a8b5e3691eae6cc559f08eee9406e548a4477ba4e67770a82b6"
+checksum = "03088793f677dce356f3ccc2edb1b314ad191ab702a5de3faf49304f7e104918"
 dependencies = [
- "maybe-uninit",
+ "cfg-if",
-]
+ "libc",
-
+ "redox_syscall",
-[[package]]
+ "winapi 0.3.9",
-name = "string"
-version = "0.2.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d24114bfcceb867ca7f71a0d3fe45d45619ec47a6fbfa98cb14e14250bfa5d6d"
-dependencies = [
- "bytes",
 ]
 
 [[package]]
@@ -696,134 +705,51 @@ dependencies = [
 
 [[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-current-thread",
- "tokio-executor",
- "tokio-io",
- "tokio-reactor",
- "tokio-threadpool",
- "tokio-timer",
-]
-
-[[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-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-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-sync"
-version = "0.1.8"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "edfe50152bc8164fcc456dab7891fa9bf8beaf01c5ee7e1dd43a397c3cf87dee"
-dependencies = [
  "fnv",
- "futures",
+ "futures-core",
+ "iovec",
+ "lazy_static",
+ "libc",
+ "memchr",
+ "mio",
+ "mio-named-pipes",
+ "mio-uds",
+ "num_cpus",
+ "pin-project-lite",
+ "signal-hook-registry",
+ "slab",
+ "tokio-macros",
+ "winapi 0.3.9",
 ]
 
 [[package]]
-name = "tokio-tcp"
+name = "tokio-macros"
-version = "0.1.4"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "98df18ed66e3b72e742f185882a9e201892407957e45fbff8da17ae7a7c51f72"
+checksum = "f0c3acc6aa564495a0f2e1d59fab677cd7f81a19994cfc7f3ad0e64301560389"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
+[[package]]
+name = "tokio-util"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "be8242891f2b6cbef26a2d7e8605133c2c554cd35b3e4948ea892d6d68436499"
 dependencies = [
  "bytes",
- "futures",
+ "futures-core",
- "iovec",
+ "futures-sink",
- "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",
+ "pin-project-lite",
- "slab",
+ "tokio",
- "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]]
@@ -835,6 +761,32 @@ 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.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6d79ca061b032d6ce30c660fded31189ca0b9922bf483cd70759f13a2d86786c"
+dependencies = [
+ "cfg-if",
+ "log",
+ "tracing-core",
+]
+
+[[package]]
+name = "tracing-core"
+version = "0.1.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4f0e00789804e99b20f12bc7003ca416309d28a6f495d6af58d1e2c2842461b5"
+dependencies = [
+ "lazy_static",
+]
+
 [[package]]
 name = "try-lock"
 version = "0.2.2"
@@ -856,6 +808,7 @@ dependencies = [
  "serde",
  "serde_derive",
  "serde_json",
+ "tokio",
  "toml",
 ]
 
@@ -882,11 +835,10 @@ checksum = "b5a972e5669d67ba988ce3dc826706fb0a8b01471c088cb0b6110b805cc36aed"
 
 [[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",
 ]

Cargo.toml

@@ -15,13 +15,14 @@ path = "src/main.rs"
 [dependencies]
 anyhow = "1.0.31"
 env_logger = "0.7.1"
-futures = "0.1"
+futures = "0.3"
-hyper = "0.12"
+hyper = "0.13"
-hyper-reverse-proxy = "0.4"
+hyper-reverse-proxy = { git = "https://github.com/felipenoris/hyper-reverse-proxy.git", commit = "d109fcbd629c62eeb5c34bec1074d737e7b58529" }
 log = "0.4.8"
 newrelic = "0.2.2"
 rand = "0.7.3"
 serde = "1.0.104"
 serde_derive = "1.0.104"
 serde_json = "1.0.55"
+tokio = { version = "0.2", features = ["full"] }
 toml = "0.5.6"

src/api.rs

@@ -0,0 +1,102 @@
+use anyhow::Result as AnyhowResult;
+use futures::TryStreamExt;
+use hyper::http::HeaderValue;
+use hyper::{Body, Request, HeaderMap, Response, StatusCode};
+use hyper::body::Bytes;
+
+/// NB: This much match the shape of SpeakRequest in the 'voder/tts_service' code.
+/// This is used for both /speak and /speak_spectrogram requests.
+#[derive(Deserialize, Debug)]
+pub struct ExternalSpeakRequest {
+  /// Slug for the speaker
+  pub speaker: String,
+
+  /// Raw text to be spoken
+  pub text: String,
+}
+
+/// The full request
+pub struct ExternalHttpRequest {
+  pub headers: HeaderMap<HeaderValue>,
+  pub speak_request: ExternalSpeakRequest,
+}
+
+#[derive(Serialize, Debug)]
+pub struct InternalSpeakRequest {
+  /// Slug for the speaker
+  pub speaker: String,
+
+  /// Raw text to be spoken
+  pub text: String,
+
+  /// Retry attempt.
+  /// The first request is "0", and all subsequent requests are incremented.
+  /// Subsequent requests do not trigger the rate limiter.
+  pub retry_attempt_number: u8,
+
+  /// Admin/debug requests get to skip the rate limiter.
+  pub skip_rate_limiter: bool,
+}
+
+pub enum ErrorType {
+  /// The proxy couldn't find a healthy host (or it died mid-connection)
+  ProxyError,
+  // The backend server reported a 500
+  ServerError,
+  /// Client is rate limited
+  RateLimitError,
+}
+
+pub struct ErrorResponse {
+  /// The type of error
+  pub error_type: ErrorType,
+  /// Description of the error
+  pub error_description: String,
+}
+
+impl ExternalHttpRequest {
+
+  pub async fn decode_request(request: Request<Body>) -> AnyhowResult<Self> {
+    let headers = request.headers().clone();
+    let (_parts, body) = request.into_parts();
+
+    let body_bytes = body_to_bytes(body).await?;
+    let speak_request = serde_json::from_slice::<ExternalSpeakRequest>(&body_bytes)?;
+
+    Ok(ExternalHttpRequest {
+      headers: headers,
+      speak_request:  speak_request,
+    })
+  }
+}
+
+impl ErrorType {
+  pub fn status_code(&self) -> StatusCode {
+    match *self {
+      ErrorType::ProxyError =>StatusCode::INTERNAL_SERVER_ERROR,
+      ErrorType::ServerError => StatusCode::INTERNAL_SERVER_ERROR,
+      ErrorType::RateLimitError => StatusCode::TOO_MANY_REQUESTS,
+    }
+  }
+}
+
+impl ErrorResponse {
+  pub fn to_response(&self) -> Response<Body> {
+    let status_code = self.error_type.status_code();
+    let body = Body::from(self.error_description.clone());
+    Response::builder()
+      .status(status_code)
+      .body(body)
+      .unwrap()
+  }
+}
+
+async fn body_to_bytes(body: Body) -> AnyhowResult<Vec<u8>> {
+  // Also: let body_bytes = hyper::body::to_bytes(body)
+  let bytes = body.try_fold(Vec::new(), |mut data, chunk| async move {
+    data.extend_from_slice(&chunk);
+    Ok(data)
+  }).await?;
+
+  Ok(bytes)
+}

src/main.rs

@@ -1,32 +1,33 @@
+#![feature(toowned_clone_into)]
+
 #[macro_use] extern crate anyhow;
 #[macro_use] extern crate log;
 #[macro_use] extern crate serde_derive;
 
+pub mod api;
 pub mod newrelic_logger;
 pub mod speak_proxy;
 
 use anyhow::Result as AnyhowResult;
 use crate::newrelic_logger::{NewRelicLogger, MaybeNewRelicTransaction};
-use crate::speak_proxy::{speak_proxy, speak_proxy_with_retry};
+use crate::speak_proxy::speak_proxy_with_retry;
 use futures::future::{self, Future};
-use hyper::Method;
 use hyper::header::HeaderValue;
 use hyper::http::header::HeaderName;
-use hyper::rt::Stream;
 use hyper::server::conn::AddrStream;
 use hyper::service::{service_fn, make_service_fn};
 use hyper::{Body, Request, Response, Server};
+use hyper::{Method, StatusCode};
 use rand::seq::IteratorRandom;
 use std::collections::HashMap;
+use std::convert::Infallible;
 use std::env;
 use std::fmt::Debug;
 use std::fmt::Display;
-use std::net::SocketAddr;
+use std::net::{SocketAddr, IpAddr};
 use std::str::FromStr;
 use std::sync::Arc;
 
-type BoxFut = Box<dyn Future<Item=Response<Body>, Error=hyper::Error> + Send>;
-
 const ENV_DEFAULT_ROUTE : &'static str = "DEFAULT_ROUTE";
 const ENV_NEWRELIC_API_KEY : &'static str = "NEWRELIC_API_KEY";
 const ENV_PROXY_CONFIG_FILE : &'static str = "PROXY_CONFIG_FILE";
@@ -88,17 +89,6 @@ fn get_env_num<T>(env_name: &str, default: T) -> AnyhowResult<T>
   }
 }
 
-fn _debug_request(req: Request<Body>) -> BoxFut {
-  let body_str = format!("{:?}", req);
-  let response = Response::new(Body::from(body_str));
-  Box::new(future::ok(response))
-}
-
-fn health_check_response(_req: Request<Body>) -> BoxFut {
-  let response = Response::new(Body::from("healthy"));
-  Box::new(future::ok(response))
-}
-
 pub struct Router {
   pub routes: HashMap<String, String>,
   pub default_route: String,
@@ -120,7 +110,66 @@ impl Router {
   }
 }
 
-fn main() -> AnyhowResult<()> {
+fn debug_request(req: Request<Body>) -> Result<Response<Body>, Infallible>  {
+  let body_str = format!("{:?}", req);
+  Ok(Response::new(Body::from(body_str)))
+}
+
+fn health_check_response(_req: Request<Body>) -> Result<Response<Body>, Infallible> {
+  Ok(Response::new(Body::from("healthy")))
+}
+
+pub async fn handle(
+  client_ip: IpAddr,
+  req: Request<Body>,
+  router: Arc<Router>,
+  newrelic_logger: Arc<NewRelicLogger>) -> Result<Response<Body>, Infallible>
+{
+  match req.uri().path() {
+    "/speak" => {
+      let nr_transaction = newrelic_logger.web_transaction("/speak");
+      let result = speak_proxy_with_retry(client_ip, req, router, nr_transaction, "/speak").await;
+      match result {
+        Ok(response) => Ok(response),
+        Err(error) => Ok(error.to_response()),
+      }
+    },
+    "/speak_spectrogram" => {
+      let nr_transaction = newrelic_logger.web_transaction("/speak_spectrogram");
+      let result = speak_proxy_with_retry(client_ip, req, router, nr_transaction, "/speak_spectrogram").await;
+      match result {
+        Ok(response) => Ok(response),
+        Err(error) => Ok(error.to_response()),
+      }
+    },
+    "/proxy_health" => {
+      let _droppable_transaction = newrelic_logger.web_transaction("/proxy_health");
+      health_check_response(req)
+    },
+    _ => {
+      let _droppable_transaction = newrelic_logger
+        .web_transaction("unmatched endpoint");
+      let forward = router.get_random_host();
+
+      info!("Forwarding to `{}` random host: {}", &forward, req.uri());
+
+      match hyper_reverse_proxy::call(client_ip, &forward, req).await {
+        Ok(response) => {
+          Ok(response)
+        },
+        Err(_) => {
+          Ok(Response::builder()
+            .status(StatusCode::INTERNAL_SERVER_ERROR)
+            .body(Body::from("unknown proxy error"))
+            .unwrap())
+        }
+      }
+    },
+  }
+}
+
+#[tokio::main]
+async fn main() {
   if env::var(ENV_RUST_LOG)
     .as_ref()
     .ok()
@@ -134,7 +183,8 @@ fn main() -> AnyhowResult<()> {
   env_logger::init();
 
   let default_route = get_env_string(ENV_DEFAULT_ROUTE, DEFAULT_DEFAULT_ROUTE);
-  let service_port = get_env_num::<u16>(ENV_SERVICE_PORT, DEFAULT_SERVICE_PORT)?;
+  let service_port = get_env_num::<u16>(ENV_SERVICE_PORT, DEFAULT_SERVICE_PORT)
+    .expect("should have env var");
   let newrelic_api_key = get_env_string(ENV_NEWRELIC_API_KEY, DEFAULT_NEWRELIC_API_KEY);
 
   info!("Default route: {}", default_route);
@@ -142,7 +192,8 @@ fn main() -> AnyhowResult<()> {
   let proxy_configs_file = get_env_string(ENV_PROXY_CONFIG_FILE, DEFAULT_PROXY_CONFIG_FILE);
   info!("Proxy config file: {}", proxy_configs_file);
 
-  let proxy_configs = ProxyConfigs::load_from_file(&proxy_configs_file)?;
+  let proxy_configs = ProxyConfigs::load_from_file(&proxy_configs_file)
+    .expect("should have configs");
   info!("Proxy configs: {:?}", proxy_configs);
 
   let mut route_map = HashMap::new();
@@ -165,49 +216,30 @@ fn main() -> AnyhowResult<()> {
 
   let newrelic_logger = Arc::new(newrelic_logger);
 
-  let make_service = make_service_fn(move |socket: &AddrStream| {
+  let make_service = make_service_fn(move |conn: &AddrStream| {
-    let remote_addr = socket.remote_addr();
+    let remote_addr = conn.remote_addr().ip();
-    info!("Got a request from: {:?}", remote_addr);
-
     let router2 = router.clone();
     let newrelic_logger2 = newrelic_logger.clone();
 
-    service_fn(move |req: Request<Body>| {
+    async move {
       let router3 = router2.clone();
       let newrelic_logger3 = newrelic_logger2.clone();
 
-      match (req.method(), req.uri().path()) {
+      Ok::<_, Infallible>(service_fn(move |req| {
-        (&Method::POST, "/speak") => {
+        let router4 = router3.clone();
-          let nr_transaction = newrelic_logger3.web_transaction("/speak");
+        let newrelic_logger4 = newrelic_logger3.clone();
-          speak_proxy_with_retry(req, remote_addr.clone(), router3, nr_transaction, "/speak")
+        handle(remote_addr, req, router4, newrelic_logger4)
-        },
+      }))
-        (&Method::POST, "/speak_spectrogram") => {
-          let nr_transaction = newrelic_logger3.web_transaction("/speak_spectrogram");
-          speak_proxy(req, remote_addr.clone(), router3, nr_transaction, "/speak_spectrogram")
-        },
-        (&Method::GET, "/proxy_health") => {
-          let _droppable_transaction = newrelic_logger3.web_transaction("/proxy_health");
-          health_check_response(req)
-        },
-        _ => {
-          let _droppable_transaction = newrelic_logger3.web_transaction("unmatched endpoint");
-          let forward = router3.get_random_host();
-          info!("Forwarding to `{}` random host: {}", &forward, req.uri());
-          hyper_reverse_proxy::call(remote_addr.ip(), &forward, req)
     }
-      }
-    })
   });
 
   let addr = ([0, 0, 0, 0], service_port).into();
 
   info!("Running server on {:?}", addr);
 
-  let server = Server::bind(&addr)
+  let server = Server::bind(&addr).serve(make_service);
-    .serve(make_service)
-    .map_err(|e| eprintln!("server error: {}", e));
 
-  hyper::rt::run(server);
+  if let Err(e) = server.await {
-
+    eprintln!("server error: {}", e);
-  Ok(())
+  }
 }

src/speak_proxy.rs

@@ -1,72 +1,45 @@
 use anyhow::Result as AnyhowResult;
-use crate::{BoxFut, Router};
+use crate::Router;
+use crate::api::{ExternalSpeakRequest, ExternalHttpRequest, InternalSpeakRequest, ErrorResponse, ErrorType};
 use crate::newrelic_logger::{NewRelicLogger, MaybeNewRelicTransaction};
+use futures::TryStreamExt;
 use futures::future::{self, Future};
-use hyper::Method;
 use hyper::header::HeaderValue;
 use hyper::http::header::HeaderName;
-use hyper::rt::Stream;
 use hyper::server::conn::AddrStream;
 use hyper::service::{service_fn, make_service_fn};
-use hyper::{Body, Request, Response, Server};
+use hyper::{Body, Request, Response, Server, HeaderMap};
+use hyper::{Method, StatusCode};
+use hyper_reverse_proxy::ProxyError;
 use rand::seq::IteratorRandom;
 use std::collections::HashMap;
-use std::env;
+use std::convert::Infallible;
 use std::fmt::Debug;
 use std::fmt::Display;
-use std::net::SocketAddr;
+use std::net::{SocketAddr, IpAddr};
 use std::str::FromStr;
 use std::sync::Arc;
-use futures::IntoFuture;
+use std::time::Duration;
+use std::{env, thread};
 
-/// NB: This much match the shape of SpeakRequest in the 'voder/tts_service' code.
+pub async fn speak_proxy_with_retry(
-/// This is used for both /speak and /speak_spectrogram requests.
+  ip_addr: IpAddr,
-#[derive(Serialize, Deserialize, Debug)]
+  request: Request<Body>,
-pub struct SpeakRequest {
-  /// Slug for the speaker
-  speaker: String,
-  /// Raw text to be spoken
-  text: String,
-}
-
-struct RequestDetails {
-  pub request_bytes: Vec<u8>,
-  pub speaker: String,
-}
-
-pub fn speak_proxy_with_retry(
-  req: Request<Body>,
-  remote_addr: SocketAddr,
   router: Arc<Router>,
-  transaction: MaybeNewRelicTransaction,
+  _transaction: MaybeNewRelicTransaction,
-  endpoint: &'static str) -> BoxFut
+  endpoint: &'static str) -> Result<Response<Body>, ErrorResponse>
 {
-  speak_proxy(req, remote_addr, router, transaction, endpoint)
+  let mut attempt : u8 = 0;
-    .then(|result| {
-      match result {
-        Ok(t) => future::ok(t),
-        Err(t) => future::err(t),
-      }
-    })
-    .boxed()
-}
 
-pub fn speak_proxy(
+  let external_http_request = ExternalHttpRequest::decode_request(request).await.unwrap();
-  req: Request<Body>,
-  remote_addr: SocketAddr,
-  router: Arc<Router>,
-  transaction: MaybeNewRelicTransaction,
-  endpoint: &'static str) -> BoxFut
-{
-  let mut headers = req.headers().clone();
 
   // This is the IP of the Digital Ocean load balancer.
-  let gateway_ip = HeaderValue::from_str(&remote_addr.ip().to_string())
+  let gateway_ip = HeaderValue::from_str(&ip_addr.to_string())
     .ok()
     .unwrap_or(HeaderValue::from_static(""));
 
   // This is the IP of the upstream client browser.
-  let upstream_client_ip = req.headers().get(HeaderName::from_static("x-forwarded-for"))
+  let upstream_client_ip = external_http_request.headers.get(HeaderName::from_static("x-forwarded-for"))
     .map(|ip| ip.to_str().unwrap_or(""))
     .unwrap_or("");
 
@@ -74,42 +47,89 @@ pub fn speak_proxy(
     .ok()
     .unwrap_or(HeaderValue::from_static(""));
 
-  headers.insert(HeaderName::from_static("forwarded"), upstream_client_ip.clone());
+  let mut headers_to_proxy = external_http_request.headers.clone();
-  headers.insert(HeaderName::from_static("x-forwarded-for"), upstream_client_ip.clone());
+
+  headers_to_proxy.insert(HeaderName::from_static("forwarded"), upstream_client_ip.clone());
+  headers_to_proxy.insert(HeaderName::from_static("x-forwarded-for"), upstream_client_ip.clone());
+
   // Unfortunately it looks like this middleware trounces the standard headers, so
   // here we put it somewhere it won't be overwritten.
-  headers.insert(HeaderName::from_static("x-voder-proxy-for"), upstream_client_ip.clone());
+  headers_to_proxy.insert(HeaderName::from_static("x-voder-proxy-for"), upstream_client_ip.clone());
-  headers.insert(HeaderName::from_static("x-gateway-ip"), gateway_ip);
+  headers_to_proxy.insert(HeaderName::from_static("x-gateway-ip"), gateway_ip);
 
-  Box::new(req.into_body().concat2().map(move |b| { // Builds a BoxedFut to return
+  // We're going to be changing the request body length, so remove the reported value
-    let request_bytes = b.as_ref();
+  // so Hyper doesn't choke on any mismatch.
+  headers_to_proxy.remove(HeaderName::from_static("content-length"));
 
-    let request = serde_json::from_slice::<SpeakRequest>(request_bytes)
+  // The proxy host is a Kubernetes load balancer. We're not doing the round robin magic here,
-      .unwrap();
+  // so it makes sense to reuse the same value for each retry.
+  let load_balancer_hostname = router.get_speaker_host(&external_http_request.speak_request.speaker);
 
-    RequestDetails {
+  loop {
-      request_bytes: request_bytes.to_vec(),
+    let internal_speak_request = InternalSpeakRequest {
-      speaker: request.speaker,
+      speaker: external_http_request.speak_request.speaker.clone(),
+      text: external_http_request.speak_request.text.clone(),
+      retry_attempt_number: attempt,
+      skip_rate_limiter: false,
+    };
+
+    let result = speak_proxy(
+      ip_addr,
+      internal_speak_request,
+      headers_to_proxy.clone(),
+      &load_balancer_hostname,
+      endpoint).await;
+
+    match result {
+      Ok(good_response) => return Ok(good_response),
+      Err(_) => {},
     }
-  }).and_then(move |request_details: RequestDetails| {
-    let proxy_host = router.get_speaker_host(&request_details.speaker);
 
-    info!("Routing {} for {} to {}", endpoint, &request_details.speaker, &proxy_host);
+    attempt += 1;
 
-    transaction.add_attribute("speaker", &request_details.speaker);
+    thread::sleep(Duration::from_millis(400)); // TODO: Needs backoff + jitter
 
-    let mut request_builder = Request::builder();
+    if attempt > 2 {
-    request_builder.method(Method::POST)
+      return Err(ErrorResponse {
+        error_type: ErrorType::ProxyError,
+        error_description: "max internal retry attempts reached".to_string(),
+      });
+    }
+  }
+}
+
+#[derive(Serialize, Debug)]
+pub struct ResponseError {
+  pub reason: String,
+}
+
+async fn speak_proxy(
+  ip_addr: IpAddr,
+  internal_speak_request: InternalSpeakRequest,
+  headers: HeaderMap<HeaderValue>,
+  load_balancer_hostname: &str,
+  endpoint: &'static str) -> Result<Response<Body>, ()>
+{
+  info!("Routing {} for {} to {} (attempt # {})", endpoint, &internal_speak_request.speaker,
+        &load_balancer_hostname, internal_speak_request.retry_attempt_number);
+
+  let mut request_builder = Request::builder()
+    .method(Method::POST)
     .uri(endpoint);
 
   for (k, v) in headers.iter() {
-      request_builder.header(k, v);
+    // NB: Handling move because of builder class
+    request_builder = request_builder.header(k, v);
   }
 
-    let new_req = request_builder
+  let json_speak_request = serde_json::to_string(&internal_speak_request).unwrap();
-      .body(Body::from(request_details.request_bytes))
+
+  let request = request_builder
+    .body(Body::from(json_speak_request))
     .unwrap();
 
-    hyper_reverse_proxy::call(remote_addr.ip(), &proxy_host, new_req)
+  match hyper_reverse_proxy::call(ip_addr, &load_balancer_hostname, request).await {
-  }))
+    Ok(result) => Ok(result),
+    Err(_result) => Err(()), // TODO: Surface this?
+  }
 }