initial JsonConnection implementation (synchronous)

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

19 files changed, 3048 insertions, 5 deletions

diff --git a/src/jsonbuffer.cpp b/src/jsonbuffer.cpp
index 7d54c83..fbabb85 100644
--- a/src/jsonbuffer.cpp
+++ b/src/jsonbuffer.cpp
@@ -43,6 +43,7 @@
 #include <QtEndian>
 #include <QJsonDocument>
 #include <QTextCodec>
+#include <QMutexLocker>
 
 #include "jsonbuffer_p.h"
 #include "qjsondocument.h"
@@ -83,6 +84,7 @@ JsonBuffer::JsonBuffer(QObject *parent)
     , mMessageAvailable(false)
     , mMessageSize(0)
     , mEnabled(true)
+    , mThreadProtection(false)
 {
 }
 
@@ -122,7 +124,10 @@ JsonBuffer::JsonBuffer(QObject *parent)
 
 void JsonBuffer::append(const QByteArray& data)
 {
-    mBuffer.append(data.data(), data.size());
+    {
+        QScopedPointer<QMutexLocker> locker(createLocker());
+        mBuffer.append(data.data(), data.size());
+    }
     if (0 < size())
         processMessages();
 }
@@ -135,7 +140,10 @@ void JsonBuffer::append(const QByteArray& data)
 
 void JsonBuffer::append(const char *data, int len)
 {
-    mBuffer.append(data, len);
+    {
+        QScopedPointer<QMutexLocker> locker(createLocker());
+        mBuffer.append(data, len);
+    }
     if (0 < size())
         processMessages();
 }
@@ -152,11 +160,14 @@ void JsonBuffer::append(const char *data, int len)
 int JsonBuffer::copyFromFd(int fd)
 {
     const int maxcopy = 1024;
+    QScopedPointer<QMutexLocker> locker(createLocker());
     uint oldSize = mBuffer.size();
     mBuffer.resize(oldSize + maxcopy);
     int n = ::read(fd, mBuffer.data()+oldSize, maxcopy);
     if (n > 0) {
         mBuffer.resize(oldSize+n);
+        if (!locker.isNull())
+            locker->unlock();
         processMessages();
     }
     else
@@ -170,6 +181,7 @@ int JsonBuffer::copyFromFd(int fd)
 
 void JsonBuffer::clear()
 {
+    QScopedPointer<QMutexLocker> locker(createLocker());
     mBuffer.clear();
     resetParser();
 }
@@ -242,6 +254,7 @@ void JsonBuffer::processMessages()
 */
 bool JsonBuffer::messageAvailable()
 {
+    QScopedPointer<QMutexLocker> locker(createLocker());
     if (mMessageAvailable) {
         // already found - no need to check again
         return true;
@@ -388,6 +401,7 @@ QJsonObject JsonBuffer::readMessage()
 {
     QJsonObject obj;
     if (messageAvailable()) {
+        QScopedPointer<QMutexLocker> locker(createLocker());
         switch (mFormat) {
         case FormatUndefined:
             break;
@@ -471,6 +485,11 @@ EncodingFormat JsonBuffer::format() const
     return mFormat;
 }
 
+QMutexLocker *JsonBuffer::createLocker()
+{
+    return mThreadProtection ? new QMutexLocker(&mMutex) : 0;
+}
+
 /*!
     \fn void JsonBuffer::readyReadMessage()
 
diff --git a/src/jsonbuffer_p.h b/src/jsonbuffer_p.h
index cae0009..cb292a4 100644
--- a/src/jsonbuffer_p.h
+++ b/src/jsonbuffer_p.h
@@ -45,9 +45,12 @@
 #include <QObject>
 #include <QByteArray>
 #include <QJsonObject>
+#include <QMutex>
 
 #include "jsonstream-global.h"
 
+class QMutexLocker;
+
 QT_BEGIN_NAMESPACE_JSONSTREAM
 
 class Q_ADDON_JSONSTREAM_EXPORT JsonBuffer : public QObject
@@ -70,9 +73,14 @@ public:
     inline bool isEnabled() const { return mEnabled; }
     inline void setEnabled(bool enable) { mEnabled = enable; }
 
+    inline void setThreadProtection(bool enable) { mThreadProtection = enable; }
+
 signals:
     void readyReadMessage();
 
+protected:
+    QMutexLocker *createLocker();
+
 private:
     void processMessages();
     bool scanUtf(int c);
@@ -92,6 +100,8 @@ private:
     bool             mMessageAvailable;
     int              mMessageSize;
     bool             mEnabled;
+    bool             mThreadProtection;
+    QMutex           mMutex;
 };
 
 inline QByteArray JsonBuffer::rawData(int _start, int _len) const
diff --git a/src/jsonconnection.cpp b/src/jsonconnection.cpp
new file mode 100644
index 0000000..d822c36
--- /dev/null
+++ b/src/jsonconnection.cpp
@@ -0,0 +1,562 @@
+/****************************************************************************
+**
+** 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 "jsonconnection.h"
+#include "jsonconnectionprocessor_p.h"
+#include "jsonendpoint.h"
+#include "jsonendpointmanager_p.h"
+#include <QThread>
+#include <QTimer>
+#include <QDebug>
+
+QT_BEGIN_NAMESPACE_JSONSTREAM
+
+class JsonConnectionPrivate
+{
+public:
+    JsonConnectionPrivate()
+        : mTcpHostPort(0)
+        , mAutoReconnectEnabled(false)
+        , mUseSeparateThread(false)
+        , mReadBufferSize(0)
+        , mWriteBufferSize(0)
+        , mManager(0)
+        , mConnected(0)
+        , mProcessorThread(0)
+        , mError(JsonConnection::NoError)
+        , mSubError(0) {}
+
+    QString     mLocalSocketName;
+    QString     mTcpHostName;
+    int         mTcpHostPort;
+    bool        mAutoReconnectEnabled;
+    bool        mUseSeparateThread;
+    qint64      mReadBufferSize;
+    qint64      mWriteBufferSize;
+    JsonEndpointManager *mManager;
+    JsonConnectionProcessor *mProcessor;
+    bool mConnected;
+    QThread *mProcessorThread;
+
+    JsonConnection::Error mError;
+    int                   mSubError;
+    QString               mErrorStr;
+
+    /*!
+      \internal
+     */
+    void createProcessorThread()
+    {
+        if (!mUseSeparateThread)
+            return;
+
+        mProcessorThread = new QThread();
+        QObject::connect(mProcessorThread, SIGNAL(finished()), mProcessor, SLOT(deleteLater()));
+        QObject::connect(mProcessorThread, SIGNAL(finished()), mProcessorThread, SLOT(deleteLater()));
+
+        mProcessor->moveToThread(mProcessorThread);
+        mProcessorThread->start();
+    }
+};
+
+/****************************************************************************/
+
+/*!
+    \class JsonConnection
+    \brief The JsonConnection class ...
+
+*/
+
+/*!
+    \enum JsonConnection::State
+
+    This enum describes the different states in which a connection can be.
+
+    \value Unconnected No connected.
+    \value Connecting Started establishing a connection.
+    \value Authenticating Started authentication process.
+    \value Connected A connection is established.
+
+    \sa JsonConnection::state()
+*/
+
+/*!
+  Constructs a \c JsonConnection object.
+ */
+
+JsonConnection::JsonConnection(QObject *parent)
+    : QObject(parent)
+    , d_ptr(new JsonConnectionPrivate())
+{
+    Q_D(JsonConnection);
+    d->mManager = new JsonEndpointManager(this);
+    d->mProcessor = new JsonConnectionProcessor();
+    d->mProcessor->setEndpointManager(d->mManager);
+    connect(d->mProcessor, SIGNAL(disconnected()), SIGNAL(disconnected()));
+    qRegisterMetaType<JsonConnection::State>("JsonConnection::State");
+    connect(d->mProcessor, SIGNAL(stateChanged(JsonConnection::State)), SIGNAL(stateChanged(JsonConnection::State)));
+    qRegisterMetaType<JsonConnection::State>("JsonConnection::Error");
+    connect(d->mProcessor, SIGNAL(error(JsonConnection::Error,int,QString)), SLOT(handleError(JsonConnection::Error,int,QString)));
+}
+
+/*!
+  Deletes the \c JsonConnection object.
+ */
+
+JsonConnection::~JsonConnection()
+{
+    Q_D(JsonConnection);
+    d->mProcessor->setEndpointManager(0);
+    if (d->mProcessorThread)
+        d->mProcessorThread->quit();
+    if (!d->mUseSeparateThread)
+        delete d->mProcessor;
+}
+
+/*!
+  Returns the state of the connection.
+*/
+JsonConnection::State JsonConnection::state() const
+{
+    Q_D(const JsonConnection);
+    return d->mProcessor->state();
+}
+
+/*!
+    Returns the type of error that last occurred.
+
+    \sa state(), errorString(), subError()
+*/
+
+JsonConnection::Error JsonConnection::error() const
+{
+    Q_D(const JsonConnection);
+    return d->mError;
+}
+
+/*!
+  Returns the additional error code of error that last occurred. This code depends on an error
+  type. It could be casted to QLocalSocket::LocalSocketError for LocalSocketError error and to
+  QAbstractSocket::SocketError for TcpSocketError error.
+
+  \sa error()
+*/
+
+int JsonConnection::subError() const
+{
+    Q_D(const JsonConnection);
+    return d->mSubError;
+}
+
+/*!
+  Returns a human-readable description of the last device error that occurred.
+
+  \sa error()
+*/
+
+QString JsonConnection::errorString() const
+{
+    Q_D(const JsonConnection);
+    return d->mErrorStr;
+}
+
+/*!
+  Returns a socket name to be used for Unix local socket connection.
+*/
+QString JsonConnection::localSocketName() const
+{
+    Q_D(const JsonConnection);
+    return d->mLocalSocketName;
+}
+
+/*!
+  Sets a socket name to be used for Unix local socket connection.
+*/
+void JsonConnection::setLocalSocketName(const QString &name)
+{
+    Q_D(JsonConnection);
+    d->mLocalSocketName = name;
+}
+
+/*!
+  Returns a host name to be used for TCP socket connection.
+*/
+QString JsonConnection::tcpHostName() const
+{
+    Q_D(const JsonConnection);
+    return d->mTcpHostName;
+}
+
+/*!
+  Sets a host name to be used for TCP socket connection.
+*/
+void JsonConnection::setTcpHostName(const QString &name)
+{
+    Q_D(JsonConnection);
+    d->mTcpHostName = name;
+}
+
+/*!
+  Returns a port number to be used for TCP socket connection.
+*/
+int JsonConnection::tcpHostPort() const
+{
+    Q_D(const JsonConnection);
+    return d->mTcpHostPort;
+}
+
+/*!
+  Sets a port number to be used for TCP socket connection.
+*/
+void JsonConnection::setTcpHostPort(int port)
+{
+    Q_D(JsonConnection);
+    if (port >= 0)
+        d->mTcpHostPort = port;
+}
+
+/*!
+  Specifies whether to reconnect when server connection is lost.
+*/
+bool JsonConnection::autoReconnectEnabled() const
+{
+    Q_D(const JsonConnection);
+    return d->mAutoReconnectEnabled;
+}
+
+/*!
+  Sets whether to reconnect when server connection is lost.
+*/
+void JsonConnection::setAutoReconnectEnabled(bool enabled)
+{
+    Q_D(JsonConnection);
+    d->mAutoReconnectEnabled = enabled;
+    d->mProcessor->setAutoReconnectEnabled(enabled);
+}
+
+/*!
+    Returns a property which value in message object will be used as an endpoint name.
+*/
+QString JsonConnection::endpointPropertyName() const
+{
+    Q_D(const JsonConnection);
+    return d->mManager->endpointPropertyName();
+}
+
+/*!
+    Sets a property which value in message object will be used as an endpoint name.
+*/
+void JsonConnection::setEndpointPropertyName(const QString &property)
+{
+    Q_D(JsonConnection);
+    if (d->mManager) {
+        d->mManager->setEndpointPropertyName(property);
+    }
+}
+
+/*!
+  Returns a maximum size of the inbound message buffer.
+ */
+qint64 JsonConnection::readBufferSize() const
+{
+    Q_D(const JsonConnection);
+    return d->mReadBufferSize;
+}
+
+/*!
+  Sets a maximum size of the inbound message buffer to \a sz thus capping a size
+  of an inbound message.
+ */
+void JsonConnection::setReadBufferSize(qint64 sz)
+{
+    if (sz >= 0) {
+        Q_D(JsonConnection);
+        d->mReadBufferSize = sz;
+
+        if (!d->mUseSeparateThread || !d->mConnected)
+            d->mProcessor->setReadBufferSize(sz);
+        else
+            QMetaObject::invokeMethod(d->mProcessor,
+                                  "setReadBufferSize",
+                                  Qt::QueuedConnection,
+                                  QGenericReturnArgument(),
+                                  Q_ARG(qint64, sz));
+    }
+}
+
+/*!
+  Returns a maximum size of the outbound message buffer.  A value of 0
+  means the buffer size is unlimited.
+ */
+qint64 JsonConnection::writeBufferSize() const
+{
+    Q_D(const JsonConnection);
+    return d->mWriteBufferSize;
+}
+
+/*!
+  Sets a maximum size of the outbound message buffer to \a sz thus capping a size
+  of an outbound message.  A value of 0 means the buffer size is unlimited.
+ */
+void JsonConnection::setWriteBufferSize(qint64 sz)
+{
+    if (sz >= 0) {
+        Q_D(JsonConnection);
+        d->mWriteBufferSize = sz;
+
+        if (!d->mUseSeparateThread || !d->mConnected)
+            d->mProcessor->setWriteBufferSize(sz);
+        else
+            QMetaObject::invokeMethod(d->mProcessor,
+                                  "setWriteBufferSize",
+                                  Qt::QueuedConnection,
+                                  QGenericReturnArgument(),
+                                  Q_ARG(qint64, sz));
+    }
+}
+
+/*!
+  Adds endpoint to a connection.
+ */
+void JsonConnection::addEndpoint(JsonEndpoint *endpoint)
+{
+    Q_D(JsonConnection);
+    d->mManager->addEndpoint(endpoint);
+    if (endpoint->connection() != this)
+        endpoint->setConnection(this);
+}
+
+/*!
+  Removes endpoint from a connection.
+ */
+void JsonConnection::removeEndpoint(JsonEndpoint *endpoint)
+{
+    Q_D(JsonConnection);
+    d->mManager->removeEndpoint(endpoint);
+}
+
+/*!
+  Sets whether to create a separate worker thread for a connection
+ */
+void JsonConnection::setUseSeparateThreadForProcessing(bool use)
+{
+    Q_D(JsonConnection);
+
+    Q_ASSERT(!d->mConnected);
+    if (!d->mConnected)
+        d->mUseSeparateThread = use;
+}
+
+/*!
+  Returns whether a separate worker thread for a connection required
+*/
+
+bool JsonConnection::useSeparateThreadForProcessing() const
+{
+    Q_D(const JsonConnection);
+    return d->mUseSeparateThread;
+}
+
+/*!
+  Returns a default endpoint without name.
+*/
+
+JsonEndpoint * JsonConnection::defaultEndpoint()
+{
+    Q_D(JsonConnection);
+    return d->mManager->defaultEndpoint();
+}
+
+/*!
+  Connect to the JsonServer over a TCP socket at \a hostname and \a port.
+  Return true if the connection is successful.
+*/
+
+bool JsonConnection::connectTCP(const QString& hostname, int port)
+{
+    bool bRet = false;
+    Q_D(JsonConnection);
+
+    if (!d->mConnected) {
+        d->mConnected = true;
+        if (d->mUseSeparateThread) {
+            d->createProcessorThread();
+        }
+        connect(d->mProcessor, SIGNAL(readBufferOverflow(qint64)), SIGNAL(readBufferOverflow(qint64)),
+                d->mUseSeparateThread ? Qt::BlockingQueuedConnection : Qt::AutoConnection);
+    }
+
+    if (!hostname.isEmpty())
+        setTcpHostName(hostname);
+    if (port > 0)
+        setTcpHostPort(port);
+
+    if (!d->mUseSeparateThread)
+        bRet = d->mProcessor->connectTCP(tcpHostName(), tcpHostPort());
+    else
+        QMetaObject::invokeMethod(d->mProcessor,
+                              "connectTCP",
+                              Qt::BlockingQueuedConnection,
+                              Q_RETURN_ARG(bool, bRet),
+                              Q_ARG(QString, tcpHostName()),
+                              Q_ARG(int, tcpHostPort()));
+    return bRet;
+}
+
+/*!
+  Connect to the JsonServer over a Unix local socket to \a socketname.
+  Return true if the connection is successful.
+ */
+bool JsonConnection::connectLocal(const QString& socketname)
+{
+    bool bRet = false;
+    Q_D(JsonConnection);
+
+    if (!d->mConnected) {
+        d->mConnected = true;
+        if (d->mUseSeparateThread) {
+            d->createProcessorThread();
+        }
+        connect(d->mProcessor, SIGNAL(readBufferOverflow(qint64)), SIGNAL(readBufferOverflow(qint64)),
+                d->mUseSeparateThread ? Qt::BlockingQueuedConnection : Qt::AutoConnection);
+    }
+
+    if (!socketname.isEmpty())
+        setLocalSocketName(socketname);
+
+    if (!d->mUseSeparateThread)
+        bRet = d->mProcessor->connectLocal(localSocketName());
+    else
+        QMetaObject::invokeMethod(d->mProcessor,
+                              "connectLocal",
+                              Qt::BlockingQueuedConnection,
+                              Q_RETURN_ARG(bool, bRet),
+                              Q_ARG(QString, localSocketName()));
+    return bRet;
+}
+
+/*!
+  Set the current stream encoding \a format.
+  This controls how messages will be sent
+*/
+
+void JsonConnection::setFormat( EncodingFormat format )
+{
+    Q_D(JsonConnection);
+    if (!d->mUseSeparateThread || !d->mConnected)
+        d->mProcessor->setFormat(format);
+    else
+        QMetaObject::invokeMethod(d->mProcessor,
+                              "setFormat",
+                              Qt::QueuedConnection,
+                              QGenericReturnArgument(),
+                              Q_ARG(int, format));
+}
+
+/*!
+  \internal
+*/
+JsonConnectionProcessor *JsonConnection::processor() const
+{
+    Q_D(const JsonConnection);
+    return d->mProcessor;
+}
+
+/*!
+  \internal
+*/
+void JsonConnection::handleError(JsonConnection::Error err, int suberr, QString str)
+{
+    Q_D(JsonConnection);
+    d->mError = err;
+    d->mSubError = suberr;
+    d->mErrorStr = str;
+    emit error(d->mError, d->mSubError);
+}
+
+/*! \fn JsonConnection::bytesWritten(qint64 bytes)
+
+    This signal is emitted every time a payload of data has been
+    written to the device. The \a bytes argument is set to the number
+    of bytes that were written in this payload.
+*/
+
+/*! \fn JsonConnection::readBufferOverflow(qint64 bytes)
+
+  This signal is emitted when the read buffer is full of data that has been read
+  from the \l{device()}, \a bytes additional bytes are available on the device,
+  but the message is not complete.  The \l{readBufferSize()} may be increased
+  to a sufficient size in a slot connected to this signal, in which case more
+  data will be read into the read buffer.  If the buffer size  is not increased,
+  the connection is closed.
+*/
+
+/*!
+    \fn void JsonConnection::stateChanged(JsonConnection::State state)
+
+    This signal is emitted whenever JsonConnection's state changes.
+    The \a state parameter is the new state.
+
+    \sa state()
+*/
+
+/*!
+    \fn void JsonConnection::error(JsonConnection::Error error, int subError)
+
+    This signal is emitted after an error occurred. The \a error
+    parameter describes the type of error that occurred and \a subError
+    contains the additional error code.
+
+    \sa error(), subError(), errorString()
+*/
+
+/*!
+    \fn void JsonConnection::disconnected()
+
+    This signal is emitted when the connection has been disconnected.
+
+    \warning If you need to delete the sender() of this signal in a slot connected
+    to it, use the \l{QObject::deleteLater()}{deleteLater()} function.
+*/
+
+#include "moc_jsonconnection.cpp"
+
+QT_END_NAMESPACE_JSONSTREAM
diff --git a/src/jsonconnection.h b/src/jsonconnection.h
new file mode 100644
index 0000000..fdd7352
--- /dev/null
+++ b/src/jsonconnection.h
@@ -0,0 +1,148 @@
+/****************************************************************************
+**
+** 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_CONNECTION_H
+#define _JSON_CONNECTION_H
+
+#include "jsonstream-global.h"
+#include <QObject>
+
+QT_BEGIN_NAMESPACE_JSONSTREAM
+
+class JsonEndpoint;
+class JsonConnectionProcessor;
+
+class JsonConnectionPrivate;
+class Q_ADDON_JSONSTREAM_EXPORT JsonConnection : public QObject
+{
+    Q_OBJECT
+    Q_PROPERTY(State state READ state)
+    Q_PROPERTY(QString localSocketName READ localSocketName WRITE setLocalSocketName)
+    Q_PROPERTY(QString tcpHostName READ tcpHostName WRITE setTcpHostName)
+    Q_PROPERTY(int tcpHostPort READ tcpHostPort WRITE setTcpHostPort)
+    Q_PROPERTY(QString endpointPropertyName READ endpointPropertyName WRITE setEndpointPropertyName)
+    Q_PROPERTY(bool autoReconnectEnabled READ autoReconnectEnabled WRITE setAutoReconnectEnabled)
+    Q_PROPERTY(bool useSeparateThreadForProcessing READ useSeparateThreadForProcessing WRITE setUseSeparateThreadForProcessing)
+    Q_PROPERTY(qint64 readBufferSize READ readBufferSize WRITE setReadBufferSize)
+    Q_PROPERTY(qint64 writeBufferSize READ writeBufferSize WRITE setWriteBufferSize)
+public:
+    JsonConnection(QObject *parent = 0);
+    ~JsonConnection();
+
+    enum State {
+        Unconnected = 0,
+        Connecting,
+        Authenticating,
+        Connected
+    };
+    Q_ENUMS(State)
+
+    State state() const;
+
+    enum Error {
+        NoError = 0,
+        UnknownError,
+        LocalSocketError,
+        TcpSocketError
+    };
+    Q_ENUMS(Error)
+
+    Error error() const;
+    int subError() const;
+    QString errorString() const;
+
+    QString localSocketName() const;
+    void setLocalSocketName(const QString &);
+
+    QString tcpHostName() const;
+    void setTcpHostName(const QString &);
+
+    int tcpHostPort() const;
+    void setTcpHostPort(int);
+
+    QString endpointPropertyName() const;
+    void setEndpointPropertyName(const QString &);
+
+    bool autoReconnectEnabled() const;
+    void setAutoReconnectEnabled(bool);
+
+    bool useSeparateThreadForProcessing() const;
+    void setUseSeparateThreadForProcessing(bool);
+
+    qint64 readBufferSize() const;
+    void setReadBufferSize(qint64);
+
+    qint64 writeBufferSize() const;
+    void setWriteBufferSize(qint64 sz);
+
+    Q_INVOKABLE bool connectTCP(const QString& hostname = QString::null, int port = 0);
+    Q_INVOKABLE bool connectLocal(const QString& socketname = QString::null);
+
+    void addEndpoint(JsonEndpoint *);
+    JsonEndpoint * defaultEndpoint();
+    void removeEndpoint(JsonEndpoint *);
+
+    void setFormat( EncodingFormat format );
+
+    JsonConnectionProcessor *processor() const;
+
+signals:
+    void bytesWritten(qint64);
+    void readBufferOverflow(qint64);
+    void stateChanged(JsonConnection::State);
+    void disconnected();
+    void error(JsonConnection::Error, int);
+
+private slots:
+    void handleError(JsonConnection::Error, int, QString);
+
+private:
+    Q_DECLARE_PRIVATE(JsonConnection)
+    QScopedPointer<JsonConnectionPrivate> d_ptr;
+
+    // forbid copy constructor
+    JsonConnection(const JsonConnection &);
+    void operator=(const JsonConnection &);
+};
+
+QT_END_NAMESPACE_JSONSTREAM
+
+#endif // _JSON_CONNECTION_H
diff --git a/src/jsonconnectionprocessor.cpp b/src/jsonconnectionprocessor.cpp
new file mode 100644
index 0000000..48dde27
--- /dev/null
+++ b/src/jsonconnectionprocessor.cpp
@@ -0,0 +1,436 @@
+/****************************************************************************
+**
+** 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 "jsonconnectionprocessor_p.h"
+#include "jsonendpointmanager_p.h"
+#include "jsonclient.h"
+#include "jsonstream.h"
+#include "jsonendpoint.h"
+#include "jsonbuffer_p.h"
+
+#include <QMap>
+#include <QThread>
+#include <QFile>
+#include <QDir>
+#include <QTimer>
+
+const int knAUTO_RECONNECTION_TIMEOUT = 5000;
+const int knSOCKET_READ_BUFFER_SIZE = 64*1024;
+
+QT_BEGIN_NAMESPACE_JSONSTREAM
+
+class JsonConnectionProcessorPrivate
+{
+public:
+    JsonConnectionProcessorPrivate()
+        : mState(JsonConnection::Unconnected)
+        , mManager(0)
+        , mDestinationEndpoint(0)
+        , mAutoReconnectEnabled(false)
+        , mExplicitDisconnect(false)
+        , mReconnectionTimer(0)
+    {}
+
+    JsonConnection::State mState;
+    JsonEndpointManager *mManager;
+    JsonStream mStream;
+    QMutex          mutex;
+    QJsonObject     mObject; // buffer for single message
+    JsonEndpoint   *mDestinationEndpoint;
+
+    QString mServerName;
+    int mPort;
+
+    // auto reconnection
+    bool mAutoReconnectEnabled;
+    bool mExplicitDisconnect;
+    QTimer *mReconnectionTimer;
+};
+
+/****************************************************************************/
+
+/*!
+    \class JsonConnectionProcessor
+    \brief The JsonConnectionProcessor class ...
+
+*/
+
+/*!
+  Constructs a \c JsonConnectionProcessor object.
+ */
+
+JsonConnectionProcessor::JsonConnectionProcessor()
+    : QObject(0)
+    , d_ptr(new JsonConnectionProcessorPrivate())
+{
+    Q_D(JsonConnectionProcessor);
+    d->mStream.setParent(this);
+    d->mStream.setThreadProtection(true);
+}
+
+/*!
+  Deletes the \c JsonConnectionProcessor object.
+ */
+
+JsonConnectionProcessor::~JsonConnectionProcessor()
+{
+    // Variant streams don't own the socket
+    Q_D(JsonConnectionProcessor);
+    QIODevice *device = d->mStream.device();
+    if (device) {
+        device->disconnect(this);
+        d->mStream.setDevice(0);
+        delete device;
+    }
+}
+
+/*!
+  Returns the state of the connection.
+*/
+JsonConnection::State JsonConnectionProcessor::state() const
+{
+    return d_func()->mState;
+}
+
+/*!
+  Sets whether to reconnect when server connection is lost.
+*/
+void JsonConnectionProcessor::setAutoReconnectEnabled(bool enabled)
+{
+    Q_D(JsonConnectionProcessor);
+    d->mAutoReconnectEnabled = enabled;
+}
+
+void JsonConnectionProcessor::setEndpointManager(JsonEndpointManager *manager)
+{
+    Q_D(JsonConnectionProcessor);
+    QMutexLocker locker(&d->mutex);
+    d->mManager = manager;
+}
+
+/*!
+  Connect to the JsonServer over a TCP socket at \a hostname and \a port.
+  Return true if the connection is successful.
+*/
+
+bool JsonConnectionProcessor::connectTCP(const QString& hostname, int port)
+{
+    Q_D(JsonConnectionProcessor);
+    if (JsonConnection::Connecting != d->mState)
+        emit stateChanged(d->mState = JsonConnection::Connecting);
+
+    QTcpSocket *socket = new QTcpSocket(this);
+    connect(socket, SIGNAL(error(QAbstractSocket::SocketError)),
+            SLOT(handleSocketError(QAbstractSocket::SocketError)));
+    socket->connectToHost(hostname, port);
+
+    if (socket->waitForConnected()) {
+        connect(socket, SIGNAL(disconnected()), SLOT(handleSocketDisconnected()));
+        d->mStream.setDevice(socket);
+        connect(&d->mStream, SIGNAL(readyReadMessage()), this, SLOT(processMessage()));
+        connect(&d->mStream, SIGNAL(bytesWritten(qint64)), this, SIGNAL(bytesWritten(qint64)));
+        connect(&d->mStream, SIGNAL(readBufferOverflow(qint64)), this, SIGNAL(readBufferOverflow(qint64)));
+        d->mServerName = hostname;
+        d->mPort = port;
+        d->mState = JsonConnection::Connected;
+        emit stateChanged(d->mState);
+        return true;
+    }
+
+    d->mState = JsonConnection::Unconnected;
+    emit stateChanged(d->mState);
+    delete socket;
+    return false;
+}
+
+/*!
+  Connect to the JsonServer over a Unix local socket to \a socketname.
+  Return true if the connection is successful.
+ */
+bool JsonConnectionProcessor::connectLocal(const QString& socketname)
+{
+    QString socketPath(socketname);
+#if defined(Q_OS_UNIX)
+    if (!socketPath.startsWith(QLatin1Char('/')))
+        socketPath.prepend(QDir::tempPath() + QLatin1Char('/'));
+#endif
+
+    if (!QFile::exists(socketPath)) {
+        qWarning() << Q_FUNC_INFO << "socket does not exist" << socketPath;
+        return false;
+    }
+
+    Q_D(JsonConnectionProcessor);
+    if (JsonConnection::Connecting != d->mState)
+        emit stateChanged(d->mState = JsonConnection::Connecting);
+
+    QLocalSocket *socket = new QLocalSocket(this);
+    connect(socket, SIGNAL(error(QLocalSocket::LocalSocketError)),
+            SLOT(handleSocketError(QLocalSocket::LocalSocketError)));
+    socket->setReadBufferSize(knSOCKET_READ_BUFFER_SIZE);
+    socket->connectToServer(socketPath);
+
+    if (socket->waitForConnected()) {
+        connect(socket, SIGNAL(disconnected()), SLOT(handleSocketDisconnected()));
+        d->mStream.setDevice(socket);
+        connect(&d->mStream, SIGNAL(readyReadMessage()), this, SLOT(processMessage()));
+        connect(&d->mStream, SIGNAL(bytesWritten(qint64)), this, SIGNAL(bytesWritten(qint64)));
+        connect(&d->mStream, SIGNAL(readBufferOverflow(qint64)), this, SIGNAL(readBufferOverflow(qint64)));
+        d->mServerName = socketname;
+        d->mPort = -1; // local socket
+        d->mState = JsonConnection::Connected;
+        emit stateChanged(d->mState);
+        return true;
+    }
+
+    d->mState = JsonConnection::Unconnected;
+    emit stateChanged(d->mState);
+    delete socket;
+    return false;
+}
+
+/*!
+  \internal
+*/
+void JsonConnectionProcessor::handleSocketDisconnected()
+{
+    Q_D(JsonConnectionProcessor);
+    QIODevice *device = d->mStream.device();
+    if (!device)
+        return;
+
+    device->disconnect(this);
+    d->mStream.setDevice(0);
+    device->deleteLater();
+
+    if (d->mAutoReconnectEnabled && !d->mExplicitDisconnect)
+    {
+        if (!d->mReconnectionTimer || !d->mReconnectionTimer->isActive())
+        {
+            if (!d->mReconnectionTimer) {
+                // create timer
+                d->mReconnectionTimer = new QTimer(this);
+                d->mReconnectionTimer->setInterval(knAUTO_RECONNECTION_TIMEOUT);
+                connect(d->mReconnectionTimer, SIGNAL(timeout()), SLOT(handleReconnect()));
+            }
+            d->mState = JsonConnection::Connecting;
+            emit stateChanged(d->mState);
+            d->mReconnectionTimer->start();
+        }
+        return;
+    }
+
+    d->mState = JsonConnection::Unconnected;
+    emit disconnected();
+    emit stateChanged(d->mState);
+}
+
+/*!
+  \internal
+*/
+void JsonConnectionProcessor::handleSocketError(QAbstractSocket::SocketError _error)
+{
+    if (QAbstractSocket *socket = qobject_cast<QAbstractSocket*>(sender())) {
+        emit error(JsonConnection::TcpSocketError, _error, socket->errorString());
+    }
+}
+
+/*!
+  \internal
+*/
+void JsonConnectionProcessor::handleSocketError(QLocalSocket::LocalSocketError _error)
+{
+    if (QLocalSocket *socket = qobject_cast<QLocalSocket*>(sender())) {
+        emit error(JsonConnection::LocalSocketError, _error, socket->errorString());
+    }
+}
+
+/*!
+  \internal
+*/
+void JsonConnectionProcessor::handleReconnect()
+{
+    Q_D(JsonConnectionProcessor);
+    d->mReconnectionTimer->stop();
+
+    if (JsonConnection::Connecting == d->mState) {
+        if (d->mPort < 0)
+            connectLocal(d->mServerName);
+        else
+            connectTCP(d->mServerName, d->mPort);
+    }
+}
+
+/*!
+  \internal
+  Handle a received readyReadMessage signal and emit the correct signals
+*/
+
+void JsonConnectionProcessor::processMessage(JsonEndpoint *destination)
+{
+    Q_D(JsonConnectionProcessor);
+    if (!destination)
+        d->mutex.lock();
+    if (!d->mDestinationEndpoint  && d->mManager) { // no message available
+        if (d->mStream.messageAvailable()) {
+            QJsonObject obj = d->mStream.readMessage();
+            if (!obj.isEmpty()) {
+                JsonEndpoint *endpoint = d->mManager->endpoint(obj);
+                if (endpoint) {
+                    // find a corresponding endpoint and notify it
+                    d->mDestinationEndpoint = endpoint;
+                    d->mObject = obj;
+
+                    if (destination != endpoint) {
+                        // do not emit a signal if destination endpoint is already processing messages
+                        if (!destination)
+                            d->mutex.unlock();
+                        // use a queued signal if we process messages in one endpoint and need to notify another
+                        QMetaObject::invokeMethod(endpoint,
+                                                  "slotReadyReadMessage",
+                                                  !destination ? Qt::AutoConnection : Qt::QueuedConnection,
+                                                  QGenericReturnArgument());
+                        return;
+                    }
+                }
+            }
+        }
+    }
+    if (!destination)
+        d->mutex.unlock();
+}
+
+/*!
+  Returns \b true if a message is available for \a endpoint to be read via \l{readMessage()}
+  or \b false otherwise.
+ */
+
+bool JsonConnectionProcessor::messageAvailable(JsonEndpoint *endpoint)
+{
+    bool ret = false;
+    if (endpoint) {
+        Q_D(JsonConnectionProcessor);
+        QMutexLocker locker(&d->mutex);
+        if (!(ret = (d->mDestinationEndpoint == endpoint))) {
+            if (!d->mDestinationEndpoint) {
+                // check stream for more if no messages available
+                processMessage(endpoint);
+                ret = (d->mDestinationEndpoint == endpoint);
+            }
+        }
+    }
+    return ret;
+}
+
+/*!
+  Returns a JSON object that has been received for \a endpoint.  If no message is
+  available, an empty JSON object is returned.
+ */
+QJsonObject JsonConnectionProcessor::readMessage(JsonEndpoint *endpoint)
+{
+    QJsonObject obj;
+    if (endpoint) {
+        Q_D(JsonConnectionProcessor);
+        QMutexLocker locker(&d->mutex);
+        if (!d->mDestinationEndpoint) {
+            // check stream for more if no messages available
+            processMessage(endpoint);
+        }
+
+        if (d->mDestinationEndpoint == endpoint) {
+            obj = d->mObject;
+            d->mObject = QJsonObject();
+            d->mDestinationEndpoint = 0;
+        }
+    }
+    return obj;
+}
+
+/*!
+  Sets a maximum size of the inbound message buffer to \a sz thus capping a size
+  of an inbound message.
+ */
+void JsonConnectionProcessor::setReadBufferSize(qint64 sz)
+{
+    if (sz >= 0) {
+        Q_D(JsonConnectionProcessor);
+        d->mStream.setReadBufferSize(sz);
+    }
+}
+
+/*!
+  Sets a maximum size of the outbound message buffer to \a sz thus capping a size
+  of an outbound message.  A value of 0 means the buffer size is unlimited.
+ */
+void JsonConnectionProcessor::setWriteBufferSize(qint64 sz)
+{
+    if (sz >= 0) {
+        Q_D(JsonConnectionProcessor);
+        d->mStream.setWriteBufferSize(sz);
+    }
+}
+
+/*!
+  Set the current stream encoding \a format.
+  This controls how messages will be sent
+*/
+
+void JsonConnectionProcessor::setFormat( int format )
+{
+    Q_D(JsonConnectionProcessor);
+    d->mStream.setFormat((EncodingFormat)format); // EncodingFormat
+}
+
+/*!
+  Send a \a message over the socket.
+  Returns true if the entire message was send/buffered or false otherwise.
+*/
+
+bool JsonConnectionProcessor::send(QJsonObject message)
+{
+    Q_D(JsonConnectionProcessor);
+    return d->mStream.send(message);
+}
+
+#include "moc_jsonconnectionprocessor_p.cpp"
+
+QT_END_NAMESPACE_JSONSTREAM
+
diff --git a/src/jsonconnectionprocessor_p.h b/src/jsonconnectionprocessor_p.h
new file mode 100644
index 0000000..9fdb3ab
--- /dev/null
+++ b/src/jsonconnectionprocessor_p.h
@@ -0,0 +1,111 @@
+/****************************************************************************
+**
+** 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_CONNECTION_PROCESSOR_H
+#define _JSON_CONNECTION_PROCESSOR_H
+
+#include "jsonstream-global.h"
+#include "jsonconnection.h"
+
+#include <QLocalSocket>
+#include <QTcpSocket>
+class QJsonObject;
+
+QT_BEGIN_NAMESPACE_JSONSTREAM
+
+#include <QLocalSocket>
+#include <QTcpSocket>
+
+class JsonEndpoint;
+class JsonEndpointManager;
+
+class JsonConnectionProcessorPrivate;
+class JsonConnectionProcessor : public QObject
+{
+    Q_OBJECT
+public:
+    JsonConnectionProcessor();
+    ~JsonConnectionProcessor();
+
+    void setEndpointManager(JsonEndpointManager *);
+
+    void setAutoReconnectEnabled(bool enabled);
+    JsonConnection::State state() const;
+
+signals:
+    void stateChanged(JsonConnection::State);
+    void readyReadMessage();
+    void disconnected();
+    void bytesWritten(qint64);
+    void readBufferOverflow(qint64);
+    void error(JsonConnection::Error,int,QString);
+
+public slots:
+    bool connectTCP(const QString&,int);
+    bool connectLocal(const QString&);
+    void setFormat(int);
+    void setReadBufferSize(qint64);
+    void setWriteBufferSize(qint64);
+    bool send(QJsonObject message);
+    bool messageAvailable(JsonEndpoint *);
+    QJsonObject readMessage(JsonEndpoint *);
+
+protected slots:
+    void processMessage(JsonEndpoint* = 0);
+    void handleSocketDisconnected();
+    void handleReconnect();
+    void handleSocketError(QAbstractSocket::SocketError);
+    void handleSocketError(QLocalSocket::LocalSocketError);
+
+protected:
+
+private:
+    Q_DECLARE_PRIVATE(JsonConnectionProcessor)
+    QScopedPointer<JsonConnectionProcessorPrivate> d_ptr;
+
+    // forbid copy constructor
+    JsonConnectionProcessor(const JsonConnectionProcessor &);
+    void operator=(const JsonConnectionProcessor &);
+};
+
+QT_END_NAMESPACE_JSONSTREAM
+
+#endif // _JSON_CONNECTION_PROCESSOR_H
diff --git a/src/jsonendpoint.cpp b/src/jsonendpoint.cpp
new file mode 100644
index 0000000..4825708
--- /dev/null
+++ b/src/jsonendpoint.cpp
@@ -0,0 +1,261 @@
+/****************************************************************************
+**
+** 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 "jsonendpoint.h"
+#include "jsonconnection.h"
+#include "jsonconnectionprocessor_p.h"
+#include <qjsonobject.h>
+#include <QVariant>
+#include <QDebug>
+
+QT_BEGIN_NAMESPACE_JSONSTREAM
+
+class JsonEndpointPrivate
+{
+public:
+    JsonEndpointPrivate()
+        : mConnection(0)
+        , mEmittedReadyRead(false)
+        , mMessageReady(false)
+    {
+    }
+
+    QString             mName;
+    JsonConnection     *mConnection;
+    bool                mEmittedReadyRead;
+    bool                mMessageReady;
+};
+
+/****************************************************************************/
+
+/*!
+    \class JsonEndpoint
+    \brief The JsonEndpoint class ...
+
+*/
+
+/*!
+  Constructs a \c JsonEndpoint object.
+ */
+
+JsonEndpoint::JsonEndpoint(const QString & name, JsonConnection *connection)
+    : QObject(0)
+    , d_ptr(new JsonEndpointPrivate())
+{
+    Q_D(JsonEndpoint);
+    d->mName = name;
+
+    setConnection(connection);
+}
+
+/*!
+  Deletes the \c JsonEndpoint object.
+ */
+
+JsonEndpoint::~JsonEndpoint()
+{
+    Q_D(JsonEndpoint);
+    if (d->mConnection)
+        d->mConnection->removeEndpoint(this);
+}
+
+/*!
+  Returns a name used for message multiplexing. A default endpoint should not
+  have a name
+ */
+QString JsonEndpoint::name() const
+{
+    Q_D(const JsonEndpoint);
+    return d->mName;
+}
+
+/*!
+  Sets a \a name used for message multiplexing. A default endpoint should not
+  have a name
+ */
+void JsonEndpoint::setName( const QString & name )
+{
+    Q_D(JsonEndpoint);
+    d->mName = name;
+}
+
+/*!
+  Returns a connection that is used by endpoint
+ */
+JsonConnection *JsonEndpoint::connection() const
+{
+    Q_D(const JsonEndpoint);
+    return d->mConnection;
+}
+
+/*!
+  Sets a \a connection to be used by endpoint
+ */
+void JsonEndpoint::setConnection(JsonConnection *connection)
+{
+    Q_D(JsonEndpoint);
+    d->mConnection = connection;
+    if (d->mConnection)
+        d->mConnection->addEndpoint(this);
+}
+
+/*!
+  Send a QVariantMap \a message over the connection.
+  Returns \b true if the entire message was sent or buffered or \b false otherwise.
+*/
+bool JsonEndpoint::send(const QVariantMap& message)
+{
+    return send(QJsonObject::fromVariantMap(message));
+}
+
+/*!
+  Send a JsonObject \a message over the connection.
+  Returns \b true if the entire message was sent or buffered or \b false otherwise.
+*/
+bool JsonEndpoint::send(const QJsonObject& message)
+{
+    bool ret = false;
+    Q_D(const JsonEndpoint);
+    if (d->mConnection) {
+        if (!d->mConnection->useSeparateThreadForProcessing()) {
+            ret = d->mConnection->processor()->send(message);
+        }
+        else {
+            QMetaObject::invokeMethod(d->mConnection->processor(),
+                                      "send",
+                                      Qt::BlockingQueuedConnection,
+                                      Q_RETURN_ARG(bool, ret),
+                                      Q_ARG(QJsonObject, message));
+        }
+    }
+    return ret;
+}
+
+/*!
+  \internal
+  Handle a notification from connection processor and emit the correct signals
+*/
+void JsonEndpoint::slotReadyReadMessage()
+{
+    Q_D(JsonEndpoint);
+    d->mMessageReady = true;
+    if (!d->mEmittedReadyRead) {
+        d->mEmittedReadyRead = true;
+        emit readyReadMessage();
+        d->mEmittedReadyRead = false;
+    }
+}
+
+
+/*!
+  Returns \b true if a message is available to be read via \l{readMessage()}
+  or \b false otherwise.
+ */
+bool JsonEndpoint::messageAvailable()
+{
+    Q_D(JsonEndpoint);
+    bool ret = d->mMessageReady;
+    if (!d->mMessageReady) {
+        // check again
+        if (d->mConnection) {
+            ret = d->mConnection->processor()->messageAvailable(this);
+            d->mMessageReady = ret;
+        }
+    }
+    return ret;
+}
+
+/*!
+  Returns a JSON object that has been received as a variant map.  If no message is
+  available, an empty JSON object is returned.
+ */
+QVariantMap JsonEndpoint::readMessageMap()
+{
+    return readMessage().toVariantMap();
+}
+
+/*!
+  Returns a JSON object that has been received.  If no message is
+  available, an empty JSON object is returned.
+ */
+QJsonObject JsonEndpoint::readMessage()
+{
+    QJsonObject obj;
+    Q_D(JsonEndpoint);
+    if (d->mConnection) {
+        obj = d->mConnection->processor()->readMessage(this);
+        d->mMessageReady = false;
+    }
+    return obj;
+}
+
+/*!
+    \fn void JsonEndpoint::readyReadMessage()
+
+    This signal is emitted once every time new data arrives on the device
+    and a message is ready. \l{readMessage()} should be used to retrieve a message
+    and \l{messageAvailable()} to check for next available messages.
+    The client code may look like this:
+
+    \code
+    ...
+    connect(endpoint, SIGNAL(readyReadMessage()), this, SLOT(processMessages()));
+    ...
+
+    void Foo::processMessages()
+    {
+        while (endpoint->messageAvailable()) {
+            QJsonObject obj = endpoint->readMessage();
+            <process message>
+        }
+    }
+    \endcode
+
+    \b readyReadMessage() is not emitted recursively; if you reenter the event loop
+    inside a slot connected to the \b readyReadMessage() signal, the signal will not
+    be reemitted.
+
+    \sa readMessage(), messageAvailable()
+*/
+
+#include "moc_jsonendpoint.cpp"
+
+QT_END_NAMESPACE_JSONSTREAM
diff --git a/src/jsonendpoint.h b/src/jsonendpoint.h
new file mode 100644
index 0000000..26b40a0
--- /dev/null
+++ b/src/jsonendpoint.h
@@ -0,0 +1,94 @@
+/****************************************************************************
+**
+** 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_ENDPOINT_H
+#define _JSON_ENDPOINT_H
+
+#include <QObject>
+#include <QJsonObject>
+#include "jsonstream-global.h"
+
+QT_BEGIN_NAMESPACE_JSONSTREAM
+
+class JsonConnection;
+
+class JsonEndpointPrivate;
+class Q_ADDON_JSONSTREAM_EXPORT JsonEndpoint : public QObject
+{
+    Q_OBJECT
+    Q_PROPERTY(QString name READ name WRITE setName)
+    Q_PROPERTY(JsonConnection* connection READ connection WRITE setConnection)
+public:
+    JsonEndpoint(const QString & = QString::null, JsonConnection * = 0);
+    virtual ~JsonEndpoint();
+
+    QString name() const;
+    void    setName( const QString & name );
+
+    JsonConnection *connection() const;
+    void setConnection(JsonConnection *);
+
+    Q_INVOKABLE bool send(const QVariantMap& message);
+    Q_INVOKABLE bool send(const QJsonObject& message);
+
+    Q_INVOKABLE bool messageAvailable();
+
+    Q_INVOKABLE QJsonObject readMessage();
+    Q_INVOKABLE QVariantMap readMessageMap();
+
+signals:
+    void readyReadMessage();
+
+protected slots:
+    void slotReadyReadMessage();
+
+private:
+    Q_DECLARE_PRIVATE(JsonEndpoint)
+    QScopedPointer<JsonEndpointPrivate> d_ptr;
+
+    // forbid copy constructor
+    JsonEndpoint(const JsonEndpoint &);
+    void operator=(const JsonEndpoint &);
+};
+
+QT_END_NAMESPACE_JSONSTREAM
+
+#endif // _JSON_ENDPOINT_H
diff --git a/src/jsonendpointmanager.cpp b/src/jsonendpointmanager.cpp
new file mode 100644
index 0000000..f12f1ec
--- /dev/null
+++ b/src/jsonendpointmanager.cpp
@@ -0,0 +1,155 @@
+/****************************************************************************
+**
+** 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 "jsonendpointmanager_p.h"
+#include "jsonendpoint.h"
+#include "jsonconnection.h"
+
+QT_BEGIN_NAMESPACE_JSONSTREAM
+
+static const QLatin1String kstrEndpointKey("endpoint");
+
+/****************************************************************************/
+
+/*!
+    \class JsonEndpointManager
+    \brief The JsonEndpointManager class ...
+
+*/
+
+/*!
+  Constructs a \c JsonEndpointManager object.
+ */
+
+JsonEndpointManager::JsonEndpointManager(JsonConnection *parent)
+    : QObject(parent), mInit(false), mEndpointPropertyName(kstrEndpointKey)
+{
+}
+
+/*!
+  Deletes the \c JsonEndpointManager object.
+ */
+
+JsonEndpointManager::~JsonEndpointManager()
+{
+    clear();
+}
+
+JsonEndpoint *JsonEndpointManager::defaultEndpoint()
+{
+    JsonEndpoint *endpoint;
+    endpoints();
+    if (mDefaultEndpoint) {
+        endpoint = mDefaultEndpoint;
+    }
+    else {
+        endpoint = new JsonEndpoint();
+        JsonConnection *connection = qobject_cast<JsonConnection *>(parent());
+        if (connection) {
+            connection->addEndpoint(endpoint);
+        }
+        mDefaultEndpoint = endpoint;
+    }
+    return endpoint;
+}
+
+void JsonEndpointManager::addEndpoint(JsonEndpoint *endpoint)
+{
+    mInit = false; // rehashing required
+    mEndpoints.insert(QString::number((ulong)endpoint), endpoint);
+}
+
+void JsonEndpointManager::removeEndpoint(JsonEndpoint *endpoint)
+{
+    mEndpoints.remove(endpoint->name());
+    endpoint->setConnection(0);
+}
+
+QHash<QString, JsonEndpoint*> & JsonEndpointManager::endpoints()
+{
+    if (!mInit) {
+        // rehash with real names
+        QList<JsonEndpoint *> lst = mEndpoints.values();
+        mEndpoints.clear();
+        mDefaultEndpoint = 0;
+        foreach (JsonEndpoint *e, lst) {
+            mEndpoints.insert(e->name(), e);
+            if (e->name().isEmpty())
+                mDefaultEndpoint = e;
+        }
+        mInit = true;
+    }
+    return mEndpoints;
+}
+
+JsonEndpoint *JsonEndpointManager::endpoint(const QJsonObject &message)
+{
+    JsonEndpoint *endpoint = 0;
+    QString str;
+    QHash<QString, JsonEndpoint*> & hash(endpoints());
+    if (mDefaultEndpoint && 1 == hash.size()) {
+        // no need to search - have only single default endpoint
+    }
+    else if (!(str = message.value(mEndpointPropertyName).toString()).isEmpty()) {
+        QHash<QString, JsonEndpoint*>::const_iterator it = hash.find(str);
+        if (it != hash.constEnd())
+            endpoint = *it;
+    }
+
+    if (endpoint == 0)
+        endpoint = mDefaultEndpoint;
+
+    return endpoint;
+}
+
+void JsonEndpointManager::clear()
+{
+    QList<JsonEndpoint *> lst = mEndpoints.values();
+    foreach (JsonEndpoint *endpoint, lst) {
+        endpoint->setConnection(0);
+    }
+    mEndpoints.clear();
+}
+
+#include "moc_jsonendpointmanager_p.cpp"
+
+QT_END_NAMESPACE_JSONSTREAM
+
diff --git a/src/jsonendpointmanager_p.h b/src/jsonendpointmanager_p.h
new file mode 100644
index 0000000..4502417
--- /dev/null
+++ b/src/jsonendpointmanager_p.h
@@ -0,0 +1,87 @@
+/****************************************************************************
+**
+** 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_ENDPOINT_MANAGER_H
+#define _JSON_ENDPOINT_MANAGER_H
+
+#include <QObject>
+#include <QHash>
+#include "jsonstream-global.h"
+
+class QJsonObject;
+
+QT_BEGIN_NAMESPACE_JSONSTREAM
+
+class JsonEndpoint;
+class JsonConnection;
+
+class JsonEndpointManagerPrivate;
+class Q_ADDON_JSONSTREAM_EXPORT JsonEndpointManager : public QObject
+{
+    Q_OBJECT
+    Q_PROPERTY(QString endpointPropertyName READ endpointPropertyName WRITE setEndpointPropertyName)
+public:
+    JsonEndpointManager(JsonConnection *parent);
+    ~JsonEndpointManager();
+
+    JsonEndpoint *defaultEndpoint();
+
+    QString endpointPropertyName() const { return mEndpointPropertyName; }
+    void setEndpointPropertyName(const QString & name) { mEndpointPropertyName = name; }
+
+    void addEndpoint(JsonEndpoint *);
+    void removeEndpoint(JsonEndpoint *);
+    void clear();
+
+    QHash<QString, JsonEndpoint*> & endpoints();
+
+    virtual JsonEndpoint *endpoint(const QJsonObject &);
+
+protected:
+    bool mInit;
+    QString mEndpointPropertyName;
+    QHash<QString, JsonEndpoint*> mEndpoints;
+    JsonEndpoint *mDefaultEndpoint;
+};
+
+QT_END_NAMESPACE_JSONSTREAM
+
+#endif // _JSON_ENDPOINT_MANAGER_H
diff --git a/src/jsonstream.cpp b/src/jsonstream.cpp
index 9a022a8..e98b501 100644
--- a/src/jsonstream.cpp
+++ b/src/jsonstream.cpp
@@ -192,6 +192,7 @@ void JsonStream::setDevice( QIODevice *device )
         disconnect(d->mDevice, SIGNAL(readyRead()), this, SLOT(dataReadyOnSocket()));
         disconnect(d->mDevice, SIGNAL(bytesWritten(qint64)), this, SIGNAL(bytesWritten(qint64)));
         disconnect(d->mDevice, SIGNAL(aboutToClose()), this, SIGNAL(aboutToClose()));
+        d->mBuffer->clear();
     }
     d->mDevice = device;
     if (device) {
@@ -434,10 +435,19 @@ QJsonObject JsonStream::readMessage()
  */
 bool JsonStream::messageAvailable()
 {
-    Q_D(JsonStream);
+    Q_D(const JsonStream);
     return d->mBuffer->messageAvailable();
 }
 
+/*!
+  internal
+ */
+void JsonStream::setThreadProtection(bool enable) const
+{
+    Q_D(const JsonStream);
+    d->mBuffer->setThreadProtection(enable);
+}
+
 /*! \fn JsonStream::bytesWritten(qint64 bytes)
 
     This signal is emitted every time a payload of data has been
diff --git a/src/jsonstream.h b/src/jsonstream.h
index 18795c1..e9db07a 100644
--- a/src/jsonstream.h
+++ b/src/jsonstream.h
@@ -106,6 +106,10 @@ protected:
     bool sendInternal(const QByteArray& byteArray);
 
 private:
+    friend class JsonConnectionProcessor;
+    void setThreadProtection(bool) const;
+
+private:
     Q_DECLARE_PRIVATE(JsonStream)
     QScopedPointer<JsonStreamPrivate> d_ptr;
 };
diff --git a/src/src.pro b/src/src.pro
index 8f35c66..7f8d52d 100644
--- a/src/src.pro
+++ b/src/src.pro
@@ -55,10 +55,14 @@ PUBLIC_HEADERS += \
     $$PWD/jsonstream-global.h \
     $$PWD/jsonserverclient.h \
     $$PWD/jsonpipe.h \
+    $$PWD/jsonconnection.h \
+    $$PWD/jsonendpoint.h \
     $$SCHEMA_PUBLIC_HEADERS
 
 HEADERS += \
    $$PWD/jsonbuffer_p.h \
+   $$PWD/jsonconnectionprocessor_p.h \
+   $$PWD/jsonendpointmanager_p.h \
    $$BSON_HEADERS \
    $$PUBLIC_HEADERS \
    $$SCHEMA_HEADERS
@@ -76,6 +80,10 @@ SOURCES += \
     $$PWD/jsonserverclient.cpp \
     $$PWD/jsonserver.cpp \
     $$PWD/jsonpipe.cpp \
-    $$SCHEMA_SOURCES
+    $$PWD/jsonconnection.cpp \
+    $$PWD/jsonconnectionprocessor.cpp \
+    $$PWD/jsonendpoint.cpp \
+    $$PWD/jsonendpointmanager.cpp \
+    $$SCHEMA_SOURCES \
 
 mac:QMAKE_FRAMEWORK_BUNDLE_NAME = $$QT.jsonstream.name
diff --git a/tests/auto/auto.pro b/tests/auto/auto.pro
index a8778ca..3e60ad6 100644
--- a/tests/auto/auto.pro
+++ b/tests/auto/auto.pro
@@ -1,2 +1,2 @@
 TEMPLATE = subdirs
-SUBDIRS = jsonstream jsonschema jsonbuffer
+SUBDIRS = jsonstream jsonschema jsonbuffer jsonconnection
diff --git a/tests/auto/jsonconnection/jsonconnection.pro b/tests/auto/jsonconnection/jsonconnection.pro
new file mode 100644
index 0000000..fba387f
--- /dev/null
+++ b/tests/auto/jsonconnection/jsonconnection.pro
@@ -0,0 +1,3 @@
+TEMPLATE = subdirs
+
+SUBDIRS = test testClient
diff --git a/tests/auto/jsonconnection/test/test.pro b/tests/auto/jsonconnection/test/test.pro
new file mode 100644
index 0000000..bcf1e01
--- /dev/null
+++ b/tests/auto/jsonconnection/test/test.pro
@@ -0,0 +1,7 @@
+CONFIG += testcase
+CONFIG -= app_bundle
+
+QT = jsonstream testlib qml
+
+SOURCES = ../tst_jsonconnection.cpp
+TARGET = ../tst_jsonconnection
diff --git a/tests/auto/jsonconnection/testClient/main.cpp b/tests/auto/jsonconnection/testClient/main.cpp
new file mode 100644
index 0000000..a0afa0b
--- /dev/null
+++ b/tests/auto/jsonconnection/testClient/main.cpp
@@ -0,0 +1,258 @@
+ /****************************************************************************
+**
+** 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 <QCoreApplication>
+#include <QStringList>
+#include <QDebug>
+#include <QJsonArray>
+
+#include <QLocalSocket>
+#include <QLocalServer>
+#include <QDebug>
+#include <QTimer>
+#include <QFile>
+
+#include "jsonstream.h"
+#include "jsonserver.h"
+
+QT_USE_NAMESPACE_JSONSTREAM
+
+QString gSocketname = "/tmp/tst_jsonstream";
+
+/****************************/
+class BasicServer : public QObject {
+    Q_OBJECT
+
+public:
+    BasicServer(const QString& socketname)
+        : socket(0), stream(0)
+    {
+        QLocalServer::removeServer(socketname);
+        QFile::remove(socketname);
+        server = new QLocalServer(this);
+        connect(server, SIGNAL(newConnection()), SLOT(handleConnection()));
+        Q_ASSERT(server->listen(socketname));
+
+        QTimer::singleShot(100, this, SLOT(ready()));
+    }
+
+    ~BasicServer() {
+        Q_ASSERT(server);
+        delete server;
+        server = NULL;
+    }
+
+    bool send(const QJsonObject& message) {
+        return stream ? stream->send(message) : false;
+    }
+
+    void disconnectFromServer(int timeout)
+    {
+        QTimer::singleShot(timeout, this, SLOT(doDisconnect()));
+    }
+
+private slots:
+    void ready()
+    {
+        fprintf(stdout, "READY\n"); // send ready command
+        fflush(stdout);
+    }
+
+    void handleConnection() {
+        socket = server->nextPendingConnection();
+        Q_ASSERT(socket);
+        stream = new JsonStream(socket);
+        stream->setParent(socket);
+        connect(socket, SIGNAL(disconnected()), SLOT(handleDisconnection()));
+        connect(stream, SIGNAL(readyReadMessage()), SLOT(processMessages()));
+    }
+
+    void processMessages() {
+        while (stream->messageAvailable()) {
+            QJsonObject obj = stream->readMessage();
+            if (!obj.isEmpty())
+                emit messageReceived(obj);
+        }
+    }
+
+    void doDisconnect() {
+        // disconnect and wait for a new connection
+        if (socket) {
+            socket->disconnect(this);
+            socket->disconnectFromServer();
+            socket->deleteLater();
+            stream->deleteLater();
+            socket = NULL;
+            stream = NULL;
+        }
+    }
+
+
+    void handleDisconnection() {
+        Q_ASSERT(socket);
+        socket->deleteLater();
+        socket = NULL;
+        stream = NULL;
+
+        exit(0);
+    }
+
+signals:
+    void messageReceived(const QJsonObject&);
+
+private:
+    QLocalServer *server;
+    QLocalSocket *socket;
+    JsonStream   *stream;
+};
+
+class Container : public QObject
+{
+    Q_OBJECT
+
+public:
+    Container();
+    void sendMessage(const QString &endpoint = QString::null);
+
+public slots:
+    void received(const QJsonObject& message);
+
+private:
+    BasicServer *mServer;
+    int         mCounter;
+};
+
+Container::Container()
+    : mCounter(0)
+{
+    qDebug() << "Creating new json server at" << gSocketname;
+    mServer = new BasicServer(gSocketname);
+    connect(mServer, SIGNAL(messageReceived(const QJsonObject&)),
+            SLOT(received(const QJsonObject&)));
+}
+
+void Container::sendMessage(const QString &endpoint)
+{
+    static int counter = 0;
+
+    QJsonObject msg;
+    msg.insert("counter", counter++);
+    msg.insert("text", QLatin1String("Standard text"));
+    msg.insert("number", mCounter++);
+    msg.insert("int", 100);
+    msg.insert("float", 100.0);
+    msg.insert("true", true);
+    msg.insert("false", false);
+    msg.insert("array", QJsonArray::fromStringList(QStringList() << "one" << "two" << "three"));
+    QJsonObject obj;
+    obj.insert("item1", QLatin1String("This is item 1"));
+    obj.insert("item2", QLatin1String("This is item 2"));
+    msg.insert("object", obj);
+
+    if (!endpoint.isEmpty())
+        msg.insert("endpoint", endpoint);
+
+    mServer->send(msg);
+}
+
+void Container::received(const QJsonObject& message)
+{
+    qDebug() << "received " << message;
+
+    QString command = message.value("command").toString();
+    if (!command.isEmpty()) {
+        qDebug() << "Received command" << command;
+        if (command == "exit")
+            exit(0);
+        else if (command == "crash")
+            exit(1);
+        else if (command == "reply")
+            sendMessage(message.value("endpoint").toString());
+        else if (command == "flurry") {
+            QJsonArray endpoints = message.value("endpoints").toArray();
+            int msgsPerEndpoint = static_cast<int> (message.value("count").toDouble());
+            int nEndpoints = endpoints.size() ? endpoints.size() : 1;
+            bool grouped = message.value("grouped").toBool();
+            for (int count = msgsPerEndpoint * nEndpoints; count ; --count ) {
+                QString endpoint;
+                if (nEndpoints > 0) {
+                    int idx = grouped ? ((count-1) / msgsPerEndpoint) : ((count-1) % nEndpoints);
+//                    qDebug() << "ZZZ idx = " << idx;
+                    endpoint = endpoints.at(idx).toString();
+                }
+//                qDebug() << "ZZZ     sending message for " << endpoint;
+                sendMessage(endpoint);
+            }
+        }
+        else if (command == "disconnect") {
+            // send message back first
+            mServer->send(message);
+
+            int timeout = message.value("timeout").toDouble();
+            mServer->disconnectFromServer(timeout);
+        }
+    }
+    else {
+        // send message back
+        mServer->send(message);
+    }
+}
+
+int
+main(int argc, char **argv)
+{
+    QCoreApplication app(argc, argv);
+    QStringList args = QCoreApplication::arguments();
+    QString progname = args.takeFirst();
+    while (args.size()) {
+        QString arg = args.at(0);
+        if (!arg.startsWith("-"))
+            break;
+        args.removeFirst();
+        if (arg == "-socket")
+            gSocketname = args.takeFirst();
+    }
+
+    Container c;
+    return app.exec();
+}
+
+#include "main.moc"
diff --git a/tests/auto/jsonconnection/testClient/testClient.pro b/tests/auto/jsonconnection/testClient/testClient.pro
new file mode 100644
index 0000000..a053251
--- /dev/null
+++ b/tests/auto/jsonconnection/testClient/testClient.pro
@@ -0,0 +1,5 @@
+CONFIG -= app_bundle
+QT += testlib jsonstream
+
+SOURCES = main.cpp
+TARGET  = testClient
diff --git a/tests/auto/jsonconnection/tst_jsonconnection.cpp b/tests/auto/jsonconnection/tst_jsonconnection.cpp
new file mode 100644
index 0000000..86e5e83
--- /dev/null
+++ b/tests/auto/jsonconnection/tst_jsonconnection.cpp
@@ -0,0 +1,865 @@
+/****************************************************************************
+**
+** 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 <QtTest>
+#include <QLocalSocket>
+#include <QLocalServer>
+#include "jsonserver.h"
+#include "jsonconnection.h"
+#include "jsonendpoint.h"
+
+#include <QtQml/qqmlengine.h>
+#include <QtQml/qqmlcomponent.h>
+#include <QtQml/qqmlcontext.h>
+
+QT_USE_NAMESPACE_JSONSTREAM
+
+Q_DECLARE_METATYPE(QJsonObject);
+Q_DECLARE_METATYPE(JsonConnection::State);
+
+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 Child : public QObject {
+    Q_OBJECT
+public:
+    Child(const QString& progname, const QStringList& arguments) {
+        process = new QProcess;
+        process->setProcessChannelMode(QProcess::MergedChannels);
+        connect(process, SIGNAL(readyReadStandardOutput()), this, SLOT(readyread()));
+        connect(process, SIGNAL(finished(int, QProcess::ExitStatus)),
+                this, SLOT(finished(int, QProcess::ExitStatus)));
+        connect(process, SIGNAL(stateChanged(QProcess::ProcessState)),
+                SLOT(stateChanged(QProcess::ProcessState)));
+        connect(process, SIGNAL(error(QProcess::ProcessError)),
+                SLOT(error(QProcess::ProcessError)));
+        process->start(progname, arguments);
+        QVERIFY(process->waitForStarted(5000));
+    }
+
+    ~Child() {
+        if (process)
+            delete process;
+        process = 0;
+    }
+
+    void waitForFinished() {
+        if (process->state() == QProcess::Running)
+            QVERIFY(process->waitForFinished(5000));
+        QVERIFY(process->exitCode() == 0);
+        delete process;
+        process = 0;
+    }
+
+protected slots:
+    void readyread() {
+        QByteArray ba = process->readAllStandardOutput();
+
+        if (ba.simplified() == "READY") {
+            qDebug() << "SERVER READY";
+            emit serverReady();
+            return;
+        }
+
+        QList<QByteArray> list = ba.split('\n');
+        foreach (const QByteArray& s, list)
+            if (s.size())
+                qDebug() << "PROCESS" << s;
+    }
+    void stateChanged(QProcess::ProcessState state) {
+        qDebug() << "Process state" << state;
+    }
+    void error(QProcess::ProcessError err) {
+        qDebug() << "Process error" << err;
+    }
+    void finished( int exitcode, QProcess::ExitStatus status ) {
+        qDebug() << Q_FUNC_INFO << exitcode << status;
+    }
+
+signals:
+    void serverReady();
+
+private:
+    QProcess *process;
+};
+
+/****************************/
+class EndpointContainer : public QObject
+{
+    Q_OBJECT
+
+public:
+    EndpointContainer(QObject *parent = 0);
+
+    JsonEndpoint *addEndpoint(const QString & name);
+
+    void sendMessage(const QString & endpointDestination = QString::null, JsonEndpoint *sender = 0);
+    void sendMessage(const QStringList & endpointDestinationList, int messagesPerEndpoint, bool grouped);
+
+    QList<QObject *> endpoints() { return mEndpoints; }
+
+    JsonConnection *connection() { return mConnection; }
+    void setConnection(JsonConnection *connection);
+
+public slots:
+    void processMessages();
+
+signals:
+    void messageReceived(const QJsonObject&, QObject *);
+
+protected:
+    void addEndpoint(JsonEndpoint *endpoint);
+
+private:
+    JsonConnection *mConnection;
+    JsonEndpoint *mStream;
+    QList<QObject *> mEndpoints;
+    int         mCounter;
+};
+
+
+EndpointContainer::EndpointContainer(QObject *parent)
+    : QObject(parent), mConnection(0), mCounter(0)
+{
+}
+
+JsonEndpoint *EndpointContainer::addEndpoint(const QString & name)
+{
+    JsonEndpoint *endpoint = new JsonEndpoint(name, mConnection);
+    endpoint->setParent(mConnection);
+    addEndpoint(endpoint);
+    return endpoint;
+}
+
+void EndpointContainer::addEndpoint(JsonEndpoint *endpoint)
+{
+    mStream = endpoint;
+    connect(mStream, SIGNAL(readyReadMessage()), SLOT(processMessages()));
+    mEndpoints.append(mStream);
+}
+
+void EndpointContainer::setConnection(JsonConnection *connection)
+{
+    mConnection = connection;
+    foreach (QObject *obj, mEndpoints) {
+        JsonEndpoint *endpoint = qobject_cast<JsonEndpoint *> (obj);
+        if (endpoint)
+            endpoint->setConnection(connection);
+    }
+}
+
+
+void EndpointContainer::sendMessage(const QString & endpointDestination, JsonEndpoint *sender)
+{
+    QJsonObject msg;
+    msg.insert("text", QLatin1String("Standard text"));
+    msg.insert("number", mCounter++);
+    msg.insert("int", 100);
+    msg.insert("float", 100.0);
+    msg.insert("true", true);
+    msg.insert("false", false);
+    msg.insert("array", QJsonArray::fromStringList(QStringList() << "one" << "two" << "three"));
+    QJsonObject obj;
+    obj.insert("item1", QLatin1String("This is item 1"));
+    obj.insert("item2", QLatin1String("This is item 2"));
+    msg.insert("object", obj);
+
+    if (!endpointDestination.isEmpty()) {
+        msg.insert("endpoint", endpointDestination);
+    }
+
+    (sender != 0 ? sender : mStream)->send(msg);
+}
+
+void EndpointContainer::sendMessage(const QStringList &endpointDestinationList,
+                                    int messagesPerEndpoint, bool grouped)
+{
+    QJsonObject msg;
+    msg.insert("command", QLatin1String("flurry"));
+    msg.insert("count", messagesPerEndpoint);
+    msg.insert("endpoints", QJsonArray::fromStringList(endpointDestinationList));
+    msg.insert("grouped", grouped);
+
+    mStream->send(msg);
+}
+
+void EndpointContainer::processMessages() {
+    QVERIFY(QThread::currentThread() == thread());
+//    qDebug() << "XXX EndpointContainer::processMessages: " << sender();
+    QObject *source(sender());
+    QVERIFY(source && mEndpoints.contains(source));
+    if (mEndpoints.contains(source)) {
+        JsonEndpoint *endpoint = qobject_cast< JsonEndpoint *>(source);
+//        qDebug() << "    XXX: came from: " << endpoint->name();
+        while (endpoint->messageAvailable()) {
+            QJsonObject obj = endpoint->readMessage();
+//            qDebug() << "    XXX: going to: " << obj.value("endpoint").toString();
+            if (!obj.isEmpty())
+                emit messageReceived(obj, source);
+        }
+    }
+}
+
+/****************************/
+class ConnectionContainer : public EndpointContainer
+{
+    Q_OBJECT
+
+public:
+    ConnectionContainer(const QString & socketName, bool bSeparateThread = false);
+
+    void doConnect();
+    void closeConnection() {
+        JsonConnection *oldConnection = connection();
+        setConnection(0);
+        delete oldConnection;
+    }
+
+signals:
+    void disconnected();
+
+private:
+    QString mSocketName;
+};
+
+ConnectionContainer::ConnectionContainer(const QString & socketName, bool bSeparateThread/*= false*/)
+    : EndpointContainer(), mSocketName(socketName)
+{
+    qDebug() << "Creating new json client at " << socketName;
+    JsonConnection *connection = new JsonConnection();
+    connection->setUseSeparateThreadForProcessing(bSeparateThread);
+    connection->setFormat(FormatUTF8);
+
+    connect(connection, SIGNAL(disconnected()), SIGNAL(disconnected()));
+
+    addEndpoint(connection->defaultEndpoint());
+    setConnection(connection);
+}
+
+void ConnectionContainer::doConnect()
+{
+    if (!connection()->connectLocal(mSocketName)) {
+        qWarning() << "Unable to connect to" << mSocketName;
+        exit(2);
+    }
+}
+
+/****************************/
+
+class tst_JsonConnection : public QObject
+{
+    Q_OBJECT
+
+private slots:
+    void initTestCase();
+
+    void connectionTest();
+    void connectionSameThreadTest();
+    void declarativeTest();
+    void declarativeDefaultTest();
+    void multipleEndpointsTest();
+    void multipleThreadTest();
+    void autoreconnectTest();
+private:
+    void registerQmlTypes();
+
+    QQmlEngine engine;
+};
+
+void tst_JsonConnection::initTestCase()
+{
+    qRegisterMetaType<QJsonObject>();
+    registerQmlTypes();
+}
+
+
+void tst_JsonConnection::connectionTest()
+{
+    QString socketname = "/tmp/tst_socket";
+
+    Child child("testClient/testClient",
+                QStringList() << "-socket" << socketname);
+
+    QSignalSpy spy0(&child, SIGNAL(serverReady()));
+    waitForSpy(spy0, 1);
+
+    ConnectionContainer c(socketname,true);
+
+    JsonEndpoint *endpoint = c.addEndpoint("test");
+    QVERIFY(endpoint->name() == "test");
+
+    QVERIFY(c.connection()->state() == JsonConnection::Unconnected);
+    c.doConnect();
+    QVERIFY(c.connection()->state() == JsonConnection::Connected);
+    c.sendMessage(endpoint->name());
+
+    QSignalSpy spy(&c, SIGNAL(messageReceived(QJsonObject,QObject *)));
+    waitForSpy(spy, 1);
+
+    QJsonObject msg = qvariant_cast<QJsonObject>(spy.last().at(0));
+    qDebug() << msg;
+
+    QVERIFY(msg.value("text").isString() && msg.value("text").toString() == QLatin1String("Standard text"));
+    QVERIFY(msg.value("int").isDouble() && msg.value("int").toDouble() == 100);
+    QVERIFY(msg.value("float").isDouble() && msg.value("float").toDouble() == 100.0);
+    QVERIFY(msg.value("true").isBool() && msg.value("true").toBool() == true);
+    QVERIFY(msg.value("false").isBool() && msg.value("false").toBool() == false);
+    QVERIFY(msg.value("array").isArray());
+    QJsonArray array = msg.value("array").toArray();
+    QVERIFY(array.size() == 3);
+    QVERIFY(array.at(0).toString() == "one");
+    QVERIFY(array.at(1).toString() == "two");
+    QVERIFY(array.at(2).toString() == "three");
+    QVERIFY(msg.value("object").isObject());
+    QJsonObject object = msg.value("object").toObject();
+    QVERIFY(object.value("item1").toString() == "This is item 1");
+    QVERIFY(object.value("item2").toString() == "This is item 2");
+
+    c.closeConnection();
+
+    child.waitForFinished();
+}
+
+void tst_JsonConnection::connectionSameThreadTest()
+{
+    QString socketname = "/tmp/tst_socket";
+
+    Child child("testClient/testClient",
+                QStringList() << "-socket" << socketname);
+
+    QSignalSpy spy0(&child, SIGNAL(serverReady()));
+    waitForSpy(spy0, 1);
+
+    ConnectionContainer c(socketname, false);
+
+    JsonEndpoint *endpoint = c.addEndpoint("test");
+    QVERIFY(endpoint->name() == "test");
+
+    c.doConnect();
+    c.sendMessage(endpoint->name());
+
+    QSignalSpy spy(&c, SIGNAL(messageReceived(QJsonObject,QObject *)));
+    waitForSpy(spy, 1);
+
+    QJsonObject msg = qvariant_cast<QJsonObject>(spy.last().at(0));
+    qDebug() << msg;
+
+    QVERIFY(msg.value("text").isString() && msg.value("text").toString() == QLatin1String("Standard text"));
+    QVERIFY(msg.value("int").isDouble() && msg.value("int").toDouble() == 100);
+    QVERIFY(msg.value("float").isDouble() && msg.value("float").toDouble() == 100.0);
+    QVERIFY(msg.value("true").isBool() && msg.value("true").toBool() == true);
+    QVERIFY(msg.value("false").isBool() && msg.value("false").toBool() == false);
+    QVERIFY(msg.value("array").isArray());
+    QJsonArray array = msg.value("array").toArray();
+    QVERIFY(array.size() == 3);
+    QVERIFY(array.at(0).toString() == "one");
+    QVERIFY(array.at(1).toString() == "two");
+    QVERIFY(array.at(2).toString() == "three");
+    QVERIFY(msg.value("object").isObject());
+    QJsonObject object = msg.value("object").toObject();
+    QVERIFY(object.value("item1").toString() == "This is item 1");
+    QVERIFY(object.value("item2").toString() == "This is item 2");
+
+    c.closeConnection();
+
+    child.waitForFinished();
+}
+
+class Watcher : public QObject
+{
+    Q_OBJECT
+public:
+    Watcher() {}
+
+signals:
+    void done();
+};
+
+void tst_JsonConnection::registerQmlTypes()
+{
+    qmlRegisterType<JsonConnection>("Qt.json.connection.test", 1,0, "JsonConnection");
+    qmlRegisterType<JsonEndpoint>("Qt.json.connection.test", 1,0, "JsonEndpoint");
+}
+
+static const char szData[] =
+        "import QtQuick 2.0 \n\
+         import Qt.json.connection.test 1.0 \n\
+            QtObject { \n\
+            id: root \n\
+            property var retmsg \n\
+            \
+            property variant prop; \n\
+            prop: JsonConnection { \n\
+                id: _connection \n\
+                localSocketName: \"/tmp/tst_socket\" \n\
+                endpointPropertyName: \"endpoint\" \n\
+            } \n\
+            \
+            property variant prope; \n\
+            prope: JsonEndpoint { \n\
+                 id: endpoint \n\
+                 name: \"endpoint\" \n\
+                 connection: _connection \n\
+                 onReadyReadMessage: { \n\
+                                retmsg = endpoint.readMessageMap(); \n\
+                                retmsg.extra = \"extra\"; \n\
+                                retmsg.int++; \n\
+                                watcher.done(); \n\
+                 } \n\
+            } \n\
+            \
+            Component.onCompleted: { \n\
+                _connection.connectLocal(); \n\
+                endpoint.send(msg); \n\
+            } \n\
+         }";
+
+
+void tst_JsonConnection::declarativeTest()
+{
+    QString socketname = "/tmp/tst_socket";
+
+    Child child("testClient/testClient",
+                QStringList() << "-socket" << socketname);
+
+    QSignalSpy spy0(&child, SIGNAL(serverReady()));
+    waitForSpy(spy0, 1);
+
+    Watcher watcher;
+    QSignalSpy spy(&watcher, SIGNAL(done()));
+    engine.rootContext()->setContextProperty("watcher", &watcher);
+
+    QJsonObject msg;
+    msg.insert("endpoint", QLatin1String("endpoint"));
+    msg.insert("text", QLatin1String("Standard text"));
+    msg.insert("number", 0);
+    msg.insert("int", 100);
+    msg.insert("float", 100.0);
+    msg.insert("true", true);
+    msg.insert("false", false);
+    msg.insert("array", QJsonArray::fromStringList(QStringList() << "one" << "two" << "three"));
+    QJsonObject obj;
+    obj.insert("item1", QLatin1String("This is item 1"));
+    obj.insert("item2", QLatin1String("This is item 2"));
+    msg.insert("object", obj);
+    engine.rootContext()->setContextProperty("msg", QVariant::fromValue(msg));
+
+    QJsonDocument document(msg);
+    QString str = document.toJson();
+    engine.rootContext()->setContextProperty("msgstr", str);
+
+    QQmlComponent component(&engine);
+    component.setData(szData, QUrl());
+
+
+    if (!component.isReady()) {
+        qDebug() << "QQmlComponent::setData error: " << component.errorString();
+    }
+    QVERIFY(component.isReady());
+
+    QObject *componentObject = component.create();
+    QVERIFY(componentObject != 0);
+
+    waitForSpy(spy, 1);
+
+    msg = QJsonObject::fromVariantMap(componentObject->property("retmsg").value<QVariantMap>());
+    QVERIFY(msg.value("extra").isString() && msg.value("extra").toString() == QLatin1String("extra"));
+    QVERIFY(msg.value("text").isString() && msg.value("text").toString() == QLatin1String("Standard text"));
+    QVERIFY(msg.value("int").isDouble() && msg.value("int").toDouble() == 101);
+    QVERIFY(msg.value("float").isDouble() && msg.value("float").toDouble() == 100.0);
+    QVERIFY(msg.value("true").isBool() && msg.value("true").toBool() == true);
+    QVERIFY(msg.value("false").isBool() && msg.value("false").toBool() == false);
+    QVERIFY(msg.value("array").isArray());
+    QJsonArray array = msg.value("array").toArray();
+    QVERIFY(array.size() == 3);
+    QVERIFY(array.at(0).toString() == "one");
+    QVERIFY(array.at(1).toString() == "two");
+    QVERIFY(array.at(2).toString() == "three");
+    QVERIFY(msg.value("object").isObject());
+    QJsonObject object = msg.value("object").toObject();
+    QVERIFY(object.value("item1").toString() == "This is item 1");
+    QVERIFY(object.value("item2").toString() == "This is item 2");
+
+    delete componentObject;
+
+    child.waitForFinished();
+
+}
+
+static const char szDefaultData[] =
+        "import QtQuick 2.0 \n\
+         import Qt.json.connection.test 1.0 \n\
+            QtObject { \n\
+            id: root \n\
+            property var retmsg \n\
+            \
+            property variant prop; \n\
+            prop: JsonConnection { \n\
+                id: _connection \n\
+                localSocketName: \"/tmp/tst_socket\" \n\
+                endpointPropertyName: \"endpoint\" \n\
+            } \n\
+            \
+            property variant prope; \n\
+            prope: JsonEndpoint { \n\
+                 id: endpoint \n\
+                 connection: _connection \n\
+                 onReadyReadMessage: { \n\
+                                retmsg = endpoint.readMessageMap(); \n\
+                                retmsg.extra = \"extra\"; \n\
+                                retmsg.int++; \n\
+                                watcher.done(); \n\
+                 } \n\
+            } \n\
+            \
+            Component.onCompleted: { \n\
+                _connection.connectLocal(); \n\
+                endpoint.send(msg); \n\
+            } \n\
+         }";
+
+
+void tst_JsonConnection::declarativeDefaultTest()
+{
+    QString socketname = "/tmp/tst_socket";
+
+    Child child("testClient/testClient",
+                QStringList() << "-socket" << socketname);
+
+    QSignalSpy spy0(&child, SIGNAL(serverReady()));
+    waitForSpy(spy0, 1);
+
+    Watcher watcher;
+    QSignalSpy spy(&watcher, SIGNAL(done()));
+    engine.rootContext()->setContextProperty("watcher", &watcher);
+
+    QJsonObject msg;
+    msg.insert("endpoint", QLatin1String("endpoint"));
+    msg.insert("text", QLatin1String("Standard text"));
+    msg.insert("number", 0);
+    msg.insert("int", 100);
+    msg.insert("float", 100.0);
+    msg.insert("true", true);
+    msg.insert("false", false);
+    msg.insert("array", QJsonArray::fromStringList(QStringList() << "one" << "two" << "three"));
+    QJsonObject obj;
+    obj.insert("item1", QLatin1String("This is item 1"));
+    obj.insert("item2", QLatin1String("This is item 2"));
+    msg.insert("object", obj);
+    engine.rootContext()->setContextProperty("msg", QVariant::fromValue(msg));
+
+    QJsonDocument document(msg);
+    QString str = document.toJson();
+    engine.rootContext()->setContextProperty("msgstr", str);
+
+    QQmlComponent component(&engine);
+    component.setData(szDefaultData, QUrl());
+
+
+    if (!component.isReady()) {
+        qDebug() << "QQmlComponent::setData error: " << component.errorString();
+    }
+    QVERIFY(component.isReady());
+
+    QObject *componentObject = component.create();
+    QVERIFY(componentObject != 0);
+
+    waitForSpy(spy, 1);
+
+    msg = QJsonObject::fromVariantMap(componentObject->property("retmsg").value<QVariantMap>());
+    QVERIFY(msg.value("extra").isString() && msg.value("extra").toString() == QLatin1String("extra"));
+    QVERIFY(msg.value("text").isString() && msg.value("text").toString() == QLatin1String("Standard text"));
+    QVERIFY(msg.value("int").isDouble() && msg.value("int").toDouble() == 101);
+    QVERIFY(msg.value("float").isDouble() && msg.value("float").toDouble() == 100.0);
+    QVERIFY(msg.value("true").isBool() && msg.value("true").toBool() == true);
+    QVERIFY(msg.value("false").isBool() && msg.value("false").toBool() == false);
+    QVERIFY(msg.value("array").isArray());
+    QJsonArray array = msg.value("array").toArray();
+    QVERIFY(array.size() == 3);
+    QVERIFY(array.at(0).toString() == "one");
+    QVERIFY(array.at(1).toString() == "two");
+    QVERIFY(array.at(2).toString() == "three");
+    QVERIFY(msg.value("object").isObject());
+    QJsonObject object = msg.value("object").toObject();
+    QVERIFY(object.value("item1").toString() == "This is item 1");
+    QVERIFY(object.value("item2").toString() == "This is item 2");
+
+    delete componentObject;
+
+    child.waitForFinished();
+
+}
+
+void tst_JsonConnection::multipleEndpointsTest()
+{
+    const int knEndpointsNumber = 5;
+    QString socketname = "/tmp/tst_socket";
+
+    Child child("testClient/testClient",
+                QStringList() << "-socket" << socketname);
+
+    QSignalSpy spy0(&child, SIGNAL(serverReady()));
+    waitForSpy(spy0, 1);
+
+    ConnectionContainer c(socketname,true);
+
+    for (int i = 0; i < knEndpointsNumber; i++) {
+        QString endpointName("test");
+        endpointName += QString::number(i);
+
+        JsonEndpoint *endpoint = c.addEndpoint(endpointName);
+        QVERIFY(endpoint->name() == endpointName);
+    }
+
+    QVERIFY(c.connection()->state() == JsonConnection::Unconnected);
+    c.doConnect();
+    QVERIFY(c.connection()->state() == JsonConnection::Connected);
+
+    for (int i = 0; i < knEndpointsNumber; i++) {
+        QString endpointName("test");
+        endpointName += QString::number(i);
+        c.sendMessage(endpointName);
+    }
+
+    QSignalSpy spy(&c, SIGNAL(messageReceived(QJsonObject,QObject *)));
+    waitForSpy(spy, knEndpointsNumber);
+
+    QList<QObject *> endpoints(c.endpoints());
+
+    QSignalSpy::const_iterator it;
+    for (it = spy.constBegin(); it != spy.constEnd(); it++) {
+        QJsonObject msg = qvariant_cast<QJsonObject>(it->at(0));
+
+        JsonEndpoint *endpoint = qobject_cast<JsonEndpoint *>(qvariant_cast<QObject *>(it->at(1)));
+        QVERIFY(endpoints.removeOne(endpoint));
+
+        QVERIFY(msg.value("endpoint").isString() && msg.value("endpoint").toString() == endpoint->name());
+        QVERIFY(msg.value("text").isString() && msg.value("text").toString() == QLatin1String("Standard text"));
+        QVERIFY(msg.value("int").isDouble() && msg.value("int").toDouble() == 100);
+        QVERIFY(msg.value("float").isDouble() && msg.value("float").toDouble() == 100.0);
+        QVERIFY(msg.value("true").isBool() && msg.value("true").toBool() == true);
+        QVERIFY(msg.value("false").isBool() && msg.value("false").toBool() == false);
+        QVERIFY(msg.value("array").isArray());
+        QJsonArray array = msg.value("array").toArray();
+        QVERIFY(array.size() == 3);
+        QVERIFY(array.at(0).toString() == "one");
+        QVERIFY(array.at(1).toString() == "two");
+        QVERIFY(array.at(2).toString() == "three");
+        QVERIFY(msg.value("object").isObject());
+        QJsonObject object = msg.value("object").toObject();
+        QVERIFY(object.value("item1").toString() == "This is item 1");
+        QVERIFY(object.value("item2").toString() == "This is item 2");
+    }
+
+    c.closeConnection();
+
+    child.waitForFinished();
+}
+
+void tst_JsonConnection::multipleThreadTest()
+{
+    const int knThreadCount = 5;
+    const int knMessagesPerThread = 10;
+    bool differentThreads = true;  // set to false to see if errors are due to threading
+
+    QString socketname = "/tmp/tst_socket";
+
+    Child child("testClient/testClient",
+                QStringList() << "-socket" << socketname);
+
+    QSignalSpy spy0(&child, SIGNAL(serverReady()));
+    waitForSpy(spy0, 1);
+
+    ConnectionContainer c(socketname,true);
+
+    QList<EndpointContainer *> containers;
+    QList<QSignalSpy *> spies;
+    QStringList endpointNames;
+    for (int i = 0; i < knThreadCount; i++) {
+        QThread *newThread = 0;
+        if (differentThreads)
+            newThread = new QThread(&c);
+        EndpointContainer *ec = new EndpointContainer();
+        ec->setConnection(c.connection());
+        if (differentThreads)
+            ec->moveToThread(newThread);
+        containers.append(ec);
+
+        QString endpointName("test");
+        endpointName += QString::number(i);
+        endpointNames.append(endpointName);
+
+        JsonEndpoint *endpoint = ec->addEndpoint(endpointName);
+        QVERIFY(endpoint->name() == endpointName);
+
+        QSignalSpy *spy = new QSignalSpy(ec, SIGNAL(messageReceived(QJsonObject,QObject*)));
+        spies.append(spy);
+
+        if (differentThreads)
+            newThread->start();
+    }
+
+    QVERIFY(c.connection()->state() == JsonConnection::Unconnected);
+    c.doConnect();
+    QVERIFY(c.connection()->state() == JsonConnection::Connected);
+
+    // grouped
+    c.sendMessage(endpointNames, knMessagesPerThread, true);
+
+    for (int spyCount = 0; spyCount < spies.count(); ++spyCount) {
+        QSignalSpy *spy = spies.at(spyCount);
+        waitForSpy(*spy, knMessagesPerThread);
+        for (int i = 0; i < spy->count(); ++i) {
+            QJsonObject msg = qvariant_cast<QJsonObject>(spy->at(i).at(0));
+//            qDebug() << "YYY: msg endpoint = " << msg.value("endpoint").toString() << "  should be: " << endpointNames.at(spyCount);
+            QVERIFY(msg.value("endpoint").toString() == endpointNames.at(spyCount));
+        }
+//        qDebug() << "YYY    done spy -- count = " << spy->count();
+    }
+
+    // staggered
+
+    foreach (QSignalSpy *spy, spies)
+        spy->clear();
+
+    c.sendMessage(endpointNames, knMessagesPerThread, false);
+
+    for (int spyCount = 0; spyCount < spies.count(); ++spyCount) {
+        QSignalSpy *spy = spies.at(spyCount);
+        waitForSpy(*spy, knMessagesPerThread);
+        for (int i = 0; i < spy->count(); ++i) {
+            QJsonObject msg = qvariant_cast<QJsonObject>(spy->at(i).at(0));
+//            qDebug() << "YYY: msg endpoint = " << msg.value("endpoint").toString() << "  should be: " << endpointNames.at(spyCount);
+            QVERIFY(msg.value("endpoint").toString() == endpointNames.at(spyCount));
+        }
+//        qDebug() << "YYY    done spy -- count = " << spy->count();
+    }
+
+    c.closeConnection();
+
+    child.waitForFinished();
+
+    qDeleteAll(containers);
+    qDeleteAll(spies);
+}
+
+void tst_JsonConnection::autoreconnectTest()
+{
+    QString socketname = "/tmp/tst_socket";
+
+    Child child("testClient/testClient",
+                QStringList() << "-socket" << socketname);
+
+    QSignalSpy spy0(&child, SIGNAL(serverReady()));
+    waitForSpy(spy0, 1);
+
+    ConnectionContainer c(socketname,true);
+    c.connection()->setAutoReconnectEnabled(true);
+
+    JsonEndpoint *endpoint = c.addEndpoint("test");
+    QVERIFY(endpoint->name() == "test");
+
+    QVERIFY(c.connection()->state() == JsonConnection::Unconnected);
+    c.doConnect();
+    QVERIFY(c.connection()->state() == JsonConnection::Connected);
+
+    QJsonObject msg;
+    msg.insert("endpoint", QLatin1String("test"));
+    msg.insert("text", QLatin1String("Disconnect message"));
+    msg.insert("command", QLatin1String("disconnect"));
+    msg.insert("timeout", 2000); // reconnect after 2 secs
+    endpoint->send(msg);
+
+    QSignalSpy spy(&c, SIGNAL(messageReceived(QJsonObject,QObject *)));
+    waitForSpy(spy, 1);
+
+    msg = qvariant_cast<QJsonObject>(spy.last().at(0));
+
+    QVERIFY(msg.value("endpoint").isString() && msg.value("endpoint").toString() == endpoint->name());
+    QVERIFY(msg.value("text").isString() && msg.value("text").toString() == QLatin1String("Disconnect message"));
+    QVERIFY(msg.value("timeout").isDouble() && msg.value("timeout").toDouble() == 2000);
+
+    // wait for disconnect -> connencting state
+    QSignalSpy spy1(c.connection(), SIGNAL(stateChanged(JsonConnection::State)));
+    waitForSpy(spy1, 1);
+    JsonConnection::State state = qvariant_cast<JsonConnection::State>(spy1.last().at(0));
+    QVERIFY(state == JsonConnection::Connecting);
+    QVERIFY(c.connection()->state() == JsonConnection::Connecting);
+
+    // wait for reconnection
+    waitForSpy(spy1, 2, 10000);
+    state = qvariant_cast<JsonConnection::State>(spy1.last().at(0));
+    QVERIFY(state == JsonConnection::Connected);
+    QVERIFY(c.connection()->state() == JsonConnection::Connected);
+
+    // send a new message after reconnection and wait for a reply
+    msg = QJsonObject();
+    msg.insert("endpoint", QLatin1String("test"));
+    msg.insert("text", QLatin1String("New message"));
+    endpoint->send(msg);
+
+    QSignalSpy spy2(&c, SIGNAL(messageReceived(QJsonObject,QObject *)));
+    waitForSpy(spy2, 1);
+
+    msg = qvariant_cast<QJsonObject>(spy2.last().at(0));
+
+    QVERIFY(msg.value("endpoint").isString() && msg.value("endpoint").toString() == endpoint->name());
+    QVERIFY(msg.value("text").isString() && msg.value("text").toString() == QLatin1String("New message"));
+
+    c.closeConnection();
+
+    child.waitForFinished();
+}
+
+QTEST_MAIN
+
+(tst_JsonConnection)
+
+#include "tst_jsonconnection.moc"