Remove the linking between QGrpcAsyncOperationBase and QAbstractGrpcChannel

- Remove redundant logic in QAbstractGrpcChannel related to Grpc
  call/stream handling. This allows to remove redundand linking
  between QGrpcAsyncOperationBase and QAbstractGrpcChannel and
  simplify the implementation.
- Unify the method that cancels call/stream, the name is abort and
  QGrpcAsyncOperationBase specifies it as pure virtual.
- Adjust tests.

examples/simplechat/simplechatengine.cpp

@@ -76,7 +76,7 @@ void SimpleChatEngine::login(const QString &name, const QString &password)
     QtProtobuf::QGrpcStreamShared stream = m_client->streamMessageList(None());
     QObject::connect(stream.get(), &QtProtobuf::QGrpcStream::error, this, [stream] {
         qCritical() << "Stream error, cancel";
-        stream->cancel();
+        stream->abort();
     });
     QObject::connect(stream.get(), &QtProtobuf::QGrpcStream::messageReceived, this, [this, name, stream]() {
         if (m_userName != name) {

src/grpc/qabstractgrpcchannel.cpp

@@ -38,26 +38,9 @@ struct QAbstractGrpcChannelPrivate {
     const QThread *thread;
 };
 
-QAbstractGrpcChannel::QAbstractGrpcChannel() : dPtr(new QAbstractGrpcChannelPrivate)
-{
-
-}
-
+QAbstractGrpcChannel::QAbstractGrpcChannel() : dPtr(new QAbstractGrpcChannelPrivate) {}
 QAbstractGrpcChannel::~QAbstractGrpcChannel() = default;
 
-void QAbstractGrpcChannel::abort(QGrpcCallReply *reply)
-{
-    assert(reply != nullptr);
-    reply->setData({});
-    reply->error({QGrpcStatus::StatusCode::Aborted, QLatin1String("Call aborted by user or timeout")});
-}
-
-void QAbstractGrpcChannel::cancel(QGrpcStream *stream)
-{
-    assert(stream != nullptr);
-    stream->finished();
-}
-
 const QThread *QAbstractGrpcChannel::thread() const
 {
     return dPtr->thread;

src/grpc/qabstractgrpcchannel.h

@@ -38,8 +38,6 @@ class QThread;
 
 namespace QtProtobuf {
 
-class QGrpcCallReply;
-class QGrpcStream;
 class QAbstractGrpcClient;
 class QAbstractProtobufSerializer;
 struct QAbstractGrpcChannelPrivate;
@@ -96,23 +94,6 @@ protected:
     QAbstractGrpcChannel();
     //! \private
     virtual ~QAbstractGrpcChannel();
-
-    /*!
-     * \private
-     * \brief Aborts async call for given \p reply
-     * \param[in] reply returned by asynchronous QAbstractGrpcChannel::call() method
-     */
-    virtual void abort(QGrpcCallReply *reply);
-
-    /*!
-     * \private
-     * \brief Cancels \p stream
-     * \param[in] stream returned by QAbstractGrpcChannel::stream() method
-     */
-    virtual void cancel(QGrpcStream *stream);
-
-    friend class QGrpcCallReply;
-    friend class QGrpcStream;
 private:
     Q_DISABLE_COPY(QAbstractGrpcChannel)
     std::unique_ptr<QAbstractGrpcChannelPrivate> dPtr;

src/grpc/qabstractgrpcclient.cpp

@@ -64,11 +64,11 @@ void QAbstractGrpcClient::attachChannel(const std::shared_ptr<QAbstractGrpcChann
         throw std::runtime_error("Call from another thread");
     }
     for (auto stream : dPtr->activeStreams) {
-        stream->cancel();
+        stream->abort();
     }
     dPtr->channel = channel;
     for (auto stream : dPtr->activeStreams) {
-        stream->cancel();
+        stream->abort();
     }
 }
 
@@ -105,7 +105,7 @@ QGrpcCallReplyShared QAbstractGrpcClient::call(const QString &method, const QByt
                                       return call(method, arg);
                                   }, Qt::BlockingQueuedConnection, &reply);
     } else if (dPtr->channel) {
-        reply.reset(new QGrpcCallReply(dPtr->channel, this), [](QGrpcCallReply *reply) { reply->deleteLater(); });
+        reply.reset(new QGrpcCallReply(this), [](QGrpcCallReply *reply) { reply->deleteLater(); });
 
         auto errorConnection = std::make_shared<QMetaObject::Connection>();
         auto finishedConnection = std::make_shared<QMetaObject::Connection>();
@@ -140,7 +140,7 @@ QGrpcStreamShared QAbstractGrpcClient::stream(const QString &method, const QByte
                                       return stream(method, arg, handler);
                                   }, Qt::BlockingQueuedConnection, &grpcStream);
     } else if (dPtr->channel) {
-        grpcStream.reset(new QGrpcStream(dPtr->channel, method, arg, handler, this), [](QGrpcStream *stream) { stream->deleteLater(); });
+        grpcStream.reset(new QGrpcStream(method, arg, handler, this), [](QGrpcStream *stream) { stream->deleteLater(); });
 
         auto it = std::find_if(std::begin(dPtr->activeStreams), std::end(dPtr->activeStreams), [grpcStream](const QGrpcStreamShared &activeStream) {
            return *activeStream == *grpcStream;

src/grpc/qabstractgrpcclient.h

@@ -46,17 +46,10 @@
  */
 namespace QtProtobuf {
 
-class QGrpcCallReply;
-class QGrpcStream;
 class QGrpcAsyncOperationBase;
 class QAbstractGrpcChannel;
 class QAbstractGrpcClientPrivate;
 
-/*!
- * \private
- */
-using StreamHandler = std::function<void(const QByteArray&)>;
-
 /*!
  * \ingroup QtGrpc
  * \brief The QAbstractGrpcClient class is bridge between gRPC clients and channels. QAbstractGrpcClient provides set of

src/grpc/qgrpcasyncoperationbase_p.h

@@ -31,7 +31,6 @@
 #include <functional>
 #include <memory>
 
-#include "qabstractgrpcchannel.h"
 #include "qabstractgrpcclient.h"
 
 #include "qtgrpcglobal.h"
@@ -74,10 +73,12 @@ public:
         m_data = data;
     }
 
+    virtual void abort() = 0;
+
 signals:
     /*!
-     * \brief The signal is emitted when reply is ready for read. Usualy called by channel when all chunks of data
-     *        recevied
+     * \brief The signal indicates the end of communication for this call. If signal emitted by
+     *        stream this means that stream is succesfully closed either by client or server.
      */
     void finished();
 
@@ -89,12 +90,10 @@ signals:
 
 protected:
     //! \private
-    QGrpcAsyncOperationBase(const std::shared_ptr<QAbstractGrpcChannel> &channel, QAbstractGrpcClient *parent) : QObject(parent)
-      , m_channel(channel) {}
+    QGrpcAsyncOperationBase(QAbstractGrpcClient *parent) : QObject(parent) {}
     //! \private
     virtual ~QGrpcAsyncOperationBase();
 
-    std::shared_ptr<QAbstractGrpcChannel> m_channel;
 private:
     QGrpcAsyncOperationBase();
     Q_DISABLE_COPY_MOVE(QGrpcAsyncOperationBase)

src/grpc/qgrpccallreply.cpp

@@ -31,9 +31,13 @@ using namespace QtProtobuf;
 
 void QGrpcCallReply::abort()
 {
+    auto abortFunc = [this]() {
+        this->setData({});
+        this->error({QGrpcStatus::StatusCode::Aborted, QLatin1String("Call aborted by user or timeout")});
+    };
     if (thread() != QThread::currentThread()) {
-        QMetaObject::invokeMethod(this, [this](){m_channel->abort(this);}, Qt::BlockingQueuedConnection);
+        QMetaObject::invokeMethod(this, abortFunc, Qt::BlockingQueuedConnection);
     } else {
-        m_channel->abort(this);
+        abortFunc();
     }
 }

src/grpc/qgrpccallreply.h

@@ -29,7 +29,6 @@
 #include <QMutex>
 #include <memory>
 
-#include "qabstractgrpcchannel.h"
 #include "qabstractgrpcclient.h"
 #include "qgrpcasyncoperationbase_p.h"
 
@@ -50,7 +49,7 @@ public:
     /*!
      * \brief Aborts this reply and try to abort call in channel
      */
-    void abort();
+    void abort() override;
 
     /*!
      * \brief Subscribe to QGrpcCallReply signals
@@ -76,7 +75,7 @@ public:
 
 protected:
     //! \private
-    QGrpcCallReply(const std::shared_ptr<QAbstractGrpcChannel> &channel, QAbstractGrpcClient *parent) : QGrpcAsyncOperationBase(channel, parent)
+    QGrpcCallReply(QAbstractGrpcClient *parent) : QGrpcAsyncOperationBase(parent)
     {}
     //! \private
     ~QGrpcCallReply() = default;

src/grpc/qgrpcstream.cpp

@@ -30,8 +30,7 @@
 
 using namespace QtProtobuf;
 
-QGrpcStream::QGrpcStream(const std::shared_ptr<QAbstractGrpcChannel> &channel, const QString &method,
-                                     const QByteArray &arg, const StreamHandler &handler, QAbstractGrpcClient *parent) : QGrpcAsyncOperationBase(channel, parent)
+QGrpcStream::QGrpcStream(const QString &method, const QByteArray &arg, const StreamHandler &handler, QAbstractGrpcClient *parent) : QGrpcAsyncOperationBase(parent)
   , m_method(method)
   , m_arg(arg)
 {
@@ -47,11 +46,11 @@ void QGrpcStream::addHandler(const StreamHandler &handler)
     }
 }
 
-void QGrpcStream::cancel()
+void QGrpcStream::abort()
 {
     if (thread() != QThread::currentThread()) {
-        QMetaObject::invokeMethod(this, [this](){m_channel->cancel(this);}, Qt::BlockingQueuedConnection);
+        QMetaObject::invokeMethod(this, &QGrpcStream::finished, Qt::BlockingQueuedConnection);
     } else {
-        m_channel->cancel(this);
+        finished();
     }
 }

src/grpc/qgrpcstream.h

@@ -29,7 +29,6 @@
 #include <QMutex>
 #include <memory>
 
-#include "qabstractgrpcchannel.h"
 #include "qabstractgrpcclient.h"
 #include "qgrpcasyncoperationbase_p.h"
 
@@ -50,7 +49,7 @@ public:
     /*!
      * \brief Cancels this stream and try to abort call in channel
      */
-    void cancel();
+    void abort() override;
 
     /*!
      * \brief Returns method for this stream
@@ -88,8 +87,8 @@ signals:
 
 protected:
     //! \private
-    QGrpcStream(const std::shared_ptr<QAbstractGrpcChannel> &channel, const QString &method,
-                      const QByteArray &arg, const StreamHandler &handler, QAbstractGrpcClient *parent);
+    QGrpcStream(const QString &method, const QByteArray &arg,
+                const StreamHandler &handler, QAbstractGrpcClient *parent);
     //! \private
     virtual ~QGrpcStream() = default;
 

src/grpc/qtgrpcglobal.h

@@ -64,10 +64,12 @@
 #endif //Q_PROTOBUF_IMPORT_QUICK_PLUGIN
 
 #include <memory>
+#include <functional>
 
 namespace QtProtobuf {
 class QGrpcCallReply;
 class QGrpcStream;
 using QGrpcCallReplyShared = std::shared_ptr<QGrpcCallReply>;
 using QGrpcStreamShared = std::shared_ptr<QGrpcStream>;
+using StreamHandler = std::function<void(const QByteArray&)>;
 }

src/grpc/quick/qquickgrpcstream.cpp

@@ -41,7 +41,7 @@ QQuickGrpcStream::QQuickGrpcStream(QObject *parent) : QObject(parent)
 QQuickGrpcStream::~QQuickGrpcStream()
 {
     if (m_stream) {
-        m_stream->cancel();
+        m_stream->abort();
     }
     delete m_returnValue;
 }
@@ -50,7 +50,7 @@ QQuickGrpcStream::~QQuickGrpcStream()
 void QQuickGrpcStream::updateStream()
 {
     if (m_stream) {
-        m_stream->cancel();
+        m_stream->abort();
         m_stream.reset();
     }
 

tests/test_grpc/clienttest.cpp

@@ -277,7 +277,7 @@ TEST_P(ClientTest, StringEchoStreamAbortTest)
         result.setTestFieldString(result.testFieldString() + ret.testFieldString());
 
         if (i == 3) {
-            stream->cancel();
+            stream->abort();
             QTimer::singleShot(4000, &waiter, &QEventLoop::quit);
         }
     });
@@ -303,7 +303,7 @@ TEST_P(ClientTest, StringEchoStreamAbortByTimerTest)
     int i = 0;
     auto stream = testClient->streamTestMethodServerStream(request);
     QTimer::singleShot(3500, stream.get(), [stream]() {
-        stream->cancel();
+        stream->abort();
     });
 
     bool isFinished = false;
@@ -621,7 +621,7 @@ TEST_P(ClientTest, MultipleStreamsCancelTest)
         isFinishedNext = true;
     });
 
-    streamNext->cancel();
+    streamNext->abort();
 
     ASSERT_TRUE(isFinished);
     ASSERT_TRUE(isFinishedNext);
@@ -643,7 +643,7 @@ TEST_P(ClientTest, MultipleStreamsCancelTest)
         isFinishedNext = true;
     });
 
-    stream->cancel();
+    stream->abort();
 
     ASSERT_TRUE(isFinished);
     ASSERT_TRUE(isFinishedNext);
@@ -801,7 +801,7 @@ TEST_P(ClientTest, StreamCancelWhileErrorTimeoutTest)
         ok = true;
         waiter.quit();
     });
-    stream->cancel();
+    stream->abort();
     stream.reset();
 
     QTimer::singleShot(5000, &waiter, &QEventLoop::quit);