1 files changed, 24 insertions, 10 deletions

diff --git a/daemon/services.cpp b/daemon/services.cpp
index d86b54a0..07fbe328 100644
--- a/daemon/services.cpp
+++ b/daemon/services.cpp
@@ -49,10 +49,12 @@
 #include "adb_utils.h"
 #include "services.h"
 #include "socket_spec.h"
+#include "sysdeps.h"
 #include "transport.h"
 
 #include "daemon/file_sync_service.h"
 #include "daemon/framebuffer_service.h"
+#include "daemon/jdwp_service.h"
 #include "daemon/logging.h"
 #include "daemon/restart_service.h"
 #include "daemon/shell_service.h"
@@ -131,7 +133,7 @@ static void spin_service(unique_fd fd) {
         return;
     }
 
-    fdevent_run_on_main_thread([fd = pipe_read.release()]() {
+    fdevent_run_on_looper([fd = pipe_read.release()]() {
         fdevent* fde = fdevent_create(
                 fd, [](int, unsigned, void*) {}, nullptr);
         fdevent_add(fde, FDE_READ);
@@ -161,9 +163,14 @@ static void spin_service(unique_fd fd) {
 }
 
 struct ServiceSocket : public asocket {
-    ServiceSocket() {
+    ServiceSocket() = delete;
+    explicit ServiceSocket(atransport* transport) {
+        CHECK(transport);
         install_local_socket(this);
+        this->transport = transport;
         this->enqueue = [](asocket* self, apacket::payload_type data) {
+            // TODO: This interface currently can't give any backpressure.
+            send_ready(self->id, self->peer->id, self->transport, data.size());
             return static_cast<ServiceSocket*>(self)->Enqueue(std::move(data));
         };
         this->ready = [](asocket* self) { return static_cast<ServiceSocket*>(self)->Ready(); };
@@ -171,6 +178,11 @@ struct ServiceSocket : public asocket {
     }
     virtual ~ServiceSocket() = default;
 
+    ServiceSocket(const ServiceSocket& copy) = delete;
+    ServiceSocket(ServiceSocket&& move) = delete;
+    ServiceSocket& operator=(const ServiceSocket& copy) = delete;
+    ServiceSocket& operator=(ServiceSocket&& move) = delete;
+
     virtual int Enqueue(apacket::payload_type data) { return -1; }
     virtual void Ready() {}
     virtual void Close() {
@@ -188,9 +200,9 @@ struct ServiceSocket : public asocket {
 };
 
 struct SinkSocket : public ServiceSocket {
-    explicit SinkSocket(size_t byte_count) {
+    explicit SinkSocket(atransport* transport, size_t byte_count)
+        : ServiceSocket(transport), bytes_left_(byte_count) {
         LOG(INFO) << "Creating new SinkSocket with capacity " << byte_count;
-        bytes_left_ = byte_count;
     }
 
     virtual ~SinkSocket() { LOG(INFO) << "SinkSocket destroyed"; }
@@ -210,9 +222,9 @@ struct SinkSocket : public ServiceSocket {
 };
 
 struct SourceSocket : public ServiceSocket {
-    explicit SourceSocket(size_t byte_count) {
+    explicit SourceSocket(atransport* transport, size_t byte_count)
+        : ServiceSocket(transport), bytes_left_(byte_count) {
         LOG(INFO) << "Creating new SourceSocket with capacity " << byte_count;
-        bytes_left_ = byte_count;
     }
 
     virtual ~SourceSocket() { LOG(INFO) << "SourceSocket destroyed"; }
@@ -235,7 +247,7 @@ struct SourceSocket : public ServiceSocket {
     size_t bytes_left_;
 };
 
-asocket* daemon_service_to_socket(std::string_view name) {
+asocket* daemon_service_to_socket(std::string_view name, atransport* transport) {
     if (name == "jdwp") {
         return create_jdwp_service_socket();
     } else if (name == "track-jdwp") {
@@ -247,13 +259,13 @@ asocket* daemon_service_to_socket(std::string_view name) {
         if (!ParseUint(&byte_count, name)) {
             return nullptr;
         }
-        return new SinkSocket(byte_count);
+        return new SinkSocket(transport, byte_count);
     } else if (android::base::ConsumePrefix(&name, "source:")) {
         uint64_t byte_count = 0;
         if (!ParseUint(&byte_count, name)) {
             return nullptr;
         }
-        return new SourceSocket(byte_count);
+        return new SourceSocket(transport, byte_count);
     }
 
     return nullptr;
@@ -308,7 +320,9 @@ unique_fd daemon_service_to_fd(std::string_view name, atransport* transport) {
     }
 #endif
 
-    if (android::base::ConsumePrefix(&name, "jdwp:")) {
+    if (android::base::ConsumePrefix(&name, "dev:")) {
+        return unique_fd{unix_open(name, O_RDWR | O_CLOEXEC)};
+    } else if (android::base::ConsumePrefix(&name, "jdwp:")) {
         pid_t pid;
         if (!ParseUint(&pid, name)) {
             return unique_fd{};