move shell to separate thread

src/CatSession.cc

@@ -31,21 +31,19 @@
 
 using namespace std;
 
+CatSession::exit_shell::exit_shell() : runtime_error("shell exited") {}
+
 CatSession::CatSession(
     shared_ptr<struct event_base> base,
     const struct sockaddr_storage& remote,
     Version version,
     shared_ptr<const PSOBBEncryption::KeyFile> bb_key_file)
-    : Shell(base),
-      log(string_printf("[CatSession:%s] ", name_for_enum(version)), proxy_server_log.min_level),
-      channel(
-          version,
-          1,
-          CatSession::dispatch_on_channel_input,
-          CatSession::dispatch_on_channel_error,
-          this,
-          "CatSession"),
+    : log(string_printf("[CatSession:%s] ", name_for_enum(version)), proxy_server_log.min_level),
+      base(base),
+      read_event(event_new(this->base.get(), 0, EV_READ | EV_PERSIST, CatSession::dispatch_read_stdin, this), event_free),
+      channel(version, 1, CatSession::dispatch_on_channel_input, CatSession::dispatch_on_channel_error, this, "CatSession"),
       bb_key_file(bb_key_file) {
+
   if (remote.ss_family != AF_INET) {
     throw runtime_error("remote is not AF_INET");
   }
@@ -64,6 +62,14 @@ CatSession::CatSession(
           reinterpret_cast<const sockaddr*>(&remote), sizeof(struct sockaddr_in)) != 0) {
     throw runtime_error(string_printf("failed to connect (%d)", EVUTIL_SOCKET_ERROR()));
   }
+
+  event_add(this->read_event.get(), nullptr);
+  this->poll.add(0, POLLIN);
+}
+
+void CatSession::execute_command(const std::string& command) {
+  string full_cmd = parse_data_string(command, nullptr, ParseDataFlags::ALLOW_FILES);
+  send_command_with_header(this->channel, full_cmd.data(), full_cmd.size());
 }
 
 void CatSession::dispatch_on_channel_input(
@@ -129,9 +135,54 @@ void CatSession::on_channel_error(short events) {
   }
 }
 
-void CatSession::print_prompt() {}
-
-void CatSession::execute_command(const std::string& command) {
-  string full_cmd = parse_data_string(command, nullptr, ParseDataFlags::ALLOW_FILES);
-  send_command_with_header(this->channel, full_cmd.data(), full_cmd.size());
+void CatSession::dispatch_read_stdin(evutil_socket_t, short, void* ctx) {
+  reinterpret_cast<CatSession*>(ctx)->read_stdin();
+}
+
+void CatSession::read_stdin() {
+  bool any_command_read = false;
+  for (;;) {
+    auto poll_result = this->poll.poll();
+    short fd_events = 0;
+    try {
+      fd_events = poll_result.at(0);
+    } catch (const out_of_range&) {
+    }
+
+    if (!(fd_events & POLLIN)) {
+      break;
+    }
+
+    string command(2048, '\0');
+    if (!fgets(command.data(), command.size(), stdin)) {
+      if (!any_command_read) {
+        // ctrl+d probably; we should exit
+        fputc('\n', stderr);
+        event_base_loopexit(this->base.get(), nullptr);
+        return;
+      } else {
+        break; // probably not EOF; just no more commands for now
+      }
+    }
+
+    // trim the extra data off the string
+    size_t len = strlen(command.c_str());
+    if (len == 0) {
+      break;
+    }
+    if (command[len - 1] == '\n') {
+      len--;
+    }
+    command.resize(len);
+    any_command_read = true;
+
+    try {
+      execute_command(command);
+    } catch (const exit_shell&) {
+      event_base_loopexit(this->base.get(), nullptr);
+      return;
+    } catch (const exception& e) {
+      fprintf(stderr, "FAILED: %s\n", e.what());
+    }
+  }
 }