Add http client (#219)

* Add mini-http-client implementation with json and bytes support

* Use anyhow for error handling in the http-client implementation

* Remove http-client ClientBuilder

* Use local HTTPS server instead of requesting external sources

* Move http-client tests as unit tests

* refactor http-client tests

* fix: update deprecated openssl functions

* fixup: tests and cfg fmts

also clippy fixes

* move http tests to test_async

* test_http_client, fix wasm

* ci: make certs first

require certs to startup

* ci: fix wasm test name

* fix: ci windows

* ci: macos, fixup cert labels

* ci: macos, fixup openssl1 vs http-server key fmt expectations

* result.rs use anyhow result

* ci: add binstall

wsocat doesn't release direct binaries so binstall is not used there

* ci: dynamically select test port for http-client

---------

Co-authored-by: Matheus Consoli <[email protected]>

.github/workflows/ci.yml

@@ -23,8 +23,15 @@ jobs:
       - uses: Swatinem/rust-cache@v2
         with:
           key: windows-latest-test
+      - uses: cargo-bins/cargo-binstall@main
+      - name: Test Setup
+        run: |
+          make certs
+          cargo binstall http-server
       - name: Test
-        run: cargo test --features task,subscriber,fixture,task_unstable,io,sync,future,net,tls,rust_tls,timer,fs,zero_copy,mmap,retry
+        run: |
+          Start-Process cmd -Args /c,"http-server --tls --tls-key certs/test-certs/server.key --tls-cert certs/test-certs/server.crt --tls-key-algorithm pkcs8"
+          cargo test --features task,subscriber,fixture,task_unstable,io,sync,future,net,tls,rust_tls,timer,fs,zero_copy,mmap,retry,http-client-json,__skip-http-client-cert-verification
   test:
     name: Check ${{ matrix.check }} on (${{ matrix.os }})
     runs-on: ${{ matrix.os }}
@@ -54,6 +61,7 @@ jobs:
       - uses: Swatinem/rust-cache@v2
         with:
           key: ${{ matrix.os }}-${{ matrix.check }}
+      - uses: cargo-bins/cargo-binstall@main
       - name: Clippy
         if: ${{ matrix.check == 'clippy' }}
         run: make check-clippy
@@ -68,7 +76,7 @@ jobs:
         if: ${{ matrix.check == 'test' && matrix.os == 'macos-latest'}}
         timeout-minutes: 15
         # macos-latest is by default on openssl 1.x
-        run: make PFX_OPTS="" test-all
+        run: make PFX_OPTS="" CERT_OPTS=cert-patch-macos test-all
       - name: cargo audit
         if: ${{ matrix.check == 'audit' }}
         timeout-minutes: 15
@@ -77,7 +85,7 @@ jobs:
           token: ${{ secrets.GITHUB_TOKEN }}
 
   wasm_test:
-    name: ${{ matrix.test.name }} on (${{ matrix.os }})
+    name: Wasm test on (${{ matrix.os }})
     runs-on: ${{ matrix.os }}
     strategy:
    #   fail-fast: false
@@ -98,6 +106,7 @@ jobs:
           toolchain: ${{ matrix.rust }}
           profile: minimal
           target: wasm32-unknown-unknown
+      - uses: cargo-bins/cargo-binstall@main
       - name: Run Wasm Test on Linux
         if: matrix.test == 'linux-wasm'
         run: |

Cargo.toml

@@ -55,8 +55,16 @@ http-client = [
     "dep:tokio",
     "dep:hyper",
     "dep:anyhow",
+    "dep:webpki-roots",
+    "dep:http",
+    "dep:serde",
+    "dep:serde_json",
+    "dep:once_cell",
 ]
 tokio1 = ["async-std/tokio1"]
+http-client-json = ["http-client"]
+# useful for tests
+__skip-http-client-cert-verification = []
 
 [dependencies]
 log = "0.4.0"
@@ -73,6 +81,11 @@ cfg-if = { version = "1.0.0", optional = true }
 tokio = { version = "1.33.0", default-features = false, optional = true }
 hyper = { version = "0.14.27", default-features = false, features = ["client", "http1", "http2"], optional = true }
 anyhow = { version = "1.0.75", optional = true }
+http = { version = "0.2.9", optional = true }
+webpki-roots = { version = "0.25.2", optional = true }
+serde = { version = "1.0.189", optional = true }
+serde_json = { version = "1.0.107", optional = true }
+once_cell = { version = "1.18.0", optional = true }
 thiserror = "1.0.20"
 
 fluvio-test-derive = { path = "async-test-derive", version = "0.1.1", optional = true }
@@ -98,6 +111,7 @@ async-std = { version = "1.12.0", default-features = false, features = ["unstabl
 ws_stream_wasm = "0.7.3"
 
 [dev-dependencies]
+async-std = { version = "1.12.0", features = ["attributes"] }
 bytes = "1.0.0"
 lazy_static = "1.2.0"
 num_cpus = "1.10.1"
@@ -107,7 +121,8 @@ tokio-util = { version = "0.7.0", features = ["codec", "compat"] }
 tokio = { version = "1.17.0", features = ["macros"] }
 flv-util = { version = "0.5.0", features = ["fixture"] }
 fluvio-test-derive = { path = "async-test-derive", version = "0.1.0" }
-fluvio-future = { path = ".", features = ["net", "fixture", "timer", "fs", "retry"] }
+fluvio-future = { path = ".", features = ["net", "fixture", "timer", "fs", "retry", "http-client-json", "__skip-http-client-cert-verification"] }
+serde = { version = "1.0.189", features = ["derive"] }
 
 [target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies]
 portpicker = "0.1.1"

Makefile

@@ -1,5 +1,6 @@
 RUST_DOCKER_IMAGE=rust:latest
-PFX_OPTS ?= "-legacy"
+CERT_OPTS ?=
+PFX_OPTS ?= ""
 
 build-all:
 	cargo build --all-features
@@ -8,8 +9,20 @@ build-all:
 certs:
 	make -C certs generate-certs PFX_OPTS=${PFX_OPTS}
 
-test-all:	certs test-derive
-	cargo test --all-features
+cert-patch-macos:
+	sed -i '' 's/RSA PRIVATE KEY/PRIVATE KEY/' certs/test-certs/server-hs.key
+
+.PHONY: test-all run-test-all
+test-all: certs test-derive setup-http-server run-test-all
+run-test-all:
+	TEST_PORT=$$(cat tmp-PORT) cargo test --all-features
+	$(MAKE) teardown-http-server
+
+.PHONY: test-http run-test-http
+test-http: certs setup-http-server run-test-http
+run-test-http:
+	TEST_PORT=$$(cat tmp-PORT) cargo test --all-features test_http_client
+	$(MAKE) teardown-http-server
 
 install-wasm-pack:
 	which wasm-pack || curl https://rustwasm.github.io/wasm-pack/installer/init.sh -sSf | sh
@@ -54,6 +67,20 @@ install-clippy:
 install-wasm32:
 	rustup target add wasm32-unknown-unknown
 
+setup-http-server: certs $(CERT_OPTS)
+	cargo binstall http-server
+	cargo binstall -y portpicker-cli
+	portpicker > tmp-PORT
+	echo Picked port $$(cat tmp-PORT)
+	http-server --tls \
+		--tls-key certs/test-certs/server-hs.key \
+		--tls-cert certs/test-certs/server.crt \
+		--tls-key-algorithm pkcs8 -p $$(cat tmp-PORT) &
+
+teardown-http-server:
+	killall http-server
+	rm -f tmp-PORT
+
 check-clippy:	install-clippy install-wasm32
 	cargo clippy --all-features
 	cargo check --target wasm32-unknown-unknown --all-features

certs/Makefile

@@ -50,7 +50,7 @@ generate-intermediate-ca-crt: generate-intermediate-ca-csr
 
 generate-server-key:
 	openssl genrsa -out test-certs/server.key 4096
-
+	cp test-certs/server.key test-certs/server-hs.key
 
 generate-server-csr:	generate-server-key
 	openssl req -new -key test-certs/server.key \

src/fs/mmap.rs

@@ -149,7 +149,7 @@ mod tests {
     #[test_async]
     async fn test_mmap_write_slice() -> Result<(), IoError> {
         let index_path = temp_dir().join("test.index");
-        ensure_clean_file(&index_path.clone());
+        ensure_clean_file(index_path.clone());
 
         let result = MemoryMappedMutFile::create(&index_path, 3).await;
         assert!(result.is_ok());
@@ -176,7 +176,7 @@ mod tests {
     #[test_async]
     async fn test_mmap_write_pair_slice() -> Result<(), IoError> {
         let index_path = temp_dir().join("pairslice.index");
-        ensure_clean_file(&index_path.clone());
+        ensure_clean_file(index_path.clone());
 
         let result = MemoryMappedMutFile::create(&index_path, 24).await;
         assert!(result.is_ok());
@@ -205,7 +205,7 @@ mod tests {
     #[test_async]
     async fn test_mmap_write_with_pos() -> Result<(), IoError> {
         let index_path = temp_dir().join("testpos.index");
-        ensure_clean_file(&index_path.clone());
+        ensure_clean_file(index_path.clone());
 
         let (mut mm_file, _) = MemoryMappedMutFile::create(&index_path, 10).await?;
 
@@ -227,7 +227,7 @@ mod tests {
     #[test_async]
     async fn test_empty_index_read_only() -> Result<(), IoError> {
         let index_path = temp_dir().join("zerosized.index");
-        ensure_clean_file(&index_path.clone());
+        ensure_clean_file(index_path.clone());
         {
             let file = OpenOptions::new()
                 .create(true)