[2945]Merge branch 'master' into trac2945

fix merge conflicts

7 files changed, 376 insertions, 8 deletions

diff --git a/src/lib/config/.gitignore b/src/lib/config/.gitignore
index c7ec9d3565..d666f2469a 100644
--- a/src/lib/config/.gitignore
+++ b/src/lib/config/.gitignore
@@ -1,2 +1,3 @@
 /config_messages.cc
 /config_messages.h
+/s-messages
diff --git a/src/lib/config/Makefile.am b/src/lib/config/Makefile.am
index b4fc2e0c89..9820f08289 100644
--- a/src/lib/config/Makefile.am
+++ b/src/lib/config/Makefile.am
@@ -6,8 +6,11 @@ AM_CPPFLAGS += -I$(top_srcdir)/src/lib/log -I$(top_builddir)/src/lib/log
 AM_CPPFLAGS += $(BOOST_INCLUDES)
 
 # Define rule to build logging source files from message file
-config_messages.h config_messages.cc: config_messages.mes
+config_messages.h config_messages.cc: s-messages
+
+s-messages: config_messages.mes
 	$(top_builddir)/src/lib/log/compiler/message $(top_srcdir)/src/lib/config/config_messages.mes
+	touch $@
 
 BUILT_SOURCES = config_messages.h config_messages.cc
 
@@ -27,4 +30,4 @@ nodist_libb10_cfgclient_la_SOURCES  = config_messages.h config_messages.cc
 # The message file should be in the distribution.
 EXTRA_DIST = config_messages.mes
 
-CLEANFILES = *.gcno *.gcda config_messages.h config_messages.cc
+CLEANFILES = *.gcno *.gcda config_messages.h config_messages.cc s-messages
diff --git a/src/lib/config/ccsession.cc b/src/lib/config/ccsession.cc
index d094ab9bf6..10bc7286f3 100644
--- a/src/lib/config/ccsession.cc
+++ b/src/lib/config/ccsession.cc
@@ -144,7 +144,7 @@ parseCommand(ConstElementPtr& arg, ConstElementPtr command) {
         command->contains(isc::cc::CC_PAYLOAD_COMMAND)) {
         ConstElementPtr cmd = command->get(isc::cc::CC_PAYLOAD_COMMAND);
         if (cmd->getType() == Element::list &&
-            cmd->size() > 0 &&
+            !cmd->empty() &&
             cmd->get(0)->getType() == Element::string) {
             if (cmd->size() > 1) {
                 arg = cmd->get(1);
@@ -595,6 +595,8 @@ ModuleCCSession::checkModuleCommand(const std::string& cmd_str,
                                  "Command given but no "
                                  "command handler for module"));
         }
+    } else if (unhandled_callback_) {
+        unhandled_callback_(cmd_str, target_module, arg);
     }
     return (ElementPtr());
 }
@@ -609,6 +611,11 @@ ModuleCCSession::checkCommand() {
             return (0);
         }
 
+        // In case it is notification, eat it.
+        if (checkNotification(routing, data)) {
+            return (0);
+        }
+
         /* ignore result messages (in case we're out of sync, to prevent
          * pingpongs */
         if (data->getType() != Element::map ||
@@ -884,5 +891,100 @@ ModuleCCSession::rpcCall(const std::string &command, const std::string &group,
     }
 }
 
+void
+ModuleCCSession::notify(const std::string& group, const std::string& name,
+                        const ConstElementPtr& params)
+{
+    const ElementPtr message(Element::createMap());
+    const ElementPtr notification(Element::createList());
+    notification->add(Element::create(name));
+    if (params) {
+        notification->add(params);
+    }
+    message->set(isc::cc::CC_PAYLOAD_NOTIFICATION, notification);
+    groupSendMsg(message, isc::cc::CC_GROUP_NOTIFICATION_PREFIX + group,
+                 isc::cc::CC_INSTANCE_WILDCARD,
+                 isc::cc::CC_TO_WILDCARD, false);
+}
+
+ModuleCCSession::NotificationID
+ModuleCCSession::subscribeNotification(const std::string& notification_group,
+                                       const NotificationCallback& callback)
+{
+    // Either insert a new empty list of callbacks or get an existing one.
+    // Either way, get the iterator for its position.
+    const std::pair<SubscribedNotifications::iterator, bool>& inserted =
+        notifications_.insert(
+            std::pair<std::string, NotificationCallbacks>(notification_group,
+                NotificationCallbacks()));
+    if (inserted.second) {
+        // It was newly inserted. In that case, we need to subscribe to the
+        // group.
+        session_.subscribe(isc::cc::CC_GROUP_NOTIFICATION_PREFIX +
+                           notification_group);
+    }
+    // Insert the callback to the chain
+    NotificationCallbacks& callbacks = inserted.first->second;
+    const NotificationCallbacks::iterator& callback_id =
+        callbacks.insert(callbacks.end(), callback);
+    // Just pack the iterators to form the ID
+    return (NotificationID(inserted.first, callback_id));
+}
+
+void
+ModuleCCSession::unsubscribeNotification(const NotificationID& notification) {
+    NotificationCallbacks& callbacks = notification.first->second;
+    // Remove the callback
+    callbacks.erase(notification.second);
+    // If it became empty, remove it from the map and unsubscribe
+    if (callbacks.empty()) {
+        session_.unsubscribe(isc::cc::CC_GROUP_NOTIFICATION_PREFIX +
+                             notification.first->first);
+        notifications_.erase(notification.first);
+    }
+}
+
+bool
+ModuleCCSession::checkNotification(const data::ConstElementPtr& envelope,
+                                   const data::ConstElementPtr& msg)
+{
+    if (msg->getType() != data::Element::map) {
+        // If it's not a map, then it's not a notification
+        return (false);
+    }
+    if (msg->contains(isc::cc::CC_PAYLOAD_NOTIFICATION)) {
+        // There's a notification inside. Extract its parameters.
+        const std::string& group =
+            envelope->get(isc::cc::CC_HEADER_GROUP)->stringValue();
+        const std::string& notification_group =
+            group.substr(std::string(isc::cc::CC_GROUP_NOTIFICATION_PREFIX).
+                         size());
+        const data::ConstElementPtr& notification =
+            msg->get(isc::cc::CC_PAYLOAD_NOTIFICATION);
+        // The first one is the event that happened
+        const std::string& event = notification->get(0)->stringValue();
+        // Any other params are second. But they may be missing
+        const data::ConstElementPtr params =
+            notification->size() == 1 ? data::ConstElementPtr() :
+            notification->get(1);
+        // Find the chain of notification callbacks
+        const SubscribedNotifications::iterator& chain_iter =
+            notifications_.find(notification_group);
+        if (chain_iter == notifications_.end()) {
+            // This means we no longer have any notifications for this group.
+            // This can happen legally as a race condition - if msgq sends
+            // us a notification, but we unsubscribe before we get to it
+            // in the input stream.
+            return (false);
+        }
+        BOOST_FOREACH(const NotificationCallback& callback,
+                      chain_iter->second) {
+            callback(event, params);
+        }
+        return (true);
+    }
+    return (false); // Not a notification
+}
+
 }
 }
diff --git a/src/lib/config/ccsession.h b/src/lib/config/ccsession.h
index 995a5cd68b..75c3ee6fff 100644
--- a/src/lib/config/ccsession.h
+++ b/src/lib/config/ccsession.h
@@ -425,6 +425,31 @@ public:
                                            params =
                                            isc::data::ConstElementPtr());
 
+    /// \brief Send a notification to subscribed users
+    ///
+    /// Send a notification message to all users subscribed to the given
+    /// notification group.
+    ///
+    /// This method does not not block.
+    ///
+    /// See docs/design/ipc-high.txt for details about notifications and
+    /// the format of messages sent.
+    ///
+    /// \throw CCSessionError for low-level communication errors.
+    /// \param notification_group This parameter (indirectly) signifies what
+    ///     users should receive the notification. Only the users that
+    ///     subscribed to notifications on the same group receive it.
+    /// \param name The name of the event to notify about (for example
+    ///     `new_group_member`).
+    /// \param params Other parameters that describe the event. This might
+    ///     be, for example, the ID of the new member and the name of the
+    ///     group. This can be any data element, but it is common for it to be
+    ///     map.
+    void notify(const std::string& notification_group,
+                const std::string& name,
+                const isc::data::ConstElementPtr& params =
+                    isc::data::ConstElementPtr());
+
     /// \brief Convenience version of rpcCall
     ///
     /// This is exactly the same as the previous version of rpcCall, except
@@ -550,6 +575,101 @@ public:
     /// \param id The id of request as returned by groupRecvMsgAsync.
     void cancelAsyncRecv(const AsyncRecvRequestID& id);
 
+    /// \brief Called when a notification comes
+    ///
+    /// The callback should be exception-free. If it raises an exception,
+    /// it'll leak through the event loop up and probably terminate the
+    /// application.
+    ///
+    /// \param event_name The identification of event type.
+    /// \param params The parameters of the event. This may be NULL
+    ///     pointer in case no parameters were sent with the event.
+    typedef boost::function<void (const std::string& event_name,
+                                  const data::ConstElementPtr& params)>
+        NotificationCallback;
+
+    /// \brief Multiple notification callbacks for the same notification
+    typedef std::list<NotificationCallback> NotificationCallbacks;
+
+    /// \brief Mapping from groups to callbacks
+    typedef std::map<std::string, NotificationCallbacks>
+        SubscribedNotifications;
+
+    /// \brief Identification of single callback
+    typedef std::pair<SubscribedNotifications::iterator,
+                      NotificationCallbacks::iterator>
+        NotificationID;
+
+    /// \brief Subscribe to a notification group
+    ///
+    /// From now on, every notification that is sent to the given group
+    /// triggers the passed callback.
+    ///
+    /// There may be multiple (independent) callbacks for the same channel.
+    /// This one adds a new one, to the end of the chain (the callbacks
+    /// are called in the same order as they were registered).
+    ///
+    /// \param notification_group The channel of notifications.
+    /// \param callback The callback to be added.
+    /// \return ID of the notification callback. It is an opaque ID and can
+    ///     be used to remove this callback.
+    NotificationID subscribeNotification(const std::string& notification_group,
+                                         const NotificationCallback& callback);
+
+    /// \brief Unsubscribe the callback from its notification group.
+    ///
+    /// Express that the desire for this callback to be executed is no longer
+    /// relevant. All the other callbacks (even for the same notification
+    /// group) are left intact.
+    ///
+    /// \param notification The ID of notification callback returned by
+    ///     subscribeNotification.
+    void unsubscribeNotification(const NotificationID& notification);
+
+    /// \brief Subscribe to a group
+    ///
+    /// Wrapper around the CCSession::subscribe.
+    void subscribe(const std::string& group) {
+        session_.subscribe(group, isc::cc::CC_INSTANCE_WILDCARD);
+    }
+
+    /// \brief Unsubscribe from a group.
+    ///
+    /// Wrapper around the CCSession::unsubscribe.
+    void unsubscribe(const std::string& group) {
+        session_.unsubscribe(group, isc::cc::CC_INSTANCE_WILDCARD);
+    }
+
+    /// \brief Callback type for unhandled commands
+    ///
+    /// The type of functions that are not handled by the ModuleCCSession
+    /// because they are not aimed at the module.
+    ///
+    /// The parameters are:
+    /// - Name of the command.
+    /// - The module it was aimed for (may be empty).
+    /// - The parameters of the command.
+    typedef boost::function<void (const std::string&, const std::string&,
+                                  const isc::data::ConstElementPtr&)>
+        UnhandledCallback;
+
+    /// \brief Register a callback for messages sent to foreign modules.
+    ///
+    /// Usually, a command aimed at foreign module (or sent directly)
+    /// is discarded. By registering a callback here, these can be
+    /// examined.
+    ///
+    /// \note A callback overwrites the previous one set.
+    /// \todo This is a temporary, unclean, solution. A more generic
+    ///     one needs to be designed. Also, a solution that is able
+    ///     to send an answer would be great.
+    ///
+    /// \param callback The new callback to use. It may be an empty
+    ///     function.
+    void setUnhandledCallback(const UnhandledCallback& callback) {
+        unhandled_callback_ = callback;
+    }
+
 private:
     ModuleSpec readModuleSpecification(const std::string& filename);
     void startCheck();
@@ -565,6 +685,8 @@ private:
     ///     otherwise.
     bool checkAsyncRecv(const data::ConstElementPtr& envelope,
                         const data::ConstElementPtr& msg);
+    bool checkNotification(const data::ConstElementPtr& envelope,
+                           const data::ConstElementPtr& msg);
     /// \brief Checks if a message with this envelope matches the request
     bool requestMatch(const AsyncRecvRequest& request,
                       const data::ConstElementPtr& envelope) const;
@@ -574,6 +696,8 @@ private:
     isc::cc::AbstractSession& session_;
     ModuleSpec module_specification_;
     AsyncRecvRequests async_recv_requests_;
+    SubscribedNotifications notifications_;
+
     isc::data::ConstElementPtr handleConfigUpdate(
         isc::data::ConstElementPtr new_config);
 
@@ -599,6 +723,8 @@ private:
                             isc::data::ConstElementPtr new_config);
 
     ModuleSpec fetchRemoteSpec(const std::string& module, bool is_filename);
+
+    UnhandledCallback unhandled_callback_;
 };
 
 /// \brief Default handler for logging config updates
diff --git a/src/lib/config/config_data.h b/src/lib/config/config_data.h
index 33df82d126..7900aa9c27 100644
--- a/src/lib/config/config_data.h
+++ b/src/lib/config/config_data.h
@@ -29,7 +29,7 @@ namespace config {
 /// point to anything defined in the .spec file)
 class DataNotFoundError : public isc::Exception {
 public:
-    DataNotFoundError(const char* file, size_t line, const std::string& what) :
+    DataNotFoundError(const char* file, size_t line, const char* what) :
         isc::Exception(file, line, what) {}
 };
 
diff --git a/src/lib/config/tests/ccsession_unittests.cc b/src/lib/config/tests/ccsession_unittests.cc
index c11cd24163..b9c70d03b7 100644
--- a/src/lib/config/tests/ccsession_unittests.cc
+++ b/src/lib/config/tests/ccsession_unittests.cc
@@ -87,6 +87,65 @@ protected:
     const std::string root_name;
 };
 
+void
+notificationCallback(std::vector<std::string>* called,
+                     const std::string& id, const std::string& notification,
+                     const ConstElementPtr& params)
+{
+    called->push_back(id);
+    EXPECT_EQ("event", notification);
+    EXPECT_TRUE(el("{\"param\": true}")->equals(*params));
+}
+
+TEST_F(CCSessionTest, receiveNotification) {
+    // Not subscribed to the group yet
+    ModuleCCSession mccs(ccspecfile("spec1.spec"), session, NULL, NULL,
+                         false, false);
+    EXPECT_FALSE(session.haveSubscription("notifications/group", "*"));
+    std::vector<std::string> called;
+    // Subscribe to the notification. Twice.
+    const ModuleCCSession::NotificationID& first =
+        mccs.subscribeNotification("group", boost::bind(&notificationCallback,
+                                                        &called, "first",
+                                                        _1, _2));
+    const ModuleCCSession::NotificationID& second =
+        mccs.subscribeNotification("group", boost::bind(&notificationCallback,
+                                                        &called, "second",
+                                                        _1, _2));
+    EXPECT_TRUE(session.haveSubscription("notifications/group", "*"));
+    EXPECT_TRUE(called.empty());
+    // Send the notification
+    const isc::data::ConstElementPtr msg = el("{"
+        "       \"notification\": ["
+        "           \"event\", {"
+        "               \"param\": true"
+        "           }"
+        "       ]"
+        "   }");
+    session.addMessage(msg, "notifications/group", "*");
+    mccs.checkCommand();
+    ASSERT_EQ(2, called.size());
+    EXPECT_EQ("first", called[0]);
+    EXPECT_EQ("second", called[1]);
+    called.clear();
+    // Unsubscribe one of them
+    mccs.unsubscribeNotification(first);
+    // We are still subscribed to the group and handle the requests
+    EXPECT_TRUE(session.haveSubscription("notifications/group", "*"));
+    // Send the notification
+    session.addMessage(msg, "notifications/group", "*");
+    mccs.checkCommand();
+    ASSERT_EQ(1, called.size());
+    EXPECT_EQ("second", called[0]);
+    // Unsubscribe the other one too. That should cancel the upstream
+    // subscription
+    mccs.unsubscribeNotification(second);
+    EXPECT_FALSE(session.haveSubscription("notifications/group", "*"));
+    // Nothing crashes if out of sync notification comes unexpected
+    session.addMessage(msg, "notifications/group", "*");
+    EXPECT_NO_THROW(mccs.checkCommand());
+}
+
 // Test we can send an RPC (command) and get an answer. The answer is success
 // in this case.
 TEST_F(CCSessionTest, rpcCallSuccess) {
@@ -117,6 +176,57 @@ TEST_F(CCSessionTest, rpcNoRecpt) {
                  RPCRecipientMissing);
 }
 
+// Test sending a notification
+TEST_F(CCSessionTest, notify) {
+    ModuleCCSession mccs(ccspecfile("spec1.spec"), session, NULL, NULL, false,
+                         false);
+    mccs.notify("group", "event", el("{\"param\": true}"));
+    const ConstElementPtr notification(el(
+        "["
+        "   \"notifications/group\","
+        "   \"*\","
+        "   {"
+        "       \"notification\": ["
+        "           \"event\", {"
+        "               \"param\": true"
+        "           }"
+        "       ]"
+        "   },"
+        "   -1"
+        "]"));
+    EXPECT_TRUE(notification->equals(*session.getMsgQueue()->get(1))) <<
+            session.getMsgQueue()->get(1)->toWire();
+}
+
+// Test sending a notification
+TEST_F(CCSessionTest, notifyNoParams) {
+    ModuleCCSession mccs(ccspecfile("spec1.spec"), session, NULL, NULL, false,
+                         false);
+    mccs.notify("group", "event");
+    const ConstElementPtr notification(el(
+        "["
+        "   \"notifications/group\","
+        "   \"*\","
+        "   {"
+        "       \"notification\": [\"event\"]"
+        "   },"
+        "   -1"
+        "]"));
+    EXPECT_TRUE(notification->equals(*session.getMsgQueue()->get(1))) <<
+            session.getMsgQueue()->get(1)->toWire();
+}
+
+// Try to subscribe and unsubscribe once again
+TEST_F(CCSessionTest, subscribe) {
+    ModuleCCSession mccs(ccspecfile("spec1.spec"), session, NULL, NULL, false,
+                         false);
+    EXPECT_FALSE(session.haveSubscription("A group", "*"));
+    mccs.subscribe("A group");
+    EXPECT_TRUE(session.haveSubscription("A group", "*"));
+    mccs.unsubscribe("A group");
+    EXPECT_FALSE(session.haveSubscription("A group", "*"));
+}
+
 TEST_F(CCSessionTest, createAnswer) {
     ConstElementPtr answer;
     answer = createAnswer();
@@ -646,6 +756,16 @@ TEST_F(CCSessionTest, remoteConfig) {
     }
 }
 
+void
+callback(std::string* command, std::string* target, ConstElementPtr *params,
+         const std::string& command_real, const std::string& target_real,
+         const ConstElementPtr& params_real)
+{
+    *command = command_real;
+    *target = target_real;
+    *params = params_real;
+}
+
 TEST_F(CCSessionTest, ignoreRemoteConfigCommands) {
     // client will ask for config
     session.getMessages()->add(createAnswer(0, el("{  }")));
@@ -681,6 +801,22 @@ TEST_F(CCSessionTest, ignoreRemoteConfigCommands) {
     result = mccs.checkCommand();
     EXPECT_EQ(0, session.getMsgQueue()->size());
     EXPECT_EQ(0, result);
+
+    // Check that we can get the ignored commands by registering a callback
+    std::string command, target;
+    ConstElementPtr params;
+    mccs.setUnhandledCallback(boost::bind(&callback, &command, &target,
+                                          &params, _1, _2, _3));
+    session.addMessage(el("{ \"command\": [ \"good_command\","
+                          "{\"param\": true} ] }"), "Spec1", "*");
+    EXPECT_EQ(1, session.getMsgQueue()->size());
+    result = mccs.checkCommand();
+    EXPECT_EQ(0, session.getMsgQueue()->size());
+    EXPECT_EQ(0, result);
+
+    EXPECT_EQ("good_command", command);
+    EXPECT_EQ("Spec1", target);
+    EXPECT_TRUE(params && el("{\"param\": true}")->equals(*params));
 }
 
 TEST_F(CCSessionTest, initializationFail) {
diff --git a/src/lib/config/tests/fake_session.cc b/src/lib/config/tests/fake_session.cc
index e6d569e732..c2fba57577 100644
--- a/src/lib/config/tests/fake_session.cc
+++ b/src/lib/config/tests/fake_session.cc
@@ -104,7 +104,7 @@ FakeSession::recvmsg(ConstElementPtr& msg, bool nonblock, int) {
     //cout << "[XX] client asks for message " << endl;
     if (messages_ &&
         messages_->getType() == Element::list &&
-        messages_->size() > 0) {
+        !messages_->empty()) {
         msg = messages_->get(0);
         messages_->remove(0);
     } else {
@@ -127,7 +127,7 @@ FakeSession::recvmsg(ConstElementPtr& env, ConstElementPtr& msg, bool nonblock,
     env = ElementPtr();
     if (messages_ &&
         messages_->getType() == Element::list &&
-        messages_->size() > 0) {
+        !messages_->empty()) {
         // do we need initial message to have env[group] and [to] too?
         msg = messages_->get(0);
         messages_->remove(0);
@@ -210,13 +210,13 @@ FakeSession::reply(ConstElementPtr envelope, ConstElementPtr newmsg) {
 
 bool
 FakeSession::hasQueuedMsgs() const {
-    return (msg_queue_ && msg_queue_->size() > 0);
+    return (msg_queue_ && !msg_queue_->empty());
 }
 
 ConstElementPtr
 FakeSession::getFirstMessage(std::string& group, std::string& to) const {
     ConstElementPtr el;
-    if (msg_queue_ && msg_queue_->size() > 0) {
+    if (msg_queue_ && !msg_queue_->empty()) {
         el = msg_queue_->get(0);
         msg_queue_->remove(0);
         group = el->get(0)->stringValue();