diff --git a/.config/nextest.toml b/.config/nextest.toml new file mode 100644 index 00000000..5a7a7f32 --- /dev/null +++ b/.config/nextest.toml @@ -0,0 +1,2 @@ +[profile.default] +retries = { backoff = "fixed", count = 2, delay = "500ms" } diff --git a/.github/cross/Dockerfile.aarch64-unknown-linux-musl-sccache b/.github/cross/Dockerfile.aarch64-unknown-linux-musl-sccache new file mode 100644 index 00000000..898412a2 --- /dev/null +++ b/.github/cross/Dockerfile.aarch64-unknown-linux-musl-sccache @@ -0,0 +1,11 @@ +FROM ghcr.io/cross-rs/aarch64-unknown-linux-musl:0.2.5 +ARG DEBIAN_FRONTEND=noninteractive + +COPY .github/cross/sccache-prebuilt.sh /sccache.sh +RUN /sccache.sh x86_64-unknown-linux-musl + +COPY .github/cross/cargo-with-stats.sh /usr/local/bin/cargo +RUN chmod +x /usr/local/bin/cargo + +ENV RUSTC_WRAPPER=/usr/bin/sccache +ENV PATH=/usr/local/bin:/usr/local/cargo/bin:/rust/bin:/usr/bin:/bin diff --git a/.github/cross/Dockerfile.arm-unknown-linux-musleabihf-sccache b/.github/cross/Dockerfile.arm-unknown-linux-musleabihf-sccache new file mode 100644 index 00000000..d8528d9a --- /dev/null +++ b/.github/cross/Dockerfile.arm-unknown-linux-musleabihf-sccache @@ -0,0 +1,11 @@ +FROM ghcr.io/cross-rs/arm-unknown-linux-musleabihf:0.2.5 +ARG DEBIAN_FRONTEND=noninteractive + +COPY .github/cross/sccache-prebuilt.sh /sccache.sh +RUN /sccache.sh x86_64-unknown-linux-musl + +COPY .github/cross/cargo-with-stats.sh /usr/local/bin/cargo +RUN chmod +x /usr/local/bin/cargo + +ENV RUSTC_WRAPPER=/usr/bin/sccache +ENV PATH=/usr/local/bin:/usr/local/cargo/bin:/rust/bin:/usr/bin:/bin diff --git a/.github/cross/Dockerfile.armv7-unknown-linux-musleabihf-sccache b/.github/cross/Dockerfile.armv7-unknown-linux-musleabihf-sccache new file mode 100644 index 00000000..0643f686 --- /dev/null +++ b/.github/cross/Dockerfile.armv7-unknown-linux-musleabihf-sccache @@ -0,0 +1,11 @@ +FROM ghcr.io/cross-rs/armv7-unknown-linux-musleabihf:0.2.5 +ARG DEBIAN_FRONTEND=noninteractive + +COPY .github/cross/sccache-prebuilt.sh /sccache.sh +RUN /sccache.sh x86_64-unknown-linux-musl + +COPY .github/cross/cargo-with-stats.sh /usr/local/bin/cargo +RUN chmod +x /usr/local/bin/cargo + +ENV RUSTC_WRAPPER=/usr/bin/sccache +ENV PATH=/usr/local/bin:/usr/local/cargo/bin:/rust/bin:/usr/bin:/bin diff --git a/.github/cross/Dockerfile.x86_64-unknown-linux-musl-sccache b/.github/cross/Dockerfile.x86_64-unknown-linux-musl-sccache new file mode 100644 index 00000000..b4c42e42 --- /dev/null +++ b/.github/cross/Dockerfile.x86_64-unknown-linux-musl-sccache @@ -0,0 +1,11 @@ +FROM ghcr.io/cross-rs/x86_64-unknown-linux-musl:0.2.5 +ARG DEBIAN_FRONTEND=noninteractive + +COPY .github/cross/sccache-prebuilt.sh /sccache.sh +RUN /sccache.sh x86_64-unknown-linux-musl + +COPY .github/cross/cargo-with-stats.sh /usr/local/bin/cargo +RUN chmod +x /usr/local/bin/cargo + +ENV RUSTC_WRAPPER=/usr/bin/sccache +ENV PATH=/usr/local/bin:/usr/local/cargo/bin:/rust/bin:/usr/bin:/bin diff --git a/.github/cross/cargo-with-stats.sh b/.github/cross/cargo-with-stats.sh new file mode 100755 index 00000000..802c9acc --- /dev/null +++ b/.github/cross/cargo-with-stats.sh @@ -0,0 +1,18 @@ +#!/bin/sh +set -e + +if [ -x /usr/local/cargo/bin/cargo ]; then + real_cargo=/usr/local/cargo/bin/cargo +elif [ -x /rust/bin/cargo ]; then + real_cargo=/rust/bin/cargo +else + echo "cargo wrapper: real cargo not found" >&2 + exit 1 +fi + +if [ "$1" = "build" ]; then + "$real_cargo" "$@" + /usr/bin/sccache --show-stats --stats-format=json +else + "$real_cargo" "$@" +fi diff --git a/.github/cross/report-sccache-stats.sh b/.github/cross/report-sccache-stats.sh new file mode 100755 index 00000000..ab9055b2 --- /dev/null +++ b/.github/cross/report-sccache-stats.sh @@ -0,0 +1,24 @@ +#!/usr/bin/env bash +set -euo pipefail + +log_file="${1:-sccache-build.log}" +stats="$(grep '^{"stats"' "$log_file" | tail -1)" +if [[ -z "$stats" ]]; then + echo "::warning title=sccache stats::No sccache stats found in build log" + exit 0 +fi + +hits="$(echo "$stats" | jq '[.stats.cache_hits.counts | to_entries[]?.value // 0] | add // 0')" +misses="$(echo "$stats" | jq '[.stats.cache_misses.counts | to_entries[]?.value // 0] | add // 0')" +executed="$(echo "$stats" | jq '.stats.requests_executed // 0')" +compile_requests="$(echo "$stats" | jq '.stats.compile_requests // 0')" +compilations="$(echo "$stats" | jq '.stats.compilations // 0')" +not_cacheable="$(echo "$stats" | jq '.stats.requests_not_cacheable // 0')" +total=$((hits + misses)) +if (( total > 0 )); then + rate=$(( hits * 100 / total )) +else + rate=0 +fi + +echo "::notice title=sccache stats::${rate}% - ${hits} hits, ${misses} misses, ${executed} executed, ${compilations} compilations, ${compile_requests} requests, ${not_cacheable} not-cacheable" diff --git a/.github/cross/sccache-prebuilt.sh b/.github/cross/sccache-prebuilt.sh new file mode 100755 index 00000000..d0f23240 --- /dev/null +++ b/.github/cross/sccache-prebuilt.sh @@ -0,0 +1,18 @@ +#!/bin/bash +set -euxo pipefail + +triple="${1:?usage: sccache-prebuilt.sh TRIPLE}" +url="https://github.com/mozilla/sccache" +version="${SCCACHE_VERSION:-v0.15.0}" + +apt-get update +apt-get install -y --no-install-recommends ca-certificates curl +rm -rf /var/lib/apt/lists/* + +td="$(mktemp -d)" +trap 'rm -rf "${td}"' EXIT +cd "${td}" + +curl -LSfs "${url}/releases/download/${version}/sccache-${version}-${triple}.tar.gz" -o sccache.tar.gz +tar xzf sccache.tar.gz +install -m 755 "sccache-${version}-${triple}/sccache" /usr/bin/sccache diff --git a/.github/workflows/arm_vehicle.yml b/.github/workflows/arm_vehicle.yml deleted file mode 100644 index 71792e58..00000000 --- a/.github/workflows/arm_vehicle.yml +++ /dev/null @@ -1,50 +0,0 @@ -name: Vehicle Arming - -on: [push, pull_request] - -jobs: - test: - runs-on: ubuntu-latest - steps: - - uses: actions/checkout@v4 - with: - submodules: "recursive" - - - name: Set up Python - uses: actions/setup-python@v5 - with: - python-version: "3.13" - - - name: Install build dependencies - Rustup - run: | - curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- --default-toolchain stable --profile minimal --target x86_64-unknown-linux-musl,wasm32-unknown-unknown -y - echo "$HOME/.cargo/bin" >> $GITHUB_PATH - - - name: Install build dependencies - Cargo-Binstall - uses: cargo-bins/cargo-binstall@main - - - name: Install build dependencies - Trunk - run: cargo binstall -y trunk - - - name: Install system dependencies - run: | - sudo apt-get update - sudo apt-get install -y build-essential - - - name: Install the latest version of uv - uses: astral-sh/setup-uv@v6 - with: - version: "latest" - - - name: Building frontend (wasm32) - working-directory: src/webpage - run: trunk build --release --locked --verbose - - - name: Build mavlink-server - run: | - cargo build - - - name: Run vehicle arming test - run: | - cd tests/python - uv run main.py diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 37e70c37..4408429b 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -6,53 +6,221 @@ env: CARGO_TERM_COLOR: always jobs: + frontend: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + submodules: "recursive" + + - name: Configure sccache + uses: actions/github-script@v7 + with: + script: | + core.exportVariable('ACTIONS_RESULTS_URL', process.env.ACTIONS_RESULTS_URL || ''); + core.exportVariable('ACTIONS_RUNTIME_TOKEN', process.env.ACTIONS_RUNTIME_TOKEN || ''); + + - uses: mozilla-actions/sccache-action@v0.0.9 + + - name: Export sccache environment + shell: bash + run: | + SCCACHE_DIR="${SCCACHE_DIR:-$HOME/.cache/sccache}" + echo "SCCACHE_DIR=$SCCACHE_DIR" >> "$GITHUB_ENV" + mkdir -p "$SCCACHE_DIR" + echo "RUSTC_WRAPPER=$SCCACHE_PATH" >> "$GITHUB_ENV" + echo "SCCACHE_GHA_ENABLED=true" >> "$GITHUB_ENV" + "$SCCACHE_PATH" --start-server + + - uses: dtolnay/rust-toolchain@stable + with: + targets: wasm32-unknown-unknown + + - uses: Swatinem/rust-cache@v2 + with: + shared-key: frontend + cache-bin: true + cache-on-failure: true + + - name: Add Cargo bin to PATH + run: echo "$CARGO_HOME/bin" >> "$GITHUB_PATH" + + - name: Install build dependencies - Cargo-Binstall + uses: cargo-bins/cargo-binstall@main + + - name: Install build dependencies - Trunk + run: | + if [[ ! -x "$CARGO_HOME/bin/trunk" ]]; then + cargo binstall -y trunk --force + fi + + - name: Building frontend (wasm32) + working-directory: src/webpage + run: | + "$CARGO_HOME/bin/trunk" build --release --locked --verbose + + - uses: actions/upload-artifact@v4 + with: + name: webpage-dist + path: src/webpage/dist + retention-days: 1 + test: + needs: frontend runs-on: ubuntu-latest + env: + SKIP_FRONTEND: 1 steps: - uses: actions/checkout@v4 with: submodules: "recursive" - - name: Install build dependencies - Rustup + - name: Configure sccache + uses: actions/github-script@v7 + with: + script: | + core.exportVariable('ACTIONS_RESULTS_URL', process.env.ACTIONS_RESULTS_URL || ''); + core.exportVariable('ACTIONS_RUNTIME_TOKEN', process.env.ACTIONS_RUNTIME_TOKEN || ''); + + - uses: mozilla-actions/sccache-action@v0.0.9 + + - name: Export sccache environment + shell: bash run: | - curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- --default-toolchain stable --profile minimal --target x86_64-unknown-linux-musl,wasm32-unknown-unknown -y - echo "$HOME/.cargo/bin" >> $GITHUB_PATH + SCCACHE_DIR="${SCCACHE_DIR:-$HOME/.cache/sccache}" + echo "SCCACHE_DIR=$SCCACHE_DIR" >> "$GITHUB_ENV" + mkdir -p "$SCCACHE_DIR" + echo "RUSTC_WRAPPER=$SCCACHE_PATH" >> "$GITHUB_ENV" + echo "SCCACHE_GHA_ENABLED=true" >> "$GITHUB_ENV" + "$SCCACHE_PATH" --start-server + + - uses: actions/download-artifact@v4 + with: + name: webpage-dist + path: src/webpage/dist + + - uses: dtolnay/rust-toolchain@stable + + - uses: Swatinem/rust-cache@v2 + with: + shared-key: test + cache-bin: true + cache-on-failure: true + + - name: Add Cargo bin to PATH + run: echo "$CARGO_HOME/bin" >> "$GITHUB_PATH" - name: Install rustfmt run: rustup component add rustfmt + - name: Check code style run: cargo fmt -- --check - name: Install build dependencies - Cargo-Binstall uses: cargo-bins/cargo-binstall@main - - name: Install build dependencies - Trunk - run: cargo binstall -y trunk + - name: Install build dependencies - cargo-nextest + run: | + if [[ ! -x "$CARGO_HOME/bin/cargo-nextest" ]]; then + cargo binstall -y cargo-nextest --force + fi - - name: Building frontend (wasm32) - working-directory: src/webpage - run: trunk build --release --locked --verbose + - name: Run Tests + run: cargo nextest run --locked --verbose - - name: Building tests - run: SKIP_FRONTEND=1 cargo test --no-run --locked --verbose + build-linux-x86_64: + needs: frontend + runs-on: ubuntu-24.04 + env: + SKIP_FRONTEND: 1 + steps: + - uses: actions/checkout@v4 + with: + submodules: "recursive" - - name: Run Tests + - name: Configure sccache + uses: actions/github-script@v7 + with: + script: | + core.exportVariable('ACTIONS_RESULTS_URL', process.env.ACTIONS_RESULTS_URL || ''); + core.exportVariable('ACTIONS_RUNTIME_TOKEN', process.env.ACTIONS_RUNTIME_TOKEN || ''); + + - uses: mozilla-actions/sccache-action@v0.0.9 + with: + disable_annotations: true + + - name: Export sccache environment + shell: bash + run: | + SCCACHE_DIR="${SCCACHE_DIR:-$HOME/.cache/sccache}" + echo "SCCACHE_DIR=$SCCACHE_DIR" >> "$GITHUB_ENV" + mkdir -p "$SCCACHE_DIR" + echo "SCCACHE_GHA_ENABLED=true" >> "$GITHUB_ENV" + + - uses: actions/download-artifact@v4 + with: + name: webpage-dist + path: src/webpage/dist + + - uses: dtolnay/rust-toolchain@stable + with: + targets: x86_64-unknown-linux-musl + + - uses: Swatinem/rust-cache@v2 + with: + shared-key: build-x86_64-unknown-linux-musl + cache-bin: true + cache-on-failure: true + + - name: Add Cargo bin to PATH + run: echo "$CARGO_HOME/bin" >> "$GITHUB_PATH" + + - name: Install build dependencies - Cargo-Binstall + uses: cargo-bins/cargo-binstall@main + + - name: Install build dependencies - Cross + run: | + if [[ ! -x "$CARGO_HOME/bin/cross" ]]; then + cargo binstall -y cross --force + fi + + - name: Building x86_64-unknown-linux-musl + run: | + set -o pipefail + cross build --release --locked --target x86_64-unknown-linux-musl --verbose 2>&1 | tee sccache-build.log + + - name: Report sccache stats + run: bash .github/cross/report-sccache-stats.sh sccache-build.log + + - name: Stage binary + shell: bash run: | - SKIP_FRONTEND=1 cargo test --verbose -- --nocapture - SKIP_FRONTEND=1 cargo test test_udpserver_receive_only --verbose -- --nocapture - SKIP_FRONTEND=1 cargo test test_udpserver_send_only --verbose -- --nocapture --ignored - SKIP_FRONTEND=1 cargo test test_udpclient_send_only --verbose -- --nocapture --ignored - SKIP_FRONTEND=1 cargo test test_tcpserver_receive_only --verbose -- --nocapture --ignored - SKIP_FRONTEND=1 cargo test test_tcpserver_send_only --verbose -- --nocapture --ignored - SKIP_FRONTEND=1 cargo test test_tcpclient_send_only --verbose -- --nocapture --ignored - SKIP_FRONTEND=1 cargo test test_tcpclient_receive_only --verbose -- --nocapture --ignored - SKIP_FRONTEND=1 cargo test test_wsserver_receive_only --verbose -- --nocapture --ignored - SKIP_FRONTEND=1 cargo test test_wsserver_send_only --verbose -- --nocapture --ignored - SKIP_FRONTEND=1 cargo test test_wsclient_send_only --verbose -- --nocapture --ignored - SKIP_FRONTEND=1 cargo test test_wsclient_receive_only --verbose -- --nocapture --ignored + mkdir -p package + cp target/x86_64-unknown-linux-musl/release/${{ github.event.repository.name }} package/${{ github.event.repository.name }} + cp package/${{ github.event.repository.name }} ${{ github.event.repository.name }}-x86_64-unknown-linux-musl + + - uses: actions/upload-artifact@v4 + with: + name: ${{ github.event.repository.name }}-x86_64-unknown-linux-musl + path: package/ + retention-days: 1 + + - uses: svenstaro/upload-release-action@v2 + name: Upload binaries to release + if: ${{ github.event_name == 'push' }} + with: + repo_token: ${{ secrets.GITHUB_TOKEN }} + file: ${{ github.event.repository.name }}-x86_64-unknown-linux-musl + asset_name: ${{ github.event.repository.name }}-x86_64-unknown-linux-musl + tag: ${{ github.ref }} + prerelease: ${{ !startsWith(github.ref, 'refs/tags/') }} + overwrite: true build: + needs: frontend runs-on: ${{ matrix.os }} + env: + SKIP_FRONTEND: 1 strategy: fail-fast: false matrix: @@ -72,9 +240,6 @@ jobs: - os: ubuntu-24.04 TARGET: armv7-unknown-linux-musleabihf - - os: ubuntu-24.04 - TARGET: x86_64-unknown-linux-musl - - os: windows-2025 TARGET: x86_64-pc-windows-msvc EXTENSION: .exe @@ -87,37 +252,75 @@ jobs: with: submodules: "recursive" - # Required for cargo-binstall - - name: Fix openssl on Windows - if: runner.os == 'Windows' - run: | - echo "VCPKG_ROOT=$env:VCPKG_INSTALLATION_ROOT" | Out-File -FilePath $env:GITHUB_ENV -Append - vcpkg install openssl:x64-windows-static-md + - name: Configure sccache + uses: actions/github-script@v7 + with: + script: | + core.exportVariable('ACTIONS_RESULTS_URL', process.env.ACTIONS_RESULTS_URL || ''); + core.exportVariable('ACTIONS_RUNTIME_TOKEN', process.env.ACTIONS_RUNTIME_TOKEN || ''); - - name: Install build dependencies - Rustup + - uses: mozilla-actions/sccache-action@v0.0.9 + with: + disable_annotations: ${{ contains(matrix.TARGET, 'linux') }} + + - name: Export sccache environment + shell: bash run: | - curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- --default-toolchain stable --profile minimal --target ${{ matrix.TARGET }},wasm32-unknown-unknown -y - echo "$HOME/.cargo/bin" >> $GITHUB_PATH + SCCACHE_DIR="${SCCACHE_DIR:-$HOME/.cache/sccache}" + echo "SCCACHE_DIR=$SCCACHE_DIR" >> "$GITHUB_ENV" + mkdir -p "$SCCACHE_DIR" + echo "SCCACHE_GHA_ENABLED=true" >> "$GITHUB_ENV" + if [[ "${{ contains(matrix.TARGET, 'linux') }}" != "true" ]]; then + echo "RUSTC_WRAPPER=$SCCACHE_PATH" >> "$GITHUB_ENV" + "$SCCACHE_PATH" --start-server + fi + + - uses: actions/download-artifact@v4 + with: + name: webpage-dist + path: src/webpage/dist + + - uses: dtolnay/rust-toolchain@stable + with: + targets: ${{ matrix.TARGET }} + + - uses: Swatinem/rust-cache@v2 + with: + shared-key: build-${{ matrix.TARGET }} + cache-bin: true + cache-on-failure: true + + - name: Add Cargo bin to PATH + if: ${{ contains(matrix.TARGET, 'linux') }} + run: echo "$CARGO_HOME/bin" >> "$GITHUB_PATH" - name: Install build dependencies - Cargo-Binstall + if: ${{ contains(matrix.TARGET, 'linux') }} uses: cargo-bins/cargo-binstall@main - - name: Install build dependencies - Cross, Trunk - run: cargo binstall -y trunk cross + - name: Install build dependencies - Cross + if: ${{ contains(matrix.TARGET, 'linux') }} + run: | + if [[ ! -x "$CARGO_HOME/bin/cross" ]]; then + cargo binstall -y cross --force + fi - - name: Building frontend (wasm32) - working-directory: src/webpage - run: trunk build --release --locked --verbose + - name: Building linux for ${{ matrix.TARGET }} + if: ${{ contains(matrix.TARGET, 'linux') }} + run: | + set -o pipefail + cross build --release --locked --target ${{ matrix.TARGET }} --verbose 2>&1 | tee sccache-build.log - - name: Building linux for ${{ matrix.os }} ${{ matrix.target }} + - name: Report sccache stats if: ${{ contains(matrix.TARGET, 'linux') }} - run: SKIP_FRONTEND=1 cross build --release --locked --target ${{matrix.target}} --verbose + run: bash .github/cross/report-sccache-stats.sh sccache-build.log - - name: Building non linux for ${{ matrix.os }} ${{ matrix.target }} + - name: Building non linux for ${{ matrix.TARGET }} if: ${{ !contains(matrix.TARGET, 'linux') }} - run: cross build --release --locked --target ${{matrix.target}} --verbose + run: cargo build --release --locked --target ${{ matrix.TARGET }} --verbose - name: Rename + shell: bash run: cp target/${{ matrix.TARGET }}/release/${{ github.event.repository.name }}${{ matrix.EXTENSION }} ${{ github.event.repository.name }}-${{ matrix.TARGET }}${{ matrix.EXTENSION }} - uses: actions/upload-artifact@v4 @@ -135,3 +338,79 @@ jobs: tag: ${{ github.ref }} prerelease: ${{ !startsWith(github.ref, 'refs/tags/') }} overwrite: true + + test-endpoints: + needs: build-linux-x86_64 + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + include: + - name: "UDP Test" + server_endpoints: "udpout:127.0.0.1:14550 udpin:127.0.0.1:14551" + client_endpoints: "udpin:127.0.0.1:14550 udpout:127.0.0.1:14551" + - name: "TCP Test" + server_endpoints: "tcpout:127.0.0.1:5761 tcpin:127.0.0.1:5760" + client_endpoints: "tcpin:127.0.0.1:5761 tcp:127.0.0.1:5760" + + steps: + - uses: actions/checkout@v4 + with: + submodules: "recursive" + + - uses: actions/download-artifact@v4 + with: + name: ${{ github.event.repository.name }}-x86_64-unknown-linux-musl + path: bin + + - name: Install the latest version of uv + uses: astral-sh/setup-uv@v6 + with: + version: "latest" + + - name: Run mavlink-server + run: | + chmod +x bin/${{ github.event.repository.name }} + ./bin/${{ github.event.repository.name }} ${{ matrix.server_endpoints }} & + sleep 5 + + - name: Run Python Heartbeat Test + run: | + cd tests/scripts + uv run test_heartbeat.py ${{ matrix.client_endpoints }} + + test-arm-vehicle: + needs: build-linux-x86_64 + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + submodules: "recursive" + + - uses: actions/download-artifact@v4 + with: + name: ${{ github.event.repository.name }}-x86_64-unknown-linux-musl + path: bin + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.13" + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y build-essential + + - name: Install the latest version of uv + uses: astral-sh/setup-uv@v6 + with: + version: "latest" + + - name: Run vehicle arming test + env: + MAVLINK_SERVER_BIN: ${{ github.workspace }}/bin/${{ github.event.repository.name }} + run: | + chmod +x "$MAVLINK_SERVER_BIN" + cd tests/python + uv run main.py diff --git a/.github/workflows/test_endpoints.yml b/.github/workflows/test_endpoints.yml deleted file mode 100644 index f195eb83..00000000 --- a/.github/workflows/test_endpoints.yml +++ /dev/null @@ -1,56 +0,0 @@ -name: Mavlink Server Test TCP/UDP - -on: [push, pull_request] - -jobs: - test-mavlink-server: - runs-on: ubuntu-latest - strategy: - fail-fast: false - matrix: - include: - - name: "UDP Test" - server_endpoints: "udpout:127.0.0.1:14550 udpin:127.0.0.1:14551" - client_endpoints: "udpin:127.0.0.1:14550 udpout:127.0.0.1:14551" - - name: "TCP Test" - server_endpoints: "tcpout:127.0.0.1:5761 tcpin:127.0.0.1:5760" - client_endpoints: "tcpin:127.0.0.1:5761 tcp:127.0.0.1:5760" - - steps: - - uses: actions/checkout@v4 - with: - submodules: "recursive" - - - name: Install build dependencies - Rustup - run: | - curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- --default-toolchain stable --profile minimal --target x86_64-unknown-linux-musl,wasm32-unknown-unknown -y - echo "$HOME/.cargo/bin" >> $GITHUB_PATH - - - name: Install the latest version of uv - uses: astral-sh/setup-uv@v6 - with: - version: "latest" - - - name: Install build dependencies - Cargo-Binstall - uses: cargo-bins/cargo-binstall@main - - - name: Install build dependencies - Trunk - run: cargo binstall -y trunk - - - name: Building frontend (wasm32) - working-directory: src/webpage - run: trunk build --release --locked --verbose - - - name: Build mavlink-server - run: | - cargo build --release --locked - - - name: Run mavlink-server - run: | - ./target/release/mavlink-server ${{ matrix.server_endpoints }} & - sleep 5 - - - name: Run Python Heartbeat Test - run: | - cd tests/scripts - uv run test_heartbeat.py ${{ matrix.client_endpoints }} diff --git a/Cargo.lock b/Cargo.lock index 2d0b0690..a423325b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -270,6 +270,15 @@ version = "1.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8c3c1a368f70d6cf7302d78f8f7093da241fb8e8807c05cc9e51a125895a6d5b" +[[package]] +name = "bit-vec" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b71798fca2c1fe1086445a7258a4bc81e6e49dcd24c8d0dd9a1e57395b603f51" +dependencies = [ + "serde", +] + [[package]] name = "bitflags" version = "1.3.2" @@ -408,10 +417,11 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" [[package]] name = "cc" -version = "1.1.37" +version = "1.2.63" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "40545c26d092346d8a8dab71ee48e7685a7a9cba76e634790c215b41a4a7b4cf" +checksum = "556e016178bb5662a08681bbe0f00f8e17631781a4dfc8c45e466e4b185ec27f" dependencies = [ + "find-msvc-tools", "shlex", ] @@ -1009,6 +1019,12 @@ dependencies = [ "windows-sys 0.59.0", ] +[[package]] +name = "find-msvc-tools" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" + [[package]] name = "fixedbitset" version = "0.5.7" @@ -2333,9 +2349,9 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[package]] name = "lz4_flex" -version = "0.11.3" +version = "0.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "75761162ae2b0e580d7e7c390558127e5f01b4194debd6221fd8c207fc80e3f5" +checksum = "8b8c72594ac26bfd34f2d99dfced2edfaddfe8a476e3ff2ca0eb293d925c4f83" dependencies = [ "twox-hash", ] @@ -2825,6 +2841,16 @@ version = "1.0.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a" +[[package]] +name = "pem" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d30c53c26bc5b31a98cd02d20f25a7c8567146caf63ed593a9d87b2775291be" +dependencies = [ + "base64", + "serde_core", +] + [[package]] name = "pem-rfc7468" version = "0.7.0" @@ -3261,6 +3287,20 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "rcgen" +version = "0.14.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57f6d249aad744e274e682777a50283a225a32705394ee6d5fcc01efa25e4055" +dependencies = [ + "pem", + "ring", + "rustls-pki-types", + "time", + "x509-parser", + "yasna", +] + [[package]] name = "redox_syscall" version = "0.5.7" @@ -3332,15 +3372,14 @@ checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c" [[package]] name = "ring" -version = "0.17.8" +version = "0.17.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c17fa4cb658e3583423e915b9f3acc01cceaee1860e33d59ebae66adc3a2dc0d" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" dependencies = [ "cc", "cfg-if", "getrandom 0.2.15", "libc", - "spin", "untrusted", "windows-sys 0.52.0", ] @@ -3853,9 +3892,9 @@ dependencies = [ [[package]] name = "shlex" -version = "1.3.0" +version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" [[package]] name = "signal-hook" @@ -5528,11 +5567,22 @@ dependencies = [ "lazy_static", "nom", "oid-registry", + "ring", "rusticata-macros", "thiserror 2.0.17", "time", ] +[[package]] +name = "yasna" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b5f6765e852b9b4dc8e2a76843e4d64d1cea8e79bcde0b6901aea8e7c7f08282" +dependencies = [ + "bit-vec", + "time", +] + [[package]] name = "yoke" version = "0.7.4" @@ -5559,9 +5609,9 @@ dependencies = [ [[package]] name = "zenoh" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0f9ff8cb89f5267b8486a69466bc42f240f1ee2d5089e72395a23094e7b74f21" +checksum = "85e22d7002ac149ef17fe400bb40a267ebbba40a83413bab03da7762256fa94e" dependencies = [ "ahash", "arc-swap", @@ -5580,7 +5630,6 @@ dependencies = [ "petgraph", "phf", "rand 0.8.5", - "ref-cast", "rustc_version", "serde", "serde_json", @@ -5612,18 +5661,18 @@ dependencies = [ [[package]] name = "zenoh-buffers" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bd9216c3d6c84b56f3e3be634e52365022038e1ac1b9f662f10d425cbf6c0fa8" +checksum = "e89c9e2427102e8efd533716f0935389a3900a818e7334004dd647ac0bd029dc" dependencies = [ "zenoh-collections", ] [[package]] name = "zenoh-codec" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "14bc6747664aa9ecf17becd6e9a29282e535a350cd7c6bd8de7bf2dc662fb93d" +checksum = "31930531a8e387160bc3680c6d62f80a201020cf4d8aa36bd46988b425a66306" dependencies = [ "tracing", "uhlc", @@ -5634,18 +5683,18 @@ dependencies = [ [[package]] name = "zenoh-collections" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d642ecfe0d85f0cd846be9bc92805d926c092a6e6c7a575b6346752f8c3ae16" +checksum = "6fc5195efe3ad44786275f559bbd6f13c6612470e9706c9a9a8b5b47388d51e1" dependencies = [ "ahash", ] [[package]] name = "zenoh-config" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "39765a5f9975aba204c99f2f65308db4952dbea8e5ac79c78ac1eaf5711e970a" +checksum = "e8672f4eaf88fd486f0503c59d19edfc25e7dd689bccd90d3cb731a5f627e0df" dependencies = [ "json5", "nonempty-collections", @@ -5668,9 +5717,9 @@ dependencies = [ [[package]] name = "zenoh-core" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85c0c1388dccf287aec4e9d5e638630dc9d536db9f1da3522889b42697723b9b" +checksum = "1525319e4d9ef2af54fc9c74abf419236e32e6535081339321f3f55e2f34ce2f" dependencies = [ "lazy_static", "tokio", @@ -5680,9 +5729,9 @@ dependencies = [ [[package]] name = "zenoh-crypto" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b433e08df3b03f2af2d23bd29a32aa5f5522c52e66d63e3d135bfa66373736dd" +checksum = "44b80a042fc71419fc4952a90c9cbcfb323c0ced048125d8b44fd362f184045f" dependencies = [ "aes", "hmac", @@ -5694,9 +5743,9 @@ dependencies = [ [[package]] name = "zenoh-keyexpr" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "19a3c47c89cb55ea45a1b3fe7d1fe8682ea93530b1fc5245257812db14b55b3d" +checksum = "80f04c82f0728f6704a1a397a04b38de5b2fd5a9a886a232cf650c9af294ba5d" dependencies = [ "getrandom 0.2.15", "hashbrown 0.16.1", @@ -5710,9 +5759,9 @@ dependencies = [ [[package]] name = "zenoh-link" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6218cecab58435f31fb8b2e185f74f35af8aedd96e8bdd3557b333206b1acfda" +checksum = "a71103cfe96a851ef5ff781d64dda95c70e70208f74e186d6d294ba21e89cc64" dependencies = [ "zenoh-config", "zenoh-link-commons", @@ -5729,15 +5778,18 @@ dependencies = [ [[package]] name = "zenoh-link-commons" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b98adc618f7edb570b9333ce583934a7c63e3a619cb49666515bfc06a000d7b6" +checksum = "8dc91a5163793f842b4b016b2d50640db5a3533370348eb796e7078c576e87cf" dependencies = [ "async-trait", "base64", + "bytes", "flume", "futures", "quinn", + "quinn-proto", + "rcgen", "rustls", "rustls-pemfile", "rustls-pki-types", @@ -5763,56 +5815,43 @@ dependencies = [ [[package]] name = "zenoh-link-quic" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2d0c25d681db958b7714370d5e1d72a523129633d9262b098e18d44824d39893" +checksum = "17d065833147be895b7091cc3e505433c2c8b3172e36c9e06e84a5779a4aa655" dependencies = [ "async-trait", - "base64", - "quinn", - "rustls", - "rustls-pemfile", "rustls-webpki", - "secrecy", "time", - "tokio", - "tokio-util", "tracing", - "webpki-roots", - "zenoh-config", "zenoh-core", "zenoh-link-commons", + "zenoh-link-quic_datagram", "zenoh-protocol", "zenoh-result", - "zenoh-util", ] [[package]] name = "zenoh-link-quic_datagram" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "30cfb427adbc8d367505b9a62a6614c9d7802e420f03466d98644bb7bf24516f" +checksum = "e81b15df31a3d0ddde9a4fb3fbc928e9a2a6d6a4de38bcf55badd3a5208947a4" dependencies = [ "async-trait", - "quinn", - "rustls", "rustls-webpki", "time", - "tokio", "tokio-util", "tracing", "zenoh-core", "zenoh-link-commons", "zenoh-protocol", "zenoh-result", - "zenoh-util", ] [[package]] name = "zenoh-link-tcp" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0f23bd5d06a0014ce5a205961d6d47c8e8d792d9fd050ae9d0c9b609a187995" +checksum = "4e912ac36902173dfc295317e001dc630e5ac9a70c2780f2d2eac500b205a700" dependencies = [ "async-trait", "socket2 0.5.7", @@ -5828,9 +5867,9 @@ dependencies = [ [[package]] name = "zenoh-link-tls" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "17544cde682dbcd2a712114786feee14295c28a9f66fc1021637355511e32fa9" +checksum = "e6f6b308ae2599f9c1344fbcd3418b2c3a3a9fe287ec39501ba834168493ba37" dependencies = [ "async-trait", "base64", @@ -5858,9 +5897,9 @@ dependencies = [ [[package]] name = "zenoh-link-udp" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "587ca1de1caa0b444106d31c26801f4e87b354293d2d37afd8f70d9e793fe35e" +checksum = "ca106ca8c3b7625e7071b9d7408955726e994c8f35fd68e00de8ac58b185d52d" dependencies = [ "async-trait", "libc", @@ -5872,6 +5911,7 @@ dependencies = [ "zenoh-buffers", "zenoh-core", "zenoh-link-commons", + "zenoh-link-quic_datagram", "zenoh-protocol", "zenoh-result", "zenoh-sync", @@ -5880,9 +5920,9 @@ dependencies = [ [[package]] name = "zenoh-link-unixsock_stream" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac46470a17af5861e07ed9d2440277e7a3dea55109b0988866b635f7daf29ac4" +checksum = "09f2b7f60cdb5ad771d1a4f8e2eda15529e96dbe81c0ad2c1015b4c194347676" dependencies = [ "async-trait", "nix 0.29.0", @@ -5899,9 +5939,9 @@ dependencies = [ [[package]] name = "zenoh-link-ws" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e4be09c3e32d510cdddf3289bc333d53471883198fca51d58248458a20df3d51" +checksum = "9a7fb54899f7fbdfc4fd02e647f9bba0d6e089cca4355acafb9499c510ebea79" dependencies = [ "async-trait", "futures-util", @@ -5920,9 +5960,9 @@ dependencies = [ [[package]] name = "zenoh-macros" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b760a458cd906ac888b37fd1abdb21a0f58ecc64cc3882f83a976cb5ca8e0632" +checksum = "9310b02a8f6dc4bd04d9ce6b318b9d00182aeeeeca60410003307d63a2569a3f" dependencies = [ "proc-macro2", "quote", @@ -5932,9 +5972,9 @@ dependencies = [ [[package]] name = "zenoh-plugin-trait" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7325b773c43a86a94f800cb971ab7e4b7e01ce76819c9c100ea783a47c3a25e4" +checksum = "e235815d14b22448aa1db948560328761530932fa7a2f9a3c14853b3f0267941" dependencies = [ "git-version", "libloading", @@ -5950,9 +5990,9 @@ dependencies = [ [[package]] name = "zenoh-protocol" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b4d3dad7aeeea780495692b195cd56515569c32b76b9dd077cc408c3ebca03f" +checksum = "eeab45020bbecc077f14f06ee8f5aee65ce760af72481e663cac58b5dbfa66dd" dependencies = [ "const_format", "rand 0.8.5", @@ -5960,23 +6000,24 @@ dependencies = [ "uhlc", "zenoh-buffers", "zenoh-keyexpr", + "zenoh-macros", "zenoh-result", ] [[package]] name = "zenoh-result" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b4dbfea68b947a790d5525bcf061e91e2fdc2798bce619851919b353a8580fa" +checksum = "cca8e65b08f211833fe31cf38d73a48c6e1d6d900914e1ddd8cb176b3355b75b" dependencies = [ "anyhow", ] [[package]] name = "zenoh-runtime" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "760a1f7880f98427ad849d600257d1455a18afe981681f362684a3f91042537e" +checksum = "dd3d0c1558f909c9a74bde5e398c5733f81eb954818451baac2a6c09f48d6a5e" dependencies = [ "lazy_static", "ron", @@ -5989,9 +6030,9 @@ dependencies = [ [[package]] name = "zenoh-shm" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "42184ae820d64d40814c0dd3b48f748c55a3ff5591c524f1bde36f21ccfda488" +checksum = "a63670c8845775b21f718401a317c7824c604915fec6d228570456bb919994e7" dependencies = [ "advisory-lock", "async-trait", @@ -6019,9 +6060,9 @@ dependencies = [ [[package]] name = "zenoh-stats" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8954c79580ed3365580d86f40657f2cd6bf4b4c72dd463de936368df8c709c65" +checksum = "e2bd56586221cca3cbb75b8bfe6f9451219a9241b0a49228b498467705d7e10d" dependencies = [ "ahash", "prometheus-client", @@ -6033,9 +6074,9 @@ dependencies = [ [[package]] name = "zenoh-sync" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "98f132137bb003f10b7fff086cb18addf8e8273b9c0d2722a53b5074c8a79965" +checksum = "9588f87db82b414a3e73d13312026be826332f53d270966e3e19f77f7fa1f06d" dependencies = [ "arc-swap", "event-listener", @@ -6048,9 +6089,9 @@ dependencies = [ [[package]] name = "zenoh-task" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1b17d10136fdabec7e21a3fcef568c210ee6a2d71cde6adcde99e9236584f3a1" +checksum = "1e2ac601598a27152b366ba1c6825216f01f77bb898f719b7752099a3594ce72" dependencies = [ "futures", "tokio", @@ -6062,13 +6103,14 @@ dependencies = [ [[package]] name = "zenoh-transport" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "50739b4c45e0963df8377abddb74701a4b708b178590eae92f27a604e25daf44" +checksum = "80800c4adc26dbe81418735068541cf39820a95ec988114f04dd014775ba7c97" dependencies = [ "async-trait", "crossbeam-utils", "flume", + "futures", "lazy_static", "lz4_flex", "rand 0.8.5", @@ -6098,9 +6140,9 @@ dependencies = [ [[package]] name = "zenoh-util" -version = "1.7.2" +version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9512987c13925d32d3331507c8807853d5b682ea8da94d0ba6534c7a8ace48aa" +checksum = "1b10369df18a781a3e675c9a2cbf54adb44c9dc2a376c1014c5e488410df2179" dependencies = [ "async-trait", "const_format", diff --git a/Cargo.toml b/Cargo.toml index a1e6eead..5d75cad6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -57,7 +57,7 @@ tracing-subscriber = { version = "0.3.22", features = ["env-filter"] } tracing-log = "0.2.0" tracing-appender = "0.2.3" toml = "0.8.19" -zenoh = { version = "1.7.2", features = ["shared-memory"] } +zenoh = { version = "1.9.0", features = ["shared-memory"] } ringbuffer = "0.15.0" strum = "0.27" strum_macros = "0.27" @@ -70,6 +70,7 @@ sysinfo = "0.35.0" [dev-dependencies] criterion = "0.5" tokio = { version = "1", features = ["full"] } +zenoh = { version = "1.9.0", features = ["shared-memory"] } [build-dependencies] vergen-gix = { version = "1.0.2", features = ["build", "cargo"] } diff --git a/Cross.toml b/Cross.toml index 97d30df1..8e458c31 100644 --- a/Cross.toml +++ b/Cross.toml @@ -7,5 +7,26 @@ image = "joaoantoniocardoso/cross-rs:armv7-unknown-linux-gnueabihf-bullseye-slim [target.aarch64-unknown-linux-gnu] image = "joaoantoniocardoso/cross-rs:aarch64-unknown-linux-gnu-bullseye-slim-with-gstreamer" +[target.x86_64-unknown-linux-musl] +dockerfile = ".github/cross/Dockerfile.x86_64-unknown-linux-musl-sccache" + +[target.arm-unknown-linux-musleabihf] +dockerfile = ".github/cross/Dockerfile.arm-unknown-linux-musleabihf-sccache" + +[target.aarch64-unknown-linux-musl] +dockerfile = ".github/cross/Dockerfile.aarch64-unknown-linux-musl-sccache" + +[target.armv7-unknown-linux-musleabihf] +dockerfile = ".github/cross/Dockerfile.armv7-unknown-linux-musleabihf-sccache" + [build.env] -passthrough = ["SKIP_FRONTEND"] +passthrough = [ + "SKIP_FRONTEND", + "SCCACHE_DIR", + "SCCACHE_GHA_ENABLED", + "SCCACHE_LOG", + "ACTIONS_RESULTS_URL", + "ACTIONS_RUNTIME_TOKEN", + "ACTIONS_CACHE_SERVICE_V2", +] +volumes = ["SCCACHE_DIR"] diff --git a/src/lib/cli.rs b/src/lib/cli.rs index 8928f0bb..ee21d8f8 100644 --- a/src/lib/cli.rs +++ b/src/lib/cli.rs @@ -298,6 +298,8 @@ mod tests { true, ), ("tlogwriter:/tmp/little_potato.tlog", true), + ("zenoh://127.0.0.1:7447", true), + ("zenohraw://127.0.0.1:7447", true), ]; for (endpoint, expected) in endpoints { diff --git a/src/lib/drivers/mod.rs b/src/lib/drivers/mod.rs index 0d03b880..138ce1ce 100644 --- a/src/lib/drivers/mod.rs +++ b/src/lib/drivers/mod.rs @@ -34,6 +34,7 @@ pub enum Type { WebSocketClient, WebSocketServer, Zenoh, + ZenohRaw, } #[derive(Copy, Clone, Debug, Default, Eq, Hash, PartialEq, strum_macros::EnumString)] @@ -270,9 +271,13 @@ pub fn endpoints() -> Vec { typ: Type::FakeSource, }, ExtInfo { - driver_ext: Box::new(zenoh::ZenohInfo), + driver_ext: Box::new(zenoh::json::ZenohInfo), typ: Type::Zenoh, }, + ExtInfo { + driver_ext: Box::new(zenoh::raw::ZenohRawInfo), + typ: Type::ZenohRaw, + }, ] } diff --git a/src/lib/drivers/tlog/reader.rs b/src/lib/drivers/tlog/reader.rs index ec812aae..2c21f8a9 100644 --- a/src/lib/drivers/tlog/reader.rs +++ b/src/lib/drivers/tlog/reader.rs @@ -279,7 +279,7 @@ mod tests { let file_v2_messages = 30437; let file_messages = file_v2_messages; let mut total_messages_read = 0; - let timeout_time = tokio::time::Duration::from_secs(1); + let timeout_time = tokio::time::Duration::from_secs(30); let res = tokio::time::timeout(timeout_time, async { loop { let messages_received_per_id = messages_received_per_id.read().await.clone(); diff --git a/src/lib/drivers/zenoh/json.rs b/src/lib/drivers/zenoh/json.rs new file mode 100644 index 00000000..9f23a80c --- /dev/null +++ b/src/lib/drivers/zenoh/json.rs @@ -0,0 +1,338 @@ +use std::{collections::HashMap, sync::Arc}; + +use anyhow::Result; +use mavlink::{self, Message}; +use tokio::sync::{RwLock, broadcast}; +use tracing::*; +use zenoh; + +use crate::{ + callbacks::{Callbacks, MessageCallback}, + drivers::{Driver, DriverInfo, generic_tasks::SendReceiveContext}, + mavlink_json::MAVLinkJSON, + protocol::Protocol, + stats::{ + accumulated::driver::{AccumulatedDriverStats, AccumulatedDriverStatsProvider}, + driver::DriverUuid, + }, +}; + +const TOPIC_PREFIX: &str = "mavlink"; + +#[derive(Debug)] +pub struct Zenoh { + name: arc_swap::ArcSwap, + uuid: DriverUuid, + on_message_input: Callbacks>, + on_message_output: Callbacks>, + stats: Arc>, +} + +pub struct ZenohBuilder(Zenoh); + +impl ZenohBuilder { + pub fn build(self) -> Zenoh { + self.0 + } + + pub fn on_message_input(self, callback: C) -> Self + where + C: MessageCallback>, + { + self.0.on_message_input.add_callback(callback.into_boxed()); + self + } + + pub fn on_message_output(self, callback: C) -> Self + where + C: MessageCallback>, + { + self.0.on_message_output.add_callback(callback.into_boxed()); + self + } +} + +impl Zenoh { + #[instrument(level = "debug")] + pub fn builder(name: &str) -> ZenohBuilder { + let name = Arc::new(name.to_string()); + + ZenohBuilder(Self { + name: arc_swap::ArcSwap::new(name.clone()), + uuid: Self::generate_uuid(&name), + on_message_input: Callbacks::default(), + on_message_output: Callbacks::default(), + stats: Arc::new(RwLock::new(AccumulatedDriverStats::new(name, &ZenohInfo))), + }) + } + + #[instrument(level = "debug", skip_all)] + async fn receive_task( + context: &SendReceiveContext, + session: Arc, + ) -> Result<()> { + let subscriber = match session + .declare_subscriber(format!("{TOPIC_PREFIX}/in")) + .await + { + Ok(subscriber) => subscriber, + Err(error) => { + return Err(anyhow::anyhow!( + "Failed to create subscriber for mavlink data: {error:?}" + )); + } + }; + + while let Ok(sample) = subscriber.recv_async().await { + let Ok(content) = json5::from_str::>( + std::str::from_utf8(&sample.payload().to_bytes()).unwrap(), + ) else { + debug!("Failed to parse message, not a valid MAVLinkMessage: {sample:?}"); + continue; + }; + + let bus_message = Arc::new(Protocol::from_mavlink_raw( + content.header.inner, + &content.message, + "zenoh", + )); + + trace!("Received message: {bus_message:?}"); + + context.stats.write().await.stats.update_input(&bus_message); + + for future in context.on_message_input.call_all(bus_message.clone()) { + if let Err(error) = future.await { + debug!("Dropping message: on_message_input callback returned error: {error:?}"); + continue; + } + } + + if let Err(error) = context.hub_sender.send(bus_message) { + error!("Failed to send message to hub: {error:?}"); + continue; + } + + trace!("Message sent to hub"); + } + + debug!("Driver receiver task stopped!"); + + Ok(()) + } + + #[instrument(level = "debug", skip_all)] + async fn send_task(context: &SendReceiveContext, session: Arc) -> Result<()> { + let mut hub_receiver = context.hub_sender.subscribe(); + let mut publishers = HashMap::new(); + + 'mainloop: loop { + let message = match hub_receiver.recv().await { + Ok(message) => message, + Err(broadcast::error::RecvError::Closed) => { + error!("Hub channel closed!"); + break; + } + Err(broadcast::error::RecvError::Lagged(count)) => { + warn!("Channel lagged by {count} messages."); + continue; + } + }; + + if message.origin.eq("zenoh") { + continue; // Don't do loopback + } + + context.stats.write().await.stats.update_output(&message); + + for future in context.on_message_output.call_all(message.clone()) { + if let Err(error) = future.await { + debug!( + "Dropping message: on_message_output callback returned error: {error:?}" + ); + continue 'mainloop; + } + } + + let Ok(mavlink_json) = message + .to_mavlink_json::() + .await + else { + continue; + }; + + let message_name = mavlink_json.message.message_name(); + + let json_string = &match json5::to_string(&mavlink_json) { + Ok(json) => json, + Err(error) => { + error!("Failed to transform mavlink message {message_name} to json: {error:?}"); + continue; + } + }; + + let out_topic_name = format!("{TOPIC_PREFIX}/out"); + Self::publish_json(&session, &mut publishers, &out_topic_name, json_string).await; + + let header = &mavlink_json.header.inner; + let message_topic_name = format!( + "mavlink/{}/{}/{}", + header.system_id, header.component_id, message_name + ); + Self::publish_json(&session, &mut publishers, &message_topic_name, json_string).await; + + // for each key inside mavlink_json and publish under topic_name/field_name + let message_value = serde_json::to_value(&mavlink_json.message).unwrap(); + for (field_name, field_value) in message_value.as_object().unwrap() { + let field_topic_name = format!("{message_topic_name}/{field_name}"); + let field_json = json5::to_string(field_value).unwrap(); + Self::publish_json(&session, &mut publishers, &field_topic_name, &field_json).await; + } + } + + debug!("Driver sender task stopped!"); + + Ok(()) + } + + async fn publish_json( + session: &zenoh::Session, + publishers: &mut HashMap>, + topic_name: &str, + payload: &str, + ) { + if !publishers.contains_key(topic_name) { + let topic = topic_name.to_string(); + let key_expr = match zenoh::key_expr::KeyExpr::try_from(topic.clone()) { + Ok(key_expr) => key_expr, + Err(error) => { + error!("Failed to create key expression for {topic_name}: {error:?}"); + return; + } + }; + + match session.declare_publisher(key_expr).await { + Ok(publisher) => { + publishers.insert(topic, publisher); + } + Err(error) => { + error!("Failed to create publisher for {topic_name}: {error:?}"); + return; + } + } + } + + let Some(publisher) = publishers.get(topic_name) else { + return; + }; + + if let Err(error) = publisher + .put(payload) + .encoding(zenoh::bytes::Encoding::APPLICATION_JSON) + .await + { + error!("Failed to send message to {topic_name}: {error:?}"); + } else { + trace!("Message sent to {topic_name}: {payload:?}"); + } + } +} + +#[async_trait::async_trait] +impl Driver for Zenoh { + #[instrument(level = "debug", skip(self, hub_sender))] + async fn run(&self, hub_sender: broadcast::Sender>) -> Result<()> { + let context = SendReceiveContext { + direction: crate::drivers::Direction::Both, + hub_sender, + on_message_output: self.on_message_output.clone(), + on_message_input: self.on_message_input.clone(), + stats: self.stats.clone(), + }; + + let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(1)); + let mut first = true; + loop { + if first { + first = false; + } else { + interval.tick().await; + } + + debug!("Trying to connect..."); + + let session = super::session().await; + + debug!("Successfully connected"); + + tokio::select! { + result = Zenoh::send_task(&context, session.clone()) => { + if let Err(error) = result { + error!("Error in send task: {error:?}"); + } + } + result = Zenoh::receive_task(&context, session) => { + if let Err(error) = result { + error!("Error in receive task: {error:?}"); + } + } + } + + debug!("Restarting connection loop..."); + } + } + + #[instrument(level = "debug", skip(self))] + fn info(&self) -> Box { + Box::new(ZenohInfo) + } + + fn name(&self) -> Arc { + self.name.load_full() + } + + fn uuid(&self) -> &DriverUuid { + &self.uuid + } +} + +#[async_trait::async_trait] +impl AccumulatedDriverStatsProvider for Zenoh { + async fn stats(&self) -> AccumulatedDriverStats { + self.stats.read().await.clone() + } + + async fn reset_stats(&self) { + let mut stats = self.stats.write().await; + stats.stats.input = None; + stats.stats.output = None + } +} + +pub struct ZenohInfo; +impl DriverInfo for ZenohInfo { + fn name(&self) -> &'static str { + "Zenoh" + } + fn valid_schemes(&self) -> &'static [&'static str] { + &["zenoh"] + } + + fn cli_example_legacy(&self) -> Vec { + let first_schema = self.valid_schemes()[0]; + + vec![format!("{first_schema}::")] + } + + fn cli_example_url(&self) -> Vec { + let first_schema = &self.valid_schemes()[0]; + vec![format!("{first_schema}://:").to_string()] + } + + fn create_endpoint_from_url(&self, url: &url::Url) -> Option> { + println!("{}", &url); + let _host = url.host_str().unwrap(); + let _port = url.port().unwrap(); + Some(Arc::new(Zenoh::builder("Zenoh").build())) + } +} diff --git a/src/lib/drivers/zenoh/mod.rs b/src/lib/drivers/zenoh/mod.rs index 19fd69f2..989112f6 100644 --- a/src/lib/drivers/zenoh/mod.rs +++ b/src/lib/drivers/zenoh/mod.rs @@ -1,335 +1,103 @@ +pub mod json; +pub mod raw; + use std::sync::Arc; -use anyhow::Result; -use mavlink::{self, Message}; -use tokio::sync::{RwLock, broadcast}; +use tokio::sync::{OnceCell, watch}; use tracing::*; use zenoh; -use crate::{ - callbacks::{Callbacks, MessageCallback}, - cli::zenoh_config_file, - drivers::{Driver, DriverInfo, generic_tasks::SendReceiveContext}, - mavlink_json::MAVLinkJSON, - protocol::Protocol, - stats::{ - accumulated::driver::{AccumulatedDriverStats, AccumulatedDriverStatsProvider}, - driver::DriverUuid, - }, -}; - -#[derive(Debug)] -pub struct Zenoh { - name: arc_swap::ArcSwap, - uuid: DriverUuid, - on_message_input: Callbacks>, - on_message_output: Callbacks>, - stats: Arc>, -} - -pub struct ZenohBuilder(Zenoh); - -impl ZenohBuilder { - pub fn build(self) -> Zenoh { - self.0 - } - - pub fn on_message_input(self, callback: C) -> Self - where - C: MessageCallback>, - { - self.0.on_message_input.add_callback(callback.into_boxed()); - self - } - - pub fn on_message_output(self, callback: C) -> Self - where - C: MessageCallback>, - { - self.0.on_message_output.add_callback(callback.into_boxed()); - self - } -} - -impl Zenoh { - #[instrument(level = "debug")] - pub fn builder(name: &str) -> ZenohBuilder { - let name = Arc::new(name.to_string()); - - ZenohBuilder(Self { - name: arc_swap::ArcSwap::new(name.clone()), - uuid: Self::generate_uuid(&name), - on_message_input: Callbacks::default(), - on_message_output: Callbacks::default(), - stats: Arc::new(RwLock::new(AccumulatedDriverStats::new(name, &ZenohInfo))), - }) - } - - #[instrument(level = "debug", skip_all)] - async fn receive_task( - context: &SendReceiveContext, - session: Arc, - ) -> Result<()> { - let subscriber = match session - .declare_subscriber(format!("{}/in", "mavlink")) - .await - { - Ok(subscriber) => subscriber, - Err(error) => { - return Err(anyhow::anyhow!( - "Failed to create subscriber for mavlink data: {error:?}" - )); - } - }; - - while let Ok(sample) = subscriber.recv_async().await { - let Ok(content) = json5::from_str::>( - std::str::from_utf8(&sample.payload().to_bytes()).unwrap(), - ) else { - debug!("Failed to parse message, not a valid MAVLinkMessage: {sample:?}"); - continue; - }; - - let bus_message = Arc::new(Protocol::from_mavlink_raw( - content.header.inner, - &content.message, - "zenoh", - )); - - trace!("Received message: {bus_message:?}"); - - context.stats.write().await.stats.update_input(&bus_message); - - for future in context.on_message_input.call_all(bus_message.clone()) { - if let Err(error) = future.await { - debug!("Dropping message: on_message_input callback returned error: {error:?}"); - continue; - } - } - - if let Err(error) = context.hub_sender.send(bus_message) { - error!("Failed to send message to hub: {error:?}"); - continue; - } - - trace!("Message sent to hub"); - } - - debug!("Driver receiver task stopped!"); - - Ok(()) - } - - #[instrument(level = "debug", skip_all)] - async fn send_task(context: &SendReceiveContext, session: Arc) -> Result<()> { - let mut hub_receiver = context.hub_sender.subscribe(); - - 'mainloop: loop { - let message = match hub_receiver.recv().await { - Ok(message) => message, - Err(broadcast::error::RecvError::Closed) => { - error!("Hub channel closed!"); - break; - } - Err(broadcast::error::RecvError::Lagged(count)) => { - warn!("Channel lagged by {count} messages."); - continue; - } - }; - - if message.origin.eq("zenoh") { - continue; // Don't do loopback - } - - context.stats.write().await.stats.update_output(&message); - - for future in context.on_message_output.call_all(message.clone()) { - if let Err(error) = future.await { - debug!( - "Dropping message: on_message_output callback returned error: {error:?}" - ); - continue 'mainloop; - } - } - - let Ok(mavlink_json) = message - .to_mavlink_json::() - .await - else { - continue; - }; - - let message_name = mavlink_json.message.message_name(); +use crate::cli::zenoh_config_file; - let json_string = &match json5::to_string(&mavlink_json) { - Ok(json) => json, - Err(error) => { - error!("Failed to transform mavlink message {message_name} to json: {error:?}"); - continue; - } - }; - - let topic_name = "mavlink/out"; - if let Err(error) = session - .put(topic_name, json_string) - .encoding(zenoh::bytes::Encoding::APPLICATION_JSON) - .await - { - error!("Failed to send message to {topic_name}: {error:?}"); - } else { - trace!("Message sent to {topic_name}: {json_string:?}"); - } - - let header = &mavlink_json.header.inner; - let topic_name = &format!( - "mavlink/{}/{}/{}", - header.system_id, header.component_id, message_name - ); - if let Err(error) = session - .put(topic_name, json_string) - .encoding(zenoh::bytes::Encoding::APPLICATION_JSON) - .await - { - error!("Failed to send message to {topic_name}: {error:?}"); - } else { - trace!("Message sent to {topic_name}: {json_string:?}"); - } - - // for each key inside mavlink_json and publish under topic_name/field_name - let message_value = serde_json::to_value(&mavlink_json.message).unwrap(); - for (field_name, field_value) in message_value.as_object().unwrap() { - let topic_name = &format!("{}/{}", topic_name, field_name); - if let Err(error) = session - .put(topic_name, json5::to_string(field_value).unwrap()) - .encoding(zenoh::bytes::Encoding::APPLICATION_JSON) - .await - { - error!("Failed to send message to {topic_name}: {error:?}"); - } - } - } - - debug!("Driver sender task stopped!"); - - Ok(()) - } -} - -#[async_trait::async_trait] -impl Driver for Zenoh { - #[instrument(level = "debug", skip(self, hub_sender))] - async fn run(&self, hub_sender: broadcast::Sender>) -> Result<()> { - let context = SendReceiveContext { - direction: crate::drivers::Direction::Both, - hub_sender, - on_message_output: self.on_message_output.clone(), - on_message_input: self.on_message_input.clone(), - stats: self.stats.clone(), - }; - - let mut config = if let Some(zenoh_config_file) = zenoh_config_file() { - zenoh::Config::from_file(zenoh_config_file) - .map_err(|error| anyhow::anyhow!("Failed to load Zenoh config file: {error:?}"))? - } else { - let mut config = zenoh::Config::default(); - config - .insert_json5("mode", r#""client""#) - .expect("Failed to insert client mode"); - config - .insert_json5("connect/endpoints", r#"["tcp/127.0.0.1:7447"]"#) - .expect("Failed to insert connect endpoints"); - config - }; +static SESSION_RX: OnceCell>>> = OnceCell::const_new(); +#[instrument(level = "debug")] +pub(crate) fn build_config() -> anyhow::Result { + let mut config = if let Some(zenoh_config_file) = zenoh_config_file() { + zenoh::Config::from_file(zenoh_config_file) + .map_err(|error| anyhow::anyhow!("Failed to load Zenoh config file: {error:?}"))? + } else { + let mut config = zenoh::Config::default(); + config + .insert_json5("mode", r#""client""#) + .expect("Failed to insert client mode"); config - .insert_json5("adminspace", r#"{"enabled": true}"#) - .expect("Failed to insert adminspace"); + .insert_json5("connect/endpoints", r#"["tcp/127.0.0.1:7447"]"#) + .expect("Failed to insert connect endpoints"); config - .insert_json5("metadata", r#"{"name": "mavlink-server"}"#) - .expect("Failed to insert metadata"); + }; - let mut first = true; - loop { - if first { - first = false; - } else { - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; - } + config + .insert_json5("adminspace", r#"{"enabled": true}"#) + .expect("Failed to insert adminspace"); + config + .insert_json5("metadata", r#"{"name": "mavlink-server"}"#) + .expect("Failed to insert metadata"); - let session = match zenoh::open(config.clone()).await { - Ok(session) => Arc::new(session), - Err(error) => { - error!("Failed to start zenoh session: {error:?}"); - continue; - } - }; + Ok(config) +} - tokio::select! { - result = Zenoh::send_task(&context, session.clone()) => { - if let Err(error) = result { - error!("Error in send task: {error:?}"); +#[instrument(level = "debug", skip_all)] +async fn init_session_manager() { + SESSION_RX + .get_or_init(|| async { + let (tx, rx) = watch::channel(None); + + tokio::spawn(async move { + let config = match build_config() { + Ok(config) => config, + Err(error) => { + error!("Failed to build Zenoh config: {error:?}"); + return; } - } - result = Zenoh::receive_task(&context, session) => { - if let Err(error) = result { - error!("Error in receive task: {error:?}"); + }; + + let mut first = true; + loop { + if first { + first = false; + } else { + tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; } - } - } - } - } - #[instrument(level = "debug", skip(self))] - fn info(&self) -> Box { - Box::new(ZenohInfo) - } - - fn name(&self) -> Arc { - self.name.load_full() - } - - fn uuid(&self) -> &DriverUuid { - &self.uuid - } -} - -#[async_trait::async_trait] -impl AccumulatedDriverStatsProvider for Zenoh { - async fn stats(&self) -> AccumulatedDriverStats { - self.stats.read().await.clone() - } + match zenoh::open(config.clone()).await { + Ok(session) => { + if tx.send(Some(Arc::new(session))).is_err() { + break; + } + + loop { + tokio::time::sleep(tokio::time::Duration::from_secs(86400)).await; + } + } + Err(error) => { + error!("Failed to start zenoh session: {error:?}"); + } + } + } + }); - async fn reset_stats(&self) { - let mut stats = self.stats.write().await; - stats.stats.input = None; - stats.stats.output = None - } + rx + }) + .await; } -pub struct ZenohInfo; -impl DriverInfo for ZenohInfo { - fn name(&self) -> &'static str { - "Zenoh" - } - fn valid_schemes(&self) -> &'static [&'static str] { - &["zenoh"] - } +#[instrument(level = "debug", skip_all)] +pub(crate) async fn session() -> Arc { + init_session_manager().await; - fn cli_example_legacy(&self) -> Vec { - let first_schema = self.valid_schemes()[0]; + let mut rx = SESSION_RX + .get() + .expect("Zenoh session manager is not initialized") + .clone(); - vec![format!("{first_schema}::")] - } - - fn cli_example_url(&self) -> Vec { - let first_schema = &self.valid_schemes()[0]; - vec![format!("{first_schema}://:").to_string()] - } + loop { + if let Some(session) = rx.borrow_and_update().clone() { + return session; + } - fn create_endpoint_from_url(&self, url: &url::Url) -> Option> { - println!("{}", &url); - let _host = url.host_str().unwrap(); - let _port = url.port().unwrap(); - Some(Arc::new(Zenoh::builder("Zenoh").build())) + if rx.changed().await.is_err() { + panic!("Zenoh session manager stopped"); + } } } diff --git a/src/lib/drivers/zenoh/raw.rs b/src/lib/drivers/zenoh/raw.rs new file mode 100644 index 00000000..ed554bf4 --- /dev/null +++ b/src/lib/drivers/zenoh/raw.rs @@ -0,0 +1,293 @@ +use std::sync::Arc; + +use anyhow::Result; +use bytes::BytesMut; +use mavlink_codec::codec::MavlinkCodec; +use tokio::sync::{RwLock, broadcast}; +use tokio_util::codec::Decoder; +use tracing::*; +use zenoh; + +use crate::{ + callbacks::{Callbacks, MessageCallback}, + drivers::{Driver, DriverInfo, generic_tasks::SendReceiveContext}, + protocol::Protocol, + stats::{ + accumulated::driver::{AccumulatedDriverStats, AccumulatedDriverStatsProvider}, + driver::DriverUuid, + }, +}; + +const TOPIC_PREFIX: &str = "mavlink_raw"; + +#[derive(Debug)] +pub struct ZenohRaw { + name: arc_swap::ArcSwap, + uuid: DriverUuid, + on_message_input: Callbacks>, + on_message_output: Callbacks>, + stats: Arc>, +} + +pub struct ZenohRawBuilder(ZenohRaw); + +impl ZenohRawBuilder { + pub fn build(self) -> ZenohRaw { + self.0 + } + + pub fn on_message_input(self, callback: C) -> Self + where + C: MessageCallback>, + { + self.0.on_message_input.add_callback(callback.into_boxed()); + self + } + + pub fn on_message_output(self, callback: C) -> Self + where + C: MessageCallback>, + { + self.0.on_message_output.add_callback(callback.into_boxed()); + self + } +} + +impl ZenohRaw { + #[instrument(level = "debug")] + pub fn builder(name: &str) -> ZenohRawBuilder { + let name = Arc::new(name.to_string()); + + ZenohRawBuilder(Self { + name: arc_swap::ArcSwap::new(name.clone()), + uuid: Self::generate_uuid(&name), + on_message_input: Callbacks::default(), + on_message_output: Callbacks::default(), + stats: Arc::new(RwLock::new(AccumulatedDriverStats::new( + name, + &ZenohRawInfo, + ))), + }) + } + + #[instrument(level = "debug", skip_all)] + async fn receive_task( + context: &SendReceiveContext, + session: Arc, + codec: &mut MavlinkCodec, + buffer: &mut BytesMut, + ) -> Result<()> { + let subscriber = match session + .declare_subscriber(format!("{TOPIC_PREFIX}/in")) + .await + { + Ok(subscriber) => subscriber, + Err(error) => { + return Err(anyhow::anyhow!( + "Failed to create subscriber for mavlink raw data: {error:?}" + )); + } + }; + + while let Ok(sample) = subscriber.recv_async().await { + buffer.extend_from_slice(&sample.payload().to_bytes()); + + loop { + let packet = match codec.decode(buffer) { + Ok(Some(Ok(packet))) => packet, + Ok(Some(Err(decode_error))) => { + error!("Failed to decode packet: {decode_error:?}"); + continue; + } + Ok(None) => break, + Err(error) => { + error!("Failed to decode packet: {error:?}"); + continue; + } + }; + + let bus_message = Arc::new(Protocol::new("zenohraw", packet)); + + trace!("Received message: {bus_message:?}"); + + context.stats.write().await.stats.update_input(&bus_message); + + for future in context.on_message_input.call_all(bus_message.clone()) { + if let Err(error) = future.await { + debug!( + "Dropping message: on_message_input callback returned error: {error:?}" + ); + continue; + } + } + + if let Err(error) = context.hub_sender.send(bus_message) { + error!("Failed to send message to hub: {error:?}"); + continue; + } + + trace!("Message sent to hub"); + } + } + + debug!("Driver receiver task stopped!"); + + Ok(()) + } + + #[instrument(level = "debug", skip_all)] + async fn send_task(context: &SendReceiveContext, session: Arc) -> Result<()> { + let mut hub_receiver = context.hub_sender.subscribe(); + let topic_name = format!("{TOPIC_PREFIX}/out"); + + let publisher = match session.declare_publisher(&topic_name).await { + Ok(publisher) => publisher, + Err(error) => { + return Err(anyhow::anyhow!( + "Failed to create publisher for mavlink raw data: {error:?}" + )); + } + }; + + 'mainloop: loop { + let message = match hub_receiver.recv().await { + Ok(message) => message, + Err(broadcast::error::RecvError::Closed) => { + error!("Hub channel closed!"); + break; + } + Err(broadcast::error::RecvError::Lagged(count)) => { + warn!("Channel lagged by {count} messages."); + continue; + } + }; + + if message.origin.eq("zenohraw") { + continue; // Don't do loopback + } + + context.stats.write().await.stats.update_output(&message); + + for future in context.on_message_output.call_all(message.clone()) { + if let Err(error) = future.await { + debug!( + "Dropping message: on_message_output callback returned error: {error:?}" + ); + continue 'mainloop; + } + } + + if let Err(error) = publisher.put(message.as_slice()).await { + error!("Failed to send message to {topic_name}: {error:?}"); + } else { + trace!("Message sent to {topic_name}: {:?}", message.as_slice()); + } + } + + debug!("Driver sender task stopped!"); + + Ok(()) + } +} + +#[async_trait::async_trait] +impl Driver for ZenohRaw { + #[instrument(level = "debug", skip(self, hub_sender))] + async fn run(&self, hub_sender: broadcast::Sender>) -> Result<()> { + let context = SendReceiveContext { + direction: crate::drivers::Direction::Both, + hub_sender, + on_message_output: self.on_message_output.clone(), + on_message_input: self.on_message_input.clone(), + stats: self.stats.clone(), + }; + + let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(1)); + let mut first = true; + loop { + if first { + first = false; + } else { + interval.tick().await; + } + + debug!("Trying to connect..."); + + let session = super::session().await; + + debug!("Successfully connected"); + + let mut codec = MavlinkCodec::::default(); + let mut buffer = BytesMut::new(); + + tokio::select! { + result = ZenohRaw::send_task(&context, session.clone()) => { + if let Err(error) = result { + error!("Error in send task: {error:?}"); + } + } + result = ZenohRaw::receive_task(&context, session, &mut codec, &mut buffer) => { + if let Err(error) = result { + error!("Error in receive task: {error:?}"); + } + } + } + + debug!("Restarting connection loop..."); + } + } + + #[instrument(level = "debug", skip(self))] + fn info(&self) -> Box { + Box::new(ZenohRawInfo) + } + + fn name(&self) -> Arc { + self.name.load_full() + } + + fn uuid(&self) -> &DriverUuid { + &self.uuid + } +} + +#[async_trait::async_trait] +impl AccumulatedDriverStatsProvider for ZenohRaw { + async fn stats(&self) -> AccumulatedDriverStats { + self.stats.read().await.clone() + } + + async fn reset_stats(&self) { + let mut stats = self.stats.write().await; + stats.stats.input = None; + stats.stats.output = None + } +} + +pub struct ZenohRawInfo; +impl DriverInfo for ZenohRawInfo { + fn name(&self) -> &'static str { + "ZenohRaw" + } + fn valid_schemes(&self) -> &'static [&'static str] { + &["zenohraw"] + } + + fn cli_example_legacy(&self) -> Vec { + let first_schema = self.valid_schemes()[0]; + + vec![format!("{first_schema}::")] + } + + fn cli_example_url(&self) -> Vec { + let first_schema = &self.valid_schemes()[0]; + vec![format!("{first_schema}://:").to_string()] + } + + fn create_endpoint_from_url(&self, url: &url::Url) -> Option> { + println!("{}", &url); + let _host = url.host_str().unwrap(); + let _port = url.port().unwrap(); + Some(Arc::new(ZenohRaw::builder("ZenohRaw").build())) + } +} diff --git a/tests/common/mod.rs b/tests/common/mod.rs new file mode 100644 index 00000000..0a88e330 --- /dev/null +++ b/tests/common/mod.rs @@ -0,0 +1,13 @@ +use std::time::Duration; + +pub fn pick_free_port() -> u16 { + std::net::TcpListener::bind("127.0.0.1:0") + .expect("Failed to bind ephemeral port") + .local_addr() + .expect("Failed to read ephemeral port") + .port() +} + +pub async fn wait_for_stats_collection() { + tokio::time::sleep(Duration::from_secs(2)).await; +} diff --git a/tests/python/server.py b/tests/python/server.py index 67b3d9e9..00a819a9 100644 --- a/tests/python/server.py +++ b/tests/python/server.py @@ -1,18 +1,28 @@ #!/usr/bin/env python3 import asyncio +import os import sys from typing import Optional from utils import handle_output async def run_server() -> Optional[asyncio.Task]: """Run the Rust server binary with specified arguments.""" + server_args = [ + "tcpclient:0.0.0.0:5760", + "udpout:0.0.0.0:14660", + "--verbose", + ] + + binary = os.environ.get("MAVLINK_SERVER_BIN") + if binary: + command = [binary, *server_args] + else: + command = ["cargo", "run", "--", *server_args] + try: process = await asyncio.create_subprocess_exec( - "cargo", "run", "--", - "tcpclient:0.0.0.0:5760", - "udpout:0.0.0.0:14660", - "--verbose", + *command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) diff --git a/tests/tcp.rs b/tests/tcp.rs index b86a3bbb..54fce83b 100644 --- a/tests/tcp.rs +++ b/tests/tcp.rs @@ -1,15 +1,19 @@ +mod common; + use anyhow::*; use clap::Parser; use mavlink_server::{cli, hub, stats}; use tracing::*; #[tokio::test(flavor = "multi_thread", worker_threads = 10)] -#[ignore] async fn test_tcpserver_receive_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "tcpclient:0.0.0.0:3333", - "tcpserver://0.0.0.0:3333?direction=receiver", + &format!("tcpclient:{bind_addr}"), + &format!("tcpserver://{bind_addr}?direction=receiver"), "--mavlink-heartbeat-frequency", "10", ])); @@ -19,7 +23,7 @@ async fn test_tcpserver_receive_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "TcpServer" { assert!(driver.stats.output.is_none()); @@ -41,12 +45,14 @@ async fn test_tcpserver_receive_only() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 10)] -#[ignore] async fn test_tcpserver_send_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "tcpclient:0.0.0.0:3333", - "tcpserver://0.0.0.0:3333?direction=sender", + &format!("tcpclient:{bind_addr}"), + &format!("tcpserver://{bind_addr}?direction=sender"), "--mavlink-heartbeat-frequency", "10", ])); @@ -56,7 +62,7 @@ async fn test_tcpserver_send_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "TcpServer" { assert!(driver.stats.output.is_some()); @@ -78,12 +84,14 @@ async fn test_tcpserver_send_only() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 10)] -#[ignore] async fn test_tcpclient_send_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "tcpclient://0.0.0.0:3333?direction=sender", - "tcpserver://0.0.0.0:3333", + &format!("tcpclient://{bind_addr}?direction=sender"), + &format!("tcpserver://{bind_addr}"), "--mavlink-heartbeat-frequency", "10", ])); @@ -93,7 +101,7 @@ async fn test_tcpclient_send_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "TcpServer" { assert!(driver.stats.output.is_some()); @@ -115,12 +123,14 @@ async fn test_tcpclient_send_only() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 10)] -#[ignore] async fn test_tcpclient_receive_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "tcpclient://0.0.0.0:3333?direction=receiver", - "tcpserver://0.0.0.0:3333", + &format!("tcpclient://{bind_addr}?direction=receiver"), + &format!("tcpserver://{bind_addr}"), "--mavlink-heartbeat-frequency", "10", ])); @@ -130,7 +140,7 @@ async fn test_tcpclient_receive_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "TcpServer" { assert!(driver.stats.output.is_some()); diff --git a/tests/udp.rs b/tests/udp.rs index 9eb56f66..dfd23290 100644 --- a/tests/udp.rs +++ b/tests/udp.rs @@ -1,3 +1,5 @@ +mod common; + use anyhow::*; use clap::Parser; use mavlink_server::{cli, hub, stats}; @@ -5,10 +7,13 @@ use tracing::*; #[tokio::test(flavor = "multi_thread", worker_threads = 10)] async fn test_udpserver_receive_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "udpclient:0.0.0.0:3333", - "udpserver://0.0.0.0:3333?direction=receiver", + &format!("udpclient:{bind_addr}"), + &format!("udpserver://{bind_addr}?direction=receiver"), "--mavlink-heartbeat-frequency", "10", ])); @@ -18,7 +23,7 @@ async fn test_udpserver_receive_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "UdpServer" { assert!(driver.stats.output.is_none()); @@ -40,12 +45,14 @@ async fn test_udpserver_receive_only() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 10)] -#[ignore] async fn test_udpserver_send_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "udpclient:0.0.0.0:3333", - "udpserver://0.0.0.0:3333?direction=sender", + &format!("udpclient:{bind_addr}"), + &format!("udpserver://{bind_addr}?direction=sender"), "--mavlink-heartbeat-frequency", "10", ])); @@ -55,7 +62,7 @@ async fn test_udpserver_send_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "UdpServer" { assert!(driver.stats.output.is_some()); @@ -77,12 +84,14 @@ async fn test_udpserver_send_only() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 10)] -#[ignore] async fn test_udpclient_send_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "udpclient://0.0.0.0:3333?direction=sender", - "udpserver://0.0.0.0:3333", + &format!("udpclient://{bind_addr}?direction=sender"), + &format!("udpserver://{bind_addr}"), "--mavlink-heartbeat-frequency", "10", ])); @@ -92,7 +101,7 @@ async fn test_udpclient_send_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "UdpServer" { assert!(driver.stats.output.is_some()); diff --git a/tests/websocket.rs b/tests/websocket.rs index 8a3121b9..06c9d511 100644 --- a/tests/websocket.rs +++ b/tests/websocket.rs @@ -1,15 +1,19 @@ +mod common; + use anyhow::*; use clap::Parser; use mavlink_server::{cli, hub, stats}; use tracing::*; #[tokio::test(flavor = "multi_thread", worker_threads = 10)] -#[ignore] async fn test_wsserver_receive_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "wsclient://0.0.0.0:9998", - "wsserver://0.0.0.0:9998?direction=receiver", + &format!("wsclient://{bind_addr}"), + &format!("wsserver://{bind_addr}?direction=receiver"), "--mavlink-heartbeat-frequency", "10", ])); @@ -19,7 +23,7 @@ async fn test_wsserver_receive_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "WsServer" { assert!(driver.stats.output.is_none()); @@ -41,12 +45,14 @@ async fn test_wsserver_receive_only() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 10)] -#[ignore] async fn test_wsserver_send_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "wsclient://0.0.0.0:9998", - "wsserver://0.0.0.0:9998?direction=sender", + &format!("wsclient://{bind_addr}"), + &format!("wsserver://{bind_addr}?direction=sender"), "--mavlink-heartbeat-frequency", "10", ])); @@ -56,7 +62,7 @@ async fn test_wsserver_send_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "WsServer" { assert!(driver.stats.output.is_some()); @@ -78,12 +84,14 @@ async fn test_wsserver_send_only() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 10)] -#[ignore] async fn test_wsclient_send_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "wsclient://0.0.0.0:9998?direction=sender", - "wsserver://0.0.0.0:9998", + &format!("wsclient://{bind_addr}?direction=sender"), + &format!("wsserver://{bind_addr}"), "--mavlink-heartbeat-frequency", "10", ])); @@ -93,7 +101,7 @@ async fn test_wsclient_send_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "WsServer" { assert!(driver.stats.output.is_some()); @@ -115,12 +123,14 @@ async fn test_wsclient_send_only() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 10)] -#[ignore] async fn test_wsclient_receive_only() -> Result<()> { + let port = common::pick_free_port(); + let bind_addr = format!("0.0.0.0:{port}"); + cli::init_with(cli::Args::parse_from(vec![ &std::env::args().next().unwrap_or_default(), // Required dummy argv[0] (program name) - "wsclient://0.0.0.0:9998?direction=receiver", - "wsserver://0.0.0.0:9998", + &format!("wsclient://{bind_addr}?direction=receiver"), + &format!("wsserver://{bind_addr}"), "--mavlink-heartbeat-frequency", "10", ])); @@ -130,7 +140,7 @@ async fn test_wsclient_receive_only() -> Result<()> { } stats::set_period(tokio::time::Duration::from_millis(100)).await?; - tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + common::wait_for_stats_collection().await; for (_uuid, driver) in stats::drivers_stats().await? { if *driver.name == "WsServer" { assert!(driver.stats.output.is_some()); diff --git a/tests/zenoh.rs b/tests/zenoh.rs new file mode 100644 index 00000000..cd239e26 --- /dev/null +++ b/tests/zenoh.rs @@ -0,0 +1,10 @@ +mod common; + +#[path = "zenoh/mod.rs"] +mod zenoh; + +#[path = "zenoh/json.rs"] +mod json; + +#[path = "zenoh/raw.rs"] +mod raw; diff --git a/tests/zenoh/json.rs b/tests/zenoh/json.rs new file mode 100644 index 00000000..d647480e --- /dev/null +++ b/tests/zenoh/json.rs @@ -0,0 +1,65 @@ +use std::{sync::Arc, time::Duration}; + +use anyhow::{Context, Result}; +use clap::Parser; +use mavlink_server::{cli, drivers::zenoh::json::Zenoh, hub, stats}; +use tracing::*; + +use crate::common::{pick_free_port, wait_for_stats_collection}; +use crate::zenoh::{spawn_zenoh_router, wait_for_router, write_zenoh_client_config}; + +#[tokio::test(flavor = "multi_thread", worker_threads = 10)] +async fn test_zenoh_bidirectional() -> Result<()> { + let driver_name = "Zenoh"; + let port = pick_free_port(); + let _router = spawn_zenoh_router(port, "mavlink").await; + wait_for_router().await; + + let zenoh_config = write_zenoh_client_config(port)?; + let zenoh_config = zenoh_config + .to_str() + .context("Zenoh config path is not valid UTF-8")?; + + cli::init_with(cli::Args::parse_from(vec![ + &std::env::args().next().unwrap_or_default(), + "--allow-no-endpoints", + "--zenoh-config-file", + zenoh_config, + "--mavlink-heartbeat-frequency", + "10", + ])); + + hub::add_driver(Arc::new(Zenoh::builder(driver_name).build())).await?; + + stats::set_period(Duration::from_millis(100)).await?; + wait_for_stats_collection().await; + + let mut matching_drivers = 0; + for (_uuid, driver) in stats::drivers_stats().await? { + if *driver.name != driver_name { + continue; + } + + matching_drivers += 1; + assert!( + driver.stats.input.is_some(), + "{driver_name} should receive mavlink over zenoh" + ); + assert!( + driver.stats.output.is_some(), + "{driver_name} should publish mavlink over zenoh" + ); + } + + assert_eq!( + matching_drivers, 1, + "expected one {driver_name} driver in stats" + ); + + for (id, driver_info) in hub::drivers().await? { + debug!("Removing driver id {id:?} ({driver_info:?})"); + hub::remove_driver(id).await?; + } + + Ok(()) +} diff --git a/tests/zenoh/mod.rs b/tests/zenoh/mod.rs new file mode 100644 index 00000000..0e468cb2 --- /dev/null +++ b/tests/zenoh/mod.rs @@ -0,0 +1,72 @@ +use std::{path::PathBuf, time::Duration}; + +use anyhow::{Context, Result}; +use zenoh; + +pub async fn spawn_zenoh_router(port: u16, topic_prefix: &str) -> tokio::task::JoinHandle<()> { + let topic_prefix = topic_prefix.to_string(); + + tokio::spawn(async move { + let mut config = zenoh::Config::default(); + config + .insert_json5("mode", r#""router""#) + .expect("Failed to insert router mode"); + config + .insert_json5("listen/endpoints", &format!(r#"["tcp/127.0.0.1:{port}"]"#)) + .expect("Failed to insert listen endpoints"); + + let session = zenoh::open(config) + .await + .expect("Failed to start zenoh router for tests"); + + spawn_zenoh_loopback(session, &topic_prefix).await; + + std::future::pending::<()>().await + }) +} + +async fn spawn_zenoh_loopback(session: zenoh::Session, topic_prefix: &str) { + let out_topic = format!("{topic_prefix}/out"); + let in_topic = format!("{topic_prefix}/in"); + + tokio::spawn(async move { + let subscriber = session + .declare_subscriber(&out_topic) + .await + .unwrap_or_else(|error| panic!("Failed to subscribe to {out_topic}: {error:?}")); + let publisher = session + .declare_publisher(&in_topic) + .await + .unwrap_or_else(|error| panic!("Failed to publish on {in_topic}: {error:?}")); + + while let Ok(sample) = subscriber.recv_async().await { + if let Err(error) = publisher.put(sample.payload().to_bytes()).await { + panic!("Failed to loop back {out_topic} -> {in_topic}: {error:?}"); + } + } + }); +} + +pub fn write_zenoh_client_config(port: u16) -> Result { + let path = std::env::temp_dir().join(format!( + "mavlink-server-zenoh-test-{}-{port}.json5", + std::process::id() + )); + + std::fs::write( + &path, + format!( + r#"{{ + mode: "client", + connect: {{ endpoints: ["tcp/127.0.0.1:{port}"] }} +}}"# + ), + ) + .with_context(|| format!("Failed to write zenoh config to {}", path.display()))?; + + Ok(path) +} + +pub async fn wait_for_router() { + tokio::time::sleep(Duration::from_millis(500)).await; +} diff --git a/tests/zenoh/raw.rs b/tests/zenoh/raw.rs new file mode 100644 index 00000000..77069424 --- /dev/null +++ b/tests/zenoh/raw.rs @@ -0,0 +1,65 @@ +use std::{sync::Arc, time::Duration}; + +use anyhow::{Context, Result}; +use clap::Parser; +use mavlink_server::{cli, drivers::zenoh::raw::ZenohRaw, hub, stats}; +use tracing::*; + +use crate::common::{pick_free_port, wait_for_stats_collection}; +use crate::zenoh::{spawn_zenoh_router, wait_for_router, write_zenoh_client_config}; + +#[tokio::test(flavor = "multi_thread", worker_threads = 10)] +async fn test_zenohraw_bidirectional() -> Result<()> { + let driver_name = "ZenohRaw"; + let port = pick_free_port(); + let _router = spawn_zenoh_router(port, "mavlink_raw").await; + wait_for_router().await; + + let zenoh_config = write_zenoh_client_config(port)?; + let zenoh_config = zenoh_config + .to_str() + .context("Zenoh config path is not valid UTF-8")?; + + cli::init_with(cli::Args::parse_from(vec![ + &std::env::args().next().unwrap_or_default(), + "--allow-no-endpoints", + "--zenoh-config-file", + zenoh_config, + "--mavlink-heartbeat-frequency", + "10", + ])); + + hub::add_driver(Arc::new(ZenohRaw::builder(driver_name).build())).await?; + + stats::set_period(Duration::from_millis(100)).await?; + wait_for_stats_collection().await; + + let mut matching_drivers = 0; + for (_uuid, driver) in stats::drivers_stats().await? { + if *driver.name != driver_name { + continue; + } + + matching_drivers += 1; + assert!( + driver.stats.input.is_some(), + "{driver_name} should receive mavlink over zenoh" + ); + assert!( + driver.stats.output.is_some(), + "{driver_name} should publish mavlink over zenoh" + ); + } + + assert_eq!( + matching_drivers, 1, + "expected one {driver_name} driver in stats" + ); + + for (id, driver_info) in hub::drivers().await? { + debug!("Removing driver id {id:?} ({driver_info:?})"); + hub::remove_driver(id).await?; + } + + Ok(()) +}