Merge pull request #7 from msgflo/on-message

Participant: Add dedicated onMessage function

Author: jonnor

include/msgflo.h

@@ -81,6 +81,8 @@ class Message {
     virtual std::string port() = 0;
 };
 
+using MessageHandler = std::function<void(Message *)>;
+
 class Engine;
 
 class Participant {
@@ -93,14 +95,21 @@ class Participant {
 
     virtual void send(std::string port, const char *data, uint64_t len) = 0;
 
+    virtual void onMessage(const MessageHandler &handler) = 0;
+
 private:
 };
 
-using MessageHandler = std::function<void(Message *)>;
-
 class Engine {
 public:
-    virtual Participant *registerParticipant(const Definition &definition, MessageHandler handler) = 0;
+    virtual Participant *registerParticipant(const Definition &definition) = 0;
+
+    // API compatibility with existing signature
+    Participant *registerParticipant(const Definition &definition, MessageHandler handler) {
+        auto p = registerParticipant(definition);
+        p->onMessage(handler);
+        return p;
+    }
 
     virtual void launch() = 0;
 protected:

src/msgflo.cpp

@@ -44,18 +44,31 @@ class DiscoveryMessage {
     Definition definition;
 };
 
+void defaultMessageHandler(msgflo::Message *msg) {
+    cout << "Warning: No message handler defined for msgflo::Participant" << endl;
+}
+
 template<typename Engine_t>
 struct ParticipantRegistrationT : public Participant {
     Engine_t *engine;
     const std::vector<Definition::Port> inports;
     const std::vector<Definition::Port> outports;
     const string id;
-    const MessageHandler handler;
+    MessageHandler handler;
     const DiscoveryMessage discoveryMessage;
 
-    ParticipantRegistrationT(Engine_t *engine, const Definition &definition, const MessageHandler &handler)
-        : engine(engine), inports(definition.inports), outports(definition.outports), id(generateId(definition)),
-          handler(handler), discoveryMessage(definition) {}
+    ParticipantRegistrationT(Engine_t *engine, const Definition &definition)
+        : engine(engine)
+        , inports(definition.inports)
+        , outports(definition.outports)
+        , id(generateId(definition))
+        , handler(defaultMessageHandler)
+        , discoveryMessage(definition)
+    {}
+
+    void onMessage(const MessageHandler &h) {
+        handler = h;
+    }
 
     virtual void send(std::string port, const json11::Json &json) override {
         send(port, json.dump());
@@ -192,9 +205,9 @@ class AmqpEngine final : public Engine, protected AbstractEngine<AmqpEngine> {
         });
     }
 
-    virtual Participant *registerParticipant(const Definition &definition, MessageHandler handler) override {
+    virtual Participant *registerParticipant(const Definition &definition) override {
         Definition d = validateDefinitionFromUser(definition);
-        registrations.emplace_back(this, d, handler);
+        registrations.emplace_back(this, d);
         return &registrations[registrations.size() - 1];
     }
 
@@ -289,9 +302,9 @@ class MosquittoEngine final : public Engine, protected mqtt_event_listener, prot
     virtual ~MosquittoEngine() {
     }
 
-    virtual Participant *registerParticipant(const Definition &definition, MessageHandler handler) override {
+    virtual Participant *registerParticipant(const Definition &definition) override {
         Definition d = validateDefinitionFromUser(definition);
-        registrations.emplace_back(this, d, handler);
+        registrations.emplace_back(this, d);
         return &registrations[registrations.size() - 1];
     }