diff --git a/.env-template b/.env-template new file mode 100644 index 0000000000..efc8eb9932 --- /dev/null +++ b/.env-template @@ -0,0 +1,11 @@ +# Redis connection +REDIS_USERNAME= +REDIS_PASSWORD= +REDIS_HOST= +REDIS_PORT= +REDIS_USES_TLS= + +# Redis pubsub +#SUBSCRIBE_TOPIC=firehose +#PUBLISH_TOPIC=unreal +#REDIS_MAX_RETRY_COUNT=3 diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml new file mode 100644 index 0000000000..d6c6f72906 --- /dev/null +++ b/.github/workflows/docker-publish.yml @@ -0,0 +1,65 @@ +# Adapted from +# https://github.com/actions/starter-workflows/ci/docker-publish.yml +# If using base images, don't forget to add READ access, eg. from a package settings page like: +# https://github.com/orgs/storytold/packages/container/docker-base-images-rust-ssl/settings/actions_access +name: Publish Docker Container to GHCR +on: + push: + branches: + # NB: Default-branch doesn't work, despite Github's documentation. + #- $default-branch + - master +env: + IMAGE_NAME: pubsub-enricher + +jobs: + # Push image to GitHub Packages. + # See also https://docs.docker.com/docker-hub/builds/ + push: + runs-on: ubuntu-latest + if: github.event_name == 'push' + permissions: + packages: write + contents: read + + steps: + - uses: actions/checkout@v2 + + - name: Login to GitHub Container Registry + uses: docker/login-action@v1 + with: + registry: ghcr.io + #username: ${{ github.repository_owner }} + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Build image + run: docker build . --file Dockerfile --tag $IMAGE_NAME --label "runnumber=${GITHUB_RUN_ID}" + + - name: Push image to GitHub Container Registry + run: | + IMAGE_ID=ghcr.io/${{ github.repository_owner }}/$IMAGE_NAME + + # Change all uppercase to lowercase + IMAGE_ID=$(echo $IMAGE_ID | tr '[A-Z]' '[a-z]') + + # Strip git ref prefix from version + VERSION=$(echo "${{ github.ref }}" | sed -e 's,.*/\(.*\),\1,') + + # Strip "v" prefix from tag name + [[ "${{ github.ref }}" == "refs/tags/"* ]] && VERSION=$(echo $VERSION | sed -e 's/^v//') + + # Use Docker `latest` tag convention + [ "$VERSION" == "$default-branch" ] && VERSION=latest + + echo IMAGE_ID=$IMAGE_ID + echo VERSION=$VERSION + + docker tag $IMAGE_NAME $IMAGE_ID:$VERSION + docker push $IMAGE_ID:$VERSION + + SHORT_SHA=$(echo ${GITHUB_SHA} | cut -c1-12) + + docker tag $IMAGE_NAME $IMAGE_ID:$SHORT_SHA + docker push $IMAGE_ID:$SHORT_SHA + diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000000..aee3b3b46c --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +*~ +.env +.idea/ +secrets.toml +secrets.yaml +target/ diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 0000000000..e2aaee7783 --- /dev/null +++ b/.gitmodules @@ -0,0 +1,3 @@ +[submodule "protos/storyteller-protos"] + path = protos/storyteller-protos + url = git@github.com:storytold/storyteller-protos.git diff --git a/Cargo.lock b/Cargo.lock new file mode 100644 index 0000000000..9abc97549e --- /dev/null +++ b/Cargo.lock @@ -0,0 +1,2051 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 3 + +[[package]] +name = "aho-corasick" +version = "0.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7404febffaa47dac81aa44dba71523c9d069b1bdc50a77db41195149e17f68e5" +dependencies = [ + "memchr", +] + +[[package]] +name = "anyhow" +version = "1.0.40" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28b2cd92db5cbd74e8e5028f7e27dd7aa3090e89e4f2a197cc7c8dfb69c7063b" + +[[package]] +name = "arc-swap" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dabe5a181f83789739c194cbe5a897dde195078fac08568d09221fd6137a7ba8" + +[[package]] +name = "arrayvec" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b62fc65de8e4e7f52534fb52b0f3ed04746ae267519eef2a83941e8085068b" + +[[package]] +name = "async-channel" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2114d64672151c0c5eaa5e131ec84a74f06e1e559830dabba01ca30605d66319" +dependencies = [ + "concurrent-queue", + "event-listener", + "futures-core", +] + +[[package]] +name = "async-executor" +version = "1.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "871f9bb5e0a22eeb7e8cf16641feb87c9dc67032ccf8ff49e772eb9941d3a965" +dependencies = [ + "async-task", + "concurrent-queue", + "fastrand", + "futures-lite", + "once_cell", + "slab", +] + +[[package]] +name = "async-global-executor" +version = "2.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9586ec52317f36de58453159d48351bc244bc24ced3effc1fce22f3d48664af6" +dependencies = [ + "async-channel", + "async-executor", + "async-io", + "async-mutex", + "blocking", + "futures-lite", + "num_cpus", + "once_cell", +] + +[[package]] +name = "async-io" +version = "1.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4bbfd5cf2794b1e908ea8457e6c45f8f8f1f6ec5f74617bf4662623f47503c3b" +dependencies = [ + "concurrent-queue", + "fastrand", + "futures-lite", + "libc", + "log", + "once_cell", + "parking", + "polling", + "slab", + "socket2", + "waker-fn", + "winapi 0.3.9", +] + +[[package]] +name = "async-lock" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6a8ea61bf9947a1007c5cada31e647dbc77b103c679858150003ba697ea798b" +dependencies = [ + "event-listener", +] + +[[package]] +name = "async-mutex" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479db852db25d9dbf6204e6cb6253698f175c15726470f78af0d918e99d6156e" +dependencies = [ + "event-listener", +] + +[[package]] +name = "async-native-tls" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e9e7a929bd34c68a82d58a4de7f86fffdaf97fb2af850162a7bb19dd7269b33" +dependencies = [ + "async-std", + "native-tls", + "thiserror", + "url", +] + +[[package]] +name = "async-std" +version = "1.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9f06685bad74e0570f5213741bea82158279a4103d988e57bfada11ad230341" +dependencies = [ + "async-channel", + "async-global-executor", + "async-io", + "async-lock", + "crossbeam-utils", + "futures-channel", + "futures-core", + "futures-io", + "futures-lite", + "gloo-timers", + "kv-log-macro", + "log", + "memchr", + "num_cpus", + "once_cell", + "pin-project-lite 0.2.6", + "pin-utils", + "slab", + "wasm-bindgen-futures", +] + +[[package]] +name = "async-task" +version = "4.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91831deabf0d6d7ec49552e489aed63b7456a7a3c46cff62adad428110b0af0" + +[[package]] +name = "async-trait" +version = "0.1.50" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b98e84bbb4cbcdd97da190ba0c58a1bb0de2c1fdf67d159e192ed766aeca722" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "atomic-waker" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "065374052e7df7ee4047b1160cca5e1467a12351a40b3da123c870ba0b8eda2a" + +[[package]] +name = "atty" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9b39be18770d11421cdb1b9947a45dd3f37e93092cbf377614828a319d5fee8" +dependencies = [ + "hermit-abi", + "libc", + "winapi 0.3.9", +] + +[[package]] +name = "autocfg" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdb031dd78e28731d87d56cc8ffef4a8f36ca26c38fe2de700543e627f8a464a" + +[[package]] +name = "base64" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3441f0f7b02788e948e47f457ca01f1d7e6d92c693bc132c22b087d3141c03ff" + +[[package]] +name = "base64" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "904dfeac50f3cdaba28fc6f57fdcddb75f49ed61346676a78c4ffe55877802fd" + +[[package]] +name = "bitflags" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693" + +[[package]] +name = "blocking" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c5e170dbede1f740736619b776d7251cb1b9095c435c34d8ca9f57fcd2f335e9" +dependencies = [ + "async-channel", + "async-task", + "atomic-waker", + "fastrand", + "futures-lite", + "once_cell", +] + +[[package]] +name = "bumpalo" +version = "3.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63396b8a4b9de3f4fdfb320ab6080762242f66a8ef174c49d8e19b674db4cdbe" + +[[package]] +name = "bytes" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e4cec68f03f32e44924783795810fa50a7035d8c8ebe78580ad7e6c703fba38" + +[[package]] +name = "bytes" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b700ce4376041dcd0a327fd0097c41095743c4c8af8887265942faf1100bd040" + +[[package]] +name = "cache-padded" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "631ae5198c9be5e753e5cc215e1bd73c2b466a3565173db433f52bb9d3e66dba" + +[[package]] +name = "cc" +version = "1.0.67" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3c69b077ad434294d3ce9f1f6143a2a4b89a8a2d54ef813d85003a4fd1137fd" + +[[package]] +name = "cfg-if" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4785bdd1c96b2a846b2bd7cc02e86b6b3dbf14e7e53446c4f54c92a361040822" + +[[package]] +name = "cfg-if" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" + +[[package]] +name = "chrono" +version = "0.4.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "670ad68c9088c2a963aaa298cb369688cf3f9465ce5e2d4ca10e6e0098a1ce73" +dependencies = [ + "libc", + "num-integer", + "num-traits", + "serde", + "time", + "winapi 0.3.9", +] + +[[package]] +name = "chrono-tz" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2554a3155fec064362507487171dcc4edc3df60cb10f3a1fb10ed8094822b120" +dependencies = [ + "chrono", + "parse-zoneinfo", + "serde", +] + +[[package]] +name = "combine" +version = "4.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc4369b5e4c0cddf64ad8981c0111e7df4f7078f4d6ba98fb31f2e17c4c57b7e" +dependencies = [ + "bytes 0.5.6", + "bytes 1.0.1", + "futures-util", + "memchr", + "pin-project-lite 0.2.6", + "tokio 0.2.25", +] + +[[package]] +name = "concurrent-queue" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30ed07550be01594c6026cff2a1d7fe9c8f683caa798e12b68694ac9e88286a3" +dependencies = [ + "cache-padded", +] + +[[package]] +name = "core-foundation" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0a89e2ae426ea83155dccf10c0fa6b1463ef6d5fcb44cee0b224a408fa640a62" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea221b5284a47e40033bf9b66f35f984ec0ea2931eb03505246cd27a963f981b" + +[[package]] +name = "crossbeam-utils" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7e9d99fa91428effe99c5c6d4634cdeba32b8cf784fc428a2a687f61a952c49" +dependencies = [ + "autocfg", + "cfg-if 1.0.0", + "lazy_static", +] + +[[package]] +name = "ctor" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e98e2ad1a782e33928b96fc3948e7c355e5af34ba4de7670fe8bac2a3b2006d" +dependencies = [ + "quote", + "syn", +] + +[[package]] +name = "dotenv" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77c90badedccf4105eca100756a0b1289e191f6fcbdadd3cee1d2f614f97da8f" + +[[package]] +name = "dtoa" +version = "0.4.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56899898ce76aaf4a0f24d914c97ea6ed976d42fec6ad33fcbb0a1103e07b2b0" + +[[package]] +name = "easyenv" +version = "0.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "852054bee794f364fbe890bd947cbf678c35226b81ede17720d6a17fc687a0d4" +dependencies = [ + "dotenv", + "env_logger", + "log", +] + +[[package]] +name = "either" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e78d4f1cc4ae33bbfc157ed5d5a5ef3bc29227303d595861deb238fcec4e9457" + +[[package]] +name = "encoding_rs" +version = "0.8.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "80df024fbc5ac80f87dfef0d9f5209a252f2a497f7f42944cff24d8253cac065" +dependencies = [ + "cfg-if 1.0.0", +] + +[[package]] +name = "env_logger" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44533bbbb3bb3c1fa17d9f2e4e38bbbaf8396ba82193c4cb1b6445d711445d36" +dependencies = [ + "atty", + "humantime", + "log", + "regex", + "termcolor", +] + +[[package]] +name = "event-listener" +version = "2.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7531096570974c3a9dcf9e4b8e1cede1ec26cf5046219fb3b9d897503b9be59" + +[[package]] +name = "expectest" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78817c0b64aab8153fbd7532339fbac1a54e43c715802715a288163ca6024978" +dependencies = [ + "num-traits", +] + +[[package]] +name = "fastrand" +version = "1.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77b705829d1e87f762c2df6da140b26af5839e1033aa84aa5f56bb688e4e1bdb" +dependencies = [ + "instant", +] + +[[package]] +name = "fixedbitset" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37ab347416e802de484e4d03c7316c48f1ecb56574dfd4a46a80f173ce1de04d" + +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + +[[package]] +name = "foreign-types" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f6f339eb8adc052cd2ca78910fda869aefa38d22d5cb648e6485e4d3fc06f3b1" +dependencies = [ + "foreign-types-shared", +] + +[[package]] +name = "foreign-types-shared" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00b0228411908ca8685dba7fc2cdd70ec9990a6e753e89b6ac91a84c40fbaf4b" + +[[package]] +name = "form_urlencoded" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5fc25a87fa4fd2094bffb06925852034d90a17f0d1e05197d4956d3555752191" +dependencies = [ + "matches", + "percent-encoding", +] + +[[package]] +name = "fuchsia-zircon" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2e9763c69ebaae630ba35f74888db465e49e259ba1bc0eda7d06f4a067615d82" +dependencies = [ + "bitflags", + "fuchsia-zircon-sys", +] + +[[package]] +name = "fuchsia-zircon-sys" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3dcaa9ae7725d12cdb85b3ad99a434db70b468c09ded17e012d86b5c1010f7a7" + +[[package]] +name = "futures" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9d5813545e459ad3ca1bff9915e9ad7f1a47dc6a91b627ce321d5863b7dd253" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce79c6a52a299137a6013061e0cf0e688fce5d7f1bc60125f520912fdb29ec25" +dependencies = [ + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "098cd1c6dda6ca01650f1a37a794245eb73181d0d4d4e955e2f3c37db7af1815" + +[[package]] +name = "futures-executor" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10f6cb7042eda00f0049b1d2080aa4b93442997ee507eb3828e8bd7577f94c9d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "365a1a1fb30ea1c03a830fdb2158f5236833ac81fa0ad12fe35b29cddc35cb04" + +[[package]] +name = "futures-lite" +version = "1.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4481d0cd0de1d204a4fa55e7d45f07b1d958abcb06714b3446438e2eff695fb" +dependencies = [ + "fastrand", + "futures-core", + "futures-io", + "memchr", + "parking", + "pin-project-lite 0.2.6", + "waker-fn", +] + +[[package]] +name = "futures-macro" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "668c6733a182cd7deb4f1de7ba3bf2120823835b3bcfbeacf7d2c4a773c1bb8b" +dependencies = [ + "proc-macro-hack", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "futures-sink" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c5629433c555de3d82861a7a4e3794a4c40040390907cfbfd7143a92a426c23" + +[[package]] +name = "futures-task" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba7aa51095076f3ba6d9a1f702f74bd05ec65f555d70d2033d55ba8d69f581bc" + +[[package]] +name = "futures-util" +version = "0.3.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3c144ad54d60f23927f0a6b6d816e4271278b64f005ad65e4e35291d2de9c025" +dependencies = [ + "futures-channel", + "futures-core", + "futures-io", + "futures-macro", + "futures-sink", + "futures-task", + "memchr", + "pin-project-lite 0.2.6", + "pin-utils", + "proc-macro-hack", + "proc-macro-nested", + "slab", +] + +[[package]] +name = "getrandom" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9495705279e7140bf035dde1f6e750c162df8b625267cd52cc44e0b156732c8" +dependencies = [ + "cfg-if 1.0.0", + "libc", + "wasi", +] + +[[package]] +name = "gloo-timers" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47204a46aaff920a1ea58b11d03dec6f704287d27561724a4631e450654a891f" +dependencies = [ + "futures-channel", + "futures-core", + "js-sys", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "google_maps" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb31e07d9172813158c681923ac044846a5fad9d1fbba035f87851af45b4235a" +dependencies = [ + "chrono", + "chrono-tz", + "log", + "percent-encoding", + "rand", + "reqwest", + "rust_decimal", + "rust_decimal_macros", + "serde", + "serde_json", +] + +[[package]] +name = "h2" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc018e188373e2777d0ef2467ebff62a08e66c3f5857b23c8fbec3018210dc00" +dependencies = [ + "bytes 1.0.1", + "fnv", + "futures-core", + "futures-sink", + "futures-util", + "http", + "indexmap", + "slab", + "tokio 1.5.0", + "tokio-util 0.6.6", + "tracing", +] + +[[package]] +name = "hashbrown" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7afe4a420e3fe79967a00898cc1f4db7c8a49a9333a29f8a4bd76a253d5cd04" + +[[package]] +name = "heck" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87cbf45460356b7deeb5e3415b5563308c0a9b057c85e12b06ad551f98d0a6ac" +dependencies = [ + "unicode-segmentation", +] + +[[package]] +name = "hermit-abi" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "322f4de77956e22ed0e5032c359a0f1273f1f7f0d79bfa3b8ffbc730d7fbcc5c" +dependencies = [ + "libc", +] + +[[package]] +name = "http" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "527e8c9ac747e28542699a951517aa9a6945af506cd1f2e1b53a576c17b6cc11" +dependencies = [ + "bytes 1.0.1", + "fnv", + "itoa", +] + +[[package]] +name = "http-body" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5dfb77c123b4e2f72a2069aeae0b4b4949cc7e966df277813fc16347e7549737" +dependencies = [ + "bytes 1.0.1", + "http", + "pin-project-lite 0.2.6", +] + +[[package]] +name = "httparse" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a1ce40d6fc9764887c2fdc7305c3dcc429ba11ff981c1509416afd5697e4437" + +[[package]] +name = "httpdate" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05842d0d43232b23ccb7060ecb0f0626922c21f30012e97b767b30afd4a5d4b9" + +[[package]] +name = "humantime" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df004cfca50ef23c36850aaaa59ad52cc70d0e90243c3c7737a4dd32dc7a3c4f" +dependencies = [ + "quick-error", +] + +[[package]] +name = "hyper" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e5f105c494081baa3bf9e200b279e27ec1623895cd504c7dbef8d0b080fcf54" +dependencies = [ + "bytes 1.0.1", + "futures-channel", + "futures-core", + "futures-util", + "h2", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project", + "socket2", + "tokio 1.5.0", + "tower-service", + "tracing", + "want", +] + +[[package]] +name = "hyper-tls" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6183ddfa99b85da61a140bea0efc93fdf56ceaa041b37d553518030827f9905" +dependencies = [ + "bytes 1.0.1", + "hyper", + "native-tls", + "tokio 1.5.0", + "tokio-native-tls", +] + +[[package]] +name = "idna" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "418a0a6fab821475f634efe3ccc45c013f742efe03d853e8d3355d5cb850ecf8" +dependencies = [ + "matches", + "unicode-bidi", + "unicode-normalization", +] + +[[package]] +name = "indexmap" +version = "1.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824845a0bf897a9042383849b02c1bc219c2383772efcd5c6f9766fa4b81aef3" +dependencies = [ + "autocfg", + "hashbrown", +] + +[[package]] +name = "instant" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61124eeebbd69b8190558df225adf7e4caafce0d743919e5d6b19652314ec5ec" +dependencies = [ + "cfg-if 1.0.0", +] + +[[package]] +name = "iovec" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2b3ea6ff95e175473f8ffe6a7eb7c00d054240321b84c57051175fe3c1e075e" +dependencies = [ + "libc", +] + +[[package]] +name = "ipnet" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47be2f14c678be2fdcab04ab1171db51b2762ce6f0a8ee87c8dd4a04ed216135" + +[[package]] +name = "itertools" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "284f18f85651fe11e8a991b2adb42cb078325c996ed026d994719efcfca1d54b" +dependencies = [ + "either", +] + +[[package]] +name = "itoa" +version = "0.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd25036021b0de88a0aff6b850051563c6516d0bf53f8638938edbb9de732736" + +[[package]] +name = "js-sys" +version = "0.3.50" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d99f9e3e84b8f67f846ef5b4cbbc3b1c29f6c759fcbce6f01aa0e73d932a24c" +dependencies = [ + "wasm-bindgen", +] + +[[package]] +name = "kernel32-sys" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7507624b29483431c0ba2d82aece8ca6cdba9382bff4ddd0f7490560c056098d" +dependencies = [ + "winapi 0.2.8", + "winapi-build", +] + +[[package]] +name = "kv-log-macro" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0de8b303297635ad57c9f5059fd9cee7a47f8e8daa09df0fcd07dd39fb22977f" +dependencies = [ + "log", +] + +[[package]] +name = "lazy_static" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2abad23fbc42b3700f2f279844dc832adb2b2eb069b2df918f455c4e18cc646" + +[[package]] +name = "libc" +version = "0.2.94" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18794a8ad5b29321f790b55d93dfba91e125cb1a9edbd4f8e3150acc771c1a5e" + +[[package]] +name = "log" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51b9bbe6c47d51fc3e1a9b945965946b4c44142ab8792c50835a980d362c2710" +dependencies = [ + "cfg-if 1.0.0", + "value-bag", +] + +[[package]] +name = "matches" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ffc5c5338469d4d3ea17d269fa8ea3512ad247247c30bd2df69e68309ed0a08" + +[[package]] +name = "memchr" +version = "2.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ee1c47aaa256ecabcaea351eae4a9b01ef39ed810004e298d2511ed284b1525" + +[[package]] +name = "mime" +version = "0.3.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a60c7ce501c71e03a9c9c0d35b861413ae925bd979cc7a4e30d060069aaac8d" + +[[package]] +name = "mio" +version = "0.6.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4afd66f5b91bf2a3bc13fad0e21caedac168ca4c707504e75585648ae80e4cc4" +dependencies = [ + "cfg-if 0.1.10", + "fuchsia-zircon", + "fuchsia-zircon-sys", + "iovec", + "kernel32-sys", + "libc", + "log", + "miow 0.2.2", + "net2", + "slab", + "winapi 0.2.8", +] + +[[package]] +name = "mio" +version = "0.7.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf80d3e903b34e0bd7282b218398aec54e082c840d9baf8339e0080a0c542956" +dependencies = [ + "libc", + "log", + "miow 0.3.7", + "ntapi", + "winapi 0.3.9", +] + +[[package]] +name = "mio-named-pipes" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0840c1c50fd55e521b247f949c241c9997709f23bd7f023b9762cd561e935656" +dependencies = [ + "log", + "mio 0.6.23", + "miow 0.3.7", + "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 0.6.23", +] + +[[package]] +name = "miow" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebd808424166322d4a38da87083bfddd3ac4c131334ed55856112eb06d46944d" +dependencies = [ + "kernel32-sys", + "net2", + "winapi 0.2.8", + "ws2_32-sys", +] + +[[package]] +name = "miow" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9f1c5b025cda876f66ef43a113f91ebc9f4ccef34843000e0adf6ebbab84e21" +dependencies = [ + "winapi 0.3.9", +] + +[[package]] +name = "multimap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a" + +[[package]] +name = "native-tls" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8d96b2e1c8da3957d58100b09f102c6d9cfdfced01b7ec5a8974044bb09dbd4" +dependencies = [ + "lazy_static", + "libc", + "log", + "openssl", + "openssl-probe", + "openssl-sys", + "schannel", + "security-framework", + "security-framework-sys", + "tempfile", +] + +[[package]] +name = "net2" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "391630d12b68002ae1e25e8f974306474966550ad82dac6886fb8910c19568ae" +dependencies = [ + "cfg-if 0.1.10", + "libc", + "winapi 0.3.9", +] + +[[package]] +name = "ntapi" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f6bb902e437b6d86e03cce10a7e2af662292c5dfef23b65899ea3ac9354ad44" +dependencies = [ + "winapi 0.3.9", +] + +[[package]] +name = "num-integer" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2cc698a63b549a70bc047073d2949cce27cd1c7b0a4a862d08a8031bc2801db" +dependencies = [ + "autocfg", + "num-traits", +] + +[[package]] +name = "num-traits" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a64b1ec5cda2586e284722486d802acf1f7dbdc623e2bfc57e65ca1cd099290" +dependencies = [ + "autocfg", +] + +[[package]] +name = "num_cpus" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05499f3756671c15885fee9034446956fff3f243d6077b91e5767df161f766b3" +dependencies = [ + "hermit-abi", + "libc", +] + +[[package]] +name = "once_cell" +version = "1.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af8b08b04175473088b46763e51ee54da5f9a164bc162f615b91bc179dbf15a3" + +[[package]] +name = "openssl" +version = "0.10.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a61075b62a23fef5a29815de7536d940aa35ce96d18ce0cc5076272db678a577" +dependencies = [ + "bitflags", + "cfg-if 1.0.0", + "foreign-types", + "libc", + "once_cell", + "openssl-sys", +] + +[[package]] +name = "openssl-probe" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77af24da69f9d9341038eba93a073b1fdaaa1b788221b00a69bce9e762cb32de" + +[[package]] +name = "openssl-sys" +version = "0.9.61" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "313752393519e876837e09e1fa183ddef0be7735868dced3196f4472d536277f" +dependencies = [ + "autocfg", + "cc", + "libc", + "pkg-config", + "vcpkg", +] + +[[package]] +name = "parking" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "427c3892f9e783d91cc128285287e70a59e206ca452770ece88a76f7a3eddd72" + +[[package]] +name = "parse-zoneinfo" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c705f256449c60da65e11ff6626e0c16a0a0b96aaa348de61376b249bc340f41" +dependencies = [ + "regex", +] + +[[package]] +name = "percent-encoding" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d4fd5641d01c8f18a23da7b6fe29298ff4b55afcccdf78973b24cf3175fee32e" + +[[package]] +name = "petgraph" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "467d164a6de56270bd7c4d070df81d07beace25012d5103ced4e9ff08d6afdb7" +dependencies = [ + "fixedbitset", + "indexmap", +] + +[[package]] +name = "pin-project" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7509cc106041c40a4518d2af7a61530e1eed0e6285296a3d8c5472806ccc4a4" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48c950132583b500556b1efd71d45b319029f2b71518d979fcc208e16b42426f" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "pin-project-lite" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "257b64915a082f7811703966789728173279bdebb956b143dbcd23f6f970a777" + +[[package]] +name = "pin-project-lite" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc0e1f259c92177c30a4c9d177246edd0a3568b25756a977d0632cf8fa37e905" + +[[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.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3831453b3449ceb48b6d9c7ad7c96d5ea673e9b470a1dc578c2ce6521230884c" + +[[package]] +name = "polling" +version = "2.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fc12d774e799ee9ebae13f4076ca003b40d18a11ac0f3641e6f899618580b7b" +dependencies = [ + "cfg-if 1.0.0", + "libc", + "log", + "wepoll-sys", + "winapi 0.3.9", +] + +[[package]] +name = "ppv-lite86" +version = "0.2.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac74c624d6b2d21f425f752262f42188365d7b8ff1aff74c82e45136510a4857" + +[[package]] +name = "proc-macro-hack" +version = "0.5.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dbf0c48bc1d91375ae5c3cd81e3722dff1abcf81a30960240640d223f59fe0e5" + +[[package]] +name = "proc-macro-nested" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc881b2c22681370c6a780e47af9840ef841837bc98118431d4e1868bd0c1086" + +[[package]] +name = "proc-macro2" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a152013215dca273577e18d2bf00fa862b89b24169fb78c4c95aeb07992c9cec" +dependencies = [ + "unicode-xid", +] + +[[package]] +name = "prost" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e6984d2f1a23009bd270b8bb56d0926810a3d483f59c987d77969e9d8e840b2" +dependencies = [ + "bytes 1.0.1", + "prost-derive", +] + +[[package]] +name = "prost-build" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32d3ebd75ac2679c2af3a92246639f9fcc8a442ee420719cc4fe195b98dd5fa3" +dependencies = [ + "bytes 1.0.1", + "heck", + "itertools", + "log", + "multimap", + "petgraph", + "prost", + "prost-types", + "tempfile", + "which", +] + +[[package]] +name = "prost-derive" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "169a15f3008ecb5160cba7d37bcd690a7601b6d30cfb87a117d45e59d52af5d4" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "prost-types" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b518d7cdd93dab1d1122cf07fa9a60771836c668dde9d9e2a139f957f0d9f1bb" +dependencies = [ + "bytes 1.0.1", + "prost", +] + +[[package]] +name = "pubsub-enricher" +version = "0.0.0" +dependencies = [ + "anyhow", + "easyenv", + "expectest", + "futures", + "google_maps", + "lazy_static", + "log", + "prost", + "prost-build", + "redis", + "regex", + "rustls", + "serde", + "serde_derive", + "tokio 0.2.25", + "toml", +] + +[[package]] +name = "quick-error" +version = "1.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1d01941d82fa2ab50be1e79e6714289dd7cde78eba4c074bc5a4374f650dfe0" + +[[package]] +name = "quote" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d0b9745dc2debf507c8422de05d7226cc1f0644216dfdfead988f9b1ab32a7" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "rand" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ef9e7e66b4468674bfcb0c81af8b7fa0bb154fa9f28eb840da5c447baeb8d7e" +dependencies = [ + "libc", + "rand_chacha", + "rand_core", + "rand_hc", +] + +[[package]] +name = "rand_chacha" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e12735cf05c9e10bf21534da50a147b924d555dc7a547c42e6bb2d5b6017ae0d" +dependencies = [ + "ppv-lite86", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34cf66eb183df1c5876e2dcf6b13d57340741e8dc255b48e40a26de954d06ae7" +dependencies = [ + "getrandom", +] + +[[package]] +name = "rand_hc" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3190ef7066a446f2e7f42e239d161e905420ccab01eb967c9eb27d21b2322a73" +dependencies = [ + "rand_core", +] + +[[package]] +name = "redis" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "95357caf2640abc54651b93c98a8df4fe1ccbf44b8e601ccdf43d5c1451f29ac" +dependencies = [ + "arc-swap", + "async-native-tls", + "async-std", + "async-trait", + "bytes 0.5.6", + "combine", + "dtoa", + "futures", + "futures-util", + "itoa", + "native-tls", + "percent-encoding", + "pin-project-lite 0.1.12", + "sha1", + "tokio 0.2.25", + "tokio-tls", + "tokio-util 0.3.1", + "url", +] + +[[package]] +name = "redox_syscall" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8270314b5ccceb518e7e578952f0b72b88222d02e8f77f5ecf7abbb673539041" +dependencies = [ + "bitflags", +] + +[[package]] +name = "regex" +version = "1.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a26af418b574bd56588335b3a3659a65725d4e636eb1016c2f9e3b38c7cc759" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.6.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5f089152e60f62d28b835fbff2cd2e8dc0baf1ac13343bef92ab7eed84548" + +[[package]] +name = "remove_dir_all" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3acd125665422973a33ac9d3dd2df85edad0f4ae9b00dafb1a05e43a9f5ef8e7" +dependencies = [ + "winapi 0.3.9", +] + +[[package]] +name = "reqwest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2296f2fac53979e8ccbc4a1136b25dcefd37be9ed7e4a1f6b05a6029c84ff124" +dependencies = [ + "base64 0.13.0", + "bytes 1.0.1", + "encoding_rs", + "futures-core", + "futures-util", + "http", + "http-body", + "hyper", + "hyper-tls", + "ipnet", + "js-sys", + "lazy_static", + "log", + "mime", + "native-tls", + "percent-encoding", + "pin-project-lite 0.2.6", + "serde", + "serde_urlencoded", + "tokio 1.5.0", + "tokio-native-tls", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "winreg", +] + +[[package]] +name = "ring" +version = "0.16.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3053cf52e236a3ed746dfc745aa9cacf1b791d846bdaf412f60a8d7d6e17c8fc" +dependencies = [ + "cc", + "libc", + "once_cell", + "spin", + "untrusted", + "web-sys", + "winapi 0.3.9", +] + +[[package]] +name = "rust_decimal" +version = "1.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d64bd27418c091ebd539b00f50ad059fc41421c87d5033c217cd76c7506d0d71" +dependencies = [ + "arrayvec", + "num-traits", + "serde", +] + +[[package]] +name = "rust_decimal_macros" +version = "1.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81dbf822f50d527a9ba351f392aadca0ac47375c3bfad5d48ecfad53cc067833" +dependencies = [ + "quote", + "rust_decimal", +] + +[[package]] +name = "rustls" +version = "0.18.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d1126dcf58e93cee7d098dbda643b5f92ed724f1f6a63007c1116eed6700c81" +dependencies = [ + "base64 0.12.3", + "log", + "ring", + "sct", + "webpki", +] + +[[package]] +name = "ryu" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71d301d4193d031abdd79ff7e3dd721168a9572ef3fe51a1517aba235bd8f86e" + +[[package]] +name = "schannel" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f05ba609c234e60bee0d547fe94a4c7e9da733d1c962cf6e59efa4cd9c8bc75" +dependencies = [ + "lazy_static", + "winapi 0.3.9", +] + +[[package]] +name = "sct" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b362b83898e0e69f38515b82ee15aa80636befe47c3b6d3d89a911e78fc228ce" +dependencies = [ + "ring", + "untrusted", +] + +[[package]] +name = "security-framework" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3670b1d2fdf6084d192bc71ead7aabe6c06aa2ea3fbd9cc3ac111fa5c2b1bd84" +dependencies = [ + "bitflags", + "core-foundation", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework-sys" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3676258fd3cfe2c9a0ec99ce3038798d847ce3e4bb17746373eb9f0f1ac16339" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "serde" +version = "1.0.125" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "558dc50e1a5a5fa7112ca2ce4effcb321b0300c0d4ccf0776a9f60cd89031171" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.125" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b093b7a2bb58203b5da3056c05b4ec1fed827dcfdb37347a8841695263b3d06d" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.64" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "799e97dc9fdae36a5c8b8f2cae9ce2ee9fdce2058c57a93e6099d919fd982f79" +dependencies = [ + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "serde_urlencoded" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edfa57a7f8d9c1d260a549e7224100f6c43d43f9103e06dd8b4095a9b2b43ce9" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "sha1" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2579985fda508104f7587689507983eadd6a6e84dd35d6d115361f530916fa0d" + +[[package]] +name = "signal-hook-registry" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16f1d0fef1604ba8f7a073c7e701f213e056707210e9020af4528e0101ce11a6" +dependencies = [ + "libc", +] + +[[package]] +name = "slab" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f173ac3d1a7e3b28003f40de0b5ce7fe2710f9b9dc3fc38664cebee46b3b6527" + +[[package]] +name = "socket2" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e3dfc207c526015c632472a77be09cf1b6e46866581aecae5cc38fb4235dea2" +dependencies = [ + "libc", + "winapi 0.3.9", +] + +[[package]] +name = "spin" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e63cff320ae2c57904679ba7cb63280a3dc4613885beafb148ee7bf9aa9042d" + +[[package]] +name = "syn" +version = "1.0.70" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9505f307c872bab8eb46f77ae357c8eba1fdacead58ee5a850116b1d7f82883" +dependencies = [ + "proc-macro2", + "quote", + "unicode-xid", +] + +[[package]] +name = "tempfile" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dac1c663cfc93810f88aed9b8941d48cabf856a1b111c29a40439018d870eb22" +dependencies = [ + "cfg-if 1.0.0", + "libc", + "rand", + "redox_syscall", + "remove_dir_all", + "winapi 0.3.9", +] + +[[package]] +name = "termcolor" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2dfed899f0eb03f32ee8c6a0aabdb8a7949659e3466561fc0adf54e26d88c5f4" +dependencies = [ + "winapi-util", +] + +[[package]] +name = "thiserror" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0f4a65597094d4483ddaed134f409b2cb7c1beccf25201a9f73c719254fa98e" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7765189610d8241a44529806d6fd1f2e0a08734313a35d5b3a556f92b381f3c0" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "time" +version = "0.1.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca8a50ef2360fbd1eeb0ecd46795a87a19024eb4b53c5dc916ca1fd95fe62438" +dependencies = [ + "libc", + "winapi 0.3.9", +] + +[[package]] +name = "tinyvec" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b5220f05bb7de7f3f53c7c065e1199b3172696fe2db9f9c4d8ad9b4ee74c342" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cda74da7e1a664f795bb1f8a87ec406fb89a02522cf6e50620d016add6dbbf5c" + +[[package]] +name = "tokio" +version = "0.2.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6703a273949a90131b290be1fe7b039d0fc884aa1935860dfcbe056f28cd8092" +dependencies = [ + "bytes 0.5.6", + "fnv", + "futures-core", + "iovec", + "lazy_static", + "libc", + "memchr", + "mio 0.6.23", + "mio-named-pipes", + "mio-uds", + "num_cpus", + "pin-project-lite 0.1.12", + "signal-hook-registry", + "slab", + "tokio-macros", + "winapi 0.3.9", +] + +[[package]] +name = "tokio" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83f0c8e7c0addab50b663055baf787d0af7f413a46e6e7fb9559a4e4db7137a5" +dependencies = [ + "autocfg", + "bytes 1.0.1", + "libc", + "memchr", + "mio 0.7.11", + "num_cpus", + "pin-project-lite 0.2.6", +] + +[[package]] +name = "tokio-macros" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e44da00bfc73a25f814cd8d7e57a68a5c31b74b3152a0a1d1f590c97ed06265a" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio-native-tls" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7d995660bd2b7f8c1568414c1126076c13fbb725c40112dc0120b78eb9b717b" +dependencies = [ + "native-tls", + "tokio 1.5.0", +] + +[[package]] +name = "tokio-tls" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a70f4fcd7b3b24fb194f837560168208f669ca8cb70d0c4b862944452396343" +dependencies = [ + "native-tls", + "tokio 0.2.25", +] + +[[package]] +name = "tokio-util" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be8242891f2b6cbef26a2d7e8605133c2c554cd35b3e4948ea892d6d68436499" +dependencies = [ + "bytes 0.5.6", + "futures-core", + "futures-sink", + "log", + "pin-project-lite 0.1.12", + "tokio 0.2.25", +] + +[[package]] +name = "tokio-util" +version = "0.6.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940a12c99365c31ea8dd9ba04ec1be183ffe4920102bb7122c2f515437601e8e" +dependencies = [ + "bytes 1.0.1", + "futures-core", + "futures-sink", + "log", + "pin-project-lite 0.2.6", + "tokio 1.5.0", +] + +[[package]] +name = "toml" +version = "0.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a31142970826733df8241ef35dc040ef98c679ab14d7c3e54d827099b3acecaa" +dependencies = [ + "serde", +] + +[[package]] +name = "tower-service" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "360dfd1d6d30e05fda32ace2c8c70e9c0a9da713275777f5a4dbb8a1893930c6" + +[[package]] +name = "tracing" +version = "0.1.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "01ebdc2bb4498ab1ab5f5b73c5803825e60199229ccba0698170e3be0e7f959f" +dependencies = [ + "cfg-if 1.0.0", + "pin-project-lite 0.2.6", + "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 = "unicode-bidi" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eeb8be209bb1c96b7c177c7420d26e04eccacb0eeae6b980e35fcb74678107e0" +dependencies = [ + "matches", +] + +[[package]] +name = "unicode-normalization" +version = "0.1.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07fbfce1c8a97d547e8b5334978438d9d6ec8c20e38f56d4a4374d181493eaef" +dependencies = [ + "tinyvec", +] + +[[package]] +name = "unicode-segmentation" +version = "1.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb0d2e7be6ae3a5fa87eed5fb451aff96f2573d2694942e40543ae0bbe19c796" + +[[package]] +name = "unicode-xid" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7fe0bb3479651439c9112f72b6c505038574c9fbb575ed1bf3b797fa39dd564" + +[[package]] +name = "untrusted" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a" + +[[package]] +name = "url" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ccd964113622c8e9322cfac19eb1004a07e636c545f325da085d5cdde6f1f8b" +dependencies = [ + "form_urlencoded", + "idna", + "matches", + "percent-encoding", +] + +[[package]] +name = "value-bag" +version = "1.0.0-alpha.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b676010e055c99033117c2343b33a40a30b91fecd6c49055ac9cd2d6c305ab1" +dependencies = [ + "ctor", +] + +[[package]] +name = "vcpkg" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cbdbff6266a24120518560b5dc983096efb98462e51d0d68169895b237be3e5d" + +[[package]] +name = "waker-fn" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d5b2c62b4012a3e1eca5a7e077d13b3bf498c4073e33ccd58626607748ceeca" + +[[package]] +name = "want" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ce8a968cb1cd110d136ff8b819a556d6fb6d919363c61534f6860c7eb172ba0" +dependencies = [ + "log", + "try-lock", +] + +[[package]] +name = "wasi" +version = "0.10.2+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fd6fbd9a79829dd1ad0cc20627bf1ed606756a7f77edff7b66b7064f9cb327c6" + +[[package]] +name = "wasm-bindgen" +version = "0.2.73" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83240549659d187488f91f33c0f8547cbfef0b2088bc470c116d1d260ef623d9" +dependencies = [ + "cfg-if 1.0.0", + "serde", + "serde_json", + "wasm-bindgen-macro", +] + +[[package]] +name = "wasm-bindgen-backend" +version = "0.2.73" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae70622411ca953215ca6d06d3ebeb1e915f0f6613e3b495122878d7ebec7dae" +dependencies = [ + "bumpalo", + "lazy_static", + "log", + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81b8b767af23de6ac18bf2168b690bed2902743ddf0fb39252e36f9e2bfc63ea" +dependencies = [ + "cfg-if 1.0.0", + "js-sys", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.73" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e734d91443f177bfdb41969de821e15c516931c3c3db3d318fa1b68975d0f6f" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.73" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d53739ff08c8a68b0fdbcd54c372b8ab800b1449ab3c9d706503bc7dd1621b2c" +dependencies = [ + "proc-macro2", + "quote", + "syn", + "wasm-bindgen-backend", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.73" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9a543ae66aa233d14bb765ed9af4a33e81b8b58d1584cf1b47ff8cd0b9e4489" + +[[package]] +name = "web-sys" +version = "0.3.50" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a905d57e488fec8861446d3393670fb50d27a262344013181c2cdf9fff5481be" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "webpki" +version = "0.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e38c0608262c46d4a56202ebabdeb094cef7e560ca7a226c6bf055188aa4ea" +dependencies = [ + "ring", + "untrusted", +] + +[[package]] +name = "wepoll-sys" +version = "3.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fcb14dea929042224824779fbc82d9fab8d2e6d3cbc0ac404de8edf489e77ff" +dependencies = [ + "cc", +] + +[[package]] +name = "which" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b55551e42cbdf2ce2bedd2203d0cc08dba002c27510f86dab6d0ce304cba3dfe" +dependencies = [ + "either", + "libc", +] + +[[package]] +name = "winapi" +version = "0.2.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "167dc9d6949a9b857f3451275e911c3f44255842c1f7a76f33c55103a909087a" + +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-build" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2d315eee3b34aca4797b2da6b13ed88266e6d612562a0c46390af8299fc699bc" + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70ec6ce85bb158151cae5e5c87f95a8e97d2c0c4b001223f33a334e3ce5de178" +dependencies = [ + "winapi 0.3.9", +] + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + +[[package]] +name = "winreg" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0120db82e8a1e0b9fb3345a539c478767c0048d842860994d96113d5b667bd69" +dependencies = [ + "winapi 0.3.9", +] + +[[package]] +name = "ws2_32-sys" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d59cefebd0c892fa2dd6de581e937301d8552cb44489cdff035c6187cb63fa5e" +dependencies = [ + "winapi 0.2.8", + "winapi-build", +] diff --git a/Cargo.toml b/Cargo.toml new file mode 100644 index 0000000000..494f8af52e --- /dev/null +++ b/Cargo.toml @@ -0,0 +1,37 @@ +[package] +name = "pubsub-enricher" +version = "0.0.0" +edition = "2018" +authors = [ + "Brandon Thomas ", + "Brandon Thomas " +] + +[[bin]] +name = "pubsub-enricher" +path = "src/main.rs" + +[dependencies] +anyhow = "1.0.38" +easyenv = "0.0.5" +futures = "0.3.14" +lazy_static = "1.4.0" +log = "0.4.14" +prost = "0.7.0" +regex = "1.4.6" +rustls = "0.18" +serde = "1.0.123" +serde_derive = "1.0.123" +tokio = { version = "0.2", features = ["full"] } +toml = "0.5.8" + +google_maps = "1.0.3" + +redis = { version = "0.17.0", features = ["tls", "async-std-tls-comp", "tokio-tls-comp", "connection-manager"] } + +[build-dependencies] +anyhow = "1.0.38" +prost-build = "0.7.0" + +[dev-dependencies] +expectest = "0.12.0" diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000000..79beeb8b55 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,49 @@ +# Custom base image +# Make sure to add this repository so it has read acces to the base image: +# https://github.com/orgs/storytold/packages/container/docker-base-images-rust-ssl/settings/actions_access +FROM ghcr.io/storytold/docker-base-images-rust-ssl:latest as build +WORKDIR /tmp + +COPY Cargo.lock . +COPY Cargo.toml . +COPY src/ ./src +COPY protos/ ./protos + +RUN $HOME/.cargo/bin/cargo fetch + +#RUN LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} $HOME/.cargo/bin/cargo build +RUN LD_LIBRARY_PATH=/usr/lib:${LD_LIBRARY_PATH} $HOME/.cargo/bin/cargo build --release + +# Final image +FROM ubuntu:xenial +WORKDIR / + +# See: https://github.com/opencontainers/image-spec/blob/master/annotations.md +LABEL org.opencontainers.image.authors='bt@brand.io, echelon@gmail.com' +LABEL org.opencontainers.image.url='https://github.com/storytold/pubsub-enricher' +LABEL org.opencontainers.image.documentation='https://github.com/storytold/pubsub-enricher' +LABEL org.opencontainers.image.source='https://github.com/storytold/pubsub-enricher' + +# NB: Comment this out for non-debug images +# TODO: Figure out how this is done elsewhere with just the static binaries +RUN apt-get update \ + && apt-get install -y \ + curl \ + wget + +#COPY --from=build /tmp/target/debug/pubsub-enricher / +COPY --from=build /tmp/target/release/pubsub-enricher / + +# SSL certs are required for crypto +COPY --from=build /etc/ssl /etc/ssl + +# Required dynamically linked libraries +COPY --from=build /lib/x86_64-linux-gnu/libssl.* /lib/x86_64-linux-gnu/ +COPY --from=build /lib/x86_64-linux-gnu/libcrypto.* /lib/x86_64-linux-gnu/ + +# Make sure all the links resolve +RUN ldd pubsub-enricher + +EXPOSE 8080 +CMD LD_LIBRARY_PATH=/usr/lib /pubsub-enricher + diff --git a/README.md b/README.md new file mode 100644 index 0000000000..1f784226a8 --- /dev/null +++ b/README.md @@ -0,0 +1,12 @@ +pubsub-enricher +=============== +Listen to pubsub firehose, enrich events, then republish. + +Protos +------ + +Protos are defined in a git submodule: + +``` +git submodule add git@github.com:storytold/storyteller-protos.git +``` diff --git a/build.rs b/build.rs new file mode 100644 index 0000000000..7822e529f0 --- /dev/null +++ b/build.rs @@ -0,0 +1,18 @@ +extern crate anyhow; // TODO: No! +extern crate prost_build; + +use std::env; + +fn main() -> anyhow::Result<()> { + // NB: Control prost output dir with env var + // Typically these land in `target/`, but I want to source control these. + env::set_var("OUT_DIR", "src/proto_codegen"); + + prost_build::compile_protos( + &[ + "protos/storyteller-protos/protos/pubsub-events/pubsub-events.proto", + "protos/storyteller-protos/protos/unreal/unreal.proto", + ], + &["protos/"])?; + Ok(()) +} diff --git a/protos/storyteller-protos b/protos/storyteller-protos new file mode 160000 index 0000000000..c61a069a6f --- /dev/null +++ b/protos/storyteller-protos @@ -0,0 +1 @@ +Subproject commit c61a069a6ffcd6fdc15684e3086c48373a670fd0 diff --git a/src/clients/mod.rs b/src/clients/mod.rs new file mode 100644 index 0000000000..0fae7044b4 --- /dev/null +++ b/src/clients/mod.rs @@ -0,0 +1,2 @@ +pub mod redis_client; +pub mod redis_subscriber; diff --git a/src/clients/redis_client.rs b/src/clients/redis_client.rs new file mode 100644 index 0000000000..c8fbb7e3fd --- /dev/null +++ b/src/clients/redis_client.rs @@ -0,0 +1,68 @@ +use anyhow::anyhow; +use crate::AnyhowResult; +use crate::protos::protos; +use crate::secrets::RedisSecrets; +use log::{info, warn, debug}; +use prost::Message; +use redis::aio::{Connection, PubSub, ConnectionManager}; +use redis::{AsyncCommands, Msg, PubSubCommands}; +use redis; +use std::thread; +use std::time::Duration; +use tokio::stream::{Stream, StreamExt}; + +pub struct RedisClient { + secrets: RedisSecrets, + + /// This manages our connection. + /// It's meant to be absent when not connected. + connection_manager: Option, + + /// Attempts to retry a Redis command. + max_retry_count: u32, +} + +impl RedisClient { + pub fn new(secrets: &RedisSecrets, retry_count: u32) -> Self { + Self { + secrets: secrets.clone(), + connection_manager: None, + max_retry_count: retry_count, + } + } + + pub async fn connect(&mut self) -> AnyhowResult<()> { + let client = redis::Client::open(self.secrets.connection_url())?; + + let connection_manager = client.get_tokio_connection_manager().await?; + self.connection_manager = Some(connection_manager); + + Ok(()) + } + + pub async fn publish_bytes(&mut self, channel: &str, message: &[u8]) -> AnyhowResult { + for i in 0 .. self.max_retry_count { + match self.try_publish_bytes(channel, message).await { + Ok(r) => return Ok(r), + Err(_) => { + debug!("Redis publish_bytes failed; retrying (attempt {})", i); + continue; + }, + } + } + Err(anyhow!("Retry count elapsed.")) + } + + async fn try_publish_bytes(&mut self, channel: &str, message: &[u8]) -> AnyhowResult { + let connection_manager = match &mut self.connection_manager { + None => { + self.connect().await?; + return Err(anyhow!("Not connected")); + }, + Some(cm) => cm, + }; + + let result = connection_manager.publish(channel, message).await?; + Ok(result) + } +} diff --git a/src/clients/redis_subscriber.rs b/src/clients/redis_subscriber.rs new file mode 100644 index 0000000000..ef50d073ba --- /dev/null +++ b/src/clients/redis_subscriber.rs @@ -0,0 +1,101 @@ +use anyhow::anyhow; +use crate::AnyhowResult; +use crate::dispatcher::Dispatcher; +use crate::protos::protos; +use crate::secrets::RedisSecrets; +use log::{info, warn}; +use prost::Message; +use redis::aio::PubSub; +use std::thread; +use std::time::Duration; +use tokio::stream::{Stream, StreamExt}; + +pub struct RedisSubscribeClient { + secrets: RedisSecrets, + + /// This manages our connection. + /// It's meant to be absent when not connected. + /// TODO: Probably need to reset this when we disconnect. + connection: Option, + + connection_failure_count: u32, + + /// Event dispatcher that handles routing and passing to handlers. + dispatcher: Dispatcher, +} + +impl RedisSubscribeClient { + pub fn new(secrets: &RedisSecrets, dispatcher: Dispatcher) -> Self { + Self { + secrets: secrets.clone(), + connection: None, + connection_failure_count: 0, + dispatcher, + } + } + + pub async fn connect(&mut self) -> AnyhowResult<()> { + let client = redis::Client::open(self.secrets.connection_url())?; + let mut connection = client.get_async_connection().await?.into_pubsub(); + + self.connection = Some(connection); + self.connection_failure_count = 0; + + Ok(()) + } + + pub async fn failure_notify_maybe_reconnect(&mut self) -> AnyhowResult<()> { + self.connection_failure_count += 1; + + if self.connection_failure_count > 5 { + println!("Attempting Redis Reconnect in 3 sec..."); + thread::sleep(Duration::from_secs(3)); + self.connect().await?; + self.connection_failure_count = 0; + } else { + println!("Not ready to reconnect. Fail Count = {}", self.connection_failure_count); + } + Ok(()) + } + + pub async fn subscribe(&mut self, topic: &str) -> AnyhowResult<()> { + let mut connection = match &mut self.connection { + None => { + return Err(anyhow!("Not connected")); + }, + Some(connection) => connection, + }; + + connection.subscribe(topic).await?; + Ok(()) + } + + pub async fn start_stream(&mut self) -> AnyhowResult<()> { + let connection = match &mut self.connection { + None => { + return Err(anyhow!("Not connected")); + }, + Some(connection) => connection, + }; + + loop { + for message in connection.on_message().next().await { + let bytes = message.get_payload_bytes(); + + let mut message_proto = match protos::PubsubEventPayloadV1::decode(bytes) { + Ok(m) => m, + Err(e) => { + warn!("Error decoding proto: {:?}", e); + continue; + }, + }; + + self.dispatcher.handle_pubsub_event(message_proto); + } + + thread::sleep(Duration::from_millis(250)); + } + + Ok(()) + } +} diff --git a/src/dispatcher.rs b/src/dispatcher.rs new file mode 100644 index 0000000000..da06ee8340 --- /dev/null +++ b/src/dispatcher.rs @@ -0,0 +1,76 @@ +use crate::protos::protos; +use lazy_static::lazy_static; +use anyhow::anyhow; +use log::{info, warn, debug}; +use regex::Regex; +use std::collections::HashMap; + +use crate::AnyhowResult; +use crate::handlers::coordinate_and_geocode_handler::CoordinateAndGeocodeHandler; +use crate::text_chat_parsers::first_pass_command_parser::FirstPassParsedCommand; +use crate::protos::inbound_proto_utils::{InboundEvent, InboundEventSource}; + +pub trait TextCommandHandler { + fn handle_text_command(&self, + command: &FirstPassParsedCommand, + event: &InboundEvent, + event_source: &InboundEventSource) -> AnyhowResult<()>; +} + +pub struct Dispatcher { + text_command_handlers: HashMap>, +} + +impl Dispatcher { + pub fn new() -> Self { + Self { + text_command_handlers: HashMap::new(), + } + } + + pub fn add_text_command_handler(&mut self, command: &str, handler: Box) { + self.text_command_handlers.insert(command.to_string(), handler); + } + + pub fn handle_pubsub_event(&self, message: protos::PubsubEventPayloadV1) -> AnyhowResult<()> { + info!("Handling Proto: {:?}", message); + + let event_source = InboundEventSource::parse_from_payload(&message)?; + let event = InboundEvent::parse_from_payload(&message)?; + + info!("Source: {:?}", event_source); + info!("Event: {:?}", event); + + match event { + InboundEvent::TwitchMessage(ref message) => { + let command = FirstPassParsedCommand::try_parse(&message.message_contents())?; + self.handle_text_command(&command, &event, &event_source)?; + } + }; + + // TODO... other event types, including non-text commands. + + Ok(()) + } + + /// Handle text-based commands. + fn handle_text_command(&self, + command: &FirstPassParsedCommand, + event: &InboundEvent, + event_source: &InboundEventSource) + -> AnyhowResult<()> + { + info!("Handle command: {}", &command.command); + + let handler = match self.text_command_handlers.get(&command.command) { + Some(h) => h, + None => { + debug!("No handler for command: {}", &command.command); + return Ok(()); + } + }; + + info!("Dispatching handler..."); + handler.handle_text_command(command, event, event_source) + } +} diff --git a/src/handlers/coordinate_and_geocode_handler.rs b/src/handlers/coordinate_and_geocode_handler.rs new file mode 100644 index 0000000000..e13396aa4f --- /dev/null +++ b/src/handlers/coordinate_and_geocode_handler.rs @@ -0,0 +1,206 @@ +use crate::AnyhowResult; +use crate::clients::redis_client::RedisClient; +use crate::dispatcher::TextCommandHandler; +use crate::text_chat_parsers::first_pass_command_parser::FirstPassParsedCommand; +use futures::executor::block_on; +use lazy_static::lazy_static; +use log::{info, warn, debug}; +use prost::Message; +use regex::Regex; +use std::sync::{RwLock, Arc, Mutex}; +use crate::protos::inbound_proto_utils::{InboundEvent, InboundEventSource}; +use crate::protos::populate_source::populate_source; +use crate::protos::protos; +use crate::protos::binary_encode_proto::binary_encode_proto; + +// TODO: maybe separate text command handling and data source concerns +// like this? Though maybe it's too early to optimize this. +pub struct CoordinateAndGeocodeHandler { + redis_client: Arc>, +} + +impl CoordinateAndGeocodeHandler { + pub fn new(redis_client: Arc>) -> Self { + Self { + redis_client, + } + } + + fn handle_lat_long(&self, + lat_long: LatLong, + event: &InboundEvent, + event_source: &InboundEventSource) -> AnyhowResult<()> + { + let mut unreal_proto = protos::UnrealEventPayloadV1::default(); + + // Cesium + let mut cesium_proto = protos::CesiumWarpRequest::default(); + cesium_proto.latitude = lat_long.latitude; + cesium_proto.longitude = lat_long.longitude; + + debug!("Payload Proto: {:?}", cesium_proto); + + unreal_proto.payload_type = protos::unreal_event_payload_v1::PayloadType::CesiumWarp as i32; + unreal_proto.payload_data = binary_encode_proto(cesium_proto)?; + unreal_proto.debug_message = "Hello from Rust!".to_string(); + + populate_source(&mut unreal_proto, event_source)?; + + let final_binary = binary_encode_proto(unreal_proto)?; + + match self.redis_client.lock() { + Ok(mut redis_client) => { + debug!("Publishing to Redis..."); + let future = redis_client.publish_bytes("unreal", &final_binary); + block_on(future); + debug!("Published to redis."); + }, + Err(_) => {}, + } + + Ok(()) + } +} + +impl TextCommandHandler for CoordinateAndGeocodeHandler { + fn handle_text_command(&self, + command: &FirstPassParsedCommand, + event: &InboundEvent, + event_source: &InboundEventSource) + -> AnyhowResult<()> + { + let maybe_lat_long = parse_lat_long(&command.unparsed_arguments); + + if let Some(lat_long) = maybe_lat_long { + self.handle_lat_long(lat_long, event, event_source); + } + + // TODO: Geocoding. + + Ok(()) + } +} + +#[derive(Clone,Debug,PartialEq)] +pub struct LatLong { + pub latitude: f64, + pub longitude: f64, +} + +/// Command payload contains the following example instruction: +/// * "34.0659° N, 84.6769° W" +/// * "33.753746, -84.386330" +pub fn parse_lat_long(input: &str) -> Option { + const NUMERAL : &'static str = r"(\-?\d+\.?\d*)\s*°?\s*([NSEW])?"; + lazy_static! { + static ref COORDINATE_REGEX: Regex = + Regex::new(&format!(r"^({})\s*,?\s*({})$", NUMERAL, NUMERAL)).expect("Regex should work"); + } + + let captures = match COORDINATE_REGEX.captures(&input) { + None => return None, + Some(caps) => caps, + }; + + let maybe_lat = captures.get(2) + .map(|m| m.as_str()) + .and_then(|s| s.parse::().ok()); + + let maybe_lat_dir = captures.get(3) + .map(|m| m.as_str()); + + let maybe_long = captures.get(5) + .map(|m| m.as_str()) + .and_then(|s| s.parse::().ok()); + + let maybe_long_dir = captures.get(6) + .map(|m| m.as_str()); + + let mut latitude = match maybe_lat { + None => return None, + Some(lat) => lat, + }; + + let mut longitude = match maybe_long { + None => return None, + Some(long) => long, + }; + + match maybe_lat_dir { + Some("S") => latitude *= -1.0, + _ => {}, + } + + match maybe_long_dir { + Some("W") => longitude *= -1.0, + _ => {}, + } + + Some(LatLong { + latitude, + longitude + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use expectest::prelude::*; + + #[test] + fn parse_lat_long_invalid() { + expect!(parse_lat_long("foo bar baz")).to(be_none()); + expect!(parse_lat_long("1 2 3")).to(be_none()); + } + + #[test] + fn parse_lat_long_raw_coordinates() { + expect!(parse_lat_long("33.753746, -84.386330")).to( + be_eq(Some(LatLong { + latitude: 33.753746, + longitude: -84.386330, + }))); + + expect!(parse_lat_long("-123, -456")).to( + be_eq(Some(LatLong { + latitude: -123.0, + longitude: -456.0, + }))); + + expect!(parse_lat_long("-99,100")).to( + be_eq(Some(LatLong { + latitude: -99.0, + longitude: 100.0, + }))); + } + + #[test] + fn parse_lat_long_directions() { + expect!(parse_lat_long("34.0659 N, 84.6769 W")).to( + be_eq(Some(LatLong { + latitude: 34.0659, + longitude: -84.6769, + }))); + + expect!(parse_lat_long("123 S, 456 E")).to( + be_eq(Some(LatLong { + latitude: -123.0, + longitude: 456.0, + }))); + } + + #[test] + fn parse_lat_long_directions_degrees() { + expect!(parse_lat_long("34.0659° N, 84.6769° W")).to( + be_eq(Some(LatLong { + latitude: 34.0659, + longitude: -84.6769, + }))); + + expect!(parse_lat_long("123°S, 777° E")).to( + be_eq(Some(LatLong { + latitude: -123.0, + longitude: 777.0, + }))); + } +} diff --git a/src/handlers/mod.rs b/src/handlers/mod.rs new file mode 100644 index 0000000000..611c22dec7 --- /dev/null +++ b/src/handlers/mod.rs @@ -0,0 +1,3 @@ +pub mod coordinate_and_geocode_handler; +pub mod spawn_handler; +pub mod vocode_handler; \ No newline at end of file diff --git a/src/handlers/spawn_handler.rs b/src/handlers/spawn_handler.rs new file mode 100644 index 0000000000..35e1de5211 --- /dev/null +++ b/src/handlers/spawn_handler.rs @@ -0,0 +1,141 @@ +use crate::AnyhowResult; +use crate::clients::redis_client::RedisClient; +use crate::dispatcher::TextCommandHandler; +use crate::protos::binary_encode_proto::binary_encode_proto; +use crate::protos::inbound_proto_utils::{InboundEvent, InboundEventSource}; +use crate::protos::protos; +use crate::text_chat_parsers::first_pass_command_parser::FirstPassParsedCommand; +use futures::executor::block_on; +use lazy_static::lazy_static; +use log::{info, warn, debug}; +use prost::Message; +use regex::Regex; +use std::sync::{RwLock, Arc, Mutex}; +use crate::protos::populate_source::populate_source; + +// TODO: maybe separate text command handling and data source concerns +// like this? Though maybe it's too early to optimize this. +pub struct SpawnHandler { + redis_client: Arc>, +} + +impl SpawnHandler { + pub fn new(redis_client: Arc>) -> Self { + Self { + redis_client, + } + } + + fn handle(&self, + spawn: SpawnCommand, + event: &InboundEvent, + event_source: &InboundEventSource) + -> AnyhowResult<()> { + + let mut unreal_proto = protos::UnrealEventPayloadV1::default(); + + // Payload + let mut spawn_proto = protos::SpawnCreatureRequest::default(); + spawn_proto.name_slug = spawn.spawn_slug; + + debug!("Payload Proto: {:?}", spawn_proto); + + unreal_proto.payload_type = protos::unreal_event_payload_v1::PayloadType::CesiumWarp as i32; + unreal_proto.payload_data = binary_encode_proto(spawn_proto)?; + unreal_proto.debug_message = "Hello from Rust!".to_string(); + + populate_source(&mut unreal_proto, event_source)?; + + let final_binary = binary_encode_proto(unreal_proto)?; + + match self.redis_client.lock() { + Ok(mut redis_client) => { + debug!("Publishing to Redis..."); + let future = redis_client.publish_bytes("unreal", &final_binary); + block_on(future); + debug!("Published to redis"); + }, + Err(_) => {}, + } + + Ok(()) + } +} + +impl TextCommandHandler for SpawnHandler { + fn handle_text_command(&self, + command: &FirstPassParsedCommand, + event: &InboundEvent, + event_source: &InboundEventSource) + -> AnyhowResult<()> + { + let maybe_spawn_command = parse_spawn(&command.unparsed_arguments); + + debug!("Maybe spawn command: {:?}", maybe_spawn_command); + + if let Some(spawn) = maybe_spawn_command { + self.handle(spawn, event, event_source); + } + Ok(()) + } +} + +#[derive(Clone,Debug,PartialEq)] +pub struct SpawnCommand { + pub spawn_slug: String, + // TODO: Args in the future. +} + +pub fn parse_spawn(input: &str) -> Option { + lazy_static! { + static ref REGEX: Regex = + Regex::new(r"^\s*([\w\-]+)\s*$").expect("Regex should work"); + } + + let captures = match REGEX.captures(&input) { + None => return None, + Some(caps) => caps, + }; + + let maybe_capture = captures.get(1) + .map(|m| m.as_str()); + + + let capture = match maybe_capture { + None => return None, + Some(cap) => cap, + }; + + Some(SpawnCommand { + spawn_slug: capture.to_string(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use expectest::prelude::*; + + #[test] + fn failure_cases() { + expect!(parse_spawn("too many words")).to(be_none()); + expect!(parse_spawn("")).to(be_none()); + expect!(parse_spawn(" ")).to(be_none()); + } + + #[test] + fn single_word() { + expect!(parse_spawn("goomba")).to( + be_eq(Some(SpawnCommand { + spawn_slug: "goomba".to_string(), + }))); + } + + #[test] + fn hyphenated_word() { + expect!(parse_spawn("like-like")).to( + be_eq(Some(SpawnCommand { + spawn_slug: "like-like".to_string(), + }))); + } +} diff --git a/src/handlers/vocode_handler.rs b/src/handlers/vocode_handler.rs new file mode 100644 index 0000000000..8dbfc16dd5 --- /dev/null +++ b/src/handlers/vocode_handler.rs @@ -0,0 +1,152 @@ +use crate::AnyhowResult; +use crate::clients::redis_client::RedisClient; +use crate::dispatcher::TextCommandHandler; +use crate::protos::binary_encode_proto::binary_encode_proto; +use crate::protos::inbound_proto_utils::{InboundEvent, InboundEventSource}; +use crate::protos::protos; +use crate::text_chat_parsers::first_pass_command_parser::FirstPassParsedCommand; +use futures::executor::block_on; +use lazy_static::lazy_static; +use log::{info, warn, debug}; +use prost::Message; +use regex::Regex; +use std::sync::{RwLock, Arc, Mutex}; +use crate::protos::populate_source::populate_source; + +// TODO: maybe separate text command handling and data source concerns +// like this? Though maybe it's too early to optimize this. +pub struct TtsHandler { + redis_client: Arc>, +} + +impl TtsHandler { + pub fn new(redis_client: Arc>) -> Self { + Self { + redis_client, + } + } + + fn handle(&self, + tts: TtsCommand, + event: &InboundEvent, + event_source: &InboundEventSource) + -> AnyhowResult<()> { + + let mut unreal_proto = protos::UnrealEventPayloadV1::default(); + + // Payload + let mut tts_proto = protos::VocodesTtsRequest::default(); + tts_proto.voice_slug = tts.speaker_slug; + tts_proto.text = tts.utterance; + + debug!("Payload Proto: {:?}", tts_proto); + + unreal_proto.payload_type = protos::unreal_event_payload_v1::PayloadType::VocodesTts as i32; + unreal_proto.payload_data = binary_encode_proto(tts_proto)?; + unreal_proto.debug_message = "Hello from Rust!".to_string(); + + populate_source(&mut unreal_proto, event_source)?; + + let final_binary = binary_encode_proto(unreal_proto)?; + + match self.redis_client.lock() { + Ok(mut redis_client) => { + debug!("Publishing to Redis..."); + let future = redis_client.publish_bytes("unreal", &final_binary); + block_on(future); + debug!("Published to redis"); + }, + Err(_) => {}, + } + + Ok(()) + } +} + +impl TextCommandHandler for TtsHandler { + fn handle_text_command(&self, + command: &FirstPassParsedCommand, + event: &InboundEvent, + event_source: &InboundEventSource) + -> AnyhowResult<()> + { + let maybe_tts_command = parse_tts(&command.unparsed_arguments); + + debug!("Maybe tts command: {:?}", maybe_tts_command); + + if let Some(tts) = maybe_tts_command { + self.handle(tts, event, event_source); + } + Ok(()) + } +} + +#[derive(Clone,Debug,PartialEq)] +pub struct TtsCommand { + pub speaker_slug: String, + pub utterance: String, +} + +pub fn parse_tts(input: &str) -> Option { + lazy_static! { + static ref REGEX: Regex = + Regex::new(r"^\s*([\w\-]+)\s+(.*)\s*$").expect("Regex should work"); + } + + let captures = match REGEX.captures(&input) { + None => return None, + Some(caps) => caps, + }; + + let maybe_speaker = captures.get(1) + .map(|m| m.as_str()); + + let maybe_utterance = captures.get(2) + .map(|m| m.as_str()); + + let speaker = match maybe_speaker { + None => return None, + Some(cap) => cap, + }; + + let utterance = match maybe_utterance { + None => return None, + Some(cap) => cap, + }; + + Some(TtsCommand { + speaker_slug: speaker.to_lowercase(), + utterance: utterance.to_string(), + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use expectest::prelude::*; + + #[test] + fn failure_cases() { + expect!(parse_tts("onewordthatstretcheson")).to(be_none()); + expect!(parse_tts("")).to(be_none()); + expect!(parse_tts(" ")).to(be_none()); + expect!(parse_tts(" !!! nope ")).to(be_none()); + } + + #[test] + fn speaker_and_sentence() { + expect!(parse_tts("sonic gotta go fast")).to( + be_eq(Some(TtsCommand { + speaker_slug: "sonic".to_string(), + utterance: "gotta go fast".to_string() + }))); + } + + fn speaker_lowercase() { + expect!(parse_tts("LINK EXCUSE ME PRINCESS")).to( + be_eq(Some(TtsCommand { + speaker_slug: "link".to_string(), + utterance: "EXCUSE ME PRINCESS".to_string() + }))); + } +} diff --git a/src/main.rs b/src/main.rs new file mode 100644 index 0000000000..cfcd5e4253 --- /dev/null +++ b/src/main.rs @@ -0,0 +1,121 @@ +#![allow(dead_code)] +#![allow(unused_mut)] +#![allow(unused_imports)] +#![warn(unused_must_use)] +//#![allow(warnings)] + +#[macro_use] +extern crate lazy_static; +#[macro_use] +extern crate serde_derive; + +mod clients; +mod dispatcher; +mod handlers; +mod protos; +mod secrets; +mod text_chat_parsers; + +use anyhow::anyhow; +use anyhow::{Context, Error}; +use log::{warn, info}; +use redis::aio::Connection; +use redis::{AsyncCommands, RedisResult}; +use std::thread; +use std::time::Duration; + +use crate::clients::redis_client::RedisClient; +use crate::clients::redis_subscriber::RedisSubscribeClient; +use crate::dispatcher::Dispatcher; +use crate::handlers::coordinate_and_geocode_handler::CoordinateAndGeocodeHandler; +use crate::secrets::{Secrets, RedisSecrets}; +use std::sync::{Arc, Mutex}; +use crate::handlers::spawn_handler::SpawnHandler; +use crate::handlers::vocode_handler::TtsHandler; + +pub type AnyhowResult = anyhow::Result; + +const ENV_SUBSCRIBE_TOPIC : &'static str = "SUBSCRIBE_TOPIC"; +const ENV_SUBSCRIBE_TOPIC_DEFAULT : &'static str = "firehose"; + +const ENV_PUBLISH_TOPIC : &'static str = "PUBLISH_TOPIC"; +const ENV_PUBLISH_TOPIC_DEFAULT : &'static str = "unreal"; + +const ENV_REDIS_MAX_RETRY_COUNT : &'static str = "REDIS_MAX_RETRY_COUNT"; +const ENV_REDIS_MAX_RETRY_COUNT_DEFAULT : u32 = 3; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + easyenv::init_all_with_default_logging(None); + + info!("Reading env config..."); + + let redis_subscribe_topic = easyenv::get_env_string_or_default( + ENV_SUBSCRIBE_TOPIC, ENV_SUBSCRIBE_TOPIC_DEFAULT); + + let redis_max_retry_count = easyenv::get_env_num( + ENV_REDIS_MAX_RETRY_COUNT, + ENV_REDIS_MAX_RETRY_COUNT_DEFAULT)?; + + //let secrets = Secrets::from_file("secrets.toml")?; + + let redis_secrets = get_redis_secrets()?; + + let mut dispatcher = Dispatcher::new(); + + let mut redis_client = RedisClient::new( + &redis_secrets, + redis_max_retry_count + ); + + info!("Connecting to Redis..."); + + redis_client.connect().await?; + + let redis_client = Arc::new(Mutex::new(redis_client)); + + dispatcher.add_text_command_handler( + "goto", + Box::new(CoordinateAndGeocodeHandler::new(redis_client.clone()))); + dispatcher.add_text_command_handler( + "spawn", + Box::new(SpawnHandler::new(redis_client.clone()))); + dispatcher.add_text_command_handler( + "tts", + Box::new(TtsHandler::new(redis_client.clone()))); + + let mut redis_pubsub_client = RedisSubscribeClient::new( + &redis_secrets, + dispatcher + ); + + info!("Initial connect to PubSub Redis..."); + + redis_pubsub_client.connect().await?; + + loop { + info!("Main loop iter... (redis pubsub connect, subscribe, start stream...)"); + redis_pubsub_client.connect().await?; + redis_pubsub_client.subscribe(&redis_subscribe_topic).await?; + redis_pubsub_client.start_stream().await?; + thread::sleep(Duration::from_secs(5)); + } + + Ok(()) +} + +fn get_redis_secrets() -> AnyhowResult { + Ok(RedisSecrets::new( + &easyenv::get_env_string_or_default( + "REDIS_USERNAME", ""), + &easyenv::get_env_string_or_default( + "REDIS_PASSWORD", ""), + &easyenv::get_env_string_or_default( + "REDIS_HOST", ""), + easyenv::get_env_num::( + "REDIS_PORT", 6379)?, + easyenv::get_env_bool_or_default( + "REDIS_USES_TLS", false), + )) +} + diff --git a/src/proto_codegen/storyteller.pubsub.rs b/src/proto_codegen/storyteller.pubsub.rs new file mode 100644 index 0000000000..098f6b8b08 --- /dev/null +++ b/src/proto_codegen/storyteller.pubsub.rs @@ -0,0 +1,181 @@ +/// Ingestion platforms are "dumb" and shouldn't have much logic in them. +/// (Granted, we might eventually grow to send them back messages.) +/// +/// The "enricher" is the smarts. It does denylisting, rate limiting, +/// service calls, transforms, etc. +/// +/// What if we need to scale? Several ideas: Sharding (redis db indices), +/// brokers with read-once semantics (our own intelligence - UUIDs with +/// transactional semantics in MySQL). +/// +/// Events coming from multiple sources +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct PubsubEventPayloadV1 { + /// Source of the event. + #[prost(enumeration="pubsub_event_payload_v1::IngestionSourceType", optional, tag="1")] + pub ingestion_source_type: ::core::option::Option, + #[prost(enumeration="pubsub_event_payload_v1::IngestionPayloadType", optional, tag="2")] + pub ingestion_payload_type: ::core::option::Option, + /// A binary-encoded proto whose type is based on IngestionSourceType + #[prost(bytes="vec", optional, tag="3")] + pub ingestion_source_data: ::core::option::Option<::prost::alloc::vec::Vec>, + /// A binary-encoded proto whose type is based on IngestionPayloadType + #[prost(bytes="vec", optional, tag="4")] + pub ingestion_payload_data: ::core::option::Option<::prost::alloc::vec::Vec>, + /// An optional debug message. + #[prost(string, optional, tag="5")] + pub debug_message: ::core::option::Option<::prost::alloc::string::String>, +} +/// Nested message and enum types in `PubsubEventPayloadV1`. +pub mod pubsub_event_payload_v1 { + /// TODO: This should all be merged in the future, but I need to settle on proto2 vs proto3 + /// And I also need to figure out package names, etc. + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] + #[repr(i32)] + pub enum IngestionSourceType { + IstDoNotUse = 0, + IstAdminPanel = 1, + IstVocodesWebsite = 2, + IstTwitch = 3, + IstTwitter = 4, + IstDiscord = 5, + IstYoutube = 6, + IstReddit = 7, + IstTiktok = 8, + } + /// These will differ from the "enriched" types. + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] + #[repr(i32)] + pub enum IngestionPayloadType { + IptDoNotUse = 0, + /// Twitch-specific events. + /// + /// This can encode lots of things. + TwitchMessage = 100, + TwitchSubscribed = 101, + /// Twitter-specific events. + TwitterFollow = 200, + TwitterRetweet = 201, + TwitterMention = 202, + /// Discord-specific events. + DiscordMessage = 300, + DiscordJoined = 301, + /// YouTube-specific events. + YoutubeSubscribed = 400, + /// Reddit-specific events. + RedditMessage = 500, + } +} +// ============================== SOURCE PROTOS ============================== // + +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitchMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(int64, optional, tag="1")] + pub user_id: ::core::option::Option, + #[prost(string, optional, tag="2")] + pub username: ::core::option::Option<::prost::alloc::string::String>, + #[prost(bool, optional, tag="3")] + pub user_is_mod: ::core::option::Option, + #[prost(bool, optional, tag="4")] + pub user_is_subscribed: ::core::option::Option, + /// Optional. Name of the channel, prefixed with hash. eg. `#vocodes`. + #[prost(string, optional, tag="5")] + pub channel: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitterMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, optional, tag="1")] + pub username: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionDiscordMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, optional, tag="1")] + pub username: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionYoutubeMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, optional, tag="1")] + pub username: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionRedditMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, optional, tag="1")] + pub username: ::core::option::Option<::prost::alloc::string::String>, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTikTokMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, optional, tag="1")] + pub username: ::core::option::Option<::prost::alloc::string::String>, +} +// ============================== TWITCH PAYLOAD PROTOS ============================== // + +/// These are raw messages from Twitch that we send to the PubSub enricher. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitchMessage { + /// TODO: DEPRECATE THESE + #[prost(int64, optional, tag="1")] + pub user_id: ::core::option::Option, + #[prost(string, optional, tag="2")] + pub username: ::core::option::Option<::prost::alloc::string::String>, + #[prost(bool, optional, tag="3")] + pub is_mod: ::core::option::Option, + #[prost(bool, optional, tag="4")] + pub is_subscribed: ::core::option::Option, + /// TODO: DEPRECATE THESE + /// Name of the channel, prefixed with hash. eg. `#vocodes`. + #[prost(string, optional, tag="5")] + pub channel: ::core::option::Option<::prost::alloc::string::String>, + /// Actual message. Doesn't need to be trimmed. + #[prost(string, optional, tag="6")] + pub message_contents: ::core::option::Option<::prost::alloc::string::String>, +} +/// Intentionally blank for now +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitchSubscribed { +} +// ============================== TWITTER PAYLOAD PROTOS ============================== // + +/// Intentionally blank for now +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitterFollow { +} +/// Intentionally blank for now +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitterRetweet { +} +/// Intentionally blank for now +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionTwitterMention { +} +// ============================== DISCORD PAYLOAD PROTOS ============================== // + +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionDiscordMessage { + /// Raw message. Needs to be trimmed + #[prost(string, optional, tag="6")] + pub message_contents: ::core::option::Option<::prost::alloc::string::String>, +} +/// Intentionally blank for now +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionDiscordJoined { +} +// ============================== YOUTUBE PAYLOAD PROTOS ============================== // + +/// Intentionally blank for now +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionYoutubeSubscribed { +} +// ============================== REDDIT PAYLOAD PROTOS ============================== // + +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct IngestionRedditMessage { + /// Raw message. Needs to be trimmed + #[prost(string, optional, tag="6")] + pub message_contents: ::core::option::Option<::prost::alloc::string::String>, +} diff --git a/src/proto_codegen/storyteller.unreal.rs b/src/proto_codegen/storyteller.unreal.rs new file mode 100644 index 0000000000..bea0e28202 --- /dev/null +++ b/src/proto_codegen/storyteller.unreal.rs @@ -0,0 +1,187 @@ +/// The payloads delivered to unreal engine over the pubsub bus. +/// These are the "enriched" events. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct UnrealEventPayloadV1 { + /// Source of the event. + #[prost(enumeration="unreal_event_payload_v1::SourceType", tag="1")] + pub source_type: i32, + /// Type of the payload. + #[prost(enumeration="unreal_event_payload_v1::PayloadType", tag="2")] + pub payload_type: i32, + /// A binary-encoded proto whose type is based on SourceType + #[prost(bytes="vec", tag="3")] + pub source_data: ::prost::alloc::vec::Vec, + /// A binary-encoded proto whose type is based on PayloadType + #[prost(bytes="vec", tag="4")] + pub payload_data: ::prost::alloc::vec::Vec, + /// An optional debug message. + #[prost(string, tag="5")] + pub debug_message: ::prost::alloc::string::String, +} +/// Nested message and enum types in `UnrealEventPayloadV1`. +pub mod unreal_event_payload_v1 { + /// If people Retweet us or mention us, we should show that in the engine! + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] + #[repr(i32)] + pub enum SourceType { + StDoNotUse = 0, + StAdminPanel = 1, + StVocodesWebsite = 2, + StTwitch = 3, + StTwitter = 4, + StDiscord = 5, + StYoutube = 6, + StReddit = 7, + StTiktok = 8, + } + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)] + #[repr(i32)] + pub enum PayloadType { + PtDoNotUse = 0, + /// Game events. These can come from many sources (control panel, Twitch, Discord) + VocodesTts = 1, + LevelWarp = 2, + CesiumWarp = 3, + CesiumTimeChange = 4, + ///CHANGE_MUSIC = 6; + SpawnCreature = 5, + /// Twitter-specific events. + TwitterFollow = 200, + TwitterRetweet = 201, + TwitterMention = 202, + /// Discord-specific events. + DiscordJoined = 300, + /// YouTube-specific events. + YoutubeSubscribed = 400, + } +} +// ============================== SOURCE PROTOS ============================== // + +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct TwitchMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(int64, tag="1")] + pub user_id: i64, + #[prost(string, tag="2")] + pub username: ::prost::alloc::string::String, + #[prost(bool, tag="3")] + pub user_is_mod: bool, + #[prost(bool, tag="4")] + pub user_is_subscribed: bool, + /// Optional. Name of the channel, prefixed with hash. eg. `#vocodes`. + #[prost(string, tag="5")] + pub channel: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct TwitterMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, tag="1")] + pub username: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct DiscordMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, tag="1")] + pub username: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct YoutubeMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, tag="1")] + pub username: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct RedditMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, tag="1")] + pub username: ::prost::alloc::string::String, +} +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct TikTokMetadata { + /// Optional. Fields describing the user that sent the message. + #[prost(string, tag="1")] + pub username: ::prost::alloc::string::String, +} +// ============================== GAME EVENT PROTOS ============================== // + +/// Change time in Cesium. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct CesiumTimeChangeRequest { + /// Required + #[prost(int64, tag="1")] + pub hour: i64, +} +/// Warp within a Cesium geospatial world. +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct CesiumWarpRequest { + /// Required + #[prost(double, tag="1")] + pub latitude: f64, + /// Required + #[prost(double, tag="2")] + pub longitude: f64, + /// TODO: Remove these. + /// Optional. Fields describing the user that sent the message. + #[prost(int64, tag="3")] + pub twitch_user_id: i64, + #[prost(string, tag="4")] + pub twitch_username: ::prost::alloc::string::String, + #[prost(bool, tag="5")] + pub twitch_user_is_mod: bool, + #[prost(bool, tag="6")] + pub twitch_user_is_subscribed: bool, + /// TODO: Remove these. + /// Optional. Name of the channel, prefixed with hash. eg. `#vocodes`. + #[prost(string, tag="7")] + pub twitch_channel: ::prost::alloc::string::String, +} +/// Change the level +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct LevelWarpRequest { + /// Required. Level to warp to. + #[prost(string, tag="1")] + pub level_slug: ::prost::alloc::string::String, +} +/// Spawn a creature +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct SpawnCreatureRequest { + /// Required. Name of the creature. + #[prost(string, tag="1")] + pub name_slug: ::prost::alloc::string::String, +} +/// Do text to speech +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct VocodesTtsRequest { + /// Required. Speaker to use. + #[prost(string, tag="1")] + pub voice_slug: ::prost::alloc::string::String, + /// Required. Text to speak. + #[prost(string, tag="2")] + pub text: ::prost::alloc::string::String, +} +// ============================== TWITTER EVENT PROTOS ============================== // + +/// Intentionally blank +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct TwitterFollow { +} +/// Intentionally blank +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct TwitterRetweet { +} +/// Intentionally blank +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct TwitterMention { +} +// ============================== DISCORD EVENT PROTOS ============================== // + +/// Intentionally blank +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct DiscordJoined { +} +// ============================== YOUTUBE EVENT PROTOS ============================== // + +/// Intentionally blank +#[derive(Clone, PartialEq, ::prost::Message)] +pub struct YoutubeSubscribed { +} diff --git a/src/protos/binary_encode_proto.rs b/src/protos/binary_encode_proto.rs new file mode 100644 index 0000000000..5670d23fd0 --- /dev/null +++ b/src/protos/binary_encode_proto.rs @@ -0,0 +1,17 @@ +use anyhow::anyhow; +use crate::AnyhowResult; + +// Binary encode a proto. +pub fn binary_encode_proto(proto: impl prost::Message) -> AnyhowResult> { + let mut buffer : Vec = Vec::with_capacity(proto.encoded_len()); + let encode_result = proto.encode(&mut buffer); + + match encode_result { + Err(e) => { + Err(anyhow!("Inner proto encode result: {:?}", e)) + } + Ok(_) => { + Ok(buffer) + } + } +} diff --git a/src/protos/inbound_proto_utils.rs b/src/protos/inbound_proto_utils.rs new file mode 100644 index 0000000000..a28915be9e --- /dev/null +++ b/src/protos/inbound_proto_utils.rs @@ -0,0 +1,70 @@ +use anyhow::anyhow; +use crate::AnyhowResult; +use crate::protos::protos::IngestionTwitchMessage; +use crate::protos::protos::IngestionTwitchMetadata; +use crate::protos::protos::PubsubEventPayloadV1; +use crate::protos::protos::pubsub_event_payload_v1::IngestionPayloadType; +use crate::protos::protos::pubsub_event_payload_v1::IngestionSourceType; +use crate::protos::protos; +use prost::Message; + +// TODO: add some type safety w/ generics and type bounds + +/// Sources of events +#[derive(Debug, Clone)] +pub enum InboundEventSource { + Twitch(IngestionTwitchMetadata), +} + +/// Events +#[derive(Debug, Clone)] +pub enum InboundEvent { + TwitchMessage(IngestionTwitchMessage), +} + +impl InboundEventSource { + pub fn parse_from_payload(event_payload: &PubsubEventPayloadV1) + -> AnyhowResult + { + match Self::parse_type_proto(event_payload) { + Some(IngestionSourceType::IstTwitch) => { + IngestionTwitchMetadata::decode(event_payload.ingestion_source_data()) + .map(|m| Self::Twitch(m)) + .map_err(|_| anyhow!("error decoding source proto")) + }, + None => Err(anyhow!("invalid source type")), + Some(_) => Err(anyhow!("source type is not yet handled")), + } + } + + fn parse_type_proto(event_payload: &PubsubEventPayloadV1) + -> Option + { + event_payload.ingestion_source_type + .and_then(|num| IngestionSourceType::from_i32(num)) + } +} + +impl InboundEvent { + pub fn parse_from_payload(event_payload: &PubsubEventPayloadV1) + -> AnyhowResult + { + match Self::parse_type_proto(event_payload) { + Some(IngestionPayloadType::TwitchMessage) => { + IngestionTwitchMessage::decode(event_payload.ingestion_payload_data()) + .map(|m| Self::TwitchMessage(m)) + .map_err(|_| anyhow!("error decoding payload proto")) + }, + None => Err(anyhow!("invalid payload type")), + Some(_) => Err(anyhow!("payload type is not yet handled")), + } + } + + fn parse_type_proto(event_payload: &PubsubEventPayloadV1) + -> Option + { + event_payload.ingestion_payload_type + .and_then(|num| IngestionPayloadType::from_i32(num)) + } +} + diff --git a/src/protos/mod.rs b/src/protos/mod.rs new file mode 100644 index 0000000000..4498e6bdca --- /dev/null +++ b/src/protos/mod.rs @@ -0,0 +1,10 @@ +pub mod binary_encode_proto; +pub mod inbound_proto_utils; +pub mod outbound_proto_utils; +pub mod populate_source; + +// Include the protos +pub mod protos { + include!("../proto_codegen/storyteller.pubsub.rs"); + include!("../proto_codegen/storyteller.unreal.rs"); +} diff --git a/src/protos/outbound_proto_utils.rs b/src/protos/outbound_proto_utils.rs new file mode 100644 index 0000000000..65ef4e366e --- /dev/null +++ b/src/protos/outbound_proto_utils.rs @@ -0,0 +1,16 @@ +use anyhow::anyhow; +use crate::AnyhowResult; +use crate::protos::protos::IngestionTwitchMessage; +use crate::protos::protos::IngestionTwitchMetadata; +use crate::protos::protos::PubsubEventPayloadV1; +use crate::protos::protos::pubsub_event_payload_v1::IngestionPayloadType; +use crate::protos::protos::pubsub_event_payload_v1::IngestionSourceType; +use crate::protos::protos; +use prost::Message; + +pub enum OutboundPayload { +} + +pub enum OutboundSourceInfo { + Twitch(), +} \ No newline at end of file diff --git a/src/protos/populate_source.rs b/src/protos/populate_source.rs new file mode 100644 index 0000000000..73c0570535 --- /dev/null +++ b/src/protos/populate_source.rs @@ -0,0 +1,31 @@ +use anyhow::anyhow; +use crate::AnyhowResult; +use crate::protos::binary_encode_proto::binary_encode_proto; +use crate::protos::protos::UnrealEventPayloadV1; +use crate::protos::protos; +use log::{info, warn, debug}; +use crate::protos::inbound_proto_utils::InboundEventSource; + +pub fn populate_source(unreal_proto: &mut UnrealEventPayloadV1, event_source: &InboundEventSource) -> AnyhowResult<()> +{ + match event_source { + InboundEventSource::Twitch(ref twitch_source) => { + let mut twitch_metadata = protos::TwitchMetadata::default(); + twitch_metadata.username = twitch_source.username.clone().unwrap_or("".to_string()); + twitch_metadata.user_id = twitch_source.user_id.clone().unwrap_or(0); + twitch_metadata.user_is_mod = twitch_source.user_is_mod.unwrap_or(false); + twitch_metadata.user_is_subscribed = twitch_source.user_is_subscribed.unwrap_or(false); + twitch_metadata.channel = twitch_source.channel.clone().unwrap_or("".to_string()); + + debug!("Source Proto: {:?}", twitch_metadata); + + unreal_proto.source_type = protos::unreal_event_payload_v1::SourceType::StTwitch as i32; + unreal_proto.source_data = binary_encode_proto(twitch_metadata)?; + }, + _ => { + return Err(anyhow!("Unknown event source proto.")); + }, + } + + Ok(()) +} diff --git a/src/secrets.rs b/src/secrets.rs new file mode 100644 index 0000000000..017b7a550d --- /dev/null +++ b/src/secrets.rs @@ -0,0 +1,94 @@ +use crate::AnyhowResult; +use std::io::Read; + +#[derive(Deserialize)] +pub struct Secrets { + pub redis: RedisSecrets, + pub twitch: TwitchSecrets, + pub google: GoogleSecrets, +} + +#[derive(Deserialize, Clone)] +pub struct RedisSecrets { + pub username: String, + pub password: String, + pub host: String, + pub port: u32, + pub uses_tls: bool, +} + +#[derive(Deserialize, Clone)] +pub struct GoogleSecrets { + pub api_key: String, +} + +#[derive(Deserialize, Clone)] +pub struct TwitchSecrets { + /// Twitch username + pub username: String, + + /// From the creator dashboard of Twitch + pub stream_key: String, + + /// Identifies our application uniquely; cannot be reassigned. + pub application_client_id: String, + + /// Application secret (generated) + pub application_client_secret: String, + + /// OAuth access token generated with URL handshake + /// NB: This changes frequently and must be regenerated. + /// Go to this URL: + /// + /// https://id.twitch.tv/oauth2/authorize?response_type=token + /// &client_id=___APPLICATION_CLIENT_ID___ + /// &redirect_uri=http://localhost&scope=viewing_activity_read%20user_read%20channel_read%20chat:read + /// &state=randomUUID + /// + /// Which will redirect to 'localhost' per the redirect_uri. + /// The oauth token will be "access_token" prepended with "oauth:" (which isn't included) + pub oauth_access_token: String, + + /// Channels to join and monitor. + pub watch_channels: Vec, +} + +impl Secrets { + pub fn from_file(filename: &str) -> AnyhowResult { + let mut file = std::fs::File::open(filename)?; + let mut buffer = String::new(); + file.read_to_string(&mut buffer)?; + let secrets = toml::from_str(&buffer)?; + Ok(secrets) + } +} + +impl RedisSecrets { + pub fn new(username: &str, password: &str, host: &str, port: u32, uses_tls: bool) -> Self { + Self { + username: username.to_string(), + password: password.to_string(), + host: host.to_string(), + port, + uses_tls, + } + } + + pub fn connection_url(&self) -> String { + let protocol = if self.uses_tls { "rediss" } else { "redis" }; + let mut auth = "".to_string(); + + if !self.username.is_empty() { + if !self.password.is_empty() { + auth = format!("{}:{}@", self.username, self.password); + } else { + auth = format!("{}@", self.username); + } + } else if !self.password.is_empty() { + auth = format!("default:{}@", self.password); + } + + format!("{}://{}{}:{}", protocol, auth, self.host, self.port) + } +} + diff --git a/src/text_chat_parsers/first_pass_command_parser.rs b/src/text_chat_parsers/first_pass_command_parser.rs new file mode 100644 index 0000000000..216777d316 --- /dev/null +++ b/src/text_chat_parsers/first_pass_command_parser.rs @@ -0,0 +1,87 @@ +use anyhow::anyhow; +use crate::AnyhowResult; +use lazy_static::lazy_static; +use log::debug; +use regex::Regex; + +pub struct FirstPassParsedCommand { + /// The first "word" / "verb" of a sentence. + /// This is all-lowercase, no spaces, `[\w\-]+` + pub command: String, + + /// The remaining portion, stripped of leading/trailing spaces. + pub unparsed_arguments: String, +} + +impl FirstPassParsedCommand { + + pub fn try_parse(message: &str) -> AnyhowResult { + lazy_static! { + static ref COMMAND_REGEX : Regex = Regex::new(r"^\s*([\w\-]+)\s+(.*)\s*$").expect("should parse"); + } + + let captures = match COMMAND_REGEX.captures(&message) { + None => return Err(anyhow!("message could not parse")), + Some(caps) => caps, + }; + + let command = match captures.get(1) { + None => return Err(anyhow!("could not get first capture group")), + Some(cmd) => cmd.as_str().trim().to_lowercase(), + }; + + debug!("Command: {}", &command); + + let unparsed_payload = match captures.get(2) { + None => return Err(anyhow!("could not get second capture group")), + Some(p) => p.as_str().trim().to_string(), + }; + + debug!("Unparsed payload: {}", &unparsed_payload); + + Ok(Self { + command: command, + unparsed_arguments: unparsed_payload, + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn failure_cases() { + assert!(FirstPassParsedCommand::try_parse("").is_err()); + assert!(FirstPassParsedCommand::try_parse(" ").is_err()); + assert!(FirstPassParsedCommand::try_parse("foo").is_err()); + assert!(FirstPassParsedCommand::try_parse("foo-bar").is_err()); + } + + #[test] + fn parse_two_words() { + let r = FirstPassParsedCommand::try_parse("foo bar"); + assert!(r.is_ok()); + let r = r.expect("Is okay"); + assert_eq!(&r.command, "foo"); + assert_eq!(&r.unparsed_arguments, "bar"); + } + + #[test] + fn parse_two_words_spaces() { + let r = FirstPassParsedCommand::try_parse(" foo \t bar \t \n "); + assert!(r.is_ok()); + let r = r.expect("Is okay"); + assert_eq!(&r.command, "foo"); + assert_eq!(&r.unparsed_arguments, "bar"); + } + + #[test] + fn parse_hyphenated_command() { + let r = FirstPassParsedCommand::try_parse("foo-bar baz"); + assert!(r.is_ok()); + let r = r.expect("Is okay"); + assert_eq!(&r.command, "foo-bar"); + assert_eq!(&r.unparsed_arguments, "baz"); + } +} diff --git a/src/text_chat_parsers/mod.rs b/src/text_chat_parsers/mod.rs new file mode 100644 index 0000000000..ad505ed485 --- /dev/null +++ b/src/text_chat_parsers/mod.rs @@ -0,0 +1 @@ +pub mod first_pass_command_parser; \ No newline at end of file