Use JsonPipe functionality from QJsonStream library

The JsonPipe class from QJsonStream abstracts sending
and receiving JSON-formatted messages over file descriptors.

Change-Id: I7ebde3dc1f176d41e492f1bdf34c6ae0e0cf8fa5
Reviewed-by: Chris Craig <ext-chris.craig@nokia.com>

5 files changed, 27 insertions, 157 deletions

diff --git a/src/core/pipelauncher.cpp b/src/core/pipelauncher.cpp
index 6099072..ef70185 100644
--- a/src/core/pipelauncher.cpp
+++ b/src/core/pipelauncher.cpp
@@ -42,6 +42,8 @@
 #include <QtEndian>
 #include <QJsonDocument>
 
+#include <jsonpipe.h>
+
 #include "pipelauncher.h"
 #include "launcherclient.h"
 #include "remoteprotocol.h"
@@ -62,79 +64,26 @@ QT_BEGIN_NAMESPACE_PROCESSMANAGER
 PipeLauncher::PipeLauncher(QObject *parent)
     : ProcessBackendManager(parent)
 {
-    m_in = new QSocketNotifier( STDIN_FILENO, QSocketNotifier::Read, this );
-    connect(m_in, SIGNAL(activated(int)), SLOT(inReady(int)));
-    m_in->setEnabled(true);
-    m_out = new QSocketNotifier( STDOUT_FILENO, QSocketNotifier::Write, this );
-    connect(m_out, SIGNAL(activated(int)), SLOT(outReady(int)));
-    m_out->setEnabled(false);
-
+    m_pipe   = new QtAddOn::JsonStream::JsonPipe(this);
     m_client = new LauncherClient(this);
+
+    connect(m_pipe, SIGNAL(messageReceived(const QJsonObject&)),
+            SLOT(receive(const QJsonObject&)));
     connect(m_client, SIGNAL(send(const QJsonObject&)),
-            SLOT(send(const QJsonObject&)));
-}
+            m_pipe, SLOT(send(const QJsonObject&)));
 
-/*!
-  \internal
- */
-void PipeLauncher::inReady(int fd)
-{
-//    qDebug() << Q_FUNC_INFO;
-    m_in->setEnabled(false);
-    const int bufsize = 1024;
-    uint oldSize = m_inbuf.size();
-    m_inbuf.resize(oldSize + bufsize);
-    int n = ::read(fd, m_inbuf.data()+oldSize, bufsize);
-    if (n > 0)
-        m_inbuf.resize(oldSize+n);
-    else
-        m_inbuf.resize(oldSize);
-    // Could check for an error here
-    // Check for a complete JSON object
-    while (m_inbuf.size() >= 12) {
-        qint32 message_size = qFromLittleEndian(((qint32 *)m_inbuf.data())[2]) + 8;
-        if (m_inbuf.size() < message_size)
-            break;
-        QByteArray msg = m_inbuf.left(message_size);
-        m_inbuf = m_inbuf.mid(message_size);
-        QJsonObject message = QJsonDocument::fromBinaryData(msg).object();
-        if (message.value(RemoteProtocol::remote()).toString() == RemoteProtocol::stop())
-            exit(0);
-        m_client->receive(message);
-    }
-    m_in->setEnabled(true);
+    m_pipe->setFds(STDIN_FILENO, STDOUT_FILENO);
 }
 
 /*!
   \internal
  */
-void PipeLauncher::outReady(int fd)
-{
-//    qDebug() << Q_FUNC_INFO;
-    m_out->setEnabled(false);
-    if (m_outbuf.size()) {
-        int n = ::write(fd, m_outbuf.data(), m_outbuf.size());
-        if (n == -1) {
-            qDebug() << "Failed to write to stdout";
-            exit(-1);
-        }
-        if (n < m_outbuf.size())
-            m_outbuf = m_outbuf.mid(n);
-        else
-            m_outbuf.clear();
-    }
-    if (m_outbuf.size())
-        m_out->setEnabled(true);
-}
 
-/*!
-  \internal
- */
-void PipeLauncher::send(const QJsonObject& object)
+void PipeLauncher::receive(const QJsonObject& message)
 {
-    QByteArray data = QJsonDocument(object).toBinaryData();
-    m_outbuf.append(data);
-    m_out->setEnabled(true);
+    if (message.value(RemoteProtocol::remote()).toString() == RemoteProtocol::stop())
+        exit(0);
+    m_client->receive(message);
 }
 
 #include "moc_pipelauncher.cpp"
diff --git a/src/core/pipelauncher.h b/src/core/pipelauncher.h
index 4688024..9a43063 100644
--- a/src/core/pipelauncher.h
+++ b/src/core/pipelauncher.h
@@ -42,9 +42,9 @@
 
 #include <QObject>
 #include <QJsonObject>
-#include <QSocketNotifier>
 
 #include "processbackendmanager.h"
+#include "jsonpipe.h"
 
 QT_BEGIN_NAMESPACE_PROCESSMANAGER
 
@@ -57,16 +57,11 @@ public:
     PipeLauncher(QObject *parent=0);
 
 private slots:
-    void inReady(int fd);
-    void outReady(int fd);
-    void send(const QJsonObject& object);
+    void receive(const QJsonObject& object);
 
 private:
-    QSocketNotifier *m_in;
-    QSocketNotifier *m_out;
-    QByteArray       m_inbuf;
-    QByteArray       m_outbuf;
-    LauncherClient  *m_client;
+    QtAddOn::JsonStream::JsonPipe *m_pipe;
+    LauncherClient                *m_client;
 };
 
 QT_END_NAMESPACE_PROCESSMANAGER
diff --git a/src/core/preforkprocessbackendfactory.cpp b/src/core/preforkprocessbackendfactory.cpp
index b2336d3..7143fe9 100644
--- a/src/core/preforkprocessbackendfactory.cpp
+++ b/src/core/preforkprocessbackendfactory.cpp
@@ -44,9 +44,6 @@
 
 #include <QDebug>
 #include <QJsonDocument>
-#include <QSocketNotifier>
-//#include <QFileInfo>
-#include <QtEndian>
 
 QT_BEGIN_NAMESPACE_PROCESSMANAGER
 
@@ -71,9 +68,10 @@ const int kPreforkTimerInterval = 1000;
 PreforkProcessBackendFactory::PreforkProcessBackendFactory(QObject *parent)
     : RemoteProcessBackendFactory(parent)
     , m_index(-1)
-    , m_in(NULL)
-    , m_out(NULL)
 {
+    m_pipe   = new QtAddOn::JsonStream::JsonPipe(this);
+    connect(m_pipe, SIGNAL(messageReceived(const QJsonObject&)),
+            SLOT(receive(const QJsonObject&)));
 }
 
 /*!
@@ -86,17 +84,8 @@ PreforkProcessBackendFactory::~PreforkProcessBackendFactory()
     if (data) {
         QJsonObject message;
         message.insert(RemoteProtocol::remote(), RemoteProtocol::stop());
-        m_outbuf.append(QJsonDocument(message).toBinaryData());
-
-        while (m_outbuf.size()) {
-            int n = ::write(data->stdin, m_outbuf.data(), m_outbuf.size());
-            if (n == -1)
-                qFatal("Failed to write to stdout");
-            if (n < m_outbuf.size())
-                m_outbuf = m_outbuf.mid(n);
-            else
-                m_outbuf.clear();
-        }
+        m_pipe->send(message);
+        m_pipe->waitForBytesWritten();
     }
 }
 
@@ -121,18 +110,7 @@ void PreforkProcessBackendFactory::setIndex(int index)
     if (index >= 0 && index < prefork->size()) {
         m_index = index;
         const PreforkChildData *data = prefork->at(index);
-
-        if (m_in)
-            delete m_in;
-        if (m_out)
-            delete m_out;
-
-        m_in = new QSocketNotifier(data->stdout, QSocketNotifier::Read, this);
-        m_out = new QSocketNotifier(data->stdin, QSocketNotifier::Write, this);
-        connect(m_in, SIGNAL(activated(int)), SLOT(inReady(int)));
-        connect(m_out, SIGNAL(activated(int)), SLOT(outReady(int)));
-        m_in->setEnabled(true);
-        m_out->setEnabled(false);
+        m_pipe->setFds(data->stdout, data->stdin);
         emit indexChanged();
     }
     else
@@ -142,53 +120,9 @@ void PreforkProcessBackendFactory::setIndex(int index)
 
 bool PreforkProcessBackendFactory::send(const QJsonObject& message)
 {
-    m_outbuf.append(QJsonDocument(message).toBinaryData());
-    m_out->setEnabled(true);
-    return true;
+    return m_pipe->send(message);
 }
 
-void PreforkProcessBackendFactory::inReady(int fd)
-{
-    m_in->setEnabled(false);
-    const int bufsize = 1024;
-    uint oldSize = m_inbuf.size();
-    m_inbuf.resize(oldSize + bufsize);
-    int n = ::read(fd, m_inbuf.data()+oldSize, bufsize);
-    if (n > 0)
-        m_inbuf.resize(oldSize+n);
-    else
-        m_inbuf.resize(oldSize);
-
-    while (m_inbuf.size() >= 12) {
-        if (QJsonDocument::BinaryFormatTag != *((uint *) m_inbuf.data()))
-            qFatal("ERROR in receive buffer: %s", m_inbuf.data());
-        qint32 message_size = qFromLittleEndian(((qint32 *)m_inbuf.data())[2]) + 8;
-        if (m_inbuf.size() < message_size)
-            break;
-        QByteArray msg = m_inbuf.left(message_size);
-        m_inbuf = m_inbuf.mid(message_size);
-        receive(QJsonDocument::fromBinaryData(msg).object());
-    }
-    m_in->setEnabled(true);
-}
-
-void PreforkProcessBackendFactory::outReady(int fd)
-{
-    m_out->setEnabled(false);
-    if (m_outbuf.size()) {
-        int n = ::write(fd, m_outbuf.data(), m_outbuf.size());
-        if (n == -1) {
-            qDebug() << "Failed to write to stdout";
-            exit(-1);
-        }
-        if (n < m_outbuf.size())
-            m_outbuf = m_outbuf.mid(n);
-        else
-            m_outbuf.clear();
-    }
-    if (m_outbuf.size())
-        m_out->setEnabled(true);
-}
 
 /*!
   Return the preforked process
diff --git a/src/core/preforkprocessbackendfactory.h b/src/core/preforkprocessbackendfactory.h
index d485478..912668c 100644
--- a/src/core/preforkprocessbackendfactory.h
+++ b/src/core/preforkprocessbackendfactory.h
@@ -40,10 +40,9 @@
 #ifndef PREFORK_PROCESS_BACKEND_FACTORY_H
 #define PREFORK_PROCESS_BACKEND_FACTORY_H
 
+#include "jsonpipe.h"
 #include "remoteprocessbackendfactory.h"
 
-class QSocketNotifier;
-
 QT_BEGIN_NAMESPACE_PROCESSMANAGER
 
 class Q_ADDON_PROCESSMANAGER_EXPORT PreforkProcessBackendFactory : public RemoteProcessBackendFactory
@@ -66,16 +65,9 @@ signals:
 protected:
     virtual bool send(const QJsonObject&);
 
-private slots:
-    void inReady(int fd);
-    void outReady(int fd);
-
 private:
     int m_index;
-    QByteArray m_inbuf;
-    QByteArray m_outbuf;
-    QSocketNotifier *m_in;
-    QSocketNotifier *m_out;
+    QtAddOn::JsonStream::JsonPipe *m_pipe;
 };
 
 QT_END_NAMESPACE_PROCESSMANAGER
diff --git a/src/core/remoteprocessbackendfactory.h b/src/core/remoteprocessbackendfactory.h
index 07e65ce..ac2dac4 100644
--- a/src/core/remoteprocessbackendfactory.h
+++ b/src/core/remoteprocessbackendfactory.h
@@ -58,8 +58,8 @@ public:
     virtual ProcessBackend *create(const ProcessInfo& info, QObject *parent);
 
 protected:
-    virtual bool send(const QJsonObject&) = 0;
-    void receive(const QJsonObject&);
+    virtual     bool send(const QJsonObject&) = 0;
+    Q_INVOKABLE void receive(const QJsonObject&);
 
 private:
     void backendDestroyed(int);