Moved exec into its own event handler.

Makefile

@@ -17,7 +17,8 @@ SRCS=\
 	src/SocketAcceptor.cpp \
 	src/Strings.cpp \
 	src/SUTAgent.cpp \
-	src/Subprocess.cpp
+	src/Subprocess.cpp \
+	src/SubprocessEventHandler.cpp
 
 OBJS=$(subst .cpp,.o,$(SRCS))
 

src/CommandEventHandler.cpp

@@ -10,6 +10,7 @@
 #include "PushFileEventHandler.h"
 #include "Strings.h"
 #include "Subprocess.h"
+#include "SubprocessEventHandler.h"
 
 #include <dirent.h>
 #include <stdint.h>
@@ -97,6 +98,20 @@ CommandEventHandler::checkDataEventHandler(PRPollDesc desc)
 }
 
 
+void
+CommandEventHandler::handleTimeout()
+{
+  if (mDataEventHandler && !mDataEventHandler->closed())
+    mDataEventHandler->handleTimeout();
+  if (mDataEventHandler->closed())
+  {
+    delete mDataEventHandler;
+    mDataEventHandler = NULL;
+    sendPrompt();
+  }
+}
+
+
 void
 CommandEventHandler::handleEvent(PRPollDesc desc)
 {
@@ -337,7 +352,7 @@ std::string
 CommandEventHandler::exec(std::vector<std::string>& args)
 {
   if (args.size() < 1)
-    return agentWarn("command not specified");
+    return agentWarnInvalidNumArgs(1);
 
   // delete double quotes from args[0], easier to parse
   if (args[0][0] == '"')
@@ -351,11 +366,12 @@ CommandEventHandler::exec(std::vector<std::string>& args)
   // handle first part separately, check if we have env vars
   bool envs = args[0].find('=') != std::string::npos;
 
-  std::vector<std::string> env_names, env_values;
+  std::vector<std::string> envNames, envValues;
   // if we have envs we have to handle them separately
   if (envs)
   {
     char envVarStr[(*argi).size() + 1];
+    envVarStr[(*argi).size()] = 0;
     (*argi).copy(envVarStr, (*argi).size());
     envVarStr[(*argi).size()] = 0;
     char *r_env;
@@ -377,8 +393,8 @@ CommandEventHandler::exec(std::vector<std::string>& args)
         continue;
 
       std::string var(env, pos), val(env + pos + 1);
-      env_names.push_back(var);
-      env_values.push_back(val);
+      envNames.push_back(var);
+      envValues.push_back(val);
 
       env = strtok_r(NULL, ",", &r_env);
     }
@@ -390,49 +406,21 @@ CommandEventHandler::exec(std::vector<std::string>& args)
   std::string prog(*argi++);
 
   // what remains are the args
-
-  // set the env vars and backup the old vals
-  std::vector<std::string> backup;
-  for (int i = 0; i < env_names.size(); ++i)
-  {
-    const char *name = env_names[i].c_str();
-    char *old = getenv(name);
-    if (!old)
-      backup.push_back("");
-    else
-      backup.push_back(std::string(old));
-    setenv(name, env_values[i].c_str(), 1);
-  }
-
   std::ostringstream to_exec;
-  to_exec << prog << " ";
+  to_exec << prog;
   for (; argi != args.end(); ++argi)
-    to_exec << *argi << " ";
+    to_exec << " " << *argi;
 
-  FILE *p = checkPopen(to_exec.str(), "r");
-
-  // get the output so pclose won't cry even on successful calls
-  char buffer[BUFSIZE];
-  std::ostringstream output;
-
-  while (fgets(buffer, BUFSIZE, p))
-    output << std::string(buffer);
-
-  int status = pclose(p);
-
-  // restore the env
-  for (int i = 0; i < env_names.size(); ++i)
+  mDataEventHandler = new SubprocessEventHandler(mBufSocket, *this,
+                                                 to_exec.str(), envNames,
+                                                 envValues);
+  if (mDataEventHandler->closed())
   {
-    const char *name = env_names[i].c_str();
-    if (backup[i].size() == 0)
-      unsetenv(name);
-    else
-      setenv(name, backup[i].c_str(), 1);
+    delete mDataEventHandler;
+    mDataEventHandler = NULL;
+    return agentWarn("failed to launch process");
   }
-
-  if (status == 0)
-    return std::string("success");
-  return agentWarn("error");
+  return "";
 }
 
 

src/CommandEventHandler.h

@@ -22,6 +22,7 @@ public:
   virtual void close();
   virtual void getPollDescs(std::vector<PRPollDesc>& descs);
   virtual void handleEvent(PRPollDesc desc);
+  virtual void handleTimeout();
   virtual std::string name() { return "CommandEventHandler"; }
 
   void handleLine(std::string line);

src/Reactor.cpp

@@ -6,6 +6,28 @@
 #include <map>
 #include "EventHandler.h"
 
+Reactor::Timeout::Timeout(PRIntervalTime _epoch, PRIntervalTime _interval,
+  EventHandler* _evtHandler)
+  : epoch(_epoch), interval(_interval), evtHandler(_evtHandler)
+{
+}
+
+
+Reactor::Timeout::Timeout(const Timeout& t)
+  : epoch(t.epoch), interval(t.interval), evtHandler(t.evtHandler)
+{
+}
+
+
+Reactor::Timeout&
+Reactor::Timeout::operator=(const Timeout& rhs)
+{
+  epoch = rhs.epoch;
+  interval = rhs.interval;
+  evtHandler = rhs.evtHandler;
+}
+
+
 bool
 Reactor::Timeout::expired()
 {
@@ -106,19 +128,22 @@ Reactor::run()
   std::vector<EventHandler*> expiredHandlers;
 
   for (std::vector<Timeout>::iterator i = mTimeouts.begin();
-       i != mTimeouts.end(); i++)
+       i != mTimeouts.end(); )
   {
     if ((*i).expired())
     {
       expiredHandlers.push_back((*i).evtHandler);
-      mTimeouts.erase(i);
-      i = mTimeouts.begin();
+      i = mTimeouts.erase(i);
     }
+    else
+      ++i;
   }
 
   for (std::vector<EventHandler*>::iterator i = expiredHandlers.begin();
        i != expiredHandlers.end(); i++)
+  {
     (*i)->handleTimeout();
+  }
 
   deleteClosed();
 }
@@ -132,7 +157,7 @@ Reactor::stop()
   {
     (*i)->close();
     delete (*i);
-    mEvtHandlers.erase(i);
+    i = mEvtHandlers.erase(i);
   }
 }
 
@@ -140,10 +165,7 @@ Reactor::stop()
 void
 Reactor::setTimeout(PRIntervalTime interval, EventHandler* evtHandler)
 {
-  Timeout t;
-  t.epoch = PR_IntervalNow();
-  t.interval = interval;
-  t.evtHandler = evtHandler;
+  Timeout t(PR_IntervalNow(), interval, evtHandler);
   mTimeouts.push_back(t);
 }
 
@@ -152,23 +174,23 @@ void
 Reactor::deleteClosed()
 {
   for (std::vector<EventHandler*>::iterator i = mEvtHandlers.begin();
-       i != mEvtHandlers.end(); i++)
+       i != mEvtHandlers.end(); )
   {
     if ((*i)->closed())
     {
       EventHandler* hdlr = *i;
-      mEvtHandlers.erase(i);
-      i = mEvtHandlers.begin();
+      i = mEvtHandlers.erase(i);
       for (std::vector<Timeout>::iterator j = mTimeouts.begin();
-           j != mTimeouts.end(); j++)
+           j != mTimeouts.end(); )
       {
         if ((*j).evtHandler == hdlr)
-        {
-          mTimeouts.erase(j);
-          j = mTimeouts.begin();
-        }
+          j = mTimeouts.erase(j);
+        else
+          ++j;
       }
       delete hdlr;
     }
+    else
+      ++i;
   }
 }

src/Reactor.h

@@ -38,6 +38,11 @@ private:
     PRIntervalTime interval;
     EventHandler* evtHandler;
 
+    Timeout(PRIntervalTime _epoch, PRIntervalTime _interval,
+            EventHandler* _evtHandler);
+    Timeout(const Timeout& t);
+    Timeout& operator=(const Timeout& rhs);
+
     bool expired();
   };
 

src/SubprocessEventHandler.cpp

@@ -0,0 +1,93 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this file,
+ * You can obtain one at http://mozilla.org/MPL/2.0/. */
+#include "SubprocessEventHandler.h"
+#include <errno.h>
+#include <fcntl.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include "BufferedSocket.h"
+#include "CommandEventHandler.h"
+#include "Logging.h"
+#include "Reactor.h"
+
+#define SUBPROCESS_BUFFER_SIZE 1024
+
+SubprocessEventHandler::SubprocessEventHandler(
+  BufferedSocket& bufSocket, CommandEventHandler& commandEventHandler,
+  std::string cmdLine,
+  std::vector<std::string>& envNames, std::vector<std::string>& envValues)
+  : mBufSocket(bufSocket), mCommandEventHandler(commandEventHandler),
+    mP(NULL), mBuffer(new char[SUBPROCESS_BUFFER_SIZE])
+{
+  // set the env vars and backup the old vals
+  std::vector<std::string> backup;
+  for (int i = 0; i < envNames.size(); ++i)
+  {
+    const char *name = envNames[i].c_str();
+    char *old = getenv(name);
+    if (!old)
+      backup.push_back("");
+    else
+      backup.push_back(std::string(old));
+    setenv(name, envValues[i].c_str(), 1);
+  }
+
+  mP = popen(cmdLine.c_str(), "r");
+
+  // restore the env
+  for (int i = 0; i < envNames.size(); ++i)
+  {
+    const char *name = envNames[i].c_str();
+    if (backup[i].size() == 0)
+      unsetenv(name);
+    else
+      setenv(name, backup[i].c_str(), 1);
+  }
+
+  if (mP == NULL)
+  {
+    close();
+    return;
+  }
+
+  fcntl(fileno(mP), F_SETFL, O_NONBLOCK);
+
+  Reactor::instance()->setTimeout(PR_MillisecondsToInterval(SUBPROCESS_POLL_PERIOD_MS), &mCommandEventHandler);
+}
+
+
+SubprocessEventHandler::~SubprocessEventHandler()
+{
+  delete[] mBuffer;
+}
+
+
+void
+SubprocessEventHandler::close()
+{
+  if (mP)
+    pclose(mP);
+  EventHandler::close();
+}
+
+
+void
+SubprocessEventHandler::handleTimeout()
+{
+  ssize_t r = read(fileno(mP), mBuffer, SUBPROCESS_BUFFER_SIZE);
+  if (r == -1 && errno != EAGAIN)
+  {
+    mBufSocket.write(agentWarn("error reading from pipe"));
+    close();
+  }
+  else if (r > 0)
+    mBufSocket.write(mBuffer, r);
+  else if (r == 0)
+    close();
+
+  if (!closed())
+    Reactor::instance()->setTimeout(PR_MillisecondsToInterval(SUBPROCESS_POLL_PERIOD_MS), &mCommandEventHandler);
+}

src/SubprocessEventHandler.h

@@ -0,0 +1,36 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this file,
+ * You can obtain one at http://mozilla.org/MPL/2.0/. */
+#ifndef negatus_subprocess_event_handler_h
+#define negatus_subprocess_event_handler_h
+
+#include "EventHandler.h"
+
+#define SUBPROCESS_POLL_PERIOD_MS 100
+
+class BufferedSocket;
+class CommandEventHandler;
+
+class SubprocessEventHandler: public EventHandler
+{
+public:
+  SubprocessEventHandler(BufferedSocket& bufSocket,
+                         CommandEventHandler& commandEventHandler,
+                         std::string cmdLine,
+                         std::vector<std::string>& envNames,
+                         std::vector<std::string>& envValues);
+  virtual ~SubprocessEventHandler();
+
+  virtual void close();
+  virtual void handleTimeout();
+  virtual std::string name() { return "SubprocessEventHandler"; }
+
+private:
+  BufferedSocket& mBufSocket;
+  CommandEventHandler& mCommandEventHandler;
+  FILE* mP;
+  char* mBuffer;
+};
+
+#endif