Add LibCarla module

LibCarla/cmake/CMakeLists.txt

@@ -0,0 +1,19 @@
+cmake_minimum_required(VERSION 3.9.0)
+project(libcarla)
+
+message(STATUS "Building ${PROJECT_NAME} version ${CARLA_VERSION}")
+
+set(libcarla_source_path "${PROJECT_SOURCE_DIR}/../source")
+
+include_directories(${libcarla_source_path})
+
+configure_file(${libcarla_source_path}/carla/Version.h.in ${libcarla_source_path}/carla/Version.h)
+
+if (CMAKE_BUILD_TYPE STREQUAL "Client")
+  add_subdirectory("client")
+elseif (CMAKE_BUILD_TYPE STREQUAL "Server")
+  add_subdirectory("server")
+  add_subdirectory("test")
+else ()
+  message(FATAL_ERROR "Unknown build type '${CMAKE_BUILD_TYPE}'")
+endif ()

LibCarla/cmake/client/CMakeLists.txt

@@ -0,0 +1,42 @@
+cmake_minimum_required(VERSION 3.9.0)
+project(libcarla-client)
+
+# Install rpclib.
+install(DIRECTORY "${RPCLIB_INCLUDE_PATH}/rpc" DESTINATION include)
+install(FILES "${RPCLIB_LIB_PATH}/librpc.a" DESTINATION lib)
+
+file(GLOB_RECURSE libcarla_client_sources
+    "${libcarla_source_path}/carla/client/*.h"
+    "${libcarla_source_path}/carla/client/*.cpp")
+
+# Create targets for debug and release in the same build type.
+foreach(target carla_client_debug carla_client)
+  add_library(${target} STATIC ${libcarla_client_sources})
+
+  target_include_directories(${target} PRIVATE
+      "${BOOST_INCLUDE_PATH}"
+      "${RPCLIB_INCLUDE_PATH}")
+
+  install(TARGETS ${target} DESTINATION lib)
+endforeach(target)
+
+# Specific options for debug.
+set_target_properties(carla_client_debug PROPERTIES COMPILE_FLAGS ${CMAKE_CXX_FLAGS_DEBUG})
+target_compile_definitions(carla_client_debug PUBLIC -DBOOST_ASIO_ENABLE_BUFFER_DEBUGGING)
+
+# Specific options for release.
+set_target_properties(carla_client PROPERTIES COMPILE_FLAGS ${CMAKE_CXX_FLAGS_RELEASE})
+
+# Install headers.
+
+file(GLOB libcarla_carla_headers "${libcarla_source_path}/carla/*.h")
+install(FILES ${libcarla_carla_headers} DESTINATION include/carla)
+
+file(GLOB libcarla_carla_client_headers "${libcarla_source_path}/carla/client/*.h")
+install(FILES ${libcarla_carla_client_headers} DESTINATION include/carla/client)
+
+file(GLOB libcarla_carla_rpc_headers "${libcarla_source_path}/carla/rpc/*.h")
+install(FILES ${libcarla_carla_rpc_headers} DESTINATION include/carla/rpc)
+
+file(GLOB libcarla_carla_streaming_headers "${libcarla_source_path}/carla/streaming/*.h")
+install(FILES ${libcarla_carla_streaming_headers} DESTINATION include/carla/streaming)

LibCarla/cmake/server/CMakeLists.txt

@@ -0,0 +1,25 @@
+cmake_minimum_required(VERSION 3.9.0)
+project(libcarla-server)
+
+# Install libc++ shared libraries.
+file(GLOB LibCXX_Libraries "${LLVM_LIB_PATH}/libc++*")
+install(FILES ${LibCXX_Libraries} DESTINATION lib)
+
+# Install rpclib.
+install(DIRECTORY "${RPCLIB_INCLUDE_PATH}/rpc" DESTINATION include)
+install(FILES "${RPCLIB_LIB_PATH}/librpc.a" DESTINATION lib)
+
+# Install headers.
+
+install(DIRECTORY "${libcarla_source_path}/compiler" DESTINATION include)
+
+file(GLOB libcarla_carla_headers "${libcarla_source_path}/carla/*.h")
+install(FILES ${libcarla_carla_headers} DESTINATION include/carla)
+
+file(GLOB libcarla_carla_rpc_headers "${libcarla_source_path}/carla/rpc/*.h")
+install(FILES ${libcarla_carla_rpc_headers} DESTINATION include/carla/rpc)
+
+file(GLOB libcarla_carla_streaming_headers "${libcarla_source_path}/carla/streaming/*.h")
+install(FILES ${libcarla_carla_streaming_headers} DESTINATION include/carla/streaming)
+
+install(DIRECTORY "${BOOST_INCLUDE_PATH}/boost" DESTINATION include)

LibCarla/cmake/test/CMakeLists.txt

@@ -0,0 +1,36 @@
+cmake_minimum_required(VERSION 3.9.0)
+project(libcarla-unit-tests)
+
+file(GLOB_RECURSE libcarla_test_sources
+    "${libcarla_source_path}/carla/profiler/*.h"
+    "${libcarla_source_path}/carla/profiler/*.cpp"
+    "${libcarla_source_path}/test/*.h"
+    "${libcarla_source_path}/test/*.cpp")
+
+link_directories(
+    ${RPCLIB_LIB_PATH}
+    ${GTEST_LIB_PATH})
+
+# Create targets for debug and release in the same build type.
+foreach(target libcarla_test_debug libcarla_test_release)
+  add_executable(${target} ${libcarla_test_sources})
+
+  target_compile_definitions(${target} PUBLIC
+      -DLIBCARLA_ENABLE_PROFILER
+      -DLIBCARLA_WITH_GTEST)
+
+  target_include_directories(${target} PRIVATE
+      "${BOOST_INCLUDE_PATH}"
+      "${RPCLIB_INCLUDE_PATH}"
+      "${GTEST_INCLUDE_PATH}")
+  target_link_libraries(${target} "-lrpc -lgtest_main -lgtest")
+
+  install(TARGETS ${target} DESTINATION test)
+endforeach(target)
+
+# Specific options for debug.
+set_target_properties(libcarla_test_debug PROPERTIES COMPILE_FLAGS ${CMAKE_CXX_FLAGS_DEBUG})
+target_compile_definitions(libcarla_test_debug PUBLIC -DBOOST_ASIO_ENABLE_BUFFER_DEBUGGING)
+
+# Specific options for release.
+set_target_properties(libcarla_test_release PROPERTIES COMPILE_FLAGS ${CMAKE_CXX_FLAGS_RELEASE})

LibCarla/source/carla/.gitignore

@@ -0,0 +1 @@
+Version.h

LibCarla/source/carla/Debug.h

@@ -0,0 +1,29 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#ifndef NDEBUG
+#  include <cassert>
+#endif // NDEBUG
+
+#ifdef NDEBUG
+#  define DEBUG_ONLY(code)
+#else
+#  define DEBUG_ONLY(code) code
+#endif // NDEBUG
+
+#define DEBUG_ASSERT(predicate) DEBUG_ONLY(assert(predicate));
+
+#ifdef LIBCARLA_WITH_GTEST
+#  include <gtest/gtest.h>
+
+#  define DEBUG_ASSERT_EQ(lhs, rhs) DEBUG_ONLY(EXPECT_EQ(lhs, rhs));DEBUG_ASSERT(lhs == rhs);
+#  define DEBUG_ASSERT_NE(lhs, rhs) DEBUG_ONLY(EXPECT_NE(lhs, rhs));DEBUG_ASSERT(lhs != rhs);
+#else
+#  define DEBUG_ASSERT_EQ(lhs, rhs) DEBUG_ASSERT((lhs) == (rhs))
+#  define DEBUG_ASSERT_NE(lhs, rhs) DEBUG_ASSERT((lhs) != (rhs))
+#endif // LIBCARLA_WITH_GTEST

LibCarla/source/carla/Logging.h

@@ -0,0 +1,148 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#define LIBCARLA_LOG_LEVEL_DEBUG     10
+#define LIBCARLA_LOG_LEVEL_INFO      20
+#define LIBCARLA_LOG_LEVEL_WARNING   30
+#define LIBCARLA_LOG_LEVEL_ERROR     40
+#define LIBCARLA_LOG_LEVEL_CRITICAL  50
+#define LIBCARLA_LOG_LEVEL_NONE     100
+
+#ifndef LIBCARLA_LOG_LEVEL
+#  ifdef NDEBUG
+#    define LIBCARLA_LOG_LEVEL LIBCARLA_LOG_LEVEL_WARNING
+#  else
+#    define LIBCARLA_LOG_LEVEL LIBCARLA_LOG_LEVEL_INFO
+#  endif // NDEBUG
+#endif // LIBCARLA_LOG_LEVEL
+
+// The following log functions are available, they are only active if
+// LIBCARLA_LOG_LEVEL is greater equal the function's log level.
+//
+//  * log_debug
+//  * log_info
+//  * log_error
+//  * log_critical
+//
+// And macros
+//
+//  * LOG_DEBUG_ONLY(/* code here */)
+//  * LOG_INFO_ONLY(/* code here */)
+
+// =============================================================================
+// -- Implementation of log functions ------------------------------------------
+// =============================================================================
+
+#include <iostream>
+
+namespace carla {
+
+namespace logging {
+
+  // https://stackoverflow.com/a/27375675
+  template <typename Arg, typename ... Args>
+  static void write_to_stream(std::ostream &out, Arg &&arg, Args && ... args) {
+    out << std::boolalpha << std::forward<Arg>(arg);
+    using expander = int[];
+    (void) expander{0, (void(out << ' ' << std::forward<Args>(args)), 0) ...};
+  }
+
+  template <typename ... Args>
+  static inline void log(Args && ... args) {
+    logging::write_to_stream(std::cout, std::forward<Args>(args) ..., '\n');
+  }
+
+} // namespace logging
+
+#if LIBCARLA_LOG_LEVEL <= LIBCARLA_LOG_LEVEL_DEBUG
+
+  template <typename ... Args>
+  static inline void log_debug(Args && ... args) {
+    logging::write_to_stream(std::cout, "DEBUG:", std::forward<Args>(args) ..., '\n');
+  }
+
+#else
+
+  template <typename ... Args>
+  static inline void log_debug(Args && ...) {}
+
+#endif
+
+#if LIBCARLA_LOG_LEVEL <= LIBCARLA_LOG_LEVEL_INFO
+
+  template <typename ... Args>
+  static inline void log_info(Args && ... args) {
+    logging::write_to_stream(std::cout, "INFO: ", std::forward<Args>(args) ..., '\n');
+  }
+
+#else
+
+  template <typename ... Args>
+  static inline void log_info(Args && ...) {}
+
+#endif
+
+#if LIBCARLA_LOG_LEVEL <= LIBCARLA_LOG_LEVEL_WARNING
+
+  template <typename ... Args>
+  static inline void log_warning(Args && ... args) {
+    logging::write_to_stream(std::cerr, "WARNING:", std::forward<Args>(args) ..., '\n');
+  }
+
+#else
+
+  template <typename ... Args>
+  static inline void log_warning(Args && ...) {}
+
+#endif
+
+#if LIBCARLA_LOG_LEVEL <= LIBCARLA_LOG_LEVEL_ERROR
+
+  template <typename ... Args>
+  static inline void log_error(Args && ... args) {
+    logging::write_to_stream(std::cerr, "ERROR:", std::forward<Args>(args) ..., '\n');
+  }
+
+#else
+
+  template <typename ... Args>
+  static inline void log_error(Args && ...) {}
+
+#endif
+
+#if LIBCARLA_LOG_LEVEL <= LIBCARLA_LOG_LEVEL_CRITICAL
+
+  template <typename ... Args>
+  static inline void log_critical(Args && ... args) {
+    logging::write_to_stream(std::cerr, "CRITICAL:", std::forward<Args>(args) ..., '\n');
+  }
+
+#else
+
+  template <typename ... Args>
+  static inline void log_critical(Args && ...) {}
+
+#endif
+
+} // namespace carla
+
+// =============================================================================
+// -- Implementation of macros -------------------------------------------------
+// =============================================================================
+
+#if LIBCARLA_LOG_LEVEL <= LIBCARLA_LOG_LEVEL_DEBUG
+#  define LOG_DEBUG_ONLY(code) code
+#else
+#  define LOG_DEBUG_ONLY(code)
+#endif
+
+#if LIBCARLA_LOG_LEVEL <= LIBCARLA_LOG_LEVEL_INFO
+#  define LOG_INFO_ONLY(code) code
+#else
+#  define LOG_INFO_ONLY(code)
+#endif

LibCarla/source/carla/NonCopyable.h

@@ -0,0 +1,21 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+namespace carla {
+
+  class NonCopyable {
+  public:
+
+    NonCopyable() = default;
+
+    NonCopyable(const NonCopyable &) = delete;
+
+    void operator=(const NonCopyable &x) = delete;
+  };
+
+} // namespace carla

LibCarla/source/carla/StopWatch.h

@@ -0,0 +1,56 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include <chrono>
+
+namespace carla {
+
+  template <typename CLOCK>
+  class StopWatchTmpl {
+    static_assert(CLOCK::is_steady, "The StopWatch's clock must be steady");
+  public:
+
+    using clock = CLOCK;
+
+    StopWatchTmpl() : _start(clock::now()), _end(), _is_running(true) {}
+
+    void Restart() {
+      _is_running = true;
+      _start = clock::now();
+    }
+
+    void Stop() {
+      _end = clock::now();
+      _is_running = false;
+    }
+
+    typename clock::duration GetDuration() const {
+      return _is_running ? clock::now() - _start : _end - _start;
+    }
+
+    template <class RESOLUTION=std::chrono::milliseconds>
+    typename RESOLUTION::rep GetElapsedTime() const {
+      return std::chrono::duration_cast<RESOLUTION>(GetDuration()).count();
+    }
+
+    bool IsRunning() const {
+      return _is_running;
+    }
+
+  private:
+
+    typename clock::time_point _start;
+
+    typename clock::time_point _end;
+
+    bool _is_running;
+  };
+
+  using StopWatch = StopWatchTmpl<std::chrono::steady_clock>;
+
+} // namespace carla

LibCarla/source/carla/ThreadGroup.h

@@ -0,0 +1,52 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include <thread>
+#include <vector>
+
+namespace carla {
+
+  class ThreadGroup {
+  public:
+
+    ThreadGroup() {}
+
+    ThreadGroup(const ThreadGroup &) = delete;
+    ThreadGroup &operator=(const ThreadGroup &) = delete;
+
+    ~ThreadGroup() {
+      JoinAll();
+    }
+
+    template <typename F>
+    void CreateThread(F &&functor) {
+      _threads.emplace_back(std::forward<F>(functor));
+    }
+
+    template <typename F>
+    void CreateThreads(size_t count, F functor) {
+      _threads.reserve(_threads.size() + count);
+      for (size_t i = 0u; i < count; ++i) {
+        CreateThread(functor);
+      }
+    }
+
+    void JoinAll() {
+      for (auto &thread : _threads) {
+        if (thread.joinable()) {
+          thread.join();
+        }
+      }
+    }
+
+  private:
+
+    std::vector<std::thread> _threads;
+  };
+
+} // namespace carla

LibCarla/source/carla/Version.h.in

@@ -0,0 +1,15 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+namespace carla {
+
+  constexpr const char *version() {
+    return "${CARLA_VERSION}";
+  }
+
+} // namespace carla

LibCarla/source/carla/client/Actor.h

@@ -0,0 +1,69 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/Debug.h"
+#include "carla/NonCopyable.h"
+#include "carla/client/Control.h"
+#include "carla/client/Memory.h"
+#include "carla/client/World.h"
+#include "carla/rpc/Actor.h"
+
+namespace carla {
+namespace client {
+
+  class Client;
+
+  class Actor
+    : public EnableSharedFromThis<Actor>,
+      private NonCopyable {
+  public:
+
+    Actor(Actor &&) = default;
+    Actor &operator=(Actor &&) = default;
+
+    auto GetId() const {
+      return _actor.id;
+    }
+
+    const std::string &GetTypeId() const {
+      return _actor.blueprint.type_id;
+    }
+
+    ActorBlueprint GetBlueprint() const {
+      return _actor.blueprint;
+    }
+
+    SharedPtr<World> GetWorld() const {
+      return _world;
+    }
+
+    void ApplyControl(const VehicleControl &control) {
+      _world->ApplyControlToActor(*this, control);
+    }
+
+    const auto &Serialize() const {
+      return _actor;
+    }
+
+  private:
+
+    friend class Client;
+
+    Actor(carla::rpc::Actor actor, SharedPtr<World> world)
+      : _actor(actor),
+        _world(std::move(world)) {
+      DEBUG_ASSERT(_world != nullptr);
+    }
+
+    carla::rpc::Actor _actor;
+
+    SharedPtr<World> _world;
+  };
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/client/ActorBlueprint.cpp

@@ -0,0 +1,37 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#include "carla/client/ActorBlueprint.h"
+
+#include <boost/algorithm/string/predicate.hpp>
+
+#ifdef _WIN32
+#  include <shlwapi.h>
+#else
+#  include <fnmatch.h>
+#endif // _WIN32
+
+namespace carla {
+namespace client {
+
+  static bool MatchWildcards(const std::string &str, const std::string &test) {
+#ifdef _WIN32
+    return PathMatchSpecA(str.c_str(), test.c_str());
+#else
+    return 0 == fnmatch(test.c_str(), str.c_str(), 0);
+#endif // _WIN32
+  }
+
+  bool ActorBlueprint::StartsWith(const std::string &test) const {
+    return boost::starts_with(GetTypeId(), test);
+  }
+
+  bool ActorBlueprint::MatchWildcards(const std::string &test) const {
+    return ::carla::client::MatchWildcards(GetTypeId(), test);
+  }
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/client/ActorBlueprint.h

@@ -0,0 +1,44 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/Debug.h"
+#include "carla/rpc/ActorBlueprint.h"
+
+namespace carla {
+namespace client {
+
+  class ActorBlueprint {
+  public:
+
+    ActorBlueprint(carla::rpc::ActorBlueprint blueprint)
+      : _blueprint(std::move(blueprint)) {}
+
+    ActorBlueprint(const ActorBlueprint &) = default;
+    ActorBlueprint(ActorBlueprint &&) = default;
+    ActorBlueprint &operator=(const ActorBlueprint &) = default;
+    ActorBlueprint &operator=(ActorBlueprint &&) = default;
+
+    const std::string &GetTypeId() const {
+      return _blueprint.type_id;
+    }
+
+    bool StartsWith(const std::string &test) const;
+
+    bool MatchWildcards(const std::string &test) const;
+
+    const auto &Serialize() const {
+      return _blueprint;
+    }
+
+  private:
+
+    carla::rpc::ActorBlueprint _blueprint;
+  };
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/client/BlueprintLibrary.h

@@ -0,0 +1,77 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/Debug.h"
+#include "carla/NonCopyable.h"
+#include "carla/client/ActorBlueprint.h"
+
+#include <algorithm>
+#include <vector>
+
+namespace carla {
+namespace client {
+
+  class Client;
+
+  class BlueprintLibrary /*: private NonCopyable*/ {
+    using list_type = std::vector<ActorBlueprint>;
+  public:
+
+    // BlueprintLibrary() = default;
+
+    // BlueprintLibrary(BlueprintLibrary &&) = default;
+    // BlueprintLibrary &operator=(BlueprintLibrary &&) = default;
+
+    using value_type = list_type::value_type;
+    using size_type = list_type::size_type;
+    using const_iterator = list_type::const_iterator;
+    using const_reference = list_type::const_reference;
+
+    BlueprintLibrary Filter(const std::string &wildcard_pattern) const {
+      list_type result;
+      std::copy_if(begin(), end(), std::back_inserter(result), [&](const auto &x) {
+        return x.MatchWildcards(wildcard_pattern);
+      });
+      return result;
+    }
+
+    const_reference operator[](size_type pos) const {
+      return _blueprints[pos];
+    }
+
+    const_iterator begin() const /*noexcept*/ {
+      return _blueprints.begin();
+    }
+
+    const_iterator end() const /*noexcept*/ {
+      return _blueprints.end();
+    }
+
+    bool empty() const /*noexcept*/ {
+      return _blueprints.empty();
+    }
+
+    size_type size() const /*noexcept*/ {
+      return _blueprints.size();
+    }
+
+  private:
+
+    friend class Client;
+
+    BlueprintLibrary(list_type blueprints)
+      : _blueprints(std::move(blueprints)) {}
+
+    BlueprintLibrary(const std::vector<carla::rpc::ActorBlueprint> &blueprints)
+      : _blueprints(blueprints.begin(), blueprints.end()) {}
+
+    list_type _blueprints;
+  };
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/client/Client.cpp

@@ -0,0 +1,37 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#include "carla/client/Client.h"
+
+#include "carla/client/Actor.h"
+#include "carla/client/Control.h"
+#include "carla/client/World.h"
+
+namespace carla {
+namespace client {
+
+  SharedPtr<World> Client::GetWorld() {
+    if (_active_world == nullptr) {
+      _active_world.reset(new World(shared_from_this()));
+    }
+    return _active_world;
+  }
+
+  SharedPtr<Actor> Client::SpawnActor(
+      const ActorBlueprint &blueprint,
+      const Transform &transform) {
+    auto actor = Call<carla::rpc::Actor>("spawn_actor", blueprint.Serialize(), transform);
+    return SharedPtr<Actor>(new Actor{actor, GetWorld()});
+  }
+
+  void Client::ApplyControlToActor(
+      const Actor &actor,
+      const VehicleControl &control) {
+    _client.call("apply_control_to_actor", actor.Serialize(), control);
+  }
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/client/Client.h

@@ -0,0 +1,78 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/NonCopyable.h"
+#include "carla/Version.h"
+#include "carla/client/BlueprintLibrary.h"
+#include "carla/client/Control.h"
+#include "carla/client/Memory.h"
+#include "carla/client/Transform.h"
+#include "carla/rpc/Client.h"
+
+#include <string>
+
+namespace carla {
+namespace client {
+
+  class Actor;
+  class ActorBlueprint;
+  class World;
+
+  class Client
+    : public EnableSharedFromThis<Client>,
+      private NonCopyable {
+  public:
+
+    template <typename ... Args>
+    explicit Client(Args && ... args)
+      : _client(std::forward<Args>(args) ...) {
+      SetTimeout(10'000);
+    }
+
+    void SetTimeout(int64_t milliseconds) {
+      _client.set_timeout(milliseconds);
+    }
+
+    template <typename T, typename ... Args>
+    T Call(const std::string &function, Args && ... args) {
+      return _client.call(function, std::forward<Args>(args) ...).template as<T>();
+    }
+
+    std::string GetClientVersion() const {
+      return ::carla::version();
+    }
+
+    std::string GetServerVersion() {
+      return Call<std::string>("version");
+    }
+
+    bool Ping() {
+      return Call<bool>("ping");
+    }
+
+    SharedPtr<World> GetWorld();
+
+    BlueprintLibrary GetBlueprintLibrary() {
+      return Call<std::vector<carla::rpc::ActorBlueprint>>("get_blueprints");
+    }
+
+    SharedPtr<Actor> SpawnActor(const ActorBlueprint &blueprint, const Transform &transform);
+
+    void ApplyControlToActor(
+        const Actor &actor,
+        const VehicleControl &control);
+
+  private:
+
+    carla::rpc::Client _client;
+
+    SharedPtr<World> _active_world;
+  };
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/client/Control.h

@@ -0,0 +1,17 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/rpc/VehicleControl.h"
+
+namespace carla {
+namespace client {
+
+  using VehicleControl = carla::rpc::VehicleControl;
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/client/Memory.h

@@ -0,0 +1,31 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+// In this namespace, we use boost::shared_ptr for now to make it compatible
+// with boost::python, but it would be nice to make an adaptor for
+// std::shared_ptr.
+#include <boost/enable_shared_from_this.hpp>
+#include <boost/make_shared.hpp>
+#include <boost/shared_ptr.hpp>
+
+namespace carla {
+namespace client {
+
+  template <typename T>
+  using EnableSharedFromThis = boost::enable_shared_from_this<T>;
+
+  template <typename T>
+  using SharedPtr = boost::shared_ptr<T>;
+
+  template <typename T, typename ... Args>
+  auto MakeShared(Args && ... args) {
+    return boost::make_shared<T>(std::forward<Args>(args) ...);
+  }
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/client/Transform.h

@@ -0,0 +1,19 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/rpc/Transform.h"
+
+namespace carla {
+namespace client {
+
+  using Location = carla::rpc::Location;
+  using Rotation = carla::rpc::Rotation;
+  using Transform = carla::rpc::Transform;
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/client/World.cpp

@@ -0,0 +1,26 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#include "carla/client/World.h"
+
+#include "carla/Logging.h"
+
+namespace carla {
+namespace client {
+
+  SharedPtr<Actor> World::TrySpawnActor(
+      const ActorBlueprint &blueprint,
+      const Transform &transform) {
+    try {
+      return SpawnActor(blueprint, transform);
+    } catch (const std::exception & DEBUG_ONLY(e)) {
+      DEBUG_ONLY(log_debug("TrySpawnActor: failed with:", e.what()));
+      return nullptr;
+    }
+  }
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/client/World.h

@@ -0,0 +1,58 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/Debug.h"
+#include "carla/NonCopyable.h"
+#include "carla/client/Client.h"
+#include "carla/client/Memory.h"
+
+namespace carla {
+namespace client {
+
+  class Actor;
+
+  class World
+    : public EnableSharedFromThis<World>,
+      private NonCopyable {
+  public:
+
+    World(World &&) = default;
+    World &operator=(World &&) = default;
+
+    BlueprintLibrary GetBlueprintLibrary() const {
+      return _parent->GetBlueprintLibrary();
+    }
+
+    SharedPtr<Actor> TrySpawnActor(
+        const ActorBlueprint &blueprint,
+        const Transform &transform);
+
+    SharedPtr<Actor> SpawnActor(
+        const ActorBlueprint &blueprint,
+        const Transform &transform) {
+      return _parent->SpawnActor(blueprint, transform);
+    }
+
+    template <typename ControlType>
+    void ApplyControlToActor(const Actor &actor, const ControlType &control) {
+      _parent->ApplyControlToActor(actor, control);
+    }
+
+  private:
+
+    friend class Client;
+
+    explicit World(SharedPtr<Client> parent) : _parent(std::move(parent)) {
+      DEBUG_ASSERT(parent != nullptr);
+    }
+
+    SharedPtr<Client> _parent;
+  };
+
+} // namespace client
+} // namespace carla

LibCarla/source/carla/profiler/Profiler.cpp

@@ -0,0 +1,88 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#ifndef LIBCARLA_ENABLE_PROFILER
+#  define LIBCARLA_ENABLE_PROFILER
+#endif // LIBCARLA_ENABLE_PROFILER
+
+#include "carla/Logging.h"
+#include "carla/Version.h"
+#include "carla/profiler/Profiler.h"
+
+#include <fstream>
+#include <iomanip>
+#include <iostream>
+#include <mutex>
+
+namespace carla {
+namespace profiler {
+namespace detail {
+
+  template <typename Arg, typename ... Args>
+  static void write_csv_to_stream(std::ostream &out, Arg &&arg, Args &&... args) {
+    out << std::boolalpha
+        << std::left << std::setw(44)
+        << std::forward<Arg>(arg)
+        << std::right
+        << std::fixed << std::setprecision(2);
+    using expander = int[];
+    (void)expander{0, (void(out << ", " << std::setw(10) << std::forward<Args>(args)),0)...};
+  }
+
+  class StaticProfiler {
+  public:
+
+    StaticProfiler(std::string filename)
+      : _filename(std::move(filename)) {
+      logging::log("PROFILER: writing profiling data to", _filename);
+      std::string header = "# LibCarla Profiler ";
+      header += carla::version();
+#ifdef NDEBUG
+      header += " (release)";
+#else
+      header += " (debug)";
+#endif // NDEBUG
+      write_to_file(std::ios_base::out, header);
+      write_line("# context", "average", "maximum", "minimum", "units", "times");
+    }
+
+    template <typename ... Args>
+    void write_line(Args &&... args) {
+      write_to_file(std::ios_base::app|std::ios_base::out, std::forward<Args>(args)...);
+    }
+
+  private:
+
+    template <typename ... Args>
+    void write_to_file(std::ios_base::openmode mode, Args &&... args) {
+      if (!_filename.empty()) {
+        static std::mutex MUTEX;
+        std::lock_guard<std::mutex> guard(MUTEX);
+        std::ofstream file(_filename, mode);
+        write_csv_to_stream(file, std::forward<Args>(args)...);
+        file << std::endl;
+      }
+    }
+
+    const std::string _filename;
+  };
+
+  ProfilerData::~ProfilerData() {
+    static StaticProfiler PROFILER{"profiler.csv"};
+    if (_count > 0u) {
+      if (_print_fps) {
+        PROFILER.write_line(_name, fps(average()), fps(minimum()), fps(maximum()), "FPS", _count);
+      } else {
+        PROFILER.write_line(_name, average(), maximum(), minimum(), "ms", _count);
+      }
+    } else {
+      log_error("profiler", _name, " was never run!");
+    }
+  }
+
+} // namespace detail
+} // namespace profiler
+} // namespace carla

LibCarla/source/carla/profiler/Profiler.h

@@ -0,0 +1,124 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#ifndef LIBCARLA_ENABLE_PROFILER
+#  define CARLA_PROFILE_SCOPE(context, profiler_name)
+#  define CARLA_PROFILE_FPS(context, profiler_name)
+#else
+
+#include "carla/StopWatch.h"
+
+#include <algorithm>
+#include <string>
+
+namespace carla {
+namespace profiler {
+namespace detail {
+
+  class ProfilerData {
+  public:
+
+    explicit ProfilerData(std::string name, bool print_fps = false)
+      : _name(std::move(name)),
+        _print_fps(print_fps) {}
+
+    ~ProfilerData();
+
+    void Annotate(const StopWatch &stop_watch) {
+      const size_t elapsed_microseconds =
+          stop_watch.GetElapsedTime<std::chrono::microseconds>();
+      ++_count;
+      _total_microseconds += elapsed_microseconds;
+      _max_elapsed = std::max(elapsed_microseconds, _max_elapsed);
+      _min_elapsed = std::min(elapsed_microseconds, _min_elapsed);
+    }
+
+    float average() const {
+      return ms(_total_microseconds) / static_cast<float>(_count);
+    }
+
+    float maximum() const {
+      return ms(_max_elapsed);
+    }
+
+    float minimum() const {
+      return ms(_min_elapsed);
+    }
+
+  private:
+
+    static inline float ms(size_t microseconds) {
+      return 1e-3f * static_cast<float>(microseconds);
+    }
+
+    static inline float fps(float milliseconds) {
+      return 1e3f / milliseconds;
+    }
+
+    const std::string _name;
+
+    const bool _print_fps;
+
+    size_t _count = 0u;
+
+    size_t _total_microseconds = 0u;
+
+    size_t _max_elapsed = 0u;
+
+    size_t _min_elapsed = -1;
+  };
+
+  class ScopedProfiler {
+  public:
+
+    explicit ScopedProfiler(ProfilerData &parent) : _profiler(parent) {}
+
+    ~ScopedProfiler() {
+      _stop_watch.Stop();
+      _profiler.Annotate(_stop_watch);
+    }
+
+  private:
+
+    ProfilerData &_profiler;
+
+    StopWatch _stop_watch;
+  };
+
+} // namespace detail
+} // namespace profiler
+} // namespace carla
+
+#ifdef LIBCARLA_WITH_GTEST
+#  define LIBCARLA_GTEST_GET_TEST_NAME() std::string(::testing::UnitTest::GetInstance()->current_test_info()->name())
+#else
+#  define LIBCARLA_GTEST_GET_TEST_NAME() std::string("")
+#endif // LIBCARLA_WITH_GTEST
+
+#define CARLA_PROFILE_SCOPE(context, profiler_name) \
+    static thread_local ::carla::profiler::detail::ProfilerData carla_profiler_ ## context ## _ ## profiler_name ## _data( \
+        LIBCARLA_GTEST_GET_TEST_NAME() + "." #context "." #profiler_name); \
+    ::carla::profiler::detail::ScopedProfiler carla_profiler_ ## context ## _ ## profiler_name ## _scoped_profiler( \
+        carla_profiler_ ## context ## _ ## profiler_name ## _data);
+
+#define CARLA_PROFILE_FPS(context, profiler_name) \
+    { \
+      static thread_local ::carla::StopWatch stop_watch; \
+      stop_watch.Stop(); \
+      static thread_local bool first_time = true; \
+      if (!first_time) { \
+        static thread_local ::carla::profiler::detail::ProfilerData profiler_data( \
+            LIBCARLA_GTEST_GET_TEST_NAME() + "." #context "." #profiler_name, true); \
+        profiler_data.Annotate(stop_watch); \
+      } else { \
+        first_time = false; \
+      } \
+      stop_watch.Restart(); \
+    }
+
+#endif // LIBCARLA_ENABLE_PROFILER

LibCarla/source/carla/rpc/Actor.h

@@ -0,0 +1,27 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/rpc/ActorBlueprint.h"
+
+namespace carla {
+namespace rpc {
+
+  class Actor {
+  public:
+
+    using id_type = uint32_t;
+
+    id_type id;
+
+    ActorBlueprint blueprint;
+
+    MSGPACK_DEFINE_ARRAY(id, blueprint);
+  };
+
+} // namespace rpc
+} // namespace carla

LibCarla/source/carla/rpc/ActorBlueprint.h

@@ -0,0 +1,49 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/rpc/MsgPack.h"
+
+#ifdef LIBCARLA_INCLUDED_FROM_UE4
+#  include "UnrealString.h"
+#endif // LIBCARLA_INCLUDED_FROM_UE4
+
+#include <string>
+
+namespace carla {
+namespace rpc {
+
+  class ActorBlueprint {
+  public:
+
+    ActorBlueprint(std::string type_id)
+      : type_id(std::move(type_id)) {}
+
+    ActorBlueprint() = default;
+    ActorBlueprint(const ActorBlueprint &) = default;
+    ActorBlueprint(ActorBlueprint &&) = default;
+    ActorBlueprint &operator=(const ActorBlueprint &) = default;
+    ActorBlueprint &operator=(ActorBlueprint &&) = default;
+
+#ifdef LIBCARLA_INCLUDED_FROM_UE4
+
+    // ActorBlueprint(const FString &Type)
+    //   : type_id(TCHAR_TO_UTF8(*Type)) {}
+
+    FString GetTypeIdAsFString() const {
+      return FString(type_id.size(), UTF8_TO_TCHAR(type_id.c_str()));
+    }
+
+#endif // LIBCARLA_INCLUDED_FROM_UE4
+
+    std::string type_id;
+
+    MSGPACK_DEFINE_ARRAY(type_id);
+  };
+
+} // namespace rpc
+} // namespace carla

LibCarla/source/carla/rpc/Client.h

@@ -0,0 +1,17 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include <rpc/client.h>
+
+namespace carla {
+namespace rpc {
+
+  using Client = ::rpc::client;
+
+} // namespace rpc
+} // namespace carla

LibCarla/source/carla/rpc/MsgPack.h

@@ -0,0 +1,9 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include <rpc/msgpack.hpp>

LibCarla/source/carla/rpc/Server.h

@@ -0,0 +1,116 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include <boost/asio/io_service.hpp>
+
+#include <rpc/server.h>
+
+namespace carla {
+namespace rpc {
+
+namespace detail {
+
+  /// Function traits based on http://rpclib.net implementation.
+  /// MIT Licensed, Copyright (c) 2015-2017, Tamás Szelei
+  template <typename T>
+  struct wrapper_function_traits : wrapper_function_traits<decltype(&T::operator())> {};
+
+  template <typename C, typename R, typename... Args>
+  struct wrapper_function_traits<R (C::*)(Args...)> : wrapper_function_traits<R (*)(Args...)> {};
+
+  template <typename C, typename R, typename... Args>
+  struct wrapper_function_traits<R (C::*)(Args...) const> : wrapper_function_traits<R (*)(Args...)> {};
+
+  template <typename R, typename... Args> struct wrapper_function_traits<R (*)(Args...)> {
+    using result_type = R;
+    using function_type = std::function<R(Args...)>;
+    using packaged_task_type = std::packaged_task<R()>;
+  };
+
+  /// Wraps @a functor into a function type with equivalent signature. The wrap
+  /// function returned, when called, posts @a functor into the io_service and
+  /// waits for it to finish.
+  ///
+  /// This way, no matter from which thread the wrap function is called, the
+  /// @a functor provided is always called from the context of the io_service.
+  ///
+  /// @warning The wrap function blocks until @a functor is executed by the
+  /// io_service.
+  template <typename F>
+  inline auto WrapSyncCall(boost::asio::io_service &io, F functor) {
+    using func_t = typename wrapper_function_traits<F>::function_type;
+    using task_t = typename wrapper_function_traits<F>::packaged_task_type;
+
+    return func_t([&io, functor=std::move(functor)](auto && ... args) {
+      // We can pass arguments by ref to the lambda because the task will be
+      // executed before this function exits.
+      task_t task([functor=std::move(functor), &args...]() {
+        return functor(std::forward<decltype(args)>(args)...);
+      });
+      auto result = task.get_future();
+      io.post([&]() mutable { task(); });
+      return result.get();
+    });
+  }
+
+} // namespace detail
+
+  /// An RPC server in which functions can be bind to run synchronously or
+  /// asynchronously.
+  ///
+  /// Use `AsyncRun` to start the worker threads, and use `SyncRunFor` to
+  /// run a slice of work in the caller's thread.
+  ///
+  /// Functions that are bind using `BindAsync` will run asynchronously in the
+  /// worker threads. Functions that are bind using `BindSync` will run within
+  /// `SyncRunFor` function.
+  class Server {
+  public:
+
+    template <typename ... Args>
+    explicit Server(Args && ... args)
+      : _server(std::forward<Args>(args) ...) {
+      _server.suppress_exceptions(true);
+    }
+
+    template <typename Functor>
+    void BindAsync(const std::string &name, Functor &&functor) {
+      _server.bind(name, std::forward<Functor>(functor));
+    }
+
+    template <typename Functor>
+    void BindSync(const std::string &name, Functor functor) {
+      _server.bind(
+          name,
+          detail::WrapSyncCall(_sync_io_service, std::move(functor)));
+    }
+
+    void AsyncRun(size_t worker_threads) {
+      _server.async_run(worker_threads);
+    }
+
+    template <typename Duration>
+    void SyncRunFor(Duration duration) {
+      _sync_io_service.reset();
+      _sync_io_service.run_for(duration);
+    }
+
+    /// @warning does not stop the game thread.
+    void Stop() {
+      _server.stop();
+    }
+
+  private:
+
+    boost::asio::io_service _sync_io_service;
+
+    ::rpc::server _server;
+  };
+
+} // namespace rpc
+} // namespace carla

LibCarla/source/carla/rpc/Transform.h

@@ -0,0 +1,102 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/rpc/MsgPack.h"
+
+#ifdef LIBCARLA_INCLUDED_FROM_UE4
+#  include "Transform.h"
+#endif // LIBCARLA_INCLUDED_FROM_UE4
+
+namespace carla {
+namespace rpc {
+
+  class Location {
+  public:
+
+    Location() = default;
+
+    Location(float ix, float iy, float iz)
+      : x(ix),
+        y(iy),
+        z(iz) {}
+
+    float x = 0.0f;
+    float y = 0.0f;
+    float z = 0.0f;
+
+#ifdef LIBCARLA_INCLUDED_FROM_UE4
+
+    Location(const FVector &vector) // from centimeters to meters.
+      : Location(1e-2f * vector.X, 1e-2f * vector.Y, 1e-2f * vector.Z) {}
+
+    operator FVector() const {
+      return FVector{1e2f * x, 1e2f * y, 1e2f * z}; // from meters to centimeters.
+    }
+
+#endif // LIBCARLA_INCLUDED_FROM_UE4
+
+    MSGPACK_DEFINE_ARRAY(x, y, z);
+  };
+
+  class Rotation {
+  public:
+
+    Rotation() = default;
+
+    Rotation(float p, float y, float r)
+      : pitch(p),
+        yaw(y),
+        roll(r) {}
+
+    float pitch = 0.0f;
+    float yaw = 0.0f;
+    float roll = 0.0f;
+
+#ifdef LIBCARLA_INCLUDED_FROM_UE4
+
+    Rotation(const FRotator &rotator)
+      : Rotation(rotator.Pitch, rotator.Yaw, rotator.Roll) {}
+
+    operator FRotator() const {
+      return FRotator{pitch, yaw, roll};
+    }
+
+#endif // LIBCARLA_INCLUDED_FROM_UE4
+
+    MSGPACK_DEFINE_ARRAY(pitch, yaw, roll);
+  };
+
+  class Transform {
+  public:
+
+    Transform() = default;
+
+    Transform(const Location &in_location, const Rotation &in_rotation)
+      : location(in_location),
+        rotation(in_rotation) {}
+
+    Location location;
+    Rotation rotation;
+
+#ifdef LIBCARLA_INCLUDED_FROM_UE4
+
+    Transform(const FTransform &transform)
+      : Transform(Location(transform.GetLocation()), Rotation(transform.Rotator())) {}
+
+    operator FTransform() const {
+      const FVector scale{1.0f, 1.0f, 1.0f};
+      return FTransform{FRotator(rotation), FVector(location), scale};
+    }
+
+#endif // LIBCARLA_INCLUDED_FROM_UE4
+
+    MSGPACK_DEFINE_ARRAY(location, rotation);
+  };
+
+} // namespace rpc
+} // namespace carla

LibCarla/source/carla/rpc/VehicleControl.h

@@ -0,0 +1,46 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/rpc/MsgPack.h"
+
+namespace carla {
+namespace rpc {
+
+  class VehicleControl {
+  public:
+
+    VehicleControl() = default;
+
+    VehicleControl(
+        float in_throttle,
+        float in_steer,
+        float in_brake,
+        bool in_hand_brake,
+        bool in_reverse)
+      : throttle(in_throttle),
+        steer(in_steer),
+        brake(in_brake),
+        hand_brake(in_hand_brake),
+        reverse(in_reverse) {}
+
+    float throttle = 0.0f;
+    float steer = 0.0f;
+    float brake = 0.0f;
+    bool hand_brake = false;
+    bool reverse = false;
+
+    MSGPACK_DEFINE_ARRAY(
+        throttle,
+        steer,
+        brake,
+        hand_brake,
+        reverse);
+  };
+
+} // namespace rpc
+} // namespace carla

LibCarla/source/carla/streaming/Client.h

@@ -0,0 +1,58 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/ThreadGroup.h"
+#include "carla/streaming/low_level/Client.h"
+#include "carla/streaming/low_level/tcp/Client.h"
+
+#include <boost/asio/io_service.hpp>
+
+namespace carla {
+namespace streaming {
+
+  using stream_token = low_level::token_type;
+
+  /// With this client you can subscribe to multiple streams.
+  class Client {
+  public:
+
+    ~Client() {
+      Stop();
+    }
+
+    template <typename Functor>
+    void Subscribe(const stream_token &token, Functor &&callback) {
+      _client.Subscribe(_io_service, token, std::forward<Functor>(callback));
+    }
+
+    void Run() {
+      _io_service.run();
+    }
+
+    void AsyncRun(size_t worker_threads) {
+      _workers.CreateThreads(worker_threads, [this]() { Run(); });
+    }
+
+    void Stop() {
+      _io_service.stop();
+      _workers.JoinAll();
+    }
+
+  private:
+
+    using underlying_client = low_level::Client<low_level::tcp::Client>;
+
+    boost::asio::io_service _io_service;
+
+    ThreadGroup _workers;
+
+    underlying_client _client;
+  };
+
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/Message.h

@@ -0,0 +1,142 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/Debug.h"
+
+#include <boost/asio/buffer.hpp>
+
+#include <array>
+#include <exception>
+#include <limits>
+#include <memory>
+#include <string>
+
+namespace carla {
+namespace streaming {
+
+namespace low_level {
+namespace tcp {
+
+  class Client; /// @todo
+
+} // namespace low_level
+} // namespace tcp
+
+  /// A message owns a buffer with raw data.
+  class Message {
+
+    // =========================================================================
+    // -- Typedefs -------------------------------------------------------------
+    // =========================================================================
+
+  public:
+
+    using value_type = unsigned char;
+
+    using size_type = uint32_t;
+
+    // =========================================================================
+    // -- Construction and assignment ------------------------------------------
+    // =========================================================================
+
+  public:
+
+    Message() = default;
+
+    explicit Message(size_type size)
+      : _size(size),
+        _data(std::make_unique<value_type[]>(_size)) {}
+
+    explicit Message(uint64_t size)
+      : Message([size]() {
+          if (size > std::numeric_limits<size_type>::max()) {
+            throw std::invalid_argument("message size too big");
+          }
+          return static_cast<size_type>(size);
+        } ()) {}
+
+    template <typename ConstBufferSequence>
+    explicit Message(ConstBufferSequence source)
+      : Message(boost::asio::buffer_size(source)) {
+      DEBUG_ONLY(auto bytes_copied = )
+      boost::asio::buffer_copy(buffer(), source);
+      DEBUG_ASSERT(bytes_copied == size());
+    }
+
+    Message(const Message &) = delete;
+    Message &operator=(const Message &) = delete;
+
+    Message(Message &&rhs) noexcept
+      : _size(rhs._size),
+        _data(std::move(rhs._data)) {
+      rhs._size = 0u;
+    }
+
+    Message &operator=(Message &&rhs) noexcept {
+      _size = rhs._size;
+      _data = std::move(rhs._data);
+      rhs._size = 0u;
+      return *this;
+    }
+
+    // =========================================================================
+    // -- Data access ----------------------------------------------------------
+    // =========================================================================
+
+  public:
+
+    bool empty() const {
+      return _size == 0u;
+    }
+
+    size_type size() const {
+      return _size;
+    }
+
+    const value_type *data() const {
+      return _data.get();
+    }
+
+    value_type *data() {
+      return _data.get();
+    }
+
+    // =========================================================================
+    // -- Conversions ----------------------------------------------------------
+    // =========================================================================
+
+  public:
+
+    boost::asio::const_buffer buffer() const {
+      return {data(), size()};
+    }
+
+    boost::asio::mutable_buffer buffer() {
+      return {data(), size()};
+    }
+
+    std::array<boost::asio::const_buffer, 2u> encode() const {
+      DEBUG_ASSERT(!empty());
+      return {boost::asio::buffer(&_size, sizeof(_size)), buffer()};
+    }
+
+    // =========================================================================
+    // -- Private members ------------------------------------------------------
+    // =========================================================================
+
+  private:
+
+    friend class low_level::tcp::Client; /// @todo
+
+    size_type _size = 0u;
+
+    std::unique_ptr<value_type[]> _data = nullptr;
+  };
+
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/Server.h

@@ -0,0 +1,65 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/ThreadGroup.h"
+#include "carla/streaming/low_level/Server.h"
+#include "carla/streaming/low_level/tcp/Server.h"
+
+#include <boost/asio/io_service.hpp>
+
+namespace carla {
+namespace streaming {
+
+  class Server {
+    using underlying_server = low_level::Server<low_level::tcp::Server>;
+  public:
+
+    using duration_type = underlying_server::duration_type;
+
+    explicit Server(uint16_t port)
+      : _server(_io_service, port) {}
+
+    explicit Server(const std::string &address, uint16_t port)
+      : _server(_io_service, address, port) {}
+
+    ~Server() {
+      Stop();
+    }
+
+    void set_timeout(duration_type timeout) {
+      _server.set_timeout(timeout);
+    }
+
+    Stream MakeStream() {
+      return _server.MakeStream();
+    }
+
+    void Run() {
+      _io_service.run();
+    }
+
+    void AsyncRun(size_t worker_threads) {
+      _workers.CreateThreads(worker_threads, [this](){ Run(); });
+    }
+
+    void Stop() {
+      _io_service.stop();
+      _workers.JoinAll();
+    }
+
+  private:
+
+    boost::asio::io_service _io_service;
+
+    underlying_server _server;
+
+    ThreadGroup _workers;
+  };
+
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/Stream.h

@@ -0,0 +1,68 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/Debug.h"
+#include "carla/streaming/Message.h"
+#include "carla/streaming/low_level/StreamState.h"
+#include "carla/streaming/low_level/Token.h"
+
+#include <boost/asio/buffer.hpp>
+
+#include <memory>
+
+namespace carla {
+namespace streaming {
+
+namespace low_level {
+
+  class Dispatcher;
+
+} // namespace low_level
+
+  using stream_token = low_level::token_type;
+
+  class Stream {
+  public:
+
+    Stream() = delete;
+
+    Stream(const Stream &) = default;
+    Stream(Stream &&) = default;
+
+    Stream &operator=(const Stream &) = default;
+    Stream &operator=(Stream &&) = default;
+
+    stream_token token() const {
+      return _shared_state->token();
+    }
+
+    template <typename ConstBufferSequence>
+    void Write(ConstBufferSequence buffer) {
+      _shared_state->Write(std::make_shared<Message>(buffer));
+    }
+
+    template <typename T>
+    Stream &operator<<(const T &rhs) {
+      Write(boost::asio::buffer(rhs));
+      return *this;
+    }
+
+  private:
+
+    friend class low_level::Dispatcher;
+
+    Stream(std::shared_ptr<low_level::StreamState> state)
+      : _shared_state(std::move(state)) {
+      DEBUG_ASSERT(_shared_state != nullptr);
+    }
+
+    std::shared_ptr<low_level::StreamState> _shared_state;
+  };
+
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/low_level/Client.h

@@ -0,0 +1,52 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/streaming/low_level/Token.h"
+
+#include <boost/asio/io_service.hpp>
+
+#include <unordered_set>
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+
+  /// Wrapper around low level clients. You can subscribe to multiple streams.
+  ///
+  /// @warning The client should not be destroyed before the @a io_service is
+  /// stopped.
+  /// @pre T has to be hashable.
+  template <typename T>
+  class Client {
+  public:
+
+    using underlying_client = T;
+
+    template <typename Functor>
+    void Subscribe(
+        boost::asio::io_service &io_service,
+        const token_type &token,
+        Functor &&callback) {
+      if (!token.protocol_is_tcp()) {
+        throw std::invalid_argument("invalid token, only TCP tokens supported");
+      }
+      _clients.emplace(
+          io_service,
+          token.to_tcp_endpoint(),
+          token.get_stream_id(),
+          std::forward<Functor>(callback));
+    }
+
+  private:
+
+    std::unordered_set<underlying_client> _clients;
+  };
+
+} // namespace low_level
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/low_level/Dispatcher.h

@@ -0,0 +1,68 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/streaming/Stream.h"
+#include "carla/streaming/low_level/Session.h"
+#include "carla/streaming/low_level/StreamState.h"
+#include "carla/streaming/low_level/Token.h"
+
+#include <exception>
+#include <memory>
+#include <mutex>
+#include <unordered_map>
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+
+  class Dispatcher {
+  public:
+
+    template <typename P>
+    explicit Dispatcher(const boost::asio::ip::basic_endpoint<P> &ep)
+      : _cached_token(0u, ep) {}
+
+    Stream MakeStream() {
+      std::lock_guard<std::mutex> lock(_mutex);
+      ++_cached_token._token.stream_id; // id zero only happens in overflow.
+      auto ptr = std::make_shared<StreamState>(_cached_token);
+      auto result = _stream_map.emplace(std::make_pair(_cached_token.get_stream_id(), ptr));
+      if (!result.second) {
+        throw std::runtime_error("failed to create stream!");
+      }
+      return ptr;
+    }
+
+    void RegisterSession(std::shared_ptr<Session> session) {
+      DEBUG_ASSERT(session != nullptr);
+      std::lock_guard<std::mutex> lock(_mutex);
+      auto search = _stream_map.find(session->get_stream_id());
+      if (search != _stream_map.end()) {
+        DEBUG_ASSERT(search->second != nullptr);
+        search->second->set_session(std::move(session));
+      } else {
+        log_error("Invalid session: no stream available with id", session->get_stream_id());
+      }
+    }
+
+  private:
+
+    // We use a mutex here, but we assume that sessions and streams won't be
+    // created too often.
+    std::mutex _mutex;
+
+    token_type _cached_token;
+
+    std::unordered_map<
+        stream_id_type,
+        std::shared_ptr<StreamState>> _stream_map;
+  };
+
+} // namespace low_level
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/low_level/Server.h

@@ -0,0 +1,60 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/streaming/low_level/Dispatcher.h"
+#include "carla/streaming/Stream.h"
+
+#include <boost/asio/io_service.hpp>
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+
+  /// Wrapper around low level servers.
+  template <typename T>
+  class Server {
+  public:
+
+    using underlying_server = T;
+
+    using duration_type = typename underlying_server::duration_type;
+    using endpoint = typename underlying_server::endpoint;
+    using protocol_type = typename underlying_server::protocol_type;
+
+    explicit Server(boost::asio::io_service &io_service, const endpoint &ep)
+      : _server(io_service, ep),
+        _dispatcher(ep) {
+      _server.Listen([this](auto session){
+        _dispatcher.RegisterSession(session);
+      });
+    }
+
+    explicit Server(boost::asio::io_service &io_service, uint16_t port)
+      : Server(io_service, endpoint(protocol_type::v4(), port)) {}
+
+    explicit Server(boost::asio::io_service &io_service, const std::string &address, uint16_t port)
+      : Server(io_service, endpoint(boost::asio::ip::address::from_string(address), port)) {}
+
+    void set_timeout(duration_type timeout) {
+      _server.set_timeout(timeout);
+    }
+
+    Stream MakeStream() {
+      return _dispatcher.MakeStream();
+    }
+
+  private:
+
+    underlying_server _server;
+
+    Dispatcher _dispatcher;
+  };
+
+} // namespace low_level
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/low_level/Session.h

@@ -0,0 +1,19 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/streaming/low_level/tcp/ServerSession.h"
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+
+  using Session = tcp::ServerSession;
+
+} // namespace low_level
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/low_level/StreamState.h

@@ -0,0 +1,74 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/streaming/Message.h"
+#include "carla/streaming/low_level/Session.h"
+#include "carla/streaming/low_level/Token.h"
+
+#include <memory>
+#include <mutex>
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+
+namespace detail {
+
+  /// Handles the synchronization of the shared session.
+  class SessionHolder {
+  public:
+
+    void set_session(std::shared_ptr<Session> session) {
+      std::lock_guard<std::mutex> guard(_mutex);
+      _session = std::move(session);
+    }
+
+  protected:
+
+    std::shared_ptr<Session> get_session() const {
+      std::lock_guard<std::mutex> guard(_mutex);
+      return _session;
+    }
+
+  private:
+
+    mutable std::mutex _mutex; /// @todo it can be atomic
+
+    std::shared_ptr<Session> _session;
+  };
+
+} // namespace detail
+
+  /// Shared state among all the copies of a stream. Provides access to the
+  /// underlying UDP session if active.
+  class StreamState
+    : public detail::SessionHolder,
+      private boost::noncopyable {
+  public:
+
+    explicit StreamState(const token_type &token) : _token(token) {}
+
+    const token_type &token() const {
+      return _token;
+    }
+
+    void Write(std::shared_ptr<const Message> message) {
+      auto session = get_session();
+      if (session != nullptr) {
+        session->Write(message);
+      }
+    }
+
+  private:
+
+    const token_type _token;
+  };
+
+} // namespace low_level
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/low_level/Token.h

@@ -0,0 +1,164 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/Debug.h"
+#include "carla/streaming/low_level/Types.h"
+
+#include <boost/asio/ip/address.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/ip/udp.hpp>
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+
+namespace detail {
+
+#pragma pack(push, 1)
+
+  struct token {
+    stream_id_type stream_id;
+
+    uint16_t port;
+
+    enum class protocol : uint8_t {
+      not_set,
+      tcp,
+      udp
+    } protocol = protocol::not_set;
+
+    enum class address : uint8_t {
+      not_set,
+      ip_v4,
+      ip_v6
+    } address_type = address::not_set;
+
+    union {
+      boost::asio::ip::address_v4::bytes_type v4;
+      boost::asio::ip::address_v6::bytes_type v6;
+    } address;
+  };
+
+#pragma pack(pop)
+
+  static_assert(
+      sizeof(token) == 24u,
+      "Size shouldn't be more than"
+      "  v6 address  : 128"
+      "  + state     :  16"
+      "  + port      :  16"
+      "  + stream id :  32"
+      "  -----------------"
+      "                192");
+
+} // namespace detail
+
+  /// Serializes a stream endpoint. Contains all the necessary information for a
+  /// client to subscribe to a stream.
+  class token_type {
+  private:
+
+    template <typename P>
+    static constexpr auto get_protocol() {
+      return std::is_same<P, boost::asio::ip::tcp>::value ?
+          detail::token::protocol::tcp :
+          detail::token::protocol::udp;
+    }
+
+    void set_address(const boost::asio::ip::address &addr) {
+      if (addr.is_v4()) {
+        _token.address_type = detail::token::address::ip_v4;
+        _token.address.v4 = addr.to_v4().to_bytes();
+      } else if (addr.is_v6()) {
+        _token.address_type = detail::token::address::ip_v6;
+        _token.address.v6 = addr.to_v6().to_bytes();
+      } else {
+        throw std::invalid_argument("invalid ip address!");
+      }
+    }
+
+    boost::asio::ip::address get_address() const {
+      if (_token.address_type == detail::token::address::ip_v4) {
+        return boost::asio::ip::address_v4(_token.address.v4);
+      }
+      return boost::asio::ip::address_v6(_token.address.v6);
+    }
+
+    template <typename P>
+    boost::asio::ip::basic_endpoint<P> get_endpoint() const {
+      DEBUG_ASSERT(is_valid());
+      DEBUG_ASSERT(get_protocol<P>() == _token.protocol);
+      return {get_address(), _token.port};
+    }
+
+    template <typename P>
+    explicit token_type(
+        stream_id_type stream_id,
+        const boost::asio::ip::basic_endpoint<P> &ep) {
+      _token.stream_id = stream_id;
+      _token.port = ep.port();
+      _token.protocol = get_protocol<P>();
+      set_address(ep.address());
+    }
+
+  public:
+
+    token_type() = default;
+    token_type(const token_type &) = default;
+
+    auto get_stream_id() const {
+      return _token.stream_id;
+    }
+
+    auto get_port() const {
+      return _token.port;
+    }
+
+    bool is_valid() const {
+      return ((_token.protocol != detail::token::protocol::not_set) &&
+             (_token.address_type != detail::token::address::not_set));
+    }
+
+    bool address_is_v4() const {
+      return _token.address_type == detail::token::address::ip_v4;
+    }
+
+    bool address_is_v6() const {
+      return _token.address_type == detail::token::address::ip_v6;
+    }
+
+    bool protocol_is_udp() const {
+      return _token.protocol == detail::token::protocol::udp;
+    }
+
+    bool protocol_is_tcp() const {
+      return _token.protocol == detail::token::protocol::tcp;
+    }
+
+    boost::asio::ip::udp::endpoint to_udp_endpoint() const {
+      return get_endpoint<boost::asio::ip::udp>();
+    }
+
+    boost::asio::ip::tcp::endpoint to_tcp_endpoint() const {
+      return get_endpoint<boost::asio::ip::tcp>();
+    }
+
+    boost::asio::const_buffer as_buffer() const {
+      return boost::asio::buffer(&_token, sizeof(_token));
+    }
+
+  private:
+
+    friend class Dispatcher;
+
+    detail::token _token;
+  };
+
+} // namespace low_level
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/low_level/Types.h

@@ -0,0 +1,21 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include <cstdint>
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+
+  using stream_id_type = uint32_t;
+
+  using message_size_type = uint32_t;
+
+} // namespace low_level
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/low_level/tcp/Client.h

@@ -0,0 +1,249 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/Debug.h"
+#include "carla/Logging.h"
+#include "carla/streaming/Message.h"
+#include "carla/streaming/low_level/Types.h"
+#include "carla/streaming/low_level/tcp/Timeout.h"
+
+#include <boost/asio/connect.hpp>
+#include <boost/asio/deadline_timer.hpp>
+#include <boost/asio/io_service.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/read.hpp>
+#include <boost/asio/strand.hpp>
+#include <boost/asio/write.hpp>
+
+#include <functional>
+#include <memory>
+#include <string>
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+namespace tcp {
+
+  class Encoder {
+  public:
+
+    boost::asio::mutable_buffer header() {
+      return boost::asio::buffer(reinterpret_cast<unsigned char *>(&_size), sizeof(_size));
+    }
+
+    boost::asio::mutable_buffer body() {
+      DEBUG_ASSERT(_size > 0u);
+      DEBUG_ASSERT(_message == nullptr);
+      _message = std::make_shared<Message>(_size);
+      return _message->buffer();
+    }
+
+    auto size() const {
+      return _size;
+    }
+
+    auto pop() {
+      return std::move(_message);
+    }
+
+  private:
+
+    message_size_type _size = 0u;
+
+    std::shared_ptr<Message> _message;
+  };
+
+  /// @warning The client should not be destroyed before the @a io_service is
+  /// stopped.
+  class Client : private boost::noncopyable {
+  public:
+
+    using endpoint = boost::asio::ip::tcp::endpoint;
+
+    template <typename Functor>
+    Client(
+        boost::asio::io_service &io_service,
+        endpoint ep,
+        stream_id_type stream_id,
+        Functor &&callback)
+      : _endpoint(std::move(ep)),
+        _stream_id(stream_id),
+        _callback(std::forward<Functor>(callback)),
+        _socket(io_service),
+        _strand(io_service),
+        _connection_timer(io_service) {
+      Connect();
+    }
+
+    ~Client() {
+      Stop();
+    }
+
+    stream_id_type get_id() const {
+      return _stream_id;
+    }
+
+    bool operator==(const Client &rhs) const {
+      return get_id() == rhs.get_id();
+    }
+
+    void Stop() {
+      _connection_timer.cancel();
+      _strand.post([this]() {
+        _done = true;
+        if (_socket.is_open()) {
+          _socket.close();
+        }
+      });
+    }
+
+  private:
+
+    /// @todo Stop inlining and make cpp files.
+
+    inline void Connect();
+
+    inline void Reconnect() {
+      _connection_timer.expires_from_now(timeout_type::seconds(1u));
+      _connection_timer.async_wait([this](boost::system::error_code ec) {
+        if (!ec) {
+          Connect();
+        }
+      });
+    }
+
+    inline void ReadData();
+
+    const endpoint _endpoint;
+
+    const stream_id_type _stream_id;
+
+    std::function<void(std::shared_ptr<Message>)> _callback;
+
+    boost::asio::ip::tcp::socket _socket;
+
+    boost::asio::io_service::strand _strand;
+
+    boost::asio::deadline_timer _connection_timer;
+
+    bool _done = false;
+  };
+
+  void Client::Connect() {
+    _strand.post([this]() {
+      if (_done) {
+        return;
+      }
+
+      using boost::system::error_code;
+
+      if (_socket.is_open()) {
+        _socket.close();
+      }
+
+      auto handle_connect = [=](error_code ec) {
+        if (!ec) {
+          // Send the stream id to subscribe to the stream.
+          log_debug("streaming client: sending stream id", _stream_id);
+          boost::asio::async_write(
+              _socket,
+              boost::asio::buffer(&_stream_id, sizeof(_stream_id)),
+              _strand.wrap([=](error_code ec, size_t DEBUG_ONLY(bytes)) {
+            if (!ec) {
+              DEBUG_ASSERT_EQ(bytes, sizeof(_stream_id));
+              // If succeeded start reading data.
+              ReadData();
+            } else {
+              // Else try again.
+              log_debug("streaming client: failed to send stream id:", ec.message());
+              Connect();
+            }
+          }));
+        } else {
+          log_debug("streaming client: connection failed:", ec.message());
+          Reconnect();
+        }
+      };
+
+      log_debug("streaming client: connecting to", _endpoint);
+      _socket.async_connect(_endpoint, _strand.wrap(handle_connect));
+    });
+  }
+
+  void Client::ReadData() {
+    _strand.post([this]() {
+      if (_done) {
+        return;
+      }
+
+      log_debug("streaming client: Client::ReadData");
+
+      auto encoder = std::make_shared<Encoder>();
+
+      auto handle_read_data = [=](boost::system::error_code ec, size_t DEBUG_ONLY(bytes)) {
+        DEBUG_ONLY(log_debug("streaming client: Client::ReadData.handle_read_data", bytes, "bytes"));
+        if (!ec) {
+          DEBUG_ASSERT_EQ(bytes, encoder->size());
+          DEBUG_ASSERT_NE(bytes, 0u);
+          // Move the buffer to the callback function and start reading the next
+          // piece of data.
+          log_debug("streaming client: success reading data, calling the callback");
+          _socket.get_io_service().post([this, encoder]() { _callback(encoder->pop()); });
+          ReadData();
+        } else {
+          // As usual, if anything fails start over from the very top.
+          log_debug("streaming client: failed to read data:", ec.message());
+          Connect();
+        }
+      };
+
+      auto handle_read_header = [=](boost::system::error_code ec, size_t DEBUG_ONLY(bytes)) {
+        DEBUG_ONLY(log_debug("streaming client: Client::ReadData.handle_read_header", bytes, "bytes"));
+        if (!ec && (encoder->size() > 0u)) {
+          DEBUG_ASSERT_EQ(bytes, sizeof(message_size_type));
+          // Now that we know the size of the coming buffer, we can allocate
+          // our buffer and start putting data into it.
+          boost::asio::async_read(
+              _socket,
+              encoder->body(),
+              _strand.wrap(handle_read_data));
+        } else {
+          log_debug("streaming client: failed to read header:", ec.message());
+          DEBUG_ONLY(log_debug("size  = ", encoder->size()));
+          DEBUG_ONLY(log_debug("bytes = ", bytes));
+          Connect();
+        }
+      };
+
+      // Read the size of the buffer that is coming.
+      boost::asio::async_read(
+          _socket,
+          encoder->header(),
+          _strand.wrap(handle_read_header));
+    });
+  }
+
+} // namespace tcp
+} // namespace low_level
+} // namespace streaming
+} // namespace carla
+
+namespace std {
+
+  // Injecting a hash function for our clients into std namespace so we can
+  // directly insert them into std::unordered_set.
+  template <>
+  struct hash<carla::streaming::low_level::tcp::Client> {
+    using argument_type = carla::streaming::low_level::tcp::Client;
+    using result_type = std::size_t;
+    result_type operator()(const argument_type &client) const noexcept {
+      return std::hash<carla::streaming::low_level::stream_id_type>()(client.get_id());
+    }
+  };
+
+} // namespace std

LibCarla/source/carla/streaming/low_level/tcp/Server.h

@@ -0,0 +1,76 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/streaming/low_level/tcp/ServerSession.h"
+
+#include <boost/asio/io_service.hpp>
+#include <boost/asio/ip/tcp.hpp>
+
+#include <atomic>
+#include <memory>
+#include <string>
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+namespace tcp {
+
+  class Server : private boost::noncopyable {
+  public:
+
+    using endpoint = boost::asio::ip::tcp::endpoint;
+    using protocol_type = endpoint::protocol_type;
+    using duration_type = ServerSession::duration_type;
+
+    explicit Server(boost::asio::io_service &io_service, endpoint ep)
+      : _acceptor(io_service, std::move(ep)),
+        _timeout(duration_type::seconds(10u)) {}
+
+    /// Set session time-out. Applies only to newly created sessions.
+    void set_timeout(duration_type timeout) {
+      _timeout = timeout;
+    }
+
+    template <typename Functor>
+    void Listen(Functor callback) {
+      log_info("starting streaming server at port", _acceptor.local_endpoint().port());
+      _acceptor.get_io_service().post([=]() { OpenSession(_timeout, callback); });
+    }
+
+  private:
+
+    template <typename Functor>
+    void OpenSession(duration_type timeout, Functor callback) {
+      using boost::system::error_code;
+
+      auto session = std::make_shared<ServerSession>(_acceptor.get_io_service(), timeout);
+
+      auto handle_query = [=](const error_code &ec) {
+        if (!ec) {
+          session->Open(callback);
+        } else {
+          log_error("tcp accept error:", ec.message());
+        }
+      };
+
+      _acceptor.async_accept(session->_socket, [=](error_code ec) {
+        // Handle query and open a new session immediately.
+        _acceptor.get_io_service().post([=]() { handle_query(ec); });
+        OpenSession(timeout, callback);
+      });
+    }
+
+    boost::asio::ip::tcp::acceptor _acceptor;
+
+    std::atomic<timeout_type> _timeout;
+  };
+
+} // namespace tcp
+} // namespace low_level
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/low_level/tcp/ServerSession.h

@@ -0,0 +1,173 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include "carla/Debug.h"
+#include "carla/Logging.h"
+#include "carla/streaming/Message.h"
+#include "carla/streaming/low_level/Types.h"
+#include "carla/streaming/low_level/tcp/Timeout.h"
+
+#include <boost/asio/deadline_timer.hpp>
+#include <boost/asio/io_service.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/asio/read.hpp>
+#include <boost/asio/strand.hpp>
+#include <boost/asio/write.hpp>
+
+#include <array>
+#include <memory>
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+namespace tcp {
+
+namespace detail {
+
+  static std::atomic_size_t SESSION_COUNTER{0u};
+
+} // namespace detail
+
+  /// A TCP server session. When a session opens, it reads from the socket a
+  /// stream id object and passes itself to the callback functor. The session
+  /// closes itself after @a timeout of inactivity is met.
+  class ServerSession
+    : public std::enable_shared_from_this<ServerSession>,
+      private boost::noncopyable {
+  public:
+
+    using socket_type = boost::asio::ip::tcp::socket;
+    using duration_type = timeout_type;
+
+    explicit ServerSession(boost::asio::io_service &io_service, duration_type timeout)
+      : _session_id(detail::SESSION_COUNTER++),
+        _socket(io_service),
+        _timeout(timeout),
+        _deadline(io_service),
+        _strand(io_service) {}
+
+    ~ServerSession() {
+      _deadline.cancel();
+    }
+
+    /// Starts the session and calls @a callback after successfully reading the
+    /// stream id.
+    ///
+    /// @pre Callback function signature:
+    /// `void(std::shared_ptr<ServerSession>)`.
+    template <typename Functor>
+    void Open(Functor callback) {
+      StartTimer();
+      auto self = shared_from_this(); // To keep myself alive.
+      _strand.post([=]() {
+
+        auto handle_query = [this, self, callback](
+            const boost::system::error_code &ec,
+            size_t DEBUG_ONLY(bytes_received)) {
+          DEBUG_ASSERT_EQ(bytes_received, sizeof(_stream_id));
+          if (!ec) {
+            log_debug("session", _session_id, "for stream", _stream_id, " started");
+            _socket.get_io_service().post([=]() { callback(self); });
+          } else {
+            log_error("session", _session_id, ": error retrieving stream id :", ec.message());
+            Close();
+          }
+        };
+
+        // Read the stream id.
+        _deadline.expires_from_now(_timeout);
+        boost::asio::async_read(
+            _socket,
+            boost::asio::buffer(&_stream_id, sizeof(_stream_id)),
+            _strand.wrap(handle_query));
+      });
+    }
+
+    stream_id_type get_stream_id() const {
+      // Note that the stream id isn't synchronized. This function should only be
+      // called from the @a callback function, and after that point the stream_id
+      // can't change.
+      return _stream_id;
+    }
+
+    /// Writes some data to the socket.
+    void Write(std::shared_ptr<const Message> message) {
+      auto self = shared_from_this();
+      _strand.post([=]() {
+
+        /// @todo has to be a better way of doing this...
+        if (_is_writing) {
+          // Repost and return;
+          Write(std::move(message));
+          return;
+        }
+        _is_writing = true;
+
+        auto handle_sent = [this, self, message](const boost::system::error_code &ec, size_t DEBUG_ONLY(bytes)) {
+          _is_writing = false;
+          if (ec) {
+            log_error("session", _session_id, ": error sending data :", ec.message());
+          } else {
+            DEBUG_ONLY(log_debug("session", _session_id, ": successfully sent", bytes, "bytes"));
+            DEBUG_ASSERT_EQ(bytes, sizeof(message_size_type) + message->size());
+          }
+        };
+
+        log_debug("session", _session_id, ": sending message of", message->size(), "bytes");
+
+        _deadline.expires_from_now(_timeout);
+        boost::asio::async_write(
+            _socket,
+            message->encode(),
+            _strand.wrap(handle_sent));
+      });
+    }
+
+    void Close() {
+      _strand.post([this, self = shared_from_this()]() {
+        if (_socket.is_open()) {
+          _socket.close();
+        }
+        log_debug("session", _session_id, "closed");
+      });
+    }
+
+  private:
+
+    void StartTimer() {
+      if (_deadline.expires_at() <= boost::asio::deadline_timer::traits_type::now()) {
+        log_debug("session", _session_id, "timed out");
+        Close();
+      } else {
+        _deadline.async_wait([self = shared_from_this()](boost::system::error_code) {
+          self->StartTimer();
+        });
+      }
+     }
+
+    friend class Server;
+
+    const size_t _session_id;
+
+    stream_id_type _stream_id = 0u;
+
+    socket_type _socket;
+
+    duration_type _timeout;
+
+    boost::asio::deadline_timer _deadline;
+
+    boost::asio::io_service::strand _strand;
+
+    bool _is_writing = false;
+  };
+
+} // namespace tcp
+} // namespace low_level
+} // namespace streaming
+} // namespace carla

LibCarla/source/carla/streaming/low_level/tcp/Timeout.h

@@ -0,0 +1,62 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include <boost/date_time/posix_time/posix_time_types.hpp>
+
+#include <chrono>
+
+namespace carla {
+namespace streaming {
+namespace low_level {
+namespace tcp {
+
+  /// Positive time-out up to milliseconds resolution.
+  class timeout_type {
+  public:
+
+    static inline timeout_type seconds(size_t timeout) {
+      return std::chrono::seconds(timeout);
+    }
+
+    static inline timeout_type milliseconds(size_t timeout) {
+      return std::chrono::milliseconds(timeout);
+    }
+
+    constexpr timeout_type() : _milliseconds(0u) {}
+
+    template <typename Rep, typename Period>
+    timeout_type(std::chrono::duration<Rep, Period> duration)
+      : _milliseconds(std::chrono::duration_cast<std::chrono::milliseconds>(duration).count()) {}
+
+    timeout_type(boost::posix_time::time_duration timeout)
+      : timeout_type(std::chrono::milliseconds(timeout.total_milliseconds())) {}
+
+    timeout_type(const timeout_type &) = default;
+    timeout_type &operator=(const timeout_type &) = default;
+
+    boost::posix_time::time_duration to_posix_time() const {
+      return boost::posix_time::milliseconds(_milliseconds);
+    }
+
+    constexpr auto to_chrono() const {
+      return std::chrono::milliseconds(_milliseconds);
+    }
+
+    operator boost::posix_time::time_duration() const {
+      return to_posix_time();
+    }
+
+  private:
+
+    size_t _milliseconds;
+  };
+
+} // namespace tcp
+} // namespace low_level
+} // namespace streaming
+} // namespace carla

LibCarla/source/compiler/disable-ue4-macros.h

@@ -0,0 +1,18 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma push_macro("check")
+#undef check
+
+#define LIBCARLA_INCLUDED_FROM_UE4
+
+#ifndef BOOST_ERROR_CODE_HEADER_ONLY
+#  define BOOST_ERROR_CODE_HEADER_ONLY
+#endif // BOOST_ERROR_CODE_HEADER_ONLY
+
+#ifndef BOOST_COROUTINES_NO_DEPRECATION_WARNING
+#  define BOOST_COROUTINES_NO_DEPRECATION_WARNING
+#endif // BOOST_COROUTINES_NO_DEPRECATION_WARNING

LibCarla/source/compiler/enable-ue4-macros.h

@@ -0,0 +1,9 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma pop_macro("check")
+
+#undef LIBCARLA_INCLUDED_FROM_UE4

LibCarla/source/test/test.h

@@ -0,0 +1,26 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#ifndef NDEBUG
+#  define LIBCARLA_LOG_LEVEL LIBCARLA_LOG_LEVEL_INFO
+#endif // NDEBUG
+
+#include "test/util/Message.h"
+
+#include <carla/Logging.h>
+#include <carla/profiler/Profiler.h>
+
+#include <gtest/gtest.h>
+
+#include <chrono>
+#include <cstdint>
+#include <iostream>
+
+constexpr uint16_t TESTING_PORT = 8080u;
+
+using namespace std::chrono_literals;

LibCarla/source/test/test_benchmark_streaming.cpp

@@ -0,0 +1,137 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#include "test.h"
+
+#include <carla/streaming/Client.h>
+#include <carla/streaming/Server.h>
+
+using namespace carla::streaming;
+
+static auto make_special_message(size_t size) {
+  std::vector<uint32_t> v(size/sizeof(uint32_t), 42u);
+  Message msg(boost::asio::buffer(v));
+  EXPECT_EQ(msg.size(), size);
+  return msg;
+}
+
+class Benchmark {
+public:
+
+  Benchmark(uint16_t port, size_t message_size)
+    : _server(port),
+      _client(),
+      _message(make_special_message(message_size)),
+      _client_callback(),
+      _work_to_do(_client_callback) {}
+
+  void AddStream() {
+    Stream stream = _server.MakeStream();
+
+    _client.Subscribe(stream.token(), [this](std::shared_ptr<Message> msg) {
+      DEBUG_ASSERT_EQ(msg->size(), _message.size());
+      DEBUG_ASSERT(*msg == _message);
+      _client_callback.post([this]() {
+        CARLA_PROFILE_FPS(client, listen_callback);
+        ++_number_of_messages_received;
+      });
+    });
+
+    _streams.push_back(stream);
+  }
+
+  void AddStreams(size_t count) {
+    for (auto i = 0u; i < count; ++i) {
+      AddStream();
+    }
+  }
+
+  void Run(size_t number_of_messages) {
+    _threads.CreateThread([this]() { _client_callback.run(); });
+    _server.AsyncRun(_streams.size());
+    _client.AsyncRun(_streams.size());
+
+    std::this_thread::sleep_for(1s); // the client needs to be ready so we make
+                                     // sure we get all the messages.
+
+    for (auto &&stream : _streams) {
+      _threads.CreateThread([=]() mutable {
+        for (auto i = 0u; i < number_of_messages; ++i) {
+          CARLA_PROFILE_SCOPE(game, write_to_stream);
+          stream << _message.buffer();
+        }
+      });
+    }
+
+    for (auto i = 0u; i < 30; ++i) {
+      if (_number_of_messages_received >= (_streams.size() * number_of_messages)) {
+        break;
+      }
+      std::cout << "received only " << _number_of_messages_received
+                << " messages, waiting..." << std::endl;
+      std::this_thread::sleep_for(1s);
+    }
+
+    _client_callback.stop();
+    _threads.JoinAll();
+
+    std::cout << _number_of_messages_received << " messages received; done.\n";
+
+    _client.Stop();
+    _server.Stop();
+  }
+
+private:
+
+  carla::ThreadGroup _threads;
+
+  Server _server;
+
+  Client _client;
+
+  const Message _message;
+
+  boost::asio::io_service _client_callback;
+
+  boost::asio::io_service::work _work_to_do;
+
+  std::vector<Stream> _streams;
+
+  std::atomic_size_t _number_of_messages_received{0u};
+};
+
+static void benchmark_image(
+    const size_t dimensions,
+    const size_t number_of_streams = 1u) {
+  constexpr auto number_of_messages = 100u;
+  Benchmark benchmark(TESTING_PORT, 4u * dimensions);
+  benchmark.AddStreams(number_of_streams);
+  benchmark.Run(number_of_messages);
+}
+
+TEST(benchmark_streaming, image_200x200) {
+  benchmark_image(200u * 200u);
+}
+
+TEST(benchmark_streaming, image_800x600) {
+  benchmark_image(800u * 600u);
+}
+
+TEST(benchmark_streaming, image_1920x1080) {
+  benchmark_image(1920u * 1080u);
+}
+
+TEST(benchmark_streaming, image_200x200_mt) {
+  benchmark_image(200u * 200u, 9u);
+}
+
+TEST(benchmark_streaming, image_800x600_mt) {
+  benchmark_image(800u * 600u, 9u);
+}
+
+TEST(benchmark_streaming, image_1920x1080_mt) {
+  benchmark_image(1920u * 1080u, 9u);
+}

LibCarla/source/test/test_miscellaneous.cpp

@@ -0,0 +1,13 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#include "test.h"
+
+#include <carla/Version.h>
+
+TEST(miscellaneous, version) {
+  std::cout << "LibCarla " << carla::version() << std::endl;
+}

LibCarla/source/test/test_rpc.cpp

@@ -0,0 +1,58 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#include "test.h"
+
+#include <carla/ThreadGroup.h>
+#include <carla/rpc/Client.h>
+#include <carla/rpc/Server.h>
+
+#include <thread>
+
+using namespace carla::rpc;
+
+TEST(rpc, compilation_tests) {
+  Server server(TESTING_PORT);
+  server.BindSync("bind00", []() { return 2.0f; });
+  server.BindSync("bind01", [](int x) { return x; });
+  server.BindSync("bind02", [](int, float) { return 0.0; });
+  server.BindSync("bind03", [](int, float, double, char) {});
+}
+
+TEST(rpc, server_bind_sync_run_on_game_thread) {
+  const auto main_thread_id = std::this_thread::get_id();
+
+  Server server(TESTING_PORT);
+
+  server.BindSync("do_the_thing", [=](int x, int y) -> int {
+    EXPECT_EQ(std::this_thread::get_id(), main_thread_id);
+    return x + y;
+  });
+
+  server.AsyncRun(1u);
+
+  std::atomic_bool done{false};
+
+  carla::ThreadGroup threads;
+  threads.CreateThread([&]() {
+    Client client("localhost", TESTING_PORT);
+    for (auto i = 0u; i < 300u; ++i) {
+      auto result = client.call("do_the_thing", i, 1).as<int>();
+      EXPECT_EQ(result, i + 1);
+    }
+    done = true;
+  });
+
+  auto i = 0u;
+  for (; i < 1'000'000u; ++i) {
+    server.SyncRunFor(2ms);
+    if (done) {
+      break;
+    }
+  }
+  std::cout << "game thread: run " << i << " slices.\n";
+  ASSERT_TRUE(done);
+}

LibCarla/source/test/test_streaming_low_level.cpp

@@ -0,0 +1,55 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#include "test.h"
+
+#include <carla/ThreadGroup.h>
+#include <carla/streaming/low_level/Client.h>
+#include <carla/streaming/low_level/Server.h>
+#include <carla/streaming/low_level/tcp/Client.h>
+#include <carla/streaming/low_level/tcp/Server.h>
+
+#include <atomic>
+
+TEST(streaming_low_level, sending_strings) {
+  using namespace util::message;
+  using namespace carla::streaming::low_level;
+
+  constexpr auto number_of_messages = 5'000u;
+  const std::string message_text = "Hello client!";
+
+  std::atomic_size_t message_count{0u};
+
+  boost::asio::io_service io_service;
+
+  Server<tcp::Server> srv(io_service, TESTING_PORT);
+  srv.set_timeout(1s);
+
+  auto stream = srv.MakeStream();
+
+  Client<tcp::Client> c;
+  c.Subscribe(io_service, stream.token(), [&](auto message) {
+    ++message_count;
+    ASSERT_NE(message, nullptr);
+    ASSERT_EQ(message->size(), message_text.size());
+    const std::string msg = as_string(*message);
+    ASSERT_EQ(msg, message_text);
+  });
+
+  carla::ThreadGroup threads;
+  threads.CreateThreads(
+      std::max(2u, std::thread::hardware_concurrency()),
+      [&]() { io_service.run(); });
+
+  for (auto i = 0u; i < number_of_messages; ++i) {
+    stream << message_text;
+  }
+
+  std::this_thread::sleep_for(1s);
+  io_service.stop();
+
+  std::cout << "client received " << message_count << " messages\n";
+}

LibCarla/source/test/test_streaming_low_level_tcp.cpp

@@ -0,0 +1,57 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#include "test.h"
+
+#include <carla/ThreadGroup.h>
+#include <carla/streaming/low_level/tcp/Client.h>
+#include <carla/streaming/low_level/tcp/Server.h>
+
+#include <atomic>
+
+TEST(streaming_low_level_tcp, small_message) {
+  using namespace util::message;
+  using namespace carla::streaming::low_level;
+  using shared_session = std::shared_ptr<tcp::ServerSession>;
+
+  boost::asio::io_service io_service;
+  tcp::Server::endpoint ep(boost::asio::ip::tcp::v4(), TESTING_PORT);
+
+  tcp::Server srv(io_service, ep);
+  srv.set_timeout(1s);
+  std::atomic_bool done{false};
+  std::atomic_size_t message_count{0u};
+
+  srv.Listen([&](std::shared_ptr<tcp::ServerSession> session) {
+    ASSERT_EQ(session->get_stream_id(), 42u);
+    const std::string msg = "Hola!";
+    auto message = std::make_shared<Message>(boost::asio::buffer(msg));
+    while (!done) {
+      session->Write(message);
+      std::this_thread::sleep_for(1ns);
+    }
+    std::cout << "done!\n";
+  });
+
+  tcp::Client c(io_service, ep, 42u, [&](std::shared_ptr<Message> message) {
+    ++message_count;
+    ASSERT_NE(message, nullptr);
+    ASSERT_EQ(message->size(), 5u);
+    const std::string msg = as_string(*message);
+    ASSERT_EQ(msg, std::string("Hola!"));
+  });
+
+  // We need at least two threads because this server loop consumes one.
+  carla::ThreadGroup threads;
+  threads.CreateThreads(
+      std::max(2u, std::thread::hardware_concurrency()),
+      [&]() { io_service.run(); });
+
+  std::this_thread::sleep_for(2s);
+  io_service.stop();
+  done = true;
+  std::cout << "client received " << message_count << " messages\n";
+}

LibCarla/source/test/test_streaming_message.cpp

@@ -0,0 +1,52 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#include "test.h"
+
+#include <carla/streaming/Message.h>
+
+#include <cstring>
+#include <string>
+#include <vector>
+
+using namespace util::message;
+
+TEST(streaming_message, to_from_string) {
+  const std::string str = "The quick brown fox jumps over the lazy dog";
+  Message msg(boost::asio::buffer(str));
+  ASSERT_EQ(msg.size(), str.size());
+  const std::string result = as_string(msg);
+  ASSERT_EQ(result, str);
+}
+
+TEST(streaming_message, to_from_vector) {
+  constexpr auto size = 1000u;
+  using T = size_t;
+  std::vector<T> v;
+  v.reserve(size);
+  for (auto i = 0u; i < size; ++i) {
+    v.emplace_back(i);
+  }
+  Message msg(boost::asio::buffer(v));
+  ASSERT_EQ(msg.size(), sizeof(T) * size);
+  auto begin = reinterpret_cast<const T *>(msg.data());
+  std::vector<T> result(begin, begin + size);
+  ASSERT_EQ(result, v);
+}
+
+TEST(streaming_message, memcpy) {
+  const std::string str = "The quick brown fox jumps over the lazy dog";
+  Message msg(str.size());
+  ASSERT_EQ(msg.size(), str.size());
+  auto buffer = msg.buffer();
+  std::memcpy(buffer.data(), str.data(), buffer.size());
+  const std::string result = as_string(msg);
+  ASSERT_EQ(result, str);
+}
+
+TEST(streaming_message, message_too_big) {
+  ASSERT_THROW(Message(4294967296ul), std::invalid_argument);
+}

LibCarla/source/test/util/Message.cpp

@@ -0,0 +1,45 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#include "Message.h"
+
+#include <climits>
+#include <random>
+
+namespace util {
+namespace message {
+
+  shared_message make_empty(size_t size) {
+    return size == 0u ?
+        std::make_shared<Message>() :
+        std::make_shared<Message>(size);
+  }
+
+  shared_message make_random(size_t size) {
+    if (size == 0u)
+      return make_empty();
+    using random_bytes_engine = std::independent_bits_engine<
+        std::random_device,
+        CHAR_BIT,
+        unsigned char>;
+    random_bytes_engine rbe;
+    auto message = make_empty(size);
+    std::generate(begin(*message), end(*message), std::ref(rbe));
+    return message;
+  }
+
+  std::string to_hex_string(const Message &msg, size_t length) {
+    length = std::min(static_cast<size_t>(msg.size()), length);
+    auto buffer = std::make_unique<char[]>(2u * length + 1u);
+    for (auto i = 0u; i < length; ++i)
+      sprintf(&buffer[2u * i], "%02x", msg.data()[i]);
+    if (length < msg.size())
+      return std::string(buffer.get()) + std::string("...");
+    return std::string(buffer.get());
+  }
+
+} // namespace message
+} // namespace util

LibCarla/source/test/util/Message.h

@@ -0,0 +1,81 @@
+// Copyright (c) 2017 Computer Vision Center (CVC) at the Universitat Autonoma
+// de Barcelona (UAB).
+//
+// This work is licensed under the terms of the MIT license.
+// For a copy, see <https://opensource.org/licenses/MIT>.
+
+#pragma once
+
+#include <carla/streaming/Message.h>
+
+#include <algorithm>
+#include <memory>
+#include <ostream>
+#include <string>
+
+namespace util {
+namespace message {
+
+  using carla::streaming::Message;
+
+  using shared_message = std::shared_ptr<Message>;
+  using const_shared_message = std::shared_ptr<const Message>;
+
+  static inline shared_message make_empty() {
+    return std::make_shared<Message>();
+  }
+
+  shared_message make_empty(size_t size);
+
+  shared_message make_random(size_t size);
+
+  template <typename T>
+  static inline shared_message make(const T &buffer) {
+    return std::make_shared<Message>(boost::asio::buffer(buffer));
+  }
+
+  static inline std::string as_string(const Message &msg) {
+    return {reinterpret_cast<const char *>(msg.data()), msg.size()};
+  }
+
+  std::string to_hex_string(const Message &msg, size_t length = 16u);
+
+} // namespace message
+} // namespace util
+
+namespace carla {
+namespace streaming {
+
+  static inline unsigned char *begin(Message &msg) {
+    return msg.data();
+  }
+
+  static inline const unsigned char *begin(const Message &msg) {
+    return msg.data();
+  }
+
+  static inline unsigned char *end(Message &msg) {
+    return msg.data() + msg.size();
+  }
+
+  static inline const unsigned char *end(const Message &msg) {
+    return msg.data() + msg.size();
+  }
+
+  static inline std::ostream &operator<<(std::ostream &out, const Message &msg) {
+    out << "[" << msg.size() << " bytes] " << util::message::to_hex_string(msg);
+    return out;
+  }
+
+  static inline bool operator==(const Message &lhs, const Message &rhs) {
+    return
+        (lhs.size() == rhs.size()) &&
+        std::equal(begin(lhs), end(lhs), begin(rhs));
+  }
+
+  static inline bool operator!=(const Message &lhs, const Message &rhs) {
+    return !(lhs == rhs);
+  }
+
+} // namespace streaming
+} // namespace carla