Added JsonPipe class

The JsonPipe class supports reading and writing
JSON objects over a pair of pipes (file descriptors).

Change-Id: I388907d38d1afb12dd53fadc3051db1db4d2f35f
Reviewed-by: Alexei Rousskikh <ext-alexei.rousskikh@nokia.com>
Reviewed-by: Chris Craig <ext-chris.craig@nokia.com>

6 files changed, 583 insertions, 1 deletions

diff --git a/src/jsonbuffer.cpp b/src/jsonbuffer.cpp
index 60bcd67..4382713 100644
--- a/src/jsonbuffer.cpp
+++ b/src/jsonbuffer.cpp
@@ -83,7 +83,8 @@ JsonBuffer::JsonBuffer(QObject *parent)
 
 void JsonBuffer::append(const QByteArray& data)
 {
-    append(data.data(), data.size());
+    mBuffer.append(data.data(), data.size());
+    processMessages();
 }
 
 /*!
@@ -95,7 +96,48 @@ void JsonBuffer::append(const QByteArray& data)
 void JsonBuffer::append(const char *data, int len)
 {
     mBuffer.append(data, len);
+    processMessages();
+}
+
+/*!
+  Copy data from a file descriptor \a fd into the buffer.
+  This function tries to eliminate extra data copy operations.
+  It assumes that the file descriptor is ready to read and
+  it does not try to read all of the data.
 
+  Returns the number of bytes read or -1 for an error condition.
+ */
+
+int JsonBuffer::copyFromFd(int fd)
+{
+    const int maxcopy = 1024;
+    uint oldSize = mBuffer.size();
+    mBuffer.resize(oldSize + maxcopy);
+    int n = ::read(fd, mBuffer.data()+oldSize, maxcopy);
+    if (n > 0) {
+        mBuffer.resize(oldSize+n);
+        processMessages();
+    }
+    else
+        mBuffer.resize(oldSize);
+    return n;
+}
+
+/*!
+    Clear the contents of the buffer.
+ */
+
+void JsonBuffer::clear()
+{
+    mBuffer.clear();
+}
+
+/*!
+  \internal
+*/
+
+void JsonBuffer::processMessages()
+{
     if (mFormat == FormatUndefined && mBuffer.size() >= 4) {
         if (strncmp("bson", mBuffer.data(), 4) == 0)
             mFormat = FormatBSON;
diff --git a/src/jsonbuffer_p.h b/src/jsonbuffer_p.h
index 0a72d13..3d72be9 100644
--- a/src/jsonbuffer_p.h
+++ b/src/jsonbuffer_p.h
@@ -57,6 +57,8 @@ public:
     JsonBuffer(QObject *parent=0);
     void append(const QByteArray& data);
     void append(const char* data, int len);
+    int  copyFromFd(int fd);
+    void clear();
 
     EncodingFormat  format() const;
 
@@ -64,6 +66,9 @@ signals:
     void objectReceived(const QJsonObject& object);
 
 private:
+    void processMessages();
+
+private:
     enum UTF8ParsingState { ParseNormal, ParseInString, ParseInBackslash };
 
     EncodingFormat   mFormat;
diff --git a/src/jsonpipe.cpp b/src/jsonpipe.cpp
new file mode 100644
index 0000000..257e103
--- /dev/null
+++ b/src/jsonpipe.cpp
@@ -0,0 +1,315 @@
+/****************************************************************************
+**
+** Copyright (C) 2011 Nokia Corporation and/or its subsidiary(-ies).
+** All rights reserved.
+** Contact: Nokia Corporation (qt-info@nokia.com)
+**
+** This file is part of the QtAddOn.JsonStream module of the Qt Toolkit.
+**
+** $QT_BEGIN_LICENSE:LGPL$
+** GNU Lesser General Public License Usage
+** This file may be used under the terms of the GNU Lesser General Public
+** License version 2.1 as published by the Free Software Foundation and
+** appearing in the file LICENSE.LGPL included in the packaging of this
+** file. Please review the following information to ensure the GNU Lesser
+** General Public License version 2.1 requirements will be met:
+** http://www.gnu.org/licenses/old-licenses/lgpl-2.1.html.
+**
+** In addition, as a special exception, Nokia gives you certain additional
+** rights. These rights are described in the Nokia Qt LGPL Exception
+** version 1.1, included in the file LGPL_EXCEPTION.txt in this package.
+**
+** GNU General Public License Usage
+** Alternatively, this file may be used under the terms of the GNU General
+** Public License version 3.0 as published by the Free Software Foundation
+** and appearing in the file LICENSE.GPL included in the packaging of this
+** file. Please review the following information to ensure the GNU General
+** Public License version 3.0 requirements will be met:
+** http://www.gnu.org/copyleft/gpl.html.
+**
+** Other Usage
+** Alternatively, this file may be used in accordance with the terms and
+** conditions contained in a signed written agreement between you and Nokia.
+**
+**
+**
+**
+**
+** $QT_END_LICENSE$
+**
+****************************************************************************/
+
+#include <QDebug>
+#include <QtEndian>
+#include <QSocketNotifier>
+#include <QElapsedTimer>
+#include <qjsondocument.h>
+#include <qjsonobject.h>
+
+#include <sys/select.h>
+#include <stdio.h>
+#include <errno.h>
+
+#include "jsonpipe.h"
+#include "jsonbuffer_p.h"
+#include "bson/qt-bson_p.h"
+
+QT_BEGIN_NAMESPACE_JSONSTREAM
+
+/****************************************************************************/
+
+/*!
+    \class JsonPipe
+    \brief The JsonPipe class serializes JSON data.
+
+    The JsonPipe class is a generic interface for serializing and deserializing
+    JSON data over pipe connections.  It is designed to support multiple serialization
+    and deserialization formats by auto-detecting the format in use.
+*/
+
+/*!
+  Constructs a \c JsonPipe object with an optional \a parent.
+ */
+
+JsonPipe::JsonPipe(QObject *parent)
+    : QObject(parent)
+    , mIn(0)
+    , mOut(0)
+    , mFormat(FormatUndefined)
+{
+    mInBuffer = new JsonBuffer(this);
+    connect(mInBuffer, SIGNAL(objectReceived(const QJsonObject&)),
+            SLOT(objectReceived(const QJsonObject&)));
+}
+
+/*!
+  Delete the \c JsonPipe object
+ */
+
+JsonPipe::~JsonPipe()
+{
+}
+
+/*!
+  Return true if writing should be possible
+*/
+
+bool JsonPipe::writeEnabled() const
+{
+    return (mOut != NULL);
+}
+
+/*!
+  Return true if more data may be read
+*/
+
+bool JsonPipe::readEnabled() const
+{
+    return (mIn != NULL);
+}
+
+/*!
+  Set the current file descriptors
+*/
+
+void JsonPipe::setFds(int in_fd, int out_fd)
+{
+    if (mIn)
+        delete mIn;
+    if (mOut)
+        delete mOut;
+
+    mIn = new QSocketNotifier(in_fd, QSocketNotifier::Read, this);
+    mOut = new QSocketNotifier(out_fd, QSocketNotifier::Write, this);
+    connect(mIn, SIGNAL(activated(int)), SLOT(inReady(int)));
+    connect(mOut, SIGNAL(activated(int)), SLOT(outReady(int)));
+    mIn->setEnabled(true);
+    mOut->setEnabled(mOutBuffer.size() > 0);
+}
+
+void JsonPipe::inReady(int fd)
+{
+    mIn->setEnabled(false);
+    int n = mInBuffer->copyFromFd(fd);
+    if (n <= 0) {
+        mInBuffer->clear();
+        mIn->deleteLater();
+        mIn = NULL;
+        emit error( (n < 0) ? ReadFailed : ReadAtEnd );
+    }
+    else
+        mIn->setEnabled(true);
+}
+
+/*!
+  \internal
+
+  Return number of byte written
+ */
+int JsonPipe::writeInternal(int fd)
+{
+    if (!mOutBuffer.size())
+        return 0;
+
+    int n = ::write(fd, mOutBuffer.data(), mOutBuffer.size());
+    if (n <= 0) {
+        mOut->deleteLater();
+        mOut = NULL;
+        // ### TODO: This emits errors in the middle of waitForBytesWritten.
+        // ### This could cause problems 'cause it gets called in destructors
+        emit error(n < 0 ? WriteFailed : WriteAtEnd);
+    }
+    else if (n < mOutBuffer.size())
+        mOutBuffer = mOutBuffer.mid(n);
+    else
+        mOutBuffer.clear();
+    return n;
+}
+
+void JsonPipe::outReady(int)
+{
+    Q_ASSERT(mOut);
+    mOut->setEnabled(false);
+    if (mOutBuffer.size()) {
+        writeInternal(mOut->socket());
+        if (mOut && !mOutBuffer.isEmpty())
+            mOut->setEnabled(true);
+    }
+}
+
+/*!
+  Send a JsonObject \a object over the pipe
+*/
+
+bool JsonPipe::send(const QJsonObject& object)
+{
+    if (!mOut)
+        return false;
+
+    QJsonDocument document(object);
+
+    switch (mFormat) {
+    case FormatUndefined:
+        mFormat = FormatQBJS;
+        // Deliberate fall through
+    case FormatQBJS:
+        mOutBuffer.append(document.toBinaryData());
+        break;
+    case FormatUTF8:
+        mOutBuffer.append(document.toJson());
+        break;
+    case FormatBSON:
+    {
+        BsonObject bson(document.toVariant().toMap());
+        mOutBuffer.append("bson");
+        mOutBuffer.append(bson.data());
+        break;
+    }
+    }
+    if (mOutBuffer.size())
+        mOut->setEnabled(true);
+    return true;
+}
+
+/*!
+  \internal
+  Handle a received Qt Binary Json \a object and emit the correct signals
+*/
+
+void JsonPipe::objectReceived(const QJsonObject& object)
+{
+    if (mFormat == FormatUndefined)
+        mFormat = mInBuffer->format();
+    emit messageReceived(object);
+}
+
+/*!
+  Return the current JsonPipe::EncodingFormat.
+ */
+
+EncodingFormat JsonPipe::format() const
+{
+    return mFormat;
+}
+
+/*!
+  Set the EncodingFormat to \a format.
+ */
+
+void JsonPipe::setFormat( EncodingFormat format )
+{
+    mFormat = format;
+}
+
+/*
+  Blocks until all of the output buffer has been written to the pipe.
+  We return true if and only if there was data to be written and it
+  was successfully written.
+ */
+
+bool JsonPipe::waitForBytesWritten(int msecs)
+{
+    if (!mOut || mOutBuffer.isEmpty())
+        return false;
+
+    mOut->setEnabled(false);
+
+    QElapsedTimer stopWatch;
+    stopWatch.start();
+
+    while (mOut && !mOutBuffer.isEmpty()) {
+        fd_set wfds;
+        FD_ZERO(&wfds);
+        FD_SET(mOut->socket(),&wfds);
+
+        int timeout = msecs - stopWatch.elapsed();
+        struct timeval tv;
+        struct timeval *tvptr = ((msecs > 0 && timeout > 0) ? &tv : NULL);
+        if (tvptr) {
+            tv.tv_sec = timeout / 1000;
+            tv.tv_usec = (timeout % 1000) * 1000;
+        }
+
+        int retval = ::select(mOut->socket() + 1, NULL, &wfds, NULL, tvptr);
+        if (retval == -1 && errno == EINTR)
+            continue;
+        if (retval <= 0)
+            break;
+        writeInternal(mOut->socket());
+    }
+
+    if (mOut && !mOutBuffer.isEmpty())
+        mOut->setEnabled(true);
+    return mOutBuffer.isEmpty();
+}
+
+/*!
+    Sends the \a map via the pipe.
+    The QVariant types allowed are restricted to basic types supported
+    by the BsonObject which is in principle bool, int, long, QString and
+    arrays and maps of them.
+
+    \sa BsonObject
+*/
+JsonPipe& operator<<( JsonPipe& s, const QJsonObject& map )
+{
+    s.send(map);
+    return s;
+}
+
+/*!
+    \fn void JsonPipe::messageReceived(const QJsonObject& message)
+    This signal is emitted when a new \a message has been received on the
+    pipe.
+*/
+
+/*!
+    \fn void JsonPipe::aboutToClose()
+    This signal is emitted when the underlying \c QIODevice is about to close.
+
+    \sa QIODevice
+*/
+
+#include "moc_jsonpipe.cpp"
+
+QT_END_NAMESPACE_JSONSTREAM
diff --git a/src/jsonpipe.h b/src/jsonpipe.h
new file mode 100644
index 0000000..c7df9a6
--- /dev/null
+++ b/src/jsonpipe.h
@@ -0,0 +1,105 @@
+/****************************************************************************
+**
+** Copyright (C) 2011 Nokia Corporation and/or its subsidiary(-ies).
+** All rights reserved.
+** Contact: Nokia Corporation (qt-info@nokia.com)
+**
+** This file is part of the QtAddOn.JsonStream module of the Qt Toolkit.
+**
+** $QT_BEGIN_LICENSE:LGPL$
+** GNU Lesser General Public License Usage
+** This file may be used under the terms of the GNU Lesser General Public
+** License version 2.1 as published by the Free Software Foundation and
+** appearing in the file LICENSE.LGPL included in the packaging of this
+** file. Please review the following information to ensure the GNU Lesser
+** General Public License version 2.1 requirements will be met:
+** http://www.gnu.org/licenses/old-licenses/lgpl-2.1.html.
+**
+** In addition, as a special exception, Nokia gives you certain additional
+** rights. These rights are described in the Nokia Qt LGPL Exception
+** version 1.1, included in the file LGPL_EXCEPTION.txt in this package.
+**
+** GNU General Public License Usage
+** Alternatively, this file may be used under the terms of the GNU General
+** Public License version 3.0 as published by the Free Software Foundation
+** and appearing in the file LICENSE.GPL included in the packaging of this
+** file. Please review the following information to ensure the GNU General
+** Public License version 3.0 requirements will be met:
+** http://www.gnu.org/copyleft/gpl.html.
+**
+** Other Usage
+** Alternatively, this file may be used in accordance with the terms and
+** conditions contained in a signed written agreement between you and Nokia.
+**
+**
+**
+**
+**
+** $QT_END_LICENSE$
+**
+****************************************************************************/
+
+#ifndef _JSON_PIPE_H
+#define _JSON_PIPE_H
+
+#include <QVariant>
+#include <QJsonObject>
+#include <QJsonDocument>
+#include "jsonstream-global.h"
+
+class QSocketNotifier;
+
+QT_BEGIN_NAMESPACE_JSONSTREAM
+
+class JsonBuffer;
+
+class Q_ADDON_JSONSTREAM_EXPORT JsonPipe : public QObject
+{
+    Q_OBJECT
+public:
+    JsonPipe(QObject *parent = 0);
+    virtual ~JsonPipe();
+
+    bool writeEnabled() const;
+    bool readEnabled() const;
+
+    Q_INVOKABLE bool send(const QJsonObject& message);
+
+    EncodingFormat format() const;
+    void           setFormat(EncodingFormat format);
+
+    void setFds(int in_fd, int out_fd);
+
+    enum PipeError { WriteFailed, WriteAtEnd, ReadFailed, ReadAtEnd };
+    Q_ENUMS(PipeError);
+
+    bool waitForBytesWritten(int msecs = 30000);
+
+signals:
+    void messageReceived(const QJsonObject& message);
+    void error(PipeError);
+
+protected slots:
+    void objectReceived(const QJsonObject& object);
+    void inReady(int fd);
+    void outReady(int fd);
+
+protected:
+    void sendInternal(const QByteArray& byteArray);
+
+private:
+    int writeInternal(int fd);
+
+private:
+    JsonBuffer      *mInBuffer;
+    QByteArray       mOutBuffer;
+    QSocketNotifier *mIn;
+    QSocketNotifier *mOut;
+    EncodingFormat   mFormat;
+};
+
+JsonPipe& operator<<( JsonPipe&, const QJsonObject& );
+
+QT_END_NAMESPACE_JSONSTREAM
+
+#endif  // _JSON_PIPE_H
diff --git a/src/src.pro b/src/src.pro
index 7551bd6..8f35c66 100644
--- a/src/src.pro
+++ b/src/src.pro
@@ -54,6 +54,7 @@ PUBLIC_HEADERS += \
     $$PWD/jsonserver.h \
     $$PWD/jsonstream-global.h \
     $$PWD/jsonserverclient.h \
+    $$PWD/jsonpipe.h \
     $$SCHEMA_PUBLIC_HEADERS
 
 HEADERS += \
@@ -74,6 +75,7 @@ SOURCES += \
     $$PWD/jsonuidrangeauthority.cpp \
     $$PWD/jsonserverclient.cpp \
     $$PWD/jsonserver.cpp \
+    $$PWD/jsonpipe.cpp \
     $$SCHEMA_SOURCES
 
 mac:QMAKE_FRAMEWORK_BUNDLE_NAME = $$QT.jsonstream.name
diff --git a/tests/auto/jsonstream/tst_jsonstream.cpp b/tests/auto/jsonstream/tst_jsonstream.cpp
index 8a47f9c..39b20ff 100644
--- a/tests/auto/jsonstream/tst_jsonstream.cpp
+++ b/tests/auto/jsonstream/tst_jsonstream.cpp
@@ -44,6 +44,7 @@
 #include <QLocalServer>
 #include "jsonserver.h"
 #include "jsonstream.h"
+#include "jsonpipe.h"
 #include "jsonuidauthority.h"
 #include "jsonuidrangeauthority.h"
 #include "schemavalidator.h"
@@ -249,11 +250,15 @@ private slots:
     void authRangeFail();
     void formatTest();
     void schemaTest();
+    void pipeTest();
+    void pipeFormatTest();
+    void pipeWaitTest();
 };
 
 void tst_JsonStream::initTestCase()
 {
     qRegisterMetaType<QJsonObject>();
+    qRegisterMetaType<JsonPipe::PipeError>("PipeError");
 }
 
 
@@ -455,6 +460,114 @@ void tst_JsonStream::schemaTest()
     child.waitForFinished();
 }
 
+void waitForSpy(QSignalSpy& spy, int count, int timeout=5000) {
+    QTime stopWatch;
+    stopWatch.restart();
+    forever {
+        if (spy.count() == count)
+            break;
+        QTestEventLoop::instance().enterLoop(1);
+        if (stopWatch.elapsed() >= timeout)
+            QFAIL("Timed out");
+    }
+}
+
+
+class Pipes {
+public:
+    Pipes() {
+        ::pipe(fd1);
+        ::pipe(fd2);
+    }
+    ~Pipes() {
+        ::close(fd1[0]);
+        ::close(fd1[1]);
+        ::close(fd2[0]);
+        ::close(fd2[1]);
+    }
+    void join(JsonPipe& jp1, JsonPipe& jp2) {
+        // fd1[0] = Read end of jp1    fd1[1] = Write end of jp2
+        // fd2[0] = Read end of jp2    fd2[1] = Write end of jp1
+        jp1.setFds(fd1[0], fd2[1]);
+        jp2.setFds(fd2[0], fd1[1]);
+    }
+    int fd1[2], fd2[2];
+};
+
+class PipeSpy {
+public:
+    PipeSpy(JsonPipe& jp)
+        : msg(&jp, SIGNAL(messageReceived(const QJsonObject&)))
+        , err(&jp, SIGNAL(error(PipeError))) {}
+    QJsonObject at(int i) { return qvariant_cast<QJsonObject>(msg.at(i).at(0)); }
+    QJsonObject last()    { return qvariant_cast<QJsonObject>(msg.last().at(0)); }
+    QSignalSpy msg, err;
+};
+
+void tst_JsonStream::pipeTest()
+{
+    Pipes pipes;
+    JsonPipe jpipe1, jpipe2;
+
+    QVERIFY(!jpipe1.writeEnabled());
+    QVERIFY(!jpipe1.readEnabled());
+    QVERIFY(!jpipe2.writeEnabled());
+    QVERIFY(!jpipe2.readEnabled());
+
+    pipes.join(jpipe1, jpipe2);
+
+    QVERIFY(jpipe1.writeEnabled());
+    QVERIFY(jpipe1.readEnabled());
+    QVERIFY(jpipe2.writeEnabled());
+    QVERIFY(jpipe2.readEnabled());
+
+    PipeSpy spy1(jpipe1);
+    PipeSpy spy2(jpipe2);
+
+    QJsonObject msg;
+    msg.insert("name", QStringLiteral("Fred"));
+    QVERIFY(jpipe1.send(msg));
+    waitForSpy(spy2.msg, 1);
+    QCOMPARE(spy2.at(0).value("name").toString(), QStringLiteral("Fred"));
+}
+
+void tst_JsonStream::pipeFormatTest()
+{
+    QList<EncodingFormat> formats = QList<EncodingFormat>() << FormatUTF8 << FormatBSON << FormatQBJS;
+
+    foreach (EncodingFormat format, formats) {
+        Pipes pipes;
+        JsonPipe jpipe1, jpipe2;
+        pipes.join(jpipe1, jpipe2);
+        PipeSpy spy(jpipe2);
+        jpipe1.setFormat(format);
+        QCOMPARE(jpipe1.format(), format);
+
+        QJsonObject msg;
+        msg.insert("name", QStringLiteral("Fred"));
+        QVERIFY(jpipe1.send(msg));
+        waitForSpy(spy.msg, 1);
+        QCOMPARE(spy.at(0).value("name").toString(), QStringLiteral("Fred"));
+        QCOMPARE(jpipe2.format(), format);
+    }
+}
+
+void tst_JsonStream::pipeWaitTest()
+{
+    Pipes pipes;
+    JsonPipe jpipe1, jpipe2;
+    pipes.join(jpipe1, jpipe2);
+
+    QJsonObject msg;
+    msg.insert("name", QStringLiteral("Jabberwocky"));
+    QVERIFY(jpipe1.send(msg));
+    QVERIFY(jpipe1.waitForBytesWritten());   // Actually push it out
+
+    ::close(pipes.fd2[1]);  // Close the write end of jp1
+    QVERIFY(jpipe1.send(msg));
+    QVERIFY(!jpipe1.waitForBytesWritten());
+}
+
 QTEST_MAIN(tst_JsonStream)
 
 #include "tst_jsonstream.moc"