Merge lp:~michihenning/unity-scopes-api/finished-status into lp:unity-scopes-api

[Michi Henning]

Added reason for why a call to finished() is made (finished, cancelled, or error).
Fixed incorrect return value from push() after cancellation.
Lots of doc fixes and other minor tidy-up.

Added reason for why a call to finished() is made (finished, cancelled, or error).
Fixed incorrect return value from push() after cancellation.
Lots of doc fixes and other minor tidy-up.

[PS Jenkins bot (ps-jenkins)]

FAILED: Continuous integration, rev:65
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-ci/20/
Executed test runs:
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-trusty-amd64-ci/20/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-trusty-armhf-ci/20/console
    FAILURE: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-trusty-i386-ci/20/console

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-ci/20/rebuild

review: Needs Fixing (continuous-integration)

[lp:~michihenning/unity-scopes-api/finished-status]

66. By Michi Henning on 2013-11-22

Fixed incomplete comment with trailing whitespace.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:66
http://jenkins.qa.ubuntu.com/job/unity-scopes-api-ci/21/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-trusty-amd64-ci/21
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-trusty-armhf-ci/21
        deb: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-trusty-armhf-ci/21/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/unity-scopes-api-trusty-i386-ci/21

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/unity-scopes-api-ci/21/rebuild

review: Approve (continuous-integration)

[Michal Hruby (mhr3)]

LGTM, though I'd like to see tests for the entire scope search use case. But that needs more infrastructure in the tests, so I'm hoping that is going to be done in a later branch.

review: Approve

[Michi Henning (michihenning)]

I'm hoping for that too, yes :-)

But, seriously, yes. At the moment, the gtest unit tests are fine, but not useful when it comes to testing end-to-end behaviour between client and server. We need something that can fire up a server, a client, run some interactions, and report back.

In part, I've been testing things to a certain extent by hand cobbling messages together in the unit tests via capnproto, but this isn't a viable strategy for the longer term.

I need a day with 48 hours :-(

=== modified file 'demo/client.cpp'
--- demo/client.cpp	2013-11-21 10:32:19 +0000
+++ demo/client.cpp	2013-11-22 04:36:14 +0000
@@ -46,9 +46,9 @@
         cout << "received result: uri=" << result.uri() << " title=" << result.title() << " category id: " << result.category()->id() << endl;
     }
 
-    virtual void finished() override
+    virtual void finished(ReceiverBase::Reason reason) override
     {
-        cout << "query complete" << endl;
+        cout << "query complete, status: " << to_string(reason) << endl;
         {
             unique_lock<decltype(mutex_)> lock(mutex_);
             query_complete_ = true;

=== modified file 'demo/scopes/scope-B/scope-B.cpp'
--- demo/scopes/scope-B/scope-B.cpp	2013-11-21 10:34:49 +0000
+++ demo/scopes/scope-B/scope-B.cpp	2013-11-22 04:36:14 +0000
@@ -56,9 +56,9 @@
         }
     }
 
-    virtual void finished() override
+    virtual void finished(Reason reason) override
     {
-        cout << "query to " << scope_name_ << " complete" << endl;
+        cout << "query to " << scope_name_ << " complete, status: " << to_string(reason) << endl;
     }
 
     Receiver(string const& scope_name, ReplyProxy const& upstream) :

=== modified file 'demo/scopes/scope-C/scope-C.cpp'
--- demo/scopes/scope-C/scope-C.cpp	2013-11-20 23:58:07 +0000
+++ demo/scopes/scope-C/scope-C.cpp	2013-11-22 04:36:14 +0000
@@ -191,7 +191,6 @@
             }
             for (int i = 1; i < 4; ++i)
             {
-                cerr << "worker thread: pushing" << endl;
                 ResultItem result(cat);
                 result.set_uri("uri");
                 result.set_title("scope-C: result " + to_string(i) + " for query \"" + query + "\"");
@@ -199,7 +198,6 @@
                 result.set_dnd_uri("dnd_uri");
                 if (!reply->push(result))
                 {
-                    cerr << "worker thread: push returned false" << endl;
                     break; // Query was cancelled
                 }
                 sleep(1);

=== modified file 'demo/scopes/scope-D/scope-D.cpp'
--- demo/scopes/scope-D/scope-D.cpp	2013-11-20 23:58:07 +0000
+++ demo/scopes/scope-D/scope-D.cpp	2013-11-22 04:36:14 +0000
@@ -143,7 +143,7 @@
         // Informational callback to let a query know when it was cancelled. The query should
         // clean up any resources it has allocated, stop pushing results, and arrange for
         // run() to return (if still active).
-        // Ignoring query cancelltion does not do any direct harm, but wastes CPU cycles: any
+        // Ignoring query cancellation does not do any direct harm, but wastes CPU cycles: any
         // results that are pushed once a query is cancelled are ignored anyway.
         // The main purpose of this callback to give queries a chance to stop doing whatever
         // work may still be in progress on a query. Note that cancellations are frequent;

=== modified file 'include/scopes/QueryBase.h'
--- include/scopes/QueryBase.h	2013-11-21 10:12:45 +0000
+++ include/scopes/QueryBase.h	2013-11-22 04:36:14 +0000
@@ -49,6 +49,8 @@
 
 // Abstract server-side base interface for a query that is executed inside a scope.
 
+// TODO: documentation
+
 class UNITY_API QueryBase : private util::NonCopyable
 {
 public:

=== modified file 'include/scopes/ReceiverBase.h'
--- include/scopes/ReceiverBase.h	2013-11-21 10:18:43 +0000
+++ include/scopes/ReceiverBase.h	2013-11-22 04:36:14 +0000
@@ -22,6 +22,7 @@
 #include <unity/util/DefinesPtrs.h>
 #include <unity/util/NonCopyable.h>
 #include <scopes/Category.h>
+#include <scopes/ReceiverBase.h>
 
 #include <string>
 
@@ -40,9 +41,13 @@
 \brief Abstract base class to receive the results of a query.
 TODO: fix doc
 The scope application code must instantiate a class derived from ReceiverBase and pass that instance as
-a parameter to the ScopeProxy::query() method. Once a query is sent, the scopes run time repeatedly
+a parameter to the Scope::query() method. Once a query is sent, the scopes run time repeatedly
 invokes the push() method, once for each result returned by the query. Once a query is complete,
 the finished() method is called once, to inform the caller that the query is complete.
+
+Calls to push() and finished() are made by an arbitrary thread.
+
+// TODO: add doc for thread pool and concurrent calls to push()
 */
 
 class UNITY_API ReceiverBase : private util::NonCopyable
@@ -61,15 +66,28 @@
 
     /**
     \brief Called once by the scopes run time for each category that is returned by a query().
-    Default implementation does nothing.
+    The default implementation does nothing. Receipt of categories may be interleaved with
+    the receipt of results, that is, there is no guarantee that the complete set of categories
+    will be provided before the first query result.
+
+    If push() throws an exception, the scopes run time calls finished() with an 'Error' reason.
     */
-
     virtual void push(Category::SCPtr category);
 
     /**
+    \brief Indicates the cause of a call to finished().
+    The Error enumerator indicates that a query terminated abnormally, for example,
+    because a scope could not be reached over the network or terminated an query
+    abnormally.
+    */
+    enum Reason { Finished, Cancelled, Error };
+
+    /**
     \brief Called once by the scopes run time after the final result for a query() was sent.
+    Exceptions thrown from finished() are ignored.
+    \param r Indicates the cause for the call to finished().
     */
-    virtual void finished() = 0;
+    virtual void finished(Reason r) = 0;
 
 protected:
     /// @cond
@@ -77,6 +95,12 @@
     /// @endcond
 };
 
+/**
+\brief Convenience function to convert a ReceiverBase::Reason enumerator to a string.
+\return Possible return values are "finished", "cancelled", and "error".
+*/
+char const* to_string(ReceiverBase::Reason reason);
+
 } // namespace scopes
 
 } // namespace api

=== modified file 'include/scopes/Registry.h'
--- include/scopes/Registry.h	2013-11-20 12:33:42 +0000
+++ include/scopes/Registry.h	2013-11-22 04:36:14 +0000
@@ -58,7 +58,7 @@
 
     /**
     \brief Returns a ScopeProxy for the scope with the given name.
-    @return If no scope with the given name exists, the returned `shared_ptr` stores a `nullptr`.
+    @return If no scope with the given name exists, find() throws NotFoundException.
     */
     ScopeProxy find(std::string const& scope_name) const;
 

=== modified file 'include/scopes/Reply.h'
--- include/scopes/Reply.h	2013-11-20 12:33:42 +0000
+++ include/scopes/Reply.h	2013-11-22 04:36:14 +0000
@@ -101,7 +101,8 @@
     friend class internal::ReplyImpl;
 
     std::unique_ptr<internal::ReplyImpl> p;
-    std::shared_ptr<internal::QueryObject> qo;  // Used to reference
+    std::shared_ptr<internal::QueryObject> qo; // Points at the corresponding QueryObject, so we can
+                                               // forward cancellation.
 };
 
 } // namespace scopes

=== modified file 'include/scopes/ScopeBase.h'
--- include/scopes/ScopeBase.h	2013-11-01 12:44:48 +0000
+++ include/scopes/ScopeBase.h	2013-11-22 04:36:14 +0000
@@ -67,7 +67,7 @@
 \class ScopeBase
 \brief Base class for a scope implementation.
 
-Scopes are accessed by the Unity run time as a shared library (one library per scope).
+Scopes are accessed by the scopes run time as a shared library (one library per scope).
 Each scope must implement a class that derives from ScopeBase, for example:
 
 ~~~
@@ -79,9 +79,9 @@
     MyScope();
     virtual ~MyScope();
 
-    virtual int start();
-    virtual void stop();
-    virtual void run();
+    virtual int start();    // Mandatory
+    virtual void stop();    // Mandatory
+    virtual void run();     // Optional
 };
 ~~~
 
@@ -90,10 +90,10 @@
  - a create function that must return a pointer to the derived instance
  - a destroy function that is passed the pointer returned by the create function
 
-Typically, the create and destroy functions will simply call new and delete, respectively. (However,
-there is no requirement that the derived class instance must be heap allocated.)
+Typically, the create and destroy functions will simply call `new` and `delete`, respectively. (However,
+there is no requirement that the derived class instance must be heap-allocated.)
 If the create function throws an exception, the destroy function will not be called. If the create function returns
-NULL, the destroy function will be called with NULL as its argument.
+NULL, the destroy function _will_ be called with NULL as its argument.
 
 Rather than hard-coding the names of the functions, use the #UNITY_API_SCOPE_CREATE_FUNCTION and
 #UNITY_API_SCOPE_DESTROY_FUNCTION macros, for example:
@@ -112,10 +112,14 @@
 }
 ~~~
 
-After the Unity run time has obtained a pointer to the class instance from the create function, it calls start(),
-which allows the scope to intialize itself. Once start() returns, incoming query requests are dispatched to the scope.
+After the scopes run time has obtained a pointer to the class instance from the create function, it calls start(),
+which allows the scope to intialize itself. This is followed by call to run(). The call to run() is made by
+a separate thread; its only purpose is to pass a thread of control to the scope, for example, to run an event loop.
 When the scope should complete its activities, the run time calls stop(). The calls to the create function, start(),
-stop(), and the destroy function) are made by the same thread. The call to run() is made by a _different_ thread.
+stop(), and the destroy function) are made by the same thread.
+
+The scope implementation, if it does not return from run(), is expected to return from run() in response to a
+call to stop() in a timely manner.
 */
 
 class UNITY_API ScopeBase : private util::NonCopyable
@@ -131,7 +135,7 @@
     static constexpr int VERSION = UNITY_SCOPES_VERSION_MAJOR;
 
     /**
-    \brief Called by the Unity run time after initialize() completes.
+    \brief Called by the scopes run time after the create function completes.
     If start() throws an exception, stop() will _not_ be called.
 
     The call to start() is made by the same thread that calls the create function.
@@ -141,16 +145,17 @@
     \param registry A proxy to the scope registry. This parameter is provided for aggregating
     scopes that need to retrieve proxies to their child scopes.
 
-    \return Any return value other than ScopeBase::VERSION will cause the Unity run time
+    \return Any return value other than ScopeBase::VERSION will cause the scopes run time
     to refuse to load the scope. The return value is used to ensure that the shared library
-    containing the scope is ABI compatible with the Unity scopes run time.
+    containing the scope is ABI compatible with the scopes scopes run time.
     */
     virtual int start(std::string const& scope_name, RegistryProxy const& registry) = 0;
 
     /**
-    \brief Called by the Unity run time when the scope should shut down.
+    \brief Called by the scopes run time when the scope should shut down.
     A scope should deallocate as many resources as possible when stop() is called, for example,
-    deallocate any caches and close network connections.
+    deallocate any caches and close network connections. In addition, if the scope implements run()
+    and did not return from run(), it must return from run() in response to the call to stop().
 
     Exceptions from stop() are ignored.
 
@@ -159,19 +164,24 @@
     virtual void stop() = 0;
 
     /**
-    \brief Called by the Unity run time to hand a thread of control to the scope.
+    \brief Called by the scopes run time after it has called start() to hand a thread of control to the scope.
     run() passes a thread of control to the scope to do with as it sees fit, for example, to run an event loop.
-
-    If run() throws an exception, stop() _will_ be called.
-
-    The call to run() is made by a separate thread (not the thread that calls the create function and start()).
+    During finalization, the scopes run time joins with the thread that called run(). This means that, if
+    the scope implementation does not return from run(), it is expected to arrange for run() to complete
+    in timely manner in response to a call to stop(). Failure to do so will cause deadlock during finalization.
+
+    If run() throws an exception, the run time handles the exception and calls stop() in response.
     */
-    virtual void run() = 0;
+    virtual void run();
 
     /**
-    \brief Called by the Unity run time when a scope needs to instantiate a query.
-    TODO: complete documentation.
-
+    \brief Called by the scopes run time when a scope needs to instantiate a query.
+    This method must return an instance that is derived from QueryBase. The implementation
+    of this method must return in a timely manner, that is, it should perform only minimal
+    initialization that is guaranteed to complete quickly. That call to create_query() is made
+    by an arbitrary thread.
+    /param q The query string to be executed by the returned object instance.
+    /param hints TODO, complete doc
     */
     virtual QueryBase::UPtr create_query(std::string const& q, VariantMap const& hints) = 0;
 
@@ -193,16 +203,18 @@
 } // namespace unity
 
 /**
-\brief The function called by the Unity run time to initialize the scope.
-It must return a pointer to a ScopeBase instance. The returned instance need not be heap-allocated.
+\brief The function called by the scopes run time to initialize the scope.
+It must return a pointer to an instance derived from ScopeBase. The returned
+instance need not be heap-allocated, but must remain in scope until the
+destroy function is called by the scopes run time.
 
 If this function throws an exception, the destroy function will _not_ be called. If this function returns NULL,
-the destroy function _will_be called with NULL as its argument.
+the destroy function _will_ be called with NULL as its argument.
 */
 extern "C" unity::api::scopes::ScopeBase* UNITY_API_SCOPE_CREATE_FUNCTION();
 
 /**
-\brief The function called by the Unity run time to finalize the scope.
+\brief The function called by the scopes run time to finalize the scope.
 The passed pointer is the pointer that was returned by the create function.
 
 Exceptions thrown by the destroy function are ignored.

=== modified file 'include/scopes/internal/MWReply.h'
--- include/scopes/internal/MWReply.h	2013-11-19 11:46:14 +0000
+++ include/scopes/internal/MWReply.h	2013-11-22 04:36:14 +0000
@@ -21,6 +21,7 @@
 
 #include <scopes/internal/MWObjectProxy.h>
 #include <scopes/internal/MWReplyProxyFwd.h>
+#include <scopes/ReceiverBase.h>
 #include <scopes/Variant.h>
 
 #include <string>
@@ -43,7 +44,7 @@
     virtual ~MWReply() noexcept;
 
     virtual void push(VariantMap const& result) = 0;
-    virtual void finished() = 0;
+    virtual void finished(ReceiverBase::Reason reason) = 0;
 
 protected:
     MWReply(MiddlewareBase* mw_base);

=== modified file 'include/scopes/internal/QueryObject.h'
--- include/scopes/internal/QueryObject.h	2013-11-01 12:44:48 +0000
+++ include/scopes/internal/QueryObject.h	2013-11-22 04:36:14 +0000
@@ -24,6 +24,7 @@
 #include <scopes/internal/MWQueryCtrlProxyFwd.h>
 #include <scopes/ReplyProxyFwd.h>
 
+#include <atomic>
 #include <mutex>
 
 namespace unity
@@ -61,6 +62,8 @@
     // Called locally only, by QueryCtrlObject.
     void cancel();
 
+    bool pushable() const noexcept; // Called locallly only, by ReplyImpl
+
     // Called by create_query(), to hold the reference count high until the run call arrives via the middleware,
     // and we can pass the shared_ptr to the ReplyImpl.
     void set_self(SPtr const& self);
@@ -70,6 +73,7 @@
     MWReplyProxy reply_;
     std::weak_ptr<Reply> reply_proxy_;
     MWQueryCtrlProxy const ctrl_;
+    std::atomic_bool pushable_;
     SPtr self_;
 };
 

=== modified file 'include/scopes/internal/ReplyImpl.h'
--- include/scopes/internal/ReplyImpl.h	2013-11-19 11:46:14 +0000
+++ include/scopes/internal/ReplyImpl.h	2013-11-22 04:36:14 +0000
@@ -23,6 +23,7 @@
 #include <scopes/internal/CategoryRegistry.h>
 #include <scopes/ReplyProxyFwd.h>
 #include <scopes/Category.h>
+#include <scopes/ReceiverBase.h>
 
 #include <atomic>
 
@@ -60,6 +61,7 @@
 
     bool push(unity::api::scopes::ResultItem const& result);
     void finished();
+    void finished(unity::api::scopes::ReceiverBase::Reason reason);
 
     static ReplyProxy create(MWReplyProxy const& mw_proxy, std::shared_ptr<QueryObject> const& qo);
 

=== modified file 'include/scopes/internal/ReplyObject.h'
--- include/scopes/internal/ReplyObject.h	2013-11-21 10:12:45 +0000
+++ include/scopes/internal/ReplyObject.h	2013-11-22 04:36:14 +0000
@@ -56,10 +56,10 @@
 
     // Remote operation implementations
     void push(VariantMap const& result) noexcept;
-    void finished() noexcept;
+    void finished(ReceiverBase::Reason reason) noexcept;
 
 private:
-    ReceiverBase::SPtr const reply_base_;
+    ReceiverBase::SPtr const receiver_base_;
     ReapItem::SPtr reap_item_;
     std::shared_ptr<CategoryRegistry> cat_registry_;
     std::atomic_bool finished_;

=== modified file 'include/scopes/internal/zmq_middleware/ZmqReply.h'
--- include/scopes/internal/zmq_middleware/ZmqReply.h	2013-11-19 11:46:14 +0000
+++ include/scopes/internal/zmq_middleware/ZmqReply.h	2013-11-22 04:36:14 +0000
@@ -45,7 +45,7 @@
     virtual ~ZmqReply() noexcept;
 
     virtual void push(VariantMap const& result) override;
-    virtual void finished() override;
+    virtual void finished(ReceiverBase::Reason reason) override;
 };
 
 } // namespace zmq_middleware

=== modified file 'src/QueryBase.cpp'
--- src/QueryBase.cpp	2013-11-21 10:12:45 +0000
+++ src/QueryBase.cpp	2013-11-22 04:36:14 +0000
@@ -50,7 +50,7 @@
 
 void QueryBase::cancel()
 {
-    p->cancel();    // Forward cancel to subquery
+    p->cancel();    // Forward cancel to subqueries
     cancelled();    // Inform this query that it was cancelled
 }
 

=== modified file 'src/ReceiverBase.cpp'
--- src/ReceiverBase.cpp	2013-11-21 10:12:45 +0000
+++ src/ReceiverBase.cpp	2013-11-22 04:36:14 +0000
@@ -18,6 +18,9 @@
 
 #include <scopes/ReceiverBase.h>
 
+#include <cassert>
+#include <unordered_map>
+
 using namespace std;
 
 namespace unity
@@ -43,6 +46,22 @@
 {
 }
 
+// Possibly overkill, but safer than using the enum as the index into an array,
+// in case the enumeration is ever added to or the enumerators get re-ordered.
+
+static unordered_map<int, char const*> const reasons =
+{
+    pair<int, char const*>(static_cast<int>(ReceiverBase::Finished),  "finished"),
+    pair<int, char const*>(static_cast<int>(ReceiverBase::Cancelled), "cancelled"),
+    pair<int, char const*>(static_cast<int>(ReceiverBase::Error),     "error")
+};
+
+char const* to_string(ReceiverBase::Reason reason)
+{
+    assert(reasons.find(static_cast<int>(reason)) != reasons.end());
+    return reasons.find(static_cast<int>(reason))->second;
+}
+
 //! @endcond
 
 } // namespace scopes

=== modified file 'src/ScopeBase.cpp'
--- src/ScopeBase.cpp	2013-11-01 12:44:48 +0000
+++ src/ScopeBase.cpp	2013-11-22 04:36:14 +0000
@@ -39,6 +39,11 @@
 
 //! @endcond
 
+void ScopeBase::run()
+{
+    // Intentionally empty: default "do nothing" implementation.
+}
+
 void ScopeBase::runtime_version(int& v_major, int& v_minor, int& v_micro) noexcept
 {
     v_major = unity::api::scopes::major_version();

=== modified file 'src/internal/QueryCtrlImpl.cpp'
--- src/internal/QueryCtrlImpl.cpp	2013-11-01 12:44:48 +0000
+++ src/internal/QueryCtrlImpl.cpp	2013-11-22 04:36:14 +0000
@@ -65,7 +65,7 @@
         // Indicate (to ourselves) that this query is complete. Calling via the MWReplyProxy ensures
         // the finished() call will be processed by a seperate server-side thread,
         // so we cannot block here.
-        reply_proxy_->finished();
+        reply_proxy_->finished(ReceiverBase::Cancelled);
     }
     catch (MiddlewareException const& e)
     {

=== modified file 'src/internal/QueryObject.cpp'
--- src/internal/QueryObject.cpp	2013-11-01 12:44:48 +0000
+++ src/internal/QueryObject.cpp	2013-11-22 04:36:14 +0000
@@ -48,7 +48,8 @@
                          MWQueryCtrlProxy const& ctrl) :
     query_base_(query_base),
     reply_(reply),
-    ctrl_(ctrl)
+    ctrl_(ctrl),
+    pushable_(true)
 {
     assert(query_base);
     assert(reply);
@@ -91,21 +92,28 @@
     }
     catch (unity::Exception const& e)
     {
+        pushable_ = false;
+        reply_->finished(ReceiverBase::Error);     // Oneway, can't block
         // TODO: log error
     }
     catch (...)
     {
+        pushable_ = false;
+        reply_->finished(ReceiverBase::Error);     // Oneway, can't block
         // TODO: log error
     }
 }
 
 void QueryObject::cancel()
 {
+    pushable_ = false;
     auto rp = reply_proxy_.lock();
     if (rp)
     {
         // Send finished() to up-stream client to tell him the query is done.
-        rp->finished();     // Oneway, can't block
+        // We send via the MWReplyProxy here because that allows passing
+        // a ReceiverBase::Reason (whereas the public ReplyProxy does not).
+        reply_->finished(ReceiverBase::Cancelled);     // Oneway, can't block
     }
 
     // Forward the cancellation to the query base (which in turn will forward it to any subqueries).
@@ -113,6 +121,11 @@
     query_base_->cancel();
 }
 
+bool QueryObject::pushable() const noexcept
+{
+    return pushable_;
+}
+
 // The point of keeping a shared_ptr to ourselves is to make sure this QueryObject cannot
 // go out of scope in between being created by the Scope, and the first ReplyProxy for this
 // query being created in QueryObject::run(). If the scope's run() method returns immediately,

=== modified file 'src/internal/ReplyImpl.cpp'
--- src/internal/ReplyImpl.cpp	2013-11-19 11:46:14 +0000
+++ src/internal/ReplyImpl.cpp	2013-11-22 04:36:14 +0000
@@ -16,11 +16,12 @@
  * Authored by: Michi Henning <[e-mail]>
  */
 
+#include <scopes/internal/ReplyImpl.h>
+
 #include <scopes/internal/MiddlewareBase.h>
 #include <scopes/internal/MWReply.h>
 #include <scopes/internal/RuntimeImpl.h>
 #include <scopes/internal/ResultItemImpl.h>
-#include <scopes/internal/ReplyImpl.h>
 #include <scopes/ResultItem.h>
 #include <scopes/ScopeExceptions.h>
 #include <unity/UnityExceptions.h>
@@ -91,8 +92,8 @@
 
 bool ReplyImpl::push(unity::api::scopes::ResultItem const& result)
 {
-    // if this is an aggregator scope, it may be pushing result items obtained
-    // from ReplyObject without registering category first.
+    // If this is an aggregator scope, it may be pushing result items obtained
+    // from ReplyObject without registering a category first.
     auto cat = cat_registry_->lookup_category(result.category()->id());
     if (cat == nullptr)
     {
@@ -115,6 +116,11 @@
 
 bool ReplyImpl::push(VariantMap const& variant_map)
 {
+    if (!qo_->pushable())
+    {
+        return false; // Query was cancelled or had an error.
+    }
+
     if (!finished_.load())
     {
         try
@@ -125,7 +131,7 @@
         catch (MiddlewareException const& e)
         {
             // TODO: log error
-            finished();
+            finished(ReceiverBase::Error);
             return false;
         }
     }
@@ -134,11 +140,16 @@
 
 void ReplyImpl::finished()
 {
+    finished(ReceiverBase::Finished);
+}
+
+void ReplyImpl::finished(ReceiverBase::Reason reason)
+{
     if (!finished_.exchange(true))
     {
         try
         {
-            mw_proxy_->finished();
+            mw_proxy_->finished(reason);
         }
         catch (MiddlewareException const& e)
         {

=== modified file 'src/internal/ReplyObject.cpp'
--- src/internal/ReplyObject.cpp	2013-11-21 10:43:04 +0000
+++ src/internal/ReplyObject.cpp	2013-11-22 04:36:14 +0000
@@ -41,13 +41,13 @@
 namespace internal
 {
 
-ReplyObject::ReplyObject(ReceiverBase::SPtr const& reply_base, RuntimeImpl const* runtime) :
-    reply_base_(reply_base),
+ReplyObject::ReplyObject(ReceiverBase::SPtr const& receiver_base, RuntimeImpl const* runtime) :
+    receiver_base_(receiver_base),
     cat_registry_(new CategoryRegistry()),
     finished_(false),
     num_push_(0)
 {
-    assert(reply_base);
+    assert(receiver_base);
     assert(runtime);
     reap_item_ = runtime->reply_reaper()->add([this] { this->disconnect(); });
 }
@@ -56,7 +56,7 @@
 {
     try
     {
-        finished();
+        finished(ReceiverBase::Finished);
     }
     catch (...)
     {
@@ -90,14 +90,14 @@
     unique_lock<mutex> lock(mutex_);
     assert(num_push_ >= 0);
     ++num_push_;
-    lock.unlock();
+    lock.unlock(); // Forward invocations to application outside synchronization
     try
     {
         auto it = result.find("category");
         if (it != result.end())
         {
             auto cat = cat_registry_->register_category(it->second.get_dict());
-            reply_base_->push(cat);
+            receiver_base_->push(cat);
         }
 
         it = result.find("result");
@@ -109,12 +109,12 @@
             if (cat == nullptr)
             {
                 // TODO: this is an internal error; log error
-                finished();
+                finished(ReceiverBase::Error);
             }
             else
             {
                 ResultItem result_item(cat, result_var);
-                reply_base_->push(std::move(result_item));      // Forward the result to the application code outside synchronization.
+                receiver_base_->push(std::move(result_item));
             }
         }
     }
@@ -123,7 +123,7 @@
         // TODO: log error
         try
         {
-            finished();
+            finished(ReceiverBase::Error);
         }
         catch (...)
         {
@@ -134,7 +134,7 @@
         // TODO: log error
         try
         {
-            finished();
+            finished(ReceiverBase::Error);
         }
         catch (...)
         {
@@ -147,7 +147,7 @@
     }
 }
 
-void ReplyObject::finished() noexcept
+void ReplyObject::finished(ReceiverBase::Reason r) noexcept
 {
     // We permit exactly one finished() call for a query. This avoids
     // a race condition where the executing down-stream query invokes
@@ -167,10 +167,10 @@
     unique_lock<mutex> lock(mutex_);
     assert(num_push_ >= 0);
     idle_.wait(lock, [this] { return num_push_ == 0; });
-    lock.unlock();
+    lock.unlock(); // Inform the application code that the query is complete outside synchronization.
     try
     {
-        reply_base_->finished();    // Inform the application code that the query is complete outside synchronization.
+        receiver_base_->finished(r);
     }
     catch (unity::Exception const& e)
     {

=== modified file 'src/internal/ScopeImpl.cpp'
--- src/internal/ScopeImpl.cpp	2013-11-21 10:12:45 +0000
+++ src/internal/ScopeImpl.cpp	2013-11-22 04:36:14 +0000
@@ -76,7 +76,7 @@
         {
             // TODO: if things go wrong, we need to make sure that the reply object
             // is disconnected from the middleware, so it gets deallocated.
-            reply->finished();
+            reply->finished(ReceiverBase::Error);
             throw;
         }
         catch (...)

=== modified file 'src/internal/ScopeLoader.cpp'
--- src/internal/ScopeLoader.cpp	2013-11-01 12:44:48 +0000
+++ src/internal/ScopeLoader.cpp	2013-11-22 04:36:14 +0000
@@ -452,6 +452,7 @@
     }
     catch (...)
     {
+        // TODO: log this
         // Run throws, we remember the exception and send a stop command to ourselves, so the stop()
         // callback on the scope will still be called.
         {

=== modified file 'src/internal/ScopeObject.cpp'
--- src/internal/ScopeObject.cpp	2013-11-18 23:53:56 +0000
+++ src/internal/ScopeObject.cpp	2013-11-22 04:36:14 +0000
@@ -76,12 +76,21 @@
     }
 
     // Ask scope to instantiate a new query.
-    QueryBase::SPtr query_base = scope_base_->create_query(q, hints);
-    if (!query_base)
-    {
-        // TODO: log error, scope returned null pointer.
+    QueryBase::SPtr query_base;
+    try
+    {
+        query_base = scope_base_->create_query(q, hints);
+        if (!query_base)
+        {
+            // TODO: log error, scope returned null pointer.
+            throw ResourceException("Scope \"" + runtime_->scope_name() +
+                                    "\" returned nullptr from create_query(\"" + q + "\")");
+        }
+    }
+    catch (...)
+    {
         throw ResourceException("Scope \"" + runtime_->scope_name() +
-                                "\" returned nullptr from create_query(\"" + q + "\")");
+                                "\" threw an exception from create_query(\"" + q + "\")");
     }
 
     MWQueryCtrlProxy ctrl_proxy;
@@ -113,7 +122,7 @@
     {
         try
         {
-            reply->finished();
+            reply->finished(ReceiverBase::Error);
         }
         catch (...)
         {
@@ -125,7 +134,7 @@
     {
         try
         {
-            reply->finished();
+            reply->finished(ReceiverBase::Error);
         }
         catch (...)
         {

=== modified file 'src/internal/zmq_middleware/ReplyI.cpp'
--- src/internal/zmq_middleware/ReplyI.cpp	2013-11-08 15:00:22 +0000
+++ src/internal/zmq_middleware/ReplyI.cpp	2013-11-22 04:36:14 +0000
@@ -74,11 +74,36 @@
 }
 
 void ReplyI::finished_(Current const&,
-                       capnp::ObjectPointer::Reader&,
+                       capnp::ObjectPointer::Reader& in_params,
                        capnproto::Response::Builder&)
 {
     auto delegate = dynamic_pointer_cast<ReplyObject>(del());
-    delegate->finished();
+    auto req = in_params.getAs<capnproto::Reply::FinishedRequest>();
+    auto r = req.getReason();
+    ReceiverBase::Reason reason;
+    switch (r)
+    {
+        case capnproto::Reply::FinishedReason::FINISHED:
+        {
+            reason = ReceiverBase::Finished;
+            break;
+        }
+        case capnproto::Reply::FinishedReason::CANCELLED:
+        {
+            reason = ReceiverBase::Cancelled;
+            break;
+        }
+        case capnproto::Reply::FinishedReason::ERROR:
+        {
+            reason = ReceiverBase::Error;
+            break;
+        }
+        default:
+        {
+            assert(false); // LCOV_EXCL_LINE
+        }
+    }
+    delegate->finished(reason);
 }
 
 } // namespace zmq_middleware

=== modified file 'src/internal/zmq_middleware/ZmqReply.cpp'
--- src/internal/zmq_middleware/ZmqReply.cpp	2013-11-19 11:46:14 +0000
+++ src/internal/zmq_middleware/ZmqReply.cpp	2013-11-22 04:36:14 +0000
@@ -74,10 +74,35 @@
     future.wait();
 }
 
-void ZmqReply::finished()
+void ZmqReply::finished(ReceiverBase::Reason reason)
 {
     capnp::MallocMessageBuilder request_builder;
-    make_request_(request_builder, "finished");
+    auto request = make_request_(request_builder, "finished");
+    auto in_params = request.initInParams().getAs<capnproto::Reply::FinishedRequest>();
+    capnproto::Reply::FinishedReason r;
+    switch (reason)
+    {
+        case ReceiverBase::Finished:
+        {
+            r = capnproto::Reply::FinishedReason::FINISHED;
+            break;
+        }
+        case ReceiverBase::Cancelled:
+        {
+            r = capnproto::Reply::FinishedReason::CANCELLED;
+            break;
+        }
+        case ReceiverBase::Error:
+        {
+            r = capnproto::Reply::FinishedReason::ERROR;
+            break;
+        }
+        default:
+        {
+            assert(false);
+        }
+    }
+    in_params.setReason(r);
 
     auto future = mw_base()->invoke_pool()->submit([&] { return this->invoke_(request_builder); });
     future.wait();

=== modified file 'src/internal/zmq_middleware/capnproto/Reply.capnp'
--- src/internal/zmq_middleware/capnproto/Reply.capnp	2013-11-04 14:10:29 +0000
+++ src/internal/zmq_middleware/capnproto/Reply.capnp	2013-11-22 04:36:14 +0000
@@ -35,21 +35,23 @@
 # Operations:
 #
 # void push(string result);
-# void finished();
+# enum FinishedReason { Finished, Cancelled, Error };
+# void finished(Reason r);
 
 struct PushRequest
 {
     result @0 : ValueDict.ValueDict;
 }
 
-struct PushResponse
+enum FinishedReason
 {
+    unused @0;
+    finished @1;
+    cancelled @2;
+    error @3;
 }
 
 struct FinishedRequest
 {
-}
-
-struct FinishedResponse
-{
+    reason @0 : FinishedReason;
 }