Add support for adb connections

This change is quite large as it includes a small rework of how some
parts of the networking layer are written.

src/CMakeLists.txt

@@ -75,10 +75,13 @@ set(SOURCES
     anbox/network/connections.h
     anbox/network/socket_connection.cpp
     anbox/network/socket_messenger.cpp
-    anbox/network/delegate_message_processor.h
+    anbox/network/delegate_message_processor.cpp
     anbox/network/delegate_connection_creator.h
+    anbox/network/base_socket_messenger.cpp
     anbox/network/local_socket_messenger.cpp
+    anbox/network/tcp_socket_messenger.cpp
     anbox/network/socket_helper.cpp
+    anbox/network/tcp_socket_connector.cpp
 
     anbox/rpc/channel.cpp
     anbox/rpc/pending_call_cache.cpp
@@ -107,6 +110,7 @@ set(SOURCES
     anbox/qemu/gsm_message_processor.cpp
     anbox/qemu/at_parser.cpp
     anbox/qemu/bootanimation_message_processor.cpp
+    anbox/qemu/adb_message_processor.cpp
 
     anbox/bridge/platform_message_processor.cpp
     anbox/bridge/platform_api_skeleton.cpp

src/anbox/cmds/run.cpp

@@ -45,7 +45,7 @@
 namespace fs = boost::filesystem;
 
 namespace {
-class NullConnectionCreator : public anbox::network::ConnectionCreator {
+class NullConnectionCreator : public anbox::network::ConnectionCreator<boost::asio::local::stream_protocol> {
 public:
     void create_connection_for(
             std::shared_ptr<boost::asio::local::stream_protocol::socket> const& socket) override {
@@ -94,10 +94,12 @@ anbox::cmds::Run::Run(const BusFactory& bus_factory)
 
         // Socket which will be used by the qemud service inside the Android
         // container for things like sensors, vibrtator etc.
+#if 0
         auto qemud_connector = std::make_shared<network::PublishedSocketConnector>(
             utils::string_format("%s/qemud", config::socket_path()),
             rt,
             std::make_shared<NullConnectionCreator>());
+#endif
 
         // The qemu pipe is used as a very fast communication channel between guest
         // and host for things like the GLES emulation/translation, the RIL or ADB.
@@ -135,7 +137,7 @@ anbox::cmds::Run::Run(const BusFactory& bus_factory)
         container::Client container(rt);
         container::Configuration container_configuration;
         container_configuration.bind_mounts = {
-            { qemud_connector->socket_file(), "/dev/qemud" },
+            // { qemud_connector->socket_file(), "/dev/qemud" },
             { qemu_pipe_connector->socket_file(), "/dev/qemu_pipe" },
             { bridge_connector->socket_file(), "/dev/anbox_bridge" },
             { config::host_input_device_path(), "/dev/input" },

src/anbox/container/service.cpp

@@ -18,7 +18,7 @@
 #include "anbox/container/service.h"
 #include "anbox/network/delegate_connection_creator.h"
 #include "anbox/network/delegate_message_processor.h"
-#include "anbox/network/socket_messenger.h"
+#include "anbox/network/local_socket_messenger.h"
 #include "anbox/qemu/null_message_processor.h"
 #include "anbox/rpc/pending_call_cache.h"
 #include "anbox/rpc/channel.h"
@@ -33,7 +33,7 @@ namespace container {
 std::shared_ptr<Service> Service::create(const std::shared_ptr<Runtime> &rt) {
     auto sp = std::make_shared<Service>(rt);
 
-    auto delegate_connector = std::make_shared<network::DelegateConnectionCreator>(
+    auto delegate_connector = std::make_shared<network::DelegateConnectionCreator<boost::asio::local::stream_protocol>>(
             [sp](std::shared_ptr<boost::asio::local::stream_protocol::socket> const &socket) {
         sp->new_client(socket);
     });
@@ -69,7 +69,7 @@ void Service::new_client(std::shared_ptr<boost::asio::local::stream_protocol::so
         return;
     }
 
-    auto const messenger = std::make_shared<network::SocketMessenger>(socket);
+    auto const messenger = std::make_shared<network::LocalSocketMessenger>(socket);
 
     DEBUG("Got connection from pid %d", messenger->creds().pid());
 
@@ -82,6 +82,7 @@ void Service::new_client(std::shared_ptr<boost::asio::local::stream_protocol::so
 
     auto const& connection = std::make_shared<network::SocketConnection>(
                 messenger, messenger, next_id(), connections_, processor);
+    connection->set_name("container-service");
 
     connections_->add(connection);
     connection->read_next_message();

src/anbox/container/service.h

@@ -42,7 +42,7 @@ private:
     std::shared_ptr<common::Dispatcher> dispatcher_;
     std::shared_ptr<network::PublishedSocketConnector> connector_;
     std::atomic<int> next_connection_id_;
-    std::shared_ptr<network::Connections<network::SocketConnection>> const connections_;
+    std::shared_ptr<network::Connections<network::SocketConnection>> connections_;
     std::shared_ptr<Container> backend_;
 };
 } // namespace container

src/anbox/graphics/opengles_message_processor.cpp

@@ -17,13 +17,12 @@
 
 #include "anbox/logger.h"
 #include "anbox/graphics/opengles_message_processor.h"
-#include "anbox/network/socket_messenger.h"
+#include "anbox/network/local_socket_messenger.h"
 #include "anbox/network/connections.h"
 #include "anbox/network/delegate_message_processor.h"
 
 namespace anbox {
 namespace graphics {
-
 OpenGlesMessageProcessor::OpenGlesMessageProcessor(const std::string &renderer_socket_path,
                                                    const std::shared_ptr<Runtime> &rt,
                                                    const std::shared_ptr<network::SocketMessenger> &messenger) :
@@ -41,17 +40,14 @@ void OpenGlesMessageProcessor::connect_and_attach(const std::string &socket_path
     auto socket = std::make_shared<boost::asio::local::stream_protocol::socket>(rt->service());
     socket->connect(boost::asio::local::stream_protocol::endpoint(socket_path));
 
-    messenger_ = std::make_shared<network::SocketMessenger>(socket);
+    messenger_ = std::make_shared<network::LocalSocketMessenger>(socket);
     renderer_ = std::make_shared<network::SocketConnection>(
-                    messenger_,
+                    messenger_, messenger_, 0, nullptr,
-                    messenger_,
-                    0,
-                    std::make_shared<network::Connections<network::SocketConnection>>(),
                     std::make_shared<network::DelegateMessageProcessor>([&](const std::vector<std::uint8_t> &data) {
         client_messenger_->send(reinterpret_cast<const char*>(data.data()), data.size());
         return true;
     }));
-
+    renderer_->set_name("opengles-renderer");
     renderer_->read_next_message();
 }
 
@@ -59,7 +55,5 @@ bool OpenGlesMessageProcessor::process_data(const std::vector<std::uint8_t> &dat
     messenger_->send(reinterpret_cast<const char*>(data.data()), data.size());
     return true;
 }
-
-
 } // namespace graphics
 } // namespace anbox

src/anbox/input/device.cpp

@@ -18,7 +18,7 @@
 #include "anbox/input/device.h"
 #include "anbox/network/delegate_connection_creator.h"
 #include "anbox/network/delegate_message_processor.h"
-#include "anbox/network/socket_messenger.h"
+#include "anbox/network/local_socket_messenger.h"
 #include "anbox/qemu/null_message_processor.h"
 #include "anbox/logger.h"
 
@@ -29,7 +29,7 @@ namespace input {
 std::shared_ptr<Device> Device::create(const std::string &path, const std::shared_ptr<Runtime> &runtime) {
     auto sp = std::make_shared<Device>();
 
-    auto delegate_connector = std::make_shared<network::DelegateConnectionCreator>(
+    auto delegate_connector = std::make_shared<network::DelegateConnectionCreator<boost::asio::local::stream_protocol>>(
             [sp](std::shared_ptr<boost::asio::local::stream_protocol::socket> const &socket) {
         sp->new_client(socket);
     });
@@ -152,11 +152,11 @@ int Device::next_id() {
 }
 
 void Device::new_client(std::shared_ptr<boost::asio::local::stream_protocol::socket> const &socket) {
-    auto const messenger = std::make_shared<network::SocketMessenger>(socket);
+    auto const messenger = std::make_shared<network::LocalSocketMessenger>(socket);
     auto const& connection = std::make_shared<network::SocketConnection>(
                 messenger, messenger, next_id(), connections_,
                 std::make_shared<qemu::NullMessageProcessor>());
-
+    connection->set_name("input-device");
     connections_->add(connection);
 
     // Send all necessary information about our device so that the remote

src/anbox/input/device.h

@@ -90,7 +90,7 @@ private:
 
     std::shared_ptr<network::PublishedSocketConnector> connector_;
     std::atomic<int> next_connection_id_;
-    std::shared_ptr<network::Connections<network::SocketConnection>> const connections_;
+    std::shared_ptr<network::Connections<network::SocketConnection>> connections_;
     Info info_;
 };
 } // namespace input

src/anbox/network/base_socket_messenger.cpp

@@ -0,0 +1,159 @@
+/*
+ * Copyright © 2013-2014 Canonical Ltd.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 3 as
+ * published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * Authored by: Kevin DuBois <kevin.dubois@canonical.com>
+ */
+
+#include "anbox/network/base_socket_messenger.h"
+#include "anbox/common/variable_length_array.h"
+#include "anbox/logger.h"
+
+#include <boost/throw_exception.hpp>
+
+#include <errno.h>
+#include <string.h>
+
+#include <stdexcept>
+
+namespace bs = boost::system;
+namespace ba = boost::asio;
+
+namespace {
+/// Buffers need to be big enough to support messages
+unsigned int const serialization_buffer_size = 2048;
+}
+
+namespace anbox {
+namespace network {
+template<typename stream_protocol>
+BaseSocketMessenger<stream_protocol>::BaseSocketMessenger() {
+}
+
+template<typename stream_protocol>
+BaseSocketMessenger<stream_protocol>::BaseSocketMessenger(std::shared_ptr<ba::basic_stream_socket<stream_protocol>> const& socket)
+{
+    setup(socket);
+}
+
+template<typename stream_protocol>
+BaseSocketMessenger<stream_protocol>::~BaseSocketMessenger()
+{
+}
+
+template<typename stream_protocol>
+void BaseSocketMessenger<stream_protocol>::setup(std::shared_ptr<ba::basic_stream_socket<stream_protocol>> const& s) {
+    socket = s;
+    socket_fd = anbox::Fd{IntOwnedFd{socket->native_handle()}};
+    socket->non_blocking(true);
+    boost::asio::socket_base::send_buffer_size option(64*1024);
+    socket->set_option(option);
+}
+
+template<typename stream_protocol>
+Credentials BaseSocketMessenger<stream_protocol>::creds() const {
+    struct ucred cr;
+    socklen_t cl = sizeof(cr);
+
+    auto status = getsockopt(socket_fd, SOL_SOCKET, SO_PEERCRED, &cr, &cl);
+
+    if (status)
+        BOOST_THROW_EXCEPTION(std::runtime_error("Failed to query client socket credentials"));
+
+    return {cr.pid, cr.uid, cr.gid};
+}
+
+template<typename stream_protocol>
+void BaseSocketMessenger<stream_protocol>::send(char const* data, size_t length)
+{
+    VariableLengthArray<serialization_buffer_size> whole_message{length};
+    std::copy(data, data + length, whole_message.data());
+
+    for (;;) {
+        try {
+            std::unique_lock<std::mutex> lg(message_lock);
+            ba::write(*socket,
+                      ba::buffer(whole_message.data(), whole_message.size()),
+                      boost::asio::transfer_all());
+        }
+        catch (const boost::system::system_error &err) {
+            if (err.code() == boost::asio::error::try_again)
+                continue;
+        }
+        break;
+    }
+}
+
+template<typename stream_protocol>
+void BaseSocketMessenger<stream_protocol>::async_receive_msg(
+    AnboxReadHandler const& handler,
+    ba::mutable_buffers_1 const& buffer)
+{
+    socket->async_read_some(buffer, handler);
+}
+
+template<typename stream_protocol>
+bs::error_code BaseSocketMessenger<stream_protocol>::receive_msg(
+    ba::mutable_buffers_1 const& buffer)
+{
+    bs::error_code e;
+    size_t nread = 0;
+
+    while (nread < ba::buffer_size(buffer))
+    {
+        nread += boost::asio::read(
+             *socket,
+             ba::mutable_buffers_1{buffer + nread},
+             e);
+
+        if (e && e != ba::error::would_block)
+            break;
+    }
+
+    return e;
+}
+
+template<typename stream_protocol>
+size_t BaseSocketMessenger<stream_protocol>::available_bytes()
+{
+    boost::asio::socket_base::bytes_readable command{true};
+    socket->io_control(command);
+    return command.get();
+}
+
+template<typename stream_protocol>
+unsigned short BaseSocketMessenger<stream_protocol>::local_port() const {
+    return 0;
+}
+
+template<typename stream_protocol>
+void BaseSocketMessenger<stream_protocol>::set_no_delay() {
+    const auto fd = socket->native();
+    int flag = 1;
+    const auto ret = ::setsockopt(fd, IPPROTO_TCP, TCP_NODELAY,
+                                  reinterpret_cast<const char*>(&flag),
+                                  sizeof(flag));
+    if (ret < 0)
+        WARNING("Failed to disable TCP delay for socket");
+}
+
+template<typename stream_protocol>
+void BaseSocketMessenger<stream_protocol>::close() {
+    socket->close();
+}
+
+template class BaseSocketMessenger<boost::asio::local::stream_protocol>;
+template class BaseSocketMessenger<boost::asio::ip::tcp>;
+} // namespace network
+} // namespace anbox

src/anbox/network/base_socket_messenger.h

@@ -0,0 +1,59 @@
+/*
+ * Copyright © 2013-2014 Canonical Ltd.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 3 as
+ * published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * Authored by: Kevin DuBois <kevin.dubois@canonical.com>
+ */
+
+#ifndef ANBOX_NETWORK_BASE_SOCKET_MESSENGER_H_
+#define ANBOX_NETWORK_BASE_SOCKET_MESSENGER_H_
+
+#include "anbox/network/socket_messenger.h"
+#include "anbox/common/fd_sets.h"
+
+#include <mutex>
+#include <boost/asio.hpp>
+
+namespace anbox {
+namespace network {
+template<typename stream_protocol>
+class BaseSocketMessenger : public SocketMessenger {
+public:
+    BaseSocketMessenger(std::shared_ptr<boost::asio::basic_stream_socket<stream_protocol>> const& socket);
+    virtual ~BaseSocketMessenger();
+
+    Credentials creds() const override;
+    unsigned short local_port() const override;
+
+    void send(char const* data, size_t length) override;
+    void async_receive_msg(AnboxReadHandler const& handle, boost::asio::mutable_buffers_1 const &buffer) override;
+    boost::system::error_code receive_msg(boost::asio::mutable_buffers_1 const& buffer) override;
+    size_t available_bytes() override;
+
+    void set_no_delay() override;
+    void close() override;
+
+protected:
+    BaseSocketMessenger();
+    void setup(std::shared_ptr<boost::asio::basic_stream_socket<stream_protocol>> const& s);
+
+private:
+    std::shared_ptr<boost::asio::basic_stream_socket<stream_protocol>> socket;
+    anbox::Fd socket_fd;
+    std::mutex message_lock;
+};
+} // namespace network
+} // namespace anbox
+
+#endif

src/anbox/network/connection_creator.h

@@ -25,10 +25,11 @@
 #include "anbox/do_not_copy_or_move.h"
 namespace anbox {
 namespace network {
+template<typename stream_protocol>
 class ConnectionCreator : public DoNotCopyOrMove {
 public:
     virtual void create_connection_for(
-        std::shared_ptr<boost::asio::local::stream_protocol::socket> const& socket) = 0;
+        std::shared_ptr<boost::asio::basic_stream_socket<stream_protocol>> const& socket) = 0;
 };
 } // namespace anbox
 } // namespace network

src/anbox/network/connections.h

@@ -29,34 +29,22 @@ template<class Connection>
 class Connections
 {
 public:
-    class Observer {
-    public:
-        virtual void connection_added(int id) = 0;
-        virtual void connection_removed(int id) = 0;
-    };
-
     Connections() {}
-    ~Connections() { clear(); }
+    ~Connections() {
-
+        clear();
-    void set_observer(const std::shared_ptr<Observer> &o)
-    {
-        observer = o;
     }
 
+
     void add(std::shared_ptr<Connection> const& connection)
     {
         std::unique_lock<std::mutex> lock(mutex);
         connections.insert({connection->id(), connection});
-        if (observer)
-            observer->connection_added(connection->id());
     }
 
     void remove(int id)
     {
         std::unique_lock<std::mutex> lock(mutex);
         connections.erase(id);
-        if (observer)
-            observer->connection_removed(id);
     }
 
     bool includes(int id) const
@@ -87,7 +75,6 @@ private:
 
     std::mutex mutex;
     std::map<int, std::shared_ptr<Connection>> connections;
-    std::shared_ptr<Observer> observer;
 };
 } // namespace anbox
 } // namespace network

src/anbox/network/delegate_connection_creator.h

@@ -25,21 +25,22 @@
 
 namespace anbox {
 namespace network {
-class DelegateConnectionCreator : public ConnectionCreator {
+template<typename stream_protocol>
+class DelegateConnectionCreator : public ConnectionCreator<stream_protocol> {
 public:
-    DelegateConnectionCreator(std::function<void(std::shared_ptr<boost::asio::local::stream_protocol::socket> const&)> delegate) :
+    DelegateConnectionCreator(std::function<void(std::shared_ptr<boost::asio::basic_stream_socket<stream_protocol>> const&)> delegate) :
         delegate_(delegate) {
     }
 
     void create_connection_for(
-            std::shared_ptr<boost::asio::local::stream_protocol::socket> const& socket) override {
+            std::shared_ptr<boost::asio::basic_stream_socket<stream_protocol>> const& socket) override {
         if (delegate_)
             delegate_(socket);
         else
             socket->close();
     }
 private:
-    std::function<void(std::shared_ptr<boost::asio::local::stream_protocol::socket> const&)> delegate_;
+    std::function<void(std::shared_ptr<boost::asio::basic_stream_socket<stream_protocol>> const&)> delegate_;
 };
 } // namespace network
 } // namespace anbox

src/anbox/network/delegate_message_processor.cpp

@@ -0,0 +1,36 @@
+/*
+ * Copyright (C) 2016 Simon Fels <morphis@gravedo.de>
+ *
+ * This program is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 3, as published
+ * by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranties of
+ * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include "anbox/network/delegate_message_processor.h"
+
+namespace anbox {
+namespace network {
+DelegateMessageProcessor::DelegateMessageProcessor(std::function<bool(const std::vector<std::uint8_t>&)> process_data) :
+    process_data_(process_data) {
+}
+
+DelegateMessageProcessor::~DelegateMessageProcessor() {
+}
+
+bool DelegateMessageProcessor::process_data(const std::vector<std::uint8_t> &data) {
+    if (!process_data_)
+        return false;
+
+    return process_data_(data);
+}
+} // namespace network
+} // namespace anbox

src/anbox/network/delegate_message_processor.h

@@ -24,24 +24,16 @@
 
 namespace anbox {
 namespace network {
-
 class DelegateMessageProcessor : public MessageProcessor {
 public:
-    DelegateMessageProcessor(std::function<bool(const std::vector<std::uint8_t>&)> delegate) :
+    DelegateMessageProcessor(std::function<bool(const std::vector<std::uint8_t>&)> process_data);
-        delegate_(delegate) {
+    ~DelegateMessageProcessor();
-    }
 
-    bool process_data(const std::vector<std::uint8_t> &data) override {
+    bool process_data(const std::vector<std::uint8_t> &data) override;
-        if (!delegate_)
-            return false;
-
-        return delegate_(data);
-    }
 
 private:
-    std::function<bool(const std::vector<std::uint8_t>&)> delegate_;
+    std::function<bool(const std::vector<std::uint8_t>&)> process_data_;
 };
-
 } // namespace network
 } // namespace anbox
 

src/anbox/network/local_socket_messenger.cpp

@@ -18,13 +18,17 @@
 #include "anbox/network/local_socket_messenger.h"
 #include "anbox/network/socket_helper.h"
 #include "anbox/utils.h"
+#include "anbox/logger.h"
 
 #include <boost/system/error_code.hpp>
 
 namespace anbox {
 namespace network {
-LocalSocketMessenger::LocalSocketMessenger(const std::string &path,
+LocalSocketMessenger::LocalSocketMessenger(std::shared_ptr<boost::asio::local::stream_protocol::socket> const &socket) :
-                                           const std::shared_ptr<Runtime> &rt) :
+    BaseSocketMessenger(socket) {
+}
+
+LocalSocketMessenger::LocalSocketMessenger(const std::string &path, const std::shared_ptr<Runtime> &rt) :
     socket_(std::make_shared<boost::asio::local::stream_protocol::socket>(rt->service())) {
 
     boost::system::error_code err;
@@ -34,26 +38,10 @@ LocalSocketMessenger::LocalSocketMessenger(const std::string &path,
         BOOST_THROW_EXCEPTION(std::runtime_error(msg));
     }
 
-    messenger_ = std::make_shared<SocketMessenger>(socket_);
+    setup(socket_);
 }
 
 LocalSocketMessenger::~LocalSocketMessenger() {
 }
-
-void LocalSocketMessenger::send(char const* data, size_t length) {
-    messenger_->send(data, length);
-}
-
-void LocalSocketMessenger::async_receive_msg(AnboxReadHandler const& handle, boost::asio::mutable_buffers_1 const &buffer) {
-    messenger_->async_receive_msg(handle, buffer);
-}
-
-boost::system::error_code LocalSocketMessenger::receive_msg(boost::asio::mutable_buffers_1 const& buffer) {
-    return messenger_->receive_msg(buffer);
-}
-
-size_t LocalSocketMessenger::available_bytes() {
-    return messenger_->available_bytes();
-}
 } // namespace network
 } // namespace anbox

src/anbox/network/local_socket_messenger.h

@@ -18,29 +18,21 @@
 #ifndef ANBOX_NETWORK_LOCAL_SOCKET_MESSENGER_H_
 #define ANBOX_NETWORK_LOCAL_SOCKET_MESSENGER_H_
 
-#include "anbox/network/socket_messenger.h"
+#include "anbox/network/base_socket_messenger.h"
-#include "anbox/network/message_sender.h"
-#include "anbox/network/message_receiver.h"
 #include "anbox/runtime.h"
 
 #include <boost/asio/local/stream_protocol.hpp>
 
 namespace anbox {
 namespace network {
-class LocalSocketMessenger : public MessageSender,
+class LocalSocketMessenger : public BaseSocketMessenger<boost::asio::local::stream_protocol> {
-                             public MessageReceiver {
 public:
+    LocalSocketMessenger(std::shared_ptr<boost::asio::local::stream_protocol::socket> const &socket);
     LocalSocketMessenger(const std::string &path, const std::shared_ptr<Runtime> &rt);
     ~LocalSocketMessenger();
 
-    void send(char const* data, size_t length) override;
-    void async_receive_msg(AnboxReadHandler const& handle, boost::asio::mutable_buffers_1 const &buffer) override;
-    boost::system::error_code receive_msg(boost::asio::mutable_buffers_1 const& buffer) override;
-    size_t available_bytes() override;
-
 private:
     std::shared_ptr<boost::asio::local::stream_protocol::socket> socket_;
-    std::shared_ptr<SocketMessenger> messenger_;
 };
 } // namespace network
 } // namespace anbox

src/anbox/network/published_socket_connector.cpp

@@ -19,16 +19,12 @@
 #include "anbox/network/connection_context.h"
 #include "anbox/network/socket_helper.h"
 
-namespace {
-
-}
-
 namespace anbox {
 namespace network {
 PublishedSocketConnector::PublishedSocketConnector(
         const std::string& socket_file,
         const std::shared_ptr<Runtime> &rt,
-        const std::shared_ptr<ConnectionCreator> &connection_creator) :
+        const std::shared_ptr<ConnectionCreator<boost::asio::local::stream_protocol>> &connection_creator) :
     socket_file_(remove_socket_if_stale(socket_file)),
     runtime_(rt),
     connection_creator_(connection_creator),

src/anbox/network/published_socket_connector.h

@@ -34,7 +34,7 @@ public:
     explicit PublishedSocketConnector(
             const std::string& socket_file,
             const std::shared_ptr<Runtime> &rt,
-            const std::shared_ptr<ConnectionCreator> &connection_creator);
+            const std::shared_ptr<ConnectionCreator<boost::asio::local::stream_protocol>> &connection_creator);
     ~PublishedSocketConnector() noexcept;
 
     std::string socket_file() const { return socket_file_; }
@@ -47,7 +47,7 @@ private:
 
     const std::string socket_file_;
     std::shared_ptr<Runtime> runtime_;
-    std::shared_ptr<ConnectionCreator> connection_creator_;
+    std::shared_ptr<ConnectionCreator<boost::asio::local::stream_protocol>> connection_creator_;
     boost::asio::local::stream_protocol::acceptor acceptor_;
 };
 } // namespace network

src/anbox/network/socket_connection.cpp

@@ -53,7 +53,8 @@ SocketConnection::~SocketConnection() noexcept
 {
 }
 
-void SocketConnection::send(char const* data, size_t length) {
+void SocketConnection::send(char const* data, size_t length)
+{
     message_sender_->send(data, length);
 }
 
@@ -68,7 +69,9 @@ void SocketConnection::on_read_size(const boost::system::error_code& error, std:
 {
     if (error)
     {
+        if (connections_)
             connections_->remove(id());
+
         BOOST_THROW_EXCEPTION(std::runtime_error(error.message()));
     }
 
@@ -77,7 +80,7 @@ void SocketConnection::on_read_size(const boost::system::error_code& error, std:
 
     if (processor_->process_data(data))
         read_next_message();
-    else
+    else if (connections_)
         connections_->remove(id());
 }
 } // namespace anbox

src/anbox/network/socket_connection.h

@@ -42,6 +42,10 @@ public:
 
     ~SocketConnection() noexcept;
 
+    void set_name(const std::string &name) { name_ = name; }
+
+    std::shared_ptr<MessageSender> message_sender() const { return message_sender_; }
+
     int id() const { return id_; }
 
     void send(char const* data, size_t length);
@@ -50,12 +54,13 @@ public:
 private:
     void on_read_size(const boost::system::error_code& ec, std::size_t bytes_read);
 
-    std::shared_ptr<MessageReceiver> const message_receiver_;
+    std::shared_ptr<MessageReceiver> message_receiver_;
-    std::shared_ptr<MessageSender>  const message_sender_;
+    std::shared_ptr<MessageSender>  message_sender_;
-    int const id_;
+    int id_;
-    std::shared_ptr<Connections<SocketConnection>> const connections_;
+    std::shared_ptr<Connections<SocketConnection>> connections_;
-    std::shared_ptr<MessageProcessor> const processor_;
+    std::shared_ptr<MessageProcessor> processor_;
     std::array<std::uint8_t, 8192> buffer_;
+    std::string name_;
 };
 } // namespace anbox
 } // namespace network

src/anbox/network/socket_messenger.cpp

@@ -1,115 +0,0 @@
-/*
- * Copyright © 2013-2014 Canonical Ltd.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License version 3 as
- * published by the Free Software Foundation.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- *
- * Authored by: Kevin DuBois <kevin.dubois@canonical.com>
- */
-
-#include "anbox/network/socket_messenger.h"
-#include "anbox/common/variable_length_array.h"
-#include "anbox/logger.h"
-
-#include <boost/throw_exception.hpp>
-
-#include <errno.h>
-#include <string.h>
-
-#include <stdexcept>
-
-namespace bs = boost::system;
-namespace ba = boost::asio;
-
-namespace {
-/// Buffers need to be big enough to support messages
-unsigned int const serialization_buffer_size = 2048;
-}
-
-namespace anbox {
-namespace network {
-SocketMessenger::SocketMessenger(std::shared_ptr<ba::local::stream_protocol::socket> const& socket)
-    : socket(socket),
-      socket_fd{IntOwnedFd{socket->native_handle()}}
-{
-    socket->non_blocking(true);
-    boost::asio::socket_base::send_buffer_size option(64*1024);
-    socket->set_option(option);
-}
-
-Credentials SocketMessenger::creds() const {
-    struct ucred cr;
-    socklen_t cl = sizeof(cr);
-
-    auto status = getsockopt(socket_fd, SOL_SOCKET, SO_PEERCRED, &cr, &cl);
-
-    if (status)
-        BOOST_THROW_EXCEPTION(std::runtime_error("Failed to query client socket credentials"));
-
-    return {cr.pid, cr.uid, cr.gid};
-}
-
-void SocketMessenger::send(char const* data, size_t length)
-{
-    VariableLengthArray<serialization_buffer_size> whole_message{length};
-    std::copy(data, data + length, whole_message.data());
-
-    for (;;) {
-        try {
-            std::unique_lock<std::mutex> lg(message_lock);
-            ba::write(*socket,
-                      ba::buffer(whole_message.data(), whole_message.size()),
-                      boost::asio::transfer_all());
-        }
-        catch (const boost::system::system_error &err) {
-            if (err.code() == boost::asio::error::try_again)
-                continue;
-        }
-        break;
-    }
-}
-
-void SocketMessenger::async_receive_msg(
-    AnboxReadHandler const& handler,
-    ba::mutable_buffers_1 const& buffer)
-{
-    socket->async_read_some(buffer, handler);
-}
-
-bs::error_code SocketMessenger::receive_msg(
-    ba::mutable_buffers_1 const& buffer)
-{
-    bs::error_code e;
-    size_t nread = 0;
-
-    while (nread < ba::buffer_size(buffer))
-    {
-        nread += boost::asio::read(
-             *socket,
-             ba::mutable_buffers_1{buffer + nread},
-             e);
-
-        if (e && e != ba::error::would_block)
-            break;
-    }
-
-    return e;
-}
-
-size_t SocketMessenger::available_bytes()
-{
-    boost::asio::socket_base::bytes_readable command{true};
-    socket->io_control(command);
-    return command.get();
-}
-} // namespace network
-} // namespace anbox

src/anbox/network/socket_messenger.h

@@ -21,7 +21,6 @@
 
 #include <mutex>
 
-#include "anbox/common/fd_sets.h"
 #include "anbox/network/message_sender.h"
 #include "anbox/network/message_receiver.h"
 #include "anbox/network/credentials.h"
@@ -31,22 +30,10 @@ namespace network {
 class SocketMessenger : public MessageSender,
                         public MessageReceiver {
 public:
-    SocketMessenger(std::shared_ptr<boost::asio::local::stream_protocol::socket> const& socket);
+    virtual Credentials creds() const = 0;
-
+    virtual unsigned short local_port() const = 0;
-    Credentials creds() const;
+    virtual void set_no_delay() = 0;
-
+    virtual void close() = 0;
-    void send(char const* data, size_t length) override;
-
-    void async_receive_msg(AnboxReadHandler const& handle, boost::asio::mutable_buffers_1 const &buffer) override;
-    boost::system::error_code receive_msg(boost::asio::mutable_buffers_1 const& buffer) override;
-    size_t available_bytes() override;
-
-
-private:
-    std::shared_ptr<boost::asio::local::stream_protocol::socket> socket;
-    anbox::Fd socket_fd;
-
-    std::mutex message_lock;
 };
 } // namespace network
 } // namespace anbox

src/anbox/network/tcp_socket_connector.cpp

@@ -0,0 +1,72 @@
+/*
+ * Copyright (C) 2016 Simon Fels <morphis@gravedo.de>
+ *
+ * This program is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 3, as published
+ * by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranties of
+ * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include "anbox/network/tcp_socket_connector.h"
+#include "anbox/network/connection_context.h"
+#include "anbox/network/socket_helper.h"
+
+namespace anbox {
+namespace network {
+TcpSocketConnector::TcpSocketConnector(
+        const boost::asio::ip::address_v4 &address,
+        unsigned short port,
+        const std::shared_ptr<Runtime> &rt,
+        const std::shared_ptr<ConnectionCreator<boost::asio::ip::tcp>> &connection_creator) :
+    address_(address),
+    port_(port),
+    runtime_(rt),
+    connection_creator_(connection_creator),
+    acceptor_(rt->service(), boost::asio::ip::tcp::endpoint(address, port)) {
+
+    start_accept();
+}
+
+TcpSocketConnector::~TcpSocketConnector() {
+    acceptor_.cancel();
+}
+
+void TcpSocketConnector::start_accept() {
+    auto socket = std::make_shared<boost::asio::ip::tcp::socket>(runtime_->service());
+
+    acceptor_.async_accept(
+        *socket,
+        [this, socket](boost::system::error_code const& err) {
+            on_new_connection(socket, err);
+        });
+}
+
+void TcpSocketConnector::on_new_connection(
+        std::shared_ptr<boost::asio::ip::tcp::socket> const& socket,
+        boost::system::error_code const& err) {
+
+
+    switch (err.value()) {
+    case boost::system::errc::success:
+        connection_creator_->create_connection_for(socket);
+        break;
+    case boost::system::errc::operation_canceled:
+        // Socket was closed so don't listen for any further incoming
+        // connection attempts.
+        return;
+    default:
+        break;
+    }
+
+    start_accept();
+}
+} // namespace network
+} // namespace anbox

src/anbox/network/tcp_socket_connector.h

@@ -0,0 +1,58 @@
+/*
+ * Copyright (C) 2016 Simon Fels <morphis@gravedo.de>
+ *
+ * This program is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 3, as published
+ * by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranties of
+ * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#ifndef ANBOX_TCPSOCKETCONNECTOR_H
+#define ANBOX_TCPSOCKETCONNECTOR_H
+
+#include <boost/asio/ip/tcp.hpp>
+
+#include "anbox/do_not_copy_or_move.h"
+#include "anbox/runtime.h"
+
+#include "anbox/network/connector.h"
+#include "anbox/network/connection_creator.h"
+
+namespace anbox {
+namespace network {
+class TcpSocketConnector : public DoNotCopyOrMove,
+                           public Connector {
+public:
+    explicit TcpSocketConnector(
+            const boost::asio::ip::address_v4 &address,
+            unsigned short port,
+            const std::shared_ptr<Runtime> &rt,
+            const std::shared_ptr<ConnectionCreator<boost::asio::ip::tcp>> &connection_creator);
+    ~TcpSocketConnector() noexcept;
+
+    unsigned short port() const { return port_; }
+
+private:
+    void start_accept();
+    void on_new_connection(
+        std::shared_ptr<boost::asio::ip::tcp::socket> const& socket,
+        boost::system::error_code const& err);
+
+    boost::asio::ip::address_v4 address_;
+    unsigned short port_;
+    std::shared_ptr<Runtime> runtime_;
+    std::shared_ptr<ConnectionCreator<boost::asio::ip::tcp>> connection_creator_;
+    boost::asio::ip::tcp::acceptor acceptor_;
+};
+} // namespace network
+} // namespace anbox
+
+#endif

src/anbox/network/tcp_socket_messenger.cpp

@@ -0,0 +1,42 @@
+/*
+ * Copyright (C) 2016 Simon Fels <morphis@gravedo.de>
+ *
+ * This program is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 3, as published
+ * by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranties of
+ * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include "anbox/network/tcp_socket_messenger.h"
+
+namespace anbox {
+namespace network {
+TcpSocketMessenger::TcpSocketMessenger(const boost::asio::ip::address_v4 &addr, unsigned short port,
+                                       const std::shared_ptr<Runtime> &rt) {
+    boost::asio::ip::tcp::endpoint endpoint(addr, port);
+    auto socket = std::make_shared<boost::asio::ip::tcp::socket>(rt->service());
+    socket->connect(endpoint);
+    setup(socket);
+    local_port_ = socket->local_endpoint().port();
+}
+
+TcpSocketMessenger::TcpSocketMessenger(std::shared_ptr<boost::asio::ip::tcp::socket> const &socket) :
+    BaseSocketMessenger(socket) {
+}
+
+TcpSocketMessenger::~TcpSocketMessenger() {
+}
+
+unsigned short TcpSocketMessenger::local_port() const {
+    return local_port_;
+}
+} // namespace network
+} // namespace anbox

src/anbox/network/tcp_socket_messenger.h

@@ -0,0 +1,43 @@
+/*
+ * Copyright (C) 2016 Simon Fels <morphis@gravedo.de>
+ *
+ * This program is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 3, as published
+ * by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranties of
+ * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#ifndef ANBOX_NETWORK_TCP_SOCKET_MESSENGER_H_
+#define ANBOX_NETWORK_TCP_SOCKET_MESSENGER_H_
+
+#include "anbox/network/base_socket_messenger.h"
+#include "anbox/runtime.h"
+
+#include <boost/asio.hpp>
+
+namespace anbox {
+namespace network {
+class TcpSocketMessenger : public BaseSocketMessenger<boost::asio::ip::tcp> {
+public:
+    TcpSocketMessenger(const boost::asio::ip::address_v4 &addr, unsigned short port,
+                       const std::shared_ptr<Runtime> &rt);
+    TcpSocketMessenger(std::shared_ptr<boost::asio::ip::tcp::socket> const &socket);
+    ~TcpSocketMessenger();
+
+    unsigned short local_port() const;
+
+private:
+    unsigned short local_port_;
+};
+} // namespace network
+} // namespace anbox
+
+#endif

src/anbox/qemu/adb_message_processor.cpp

@@ -0,0 +1,185 @@
+/*
+ * Copyright (C) 2016 Simon Fels <morphis@gravedo.de>
+ *
+ * This program is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 3, as published
+ * by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranties of
+ * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#include "anbox/logger.h"
+#include "anbox/qemu/adb_message_processor.h"
+#include "anbox/network/delegate_message_processor.h"
+#include "anbox/network/tcp_socket_messenger.h"
+#include "anbox/network/delegate_connection_creator.h"
+
+#include <fstream>
+#include <functional>
+
+namespace {
+const unsigned short default_adb_client_port{5037};
+const unsigned short default_host_listen_port{6664};
+const std::string accept_command{"accept"};
+const std::string ok_command{"ok"};
+const std::string ko_command{"ko"};
+const std::string start_command{"start"};
+const boost::posix_time::seconds default_adb_wait_time{1};
+static std::mutex active_instance;
+}
+
+using namespace std::placeholders;
+
+namespace anbox {
+namespace qemu {
+AdbMessageProcessor::AdbMessageProcessor(const std::shared_ptr<Runtime> &rt,
+                                         const std::shared_ptr<network::SocketMessenger> &messenger) :
+    runtime_(rt),
+    state_(waiting_for_guest_accept_command),
+    expected_command_(accept_command),
+    messenger_(messenger),
+    host_notify_timer_(rt->service()) {
+}
+
+AdbMessageProcessor::~AdbMessageProcessor() {
+    host_connector_.reset();
+    active_instance.unlock();
+}
+
+void AdbMessageProcessor::advance_state() {
+    switch (state_) {
+    case waiting_for_guest_accept_command:
+        // Try to get a lock here as if we already have another processor
+        // running we don't have to do anything here until that one is done.
+        // The container directly starts a second connection once the first
+        // one is established but will not use it until the active one is closed.
+        active_instance.lock();
+
+        wait_for_host_connection();
+        break;
+    case waiting_for_host_connection:
+        messenger_->send(reinterpret_cast<const char*>(ok_command.data()), ok_command.size());
+        state_ = waiting_for_guest_start_command;
+        expected_command_ = start_command;
+        break;
+    case waiting_for_guest_start_command:
+        state_ = proxying_data;
+        read_next_host_message();
+        break;
+    case proxying_data:
+        break;
+    case closed_by_host:
+        // Close the connection to the container as our adb host connection
+        // turned down. The container will try to establish a connection
+        // immediately again and we will handle that by waiting for the
+        // host adb to run up again.
+        messenger_->close();
+        break;
+    case closed_by_container:
+        // In this case the container will close the pipe connection and this
+        // message processor will be deleted once the owning socket connection
+        // is closed.
+        break;
+    default:
+        break;
+    }
+}
+
+void AdbMessageProcessor::wait_for_host_connection() {
+    if (!host_connector_) {
+        host_connector_ = std::make_shared<network::TcpSocketConnector>(
+                    boost::asio::ip::address_v4::from_string("127.0.0.1"),
+                    default_host_listen_port,
+                    runtime_,
+                    std::make_shared<network::DelegateConnectionCreator<boost::asio::ip::tcp>>(
+                        std::bind(&AdbMessageProcessor::on_host_connection, this, _1)));
+    }
+
+    try {
+        // Notify the adb host instance so that it knows on which port our
+        // proxy is waiting for incoming connections.
+        auto messenger = std::make_shared<network::TcpSocketMessenger>(boost::asio::ip::address_v4::from_string("127.0.0.1"),
+                                                                       default_adb_client_port,
+                                                                       runtime_);
+        auto message = utils::string_format("host:emulator:%d", default_host_listen_port);
+        auto handshake = utils::string_format("%04x%s", message.size(), message.c_str());
+        messenger->send(handshake.data(), handshake.size());
+    } catch (std::exception&) {
+        // Try again later when the host adb service is maybe available
+        host_notify_timer_.expires_from_now(default_adb_wait_time);
+        host_notify_timer_.async_wait([&](const boost::system::error_code&) {
+            wait_for_host_connection();
+        });
+    }
+}
+
+void AdbMessageProcessor::on_host_connection(std::shared_ptr<boost::asio::basic_stream_socket<boost::asio::ip::tcp>> const &socket) {
+    host_messenger_ = std::make_shared<network::TcpSocketMessenger>(socket);
+
+    // set_no_delay() reduces the latency of sending data, at the cost
+    // of creating more TCP packets on the connection. It's useful when
+    // doing lots of small send() calls, like the ADB protocol requires.
+    // And since this is on localhost, the packet increase should not be
+    // noticeable.
+    host_messenger_->set_no_delay();
+
+    // Let adb inside the container know that we have a connection to
+    // the adb host instance
+    messenger_->send(reinterpret_cast<const char*>(ok_command.data()), ok_command.size());
+
+    state_ = waiting_for_guest_start_command;
+    expected_command_ = start_command;
+}
+
+void AdbMessageProcessor::read_next_host_message() {
+    auto callback = std::bind(&AdbMessageProcessor::on_host_read_size,
+                              this, _1, _2);
+    host_messenger_->async_receive_msg(callback, boost::asio::buffer(host_buffer_));
+}
+
+void AdbMessageProcessor::on_host_read_size(const boost::system::error_code& error, std::size_t bytes_read) {
+    if (error) {
+        // If messenger is still alive then close the connection which will
+        // trigger the terminate of our processor instance too.
+        if (messenger_)
+            messenger_->close();
+
+        return;
+    }
+
+    messenger_->send(reinterpret_cast<const char*>(host_buffer_.data()), bytes_read);
+    read_next_host_message();
+}
+
+bool AdbMessageProcessor::process_data(const std::vector<std::uint8_t> &data) {
+    if (state_ == proxying_data) {
+        host_messenger_->send(reinterpret_cast<const char*>(data.data()), data.size());
+        return true;
+    }
+
+    for (const auto &byte : data)
+        buffer_.push_back(byte);
+
+    if (expected_command_.size() > 0 && buffer_.size() >= expected_command_.size()) {
+        if (::memcmp(buffer_.data(), expected_command_.data(), data.size()) != 0) {
+            // We got not the command we expected and will terminate here
+            return false;
+        }
+
+        buffer_.erase(buffer_.begin(), buffer_.begin() + expected_command_.size());
+        expected_command_.clear();
+
+        advance_state();
+    }
+
+    return true;
+}
+} // namespace qemu
+} // namespace anbox

src/anbox/qemu/adb_message_processor.h

@@ -0,0 +1,70 @@
+/*
+ * Copyright (C) 2016 Simon Fels <morphis@gravedo.de>
+ *
+ * This program is free software: you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 3, as published
+ * by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranties of
+ * MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ * PURPOSE.  See the GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License along
+ * with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+#ifndef ANBOX_QEMU_ADBD_MESSAGE_PROCESSOR_H_
+#define ANBOX_QEMU_ADBD_MESSAGE_PROCESSOR_H_
+
+#include "anbox/runtime.h"
+#include "anbox/network/message_processor.h"
+#include "anbox/network/socket_messenger.h"
+#include "anbox/network/socket_connection.h"
+#include "anbox/network/tcp_socket_connector.h"
+#include "anbox/network/tcp_socket_messenger.h"
+
+#include <boost/asio.hpp>
+
+namespace anbox {
+namespace qemu {
+class AdbMessageProcessor : public network::MessageProcessor {
+public:
+    AdbMessageProcessor(const std::shared_ptr<Runtime> &rt,
+                         const std::shared_ptr<network::SocketMessenger> &messenger);
+    ~AdbMessageProcessor();
+
+    bool process_data(const std::vector<std::uint8_t> &data) override;
+
+private:
+    enum State {
+        waiting_for_guest_accept_command,
+        waiting_for_host_connection,
+        waiting_for_guest_start_command,
+        proxying_data,
+        closed_by_container,
+        closed_by_host,
+    };
+
+    void advance_state();
+
+    void wait_for_host_connection();
+    void on_host_connection(std::shared_ptr<boost::asio::basic_stream_socket<boost::asio::ip::tcp>> const &socket);
+    void read_next_host_message();
+    void on_host_read_size(const boost::system::error_code& error, std::size_t bytes_read);
+
+    std::shared_ptr<Runtime> runtime_;
+    State state_ = waiting_for_guest_accept_command;
+    std::string expected_command_;
+    std::shared_ptr<network::SocketMessenger> messenger_;
+    std::vector<std::uint8_t> buffer_;
+    std::shared_ptr<network::TcpSocketConnector> host_connector_;
+    std::shared_ptr<network::TcpSocketMessenger> host_messenger_;
+    std::array<std::uint8_t, 8192> host_buffer_;
+    boost::asio::deadline_timer host_notify_timer_;
+};
+} // namespace graphics
+} // namespace anbox
+
+#endif

src/anbox/qemu/pipe_connection_creator.cpp

@@ -18,7 +18,7 @@
 #include <string>
 
 #include "anbox/logger.h"
-#include "anbox/network/socket_messenger.h"
+#include "anbox/network/local_socket_messenger.h"
 #include "anbox/graphics/opengles_message_processor.h"
 #include "anbox/qemu/pipe_connection_creator.h"
 #include "anbox/qemu/boot_properties_message_processor.h"
@@ -29,9 +29,37 @@
 #include "anbox/qemu/fingerprint_message_processor.h"
 #include "anbox/qemu/gsm_message_processor.h"
 #include "anbox/qemu/bootanimation_message_processor.h"
+#include "anbox/qemu/adb_message_processor.h"
 
 namespace ba = boost::asio;
 
+namespace {
+std::string client_type_to_string(const anbox::qemu::PipeConnectionCreator::client_type &type) {
+    switch (type) {
+    case anbox::qemu::PipeConnectionCreator::client_type::opengles:
+        return "opengles";
+    case anbox::qemu::PipeConnectionCreator::client_type::qemud_boot_properties:
+        return "boot-properties";
+    case anbox::qemu::PipeConnectionCreator::client_type::qemud_hw_control:
+        return "hw-control";
+    case anbox::qemu::PipeConnectionCreator::client_type::qemud_sensors:
+        return "sensors";
+    case anbox::qemu::PipeConnectionCreator::client_type::qemud_camera:
+        return "camera";
+    case anbox::qemu::PipeConnectionCreator::client_type::qemud_fingerprint:
+        return "fingerprint";
+    case anbox::qemu::PipeConnectionCreator::client_type::qemud_gsm:
+        return "gsm";
+    case anbox::qemu::PipeConnectionCreator::client_type::qemud_adb:
+        return "adb";
+    case anbox::qemu::PipeConnectionCreator::client_type::bootanimation:
+        return "boot-animation";
+    case anbox::qemu::PipeConnectionCreator::client_type::invalid:
+        break;
+    }
+    return "unknown";
+}
+}
 namespace anbox {
 namespace qemu {
 PipeConnectionCreator::PipeConnectionCreator(const std::shared_ptr<Runtime> &rt,
@@ -50,7 +78,7 @@ PipeConnectionCreator::~PipeConnectionCreator() {
 void PipeConnectionCreator::create_connection_for(
         std::shared_ptr<boost::asio::local::stream_protocol::socket> const& socket) {
 
-    auto const messenger = std::make_shared<network::SocketMessenger>(socket);
+    auto const messenger = std::make_shared<network::LocalSocketMessenger>(socket);
     const auto type = identify_client(messenger);
     auto const processor = create_processor(type, messenger);
     if (!processor)
@@ -58,6 +86,7 @@ void PipeConnectionCreator::create_connection_for(
 
     auto const& connection = std::make_shared<network::SocketConnection>(
                 messenger, messenger, next_id(), connections_, processor);
+    connection->set_name(client_type_to_string(type));
     connections_->add(connection);
     connection->read_next_message();
 }
@@ -100,6 +129,8 @@ PipeConnectionCreator::client_type PipeConnectionCreator::identify_client(
         return client_type::qemud_gsm;
     else if (utils::string_starts_with(identifier_and_args, "pipe:anbox:bootanimation"))
         return client_type::bootanimation;
+    else if (utils::string_starts_with(identifier_and_args, "pipe:qemud:adb"))
+        return client_type::qemud_adb;
 
     return client_type::invalid;
 }
@@ -121,6 +152,8 @@ std::shared_ptr<network::MessageProcessor> PipeConnectionCreator::create_process
         return std::make_shared<qemu::GsmMessageProcessor>(messenger);
     else if (type == client_type::bootanimation)
         return std::make_shared<qemu::BootAnimationMessageProcessor>(messenger, boot_animation_icon_path_);
+    else if (type == client_type::qemud_adb)
+        return std::make_shared<qemu::AdbMessageProcessor>(runtime_, messenger);
 
     return std::make_shared<qemu::NullMessageProcessor>();
 }

src/anbox/qemu/pipe_connection_creator.h

@@ -31,7 +31,7 @@
 
 namespace anbox {
 namespace qemu {
-class PipeConnectionCreator : public network::ConnectionCreator {
+class PipeConnectionCreator : public network::ConnectionCreator<boost::asio::local::stream_protocol> {
 public:
     PipeConnectionCreator(
             const std::shared_ptr<Runtime> &rt,
@@ -40,7 +40,7 @@ public:
     ~PipeConnectionCreator() noexcept;
 
     void create_connection_for(
-        std::shared_ptr<boost::asio::local::stream_protocol::socket> const& socket) override;
+        std::shared_ptr<boost::asio::basic_stream_socket<boost::asio::local::stream_protocol>> const& socket) override;
 
     enum class client_type {
         invalid,
@@ -51,6 +51,7 @@ public:
         qemud_camera,
         qemud_fingerprint,
         qemud_gsm,
+        qemud_adb,
         bootanimation,
     };
 

src/anbox/rpc/connection_creator.cpp

@@ -18,7 +18,7 @@
 
 #include "anbox/rpc/connection_creator.h"
 #include "anbox/rpc/message_processor.h"
-#include "anbox/network/socket_messenger.h"
+#include "anbox/network/local_socket_messenger.h"
 #include "anbox/logger.h"
 
 #include <string>
@@ -47,11 +47,12 @@ void ConnectionCreator::create_connection_for(
         return;
     }
 
-    auto const messenger = std::make_shared<network::SocketMessenger>(socket);
+    auto const messenger = std::make_shared<network::LocalSocketMessenger>(socket);
     auto const processor = message_processor_factory_(messenger);
 
     auto const& connection = std::make_shared<network::SocketConnection>(
                 messenger, messenger, next_id(), connections_, processor);
+    connection->set_name("rpc");
     connections_->add(connection);
     connection->read_next_message();
 }

src/anbox/rpc/connection_creator.h

@@ -31,7 +31,7 @@
 
 namespace anbox {
 namespace rpc {
-class ConnectionCreator : public network::ConnectionCreator {
+class ConnectionCreator : public network::ConnectionCreator<boost::asio::local::stream_protocol> {
 public:
     typedef std::function<std::shared_ptr<network::MessageProcessor>(
             const std::shared_ptr<network::MessageSender>&)> MessageProcessorFactory;
@@ -41,7 +41,7 @@ public:
     ~ConnectionCreator() noexcept;
 
     void create_connection_for(
-        std::shared_ptr<boost::asio::local::stream_protocol::socket> const& socket) override;
+        std::shared_ptr<boost::asio::basic_stream_socket<boost::asio::local::stream_protocol>> const& socket) override;
 
 private:
     int next_id();