Update C++ example implementation (#491)

### Public-Facing Changes

Update C++ example implementation

### Description
- Allows choosing between boost asio and standalone asio (fixes #483) 
- Adds support for `asset` capability
- Some refactoring

---------

Co-authored-by: Jacob Bandes-Storch <[email protected]>

.github/workflows/ci.yml

@@ -74,19 +74,14 @@ jobs:
 
   cpp:
     runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        make_cmd: [format-check, build, build-cpp20, build-boost-asio]
     defaults:
       run:
         working-directory: cpp
+    name: cpp (${{ matrix.make_cmd }})
     steps:
       - uses: actions/checkout@v3
-      - run: make format-check
-      - run: make build
-
-  cpp-std-20:
-    runs-on: ubuntu-latest
-    defaults:
-      run:
-        working-directory: cpp
-    steps:
-      - uses: actions/checkout@v3
-      - run: make build-cpp20
+      - run: make ${{ matrix.make_cmd }}

cpp/Makefile

@@ -4,6 +4,10 @@ default: build
 build:
 	docker compose build --build-arg CPPSTD=17
 
+.PHONY: build-boost-asio
+build-boost-asio:
+	docker compose build --build-arg CPPSTD=17 --build-arg ASIO=boost
+
 .PHONY: build-cpp20
 build-cpp20:
 	docker compose build --build-arg CPPSTD=20

cpp/dev.Dockerfile

@@ -24,24 +24,18 @@ WORKDIR /src
 FROM base as build
 RUN pip --no-cache-dir install conan
 RUN conan profile detect --force
-
-FROM build as build_examples
-COPY ./foxglove-websocket/conanfile.py /src/foxglove-websocket/conanfile.py
-ARG CPPSTD=17
-RUN conan install foxglove-websocket -s compiler.cppstd=$CPPSTD --build=missing
 COPY ./foxglove-websocket /src/foxglove-websocket/
-RUN conan create foxglove-websocket -s compiler.cppstd=$CPPSTD
-COPY ./examples/conanfile.py /src/examples/conanfile.py
-RUN conan install examples --output-folder examples/build --build=missing -s compiler.cppstd=$CPPSTD -s build_type=Debug
+ARG CPPSTD=17
+ARG ASIO=standalone
+RUN conan create foxglove-websocket -s compiler.cppstd=$CPPSTD --build=missing -o foxglove-websocket*:asio=$ASIO
 
-FROM build_examples AS examples
-COPY --from=build_examples /src /src
+FROM build as build_examples
+COPY --from=build /root/.conan2 /root/.conan2
 COPY ./examples /src/examples
-COPY --from=build_examples /src/examples/build/ /src/examples/build/
-RUN conan build examples --output-folder examples/ -s compiler.cppstd=$CPPSTD -s build_type=Debug
+RUN conan build examples --output-folder examples/ -s compiler.cppstd=$CPPSTD --build=missing
 
-FROM examples AS example_server_protobuf
-CMD ["examples/build/Debug/example_server_protobuf"]
+FROM build_examples AS example_server_protobuf
+CMD ["examples/build/Release/example_server_protobuf"]
 
-FROM examples AS example_server_flatbuffers
-CMD ["examples/build/Debug/example_server_flatbuffers", "/src/examples/autogenerated_flatbuffers/SceneUpdate.bfbs"]
+FROM build_examples AS example_server_flatbuffers
+CMD ["examples/build/Release/example_server_flatbuffers", "/src/examples/autogenerated_flatbuffers/SceneUpdate.bfbs"]

cpp/examples/conanfile.py

@@ -4,14 +4,14 @@
 
 class FoxgloveWebSocketExamplesConan(ConanFile):
     name = "foxglove-websocket-example"
-    version = "1.0.0"
+    version = "1.1.0"
     settings = "os", "compiler", "build_type", "arch"
     exports_sources = "CMakeLists.txt", "src/*", "proto/*"
     generators = "CMakeDeps"
 
     def requirements(self):
         self.requires("flatbuffers/23.5.26")
-        self.requires("foxglove-websocket/1.0.0")
+        self.requires("foxglove-websocket/1.1.0")
         self.requires("protobuf/3.21.4")
         self.requires("zlib/1.2.13")
 

cpp/examples/src/example_server_flatbuffers.cpp

@@ -1,3 +1,4 @@
+#include <foxglove/websocket/base64.hpp>
 #include <foxglove/websocket/websocket_notls.hpp>
 #include <foxglove/websocket/websocket_server.hpp>
 
@@ -12,7 +13,6 @@
 #include <unordered_set>
 
 #include "SceneUpdate_generated.h"
-#include "base64.hpp"
 #include "flatbuffers/flatbuffers.h"
 
 namespace foxglove {
@@ -78,13 +78,13 @@ int main(int argc, char** argv) {
     .topic = "example_msg",
     .encoding = "flatbuffer",
     .schemaName = "foxglove.SceneUpdate",
-    .schema = Base64Encode(getFileContents(sceneUpdateBfbsPath)),
+    .schema = foxglove::base64Encode(getFileContents(sceneUpdateBfbsPath)),
   }});
   const auto chanId = channelIds.front();
 
   bool running = true;
 
-  asio::signal_set signals(server->getEndpoint().get_io_service(), SIGINT);
+  websocketpp::lib::asio::signal_set signals(server->getEndpoint().get_io_service(), SIGINT);
   signals.async_wait([&](std::error_code const& ec, int sig) {
     if (ec) {
       std::cerr << "signal error: " << ec.message() << std::endl;

cpp/examples/src/example_server_protobuf.cpp

@@ -1,3 +1,4 @@
+#include <foxglove/websocket/base64.hpp>
 #include <foxglove/websocket/websocket_notls.hpp>
 #include <foxglove/websocket/websocket_server.hpp>
 
@@ -13,7 +14,6 @@
 #include <thread>
 #include <unordered_set>
 
-#include "base64.hpp"
 #include "foxglove/SceneUpdate.pb.h"
 
 namespace foxglove {
@@ -80,13 +80,13 @@ int main() {
     .topic = "example_msg",
     .encoding = "protobuf",
     .schemaName = foxglove::SceneUpdate::descriptor()->full_name(),
-    .schema = Base64Encode(SerializeFdSet(foxglove::SceneUpdate::descriptor())),
+    .schema = foxglove::base64Encode(SerializeFdSet(foxglove::SceneUpdate::descriptor())),
   }});
   const auto chanId = channelIds.front();
 
   bool running = true;
 
-  asio::signal_set signals(server->getEndpoint().get_io_service(), SIGINT);
+  websocketpp::lib::asio::signal_set signals(server->getEndpoint().get_io_service(), SIGINT);
   signals.async_wait([&](std::error_code const& ec, int sig) {
     if (ec) {
       std::cerr << "signal error: " << ec.message() << std::endl;

cpp/foxglove-websocket/CMakeLists.txt

@@ -3,11 +3,10 @@ project(FoxgloveWebSocket CXX)
 
 find_package(nlohmann_json REQUIRED)
 find_package(websocketpp REQUIRED)
-find_package(asio REQUIRED)
 
-add_library(foxglove_websocket src/parameter.cpp src/serialization.cpp)
+add_library(foxglove_websocket src/base64.cpp src/parameter.cpp src/serialization.cpp)
 target_include_directories(foxglove_websocket PUBLIC include)
-target_link_libraries(foxglove_websocket nlohmann_json::nlohmann_json websocketpp::websocketpp asio::asio)
+target_link_libraries(foxglove_websocket nlohmann_json::nlohmann_json websocketpp::websocketpp)
 set_target_properties(foxglove_websocket PROPERTIES CXX_STANDARD 17 CXX_STANDARD_REQUIRED ON)
 
 install(TARGETS foxglove_websocket)

cpp/foxglove-websocket/conanfile.py

@@ -5,7 +5,7 @@
 
 class FoxgloveWebSocketConan(ConanFile):
     name = "foxglove-websocket"
-    version = "1.0.0"
+    version = "1.1.0"
     url = "https://github.com/foxglove/ws-protocol"
     homepage = "https://github.com/foxglove/ws-protocol"
     description = "A C++ server implementation of the Foxglove WebSocket Protocol"
@@ -15,6 +15,8 @@ class FoxgloveWebSocketConan(ConanFile):
     settings = ("os", "compiler", "build_type", "arch")
     generators = "CMakeDeps"
     exports_sources = "CMakeLists.txt", "LICENSE", "src/*", "include/*"
+    options = {"asio": ["standalone", "boost"]}
+    default_options = {"asio": "standalone"}
 
     def validate(self):
         check_min_cppstd(self, "17")
@@ -24,7 +26,8 @@ def requirements(self):
         self.requires("websocketpp/0.8.2", transitive_headers=True)
 
     def configure(self):
-        self.options["websocketpp"].asio = "standalone"
+        if self.options.asio == "standalone":
+            self.options["websocketpp"].asio = "standalone"
 
     def layout(self):
         cmake_layout(self)

cpp/foxglove-websocket/include/foxglove/websocket/base64.hpp

@@ -0,0 +1,14 @@
+#pragma once
+
+#include <cstdint>
+#include <string>
+#include <string_view>
+#include <vector>
+
+namespace foxglove {
+
+std::string base64Encode(const std::string_view& input);
+
+std::vector<unsigned char> base64Decode(const std::string& input);
+
+}  // namespace foxglove

cpp/foxglove-websocket/include/foxglove/websocket/callback_queue.hpp

@@ -0,0 +1,81 @@
+#pragma once
+
+#include <atomic>
+#include <condition_variable>
+#include <deque>
+#include <functional>
+#include <mutex>
+#include <thread>
+#include <vector>
+
+#include "websocket_logging.hpp"
+
+namespace foxglove {
+
+class CallbackQueue {
+public:
+  CallbackQueue(LogCallback logCallback, size_t numThreads = 1)
+      : _logCallback(logCallback)
+      , _quit(false) {
+    for (size_t i = 0; i < numThreads; ++i) {
+      _workerThreads.push_back(std::thread(&CallbackQueue::doWork, this));
+    }
+  }
+
+  ~CallbackQueue() {
+    stop();
+  }
+
+  void stop() {
+    _quit = true;
+    _cv.notify_all();
+    for (auto& thread : _workerThreads) {
+      thread.join();
+    }
+  }
+
+  void addCallback(std::function<void(void)> cb) {
+    if (_quit) {
+      return;
+    }
+    std::unique_lock<std::mutex> lock(_mutex);
+    _callbackQueue.push_back(cb);
+    _cv.notify_one();
+  }
+
+private:
+  void doWork() {
+    while (!_quit) {
+      std::unique_lock<std::mutex> lock(_mutex);
+      _cv.wait(lock, [this] {
+        return (_quit || !_callbackQueue.empty());
+      });
+      if (_quit) {
+        break;
+      } else if (!_callbackQueue.empty()) {
+        std::function<void(void)> cb = _callbackQueue.front();
+        _callbackQueue.pop_front();
+        lock.unlock();
+        try {
+          cb();
+        } catch (const std::exception& ex) {
+          // Should never get here if we catch all exceptions in the callbacks.
+          const std::string msg =
+            std::string("Caught unhandled exception in calback_queue") + ex.what();
+          _logCallback(WebSocketLogLevel::Error, msg.c_str());
+        } catch (...) {
+          _logCallback(WebSocketLogLevel::Error, "Caught unhandled exception in calback_queue");
+        }
+      }
+    }
+  }
+
+  LogCallback _logCallback;
+  std::atomic<bool> _quit;
+  std::mutex _mutex;
+  std::condition_variable _cv;
+  std::deque<std::function<void(void)>> _callbackQueue;
+  std::vector<std::thread> _workerThreads;
+};
+
+}  // namespace foxglove

cpp/foxglove-websocket/include/foxglove/websocket/common.hpp

@@ -2,6 +2,7 @@
 
 #include <array>
 #include <cstring>
+#include <optional>
 #include <stdint.h>
 #include <string>
 #include <vector>
@@ -15,10 +16,11 @@ constexpr char CAPABILITY_PARAMETERS[] = "parameters";
 constexpr char CAPABILITY_PARAMETERS_SUBSCRIBE[] = "parametersSubscribe";
 constexpr char CAPABILITY_SERVICES[] = "services";
 constexpr char CAPABILITY_CONNECTION_GRAPH[] = "connectionGraph";
+constexpr char CAPABILITY_ASSETS[] = "assets";
 
-constexpr std::array DEFAULT_CAPABILITIES = {
+constexpr std::array<const char*, 6> DEFAULT_CAPABILITIES = {
   CAPABILITY_CLIENT_PUBLISH, CAPABILITY_CONNECTION_GRAPH, CAPABILITY_PARAMETERS_SUBSCRIBE,
-  CAPABILITY_PARAMETERS,     CAPABILITY_SERVICES,
+  CAPABILITY_PARAMETERS,     CAPABILITY_SERVICES,         CAPABILITY_ASSETS,
 };
 
 using ChannelId = uint32_t;
@@ -30,6 +32,7 @@ enum class BinaryOpcode : uint8_t {
   MESSAGE_DATA = 1,
   TIME_DATA = 2,
   SERVICE_CALL_RESPONSE = 3,
+  FETCH_ASSET_RESPONSE = 4,
 };
 
 enum class ClientBinaryOpcode : uint8_t {
@@ -50,10 +53,11 @@ struct ChannelWithoutId {
   std::string encoding;
   std::string schemaName;
   std::string schema;
+  std::optional<std::string> schemaEncoding;
 
   bool operator==(const ChannelWithoutId& other) const {
     return topic == other.topic && encoding == other.encoding && schemaName == other.schemaName &&
-           schema == other.schema;
+           schema == other.schema && schemaEncoding == other.schemaEncoding;
   }
 };
 
@@ -143,4 +147,16 @@ struct ServiceResponse {
 
 using ServiceRequest = ServiceResponse;
 
+enum class FetchAssetStatus : uint8_t {
+  Success = 0,
+  Error = 1,
+};
+
+struct FetchAssetResponse {
+  uint32_t requestId;
+  FetchAssetStatus status;
+  std::string errorMessage;
+  std::vector<uint8_t> data;
+};
+
 }  // namespace foxglove