Merge pull request #351 from singnet/senna-320-2

[#320] Implementation of DAS service bus

Author: andre-senna

src/distributed_algorithm_node/BusNode.cc

@@ -88,6 +88,7 @@ void BusNode::send_bus_command(const string& command, const vector<string>& args
 void BusNode::take_ownership(const set<string>& commands) {
     for (auto command : commands) {
         this->bus.set_ownership(command, node_id());
+        this->my_commands.insert(command);
     }
     broadcast_my_commands();
 }

src/distributed_algorithm_node/BusNode.h

@@ -1,5 +1,4 @@
-#ifndef _DISTRIBUTED_ALGORITHM_NODE_STARNODE_H
-#define _DISTRIBUTED_ALGORITHM_NODE_STARNODE_H
+#pragma once
 
 #include <string>
 
@@ -203,5 +202,3 @@ class SetCommandOwnership : public Message {
 };
 
 }  // namespace distributed_algorithm_node
-
-#endif  // _DISTRIBUTED_ALGORITHM_NODE_STARNODE_H

src/distributed_algorithm_node/StarNode.h

@@ -55,7 +55,7 @@ class StarNode : public DistributedAlgorithmNode {
      *
      * @param node_id ID of the newly inserted node.
      */
-    void node_joined_network(const string& node_id);
+    virtual void node_joined_network(const string& node_id);
 
     /**
      * Method called when a leadershipo election is requested.

src/service_bus/BUILD

@@ -8,7 +8,9 @@ cc_library(
   deps = [
     ":service_bus_singleton",
     ":service_bus",
+    ":bus_command_proxy",
     ":bus_command_processor",
+    "//distributed_algorithm_node:distributed_algorithm_node_lib",
     "//commons:commons_lib",
   ],
 )
@@ -34,11 +36,22 @@ cc_library(
   ],
 )
 
+cc_library(
+  name = "bus_command_proxy",
+  srcs = ["BusCommandProxy.cc"],
+  hdrs = ["BusCommandProxy.h"],
+  deps = [
+    "//distributed_algorithm_node:bus_node",
+    "//commons:commons_lib",
+  ],
+)
+
 cc_library(
   name = "bus_command_processor",
   srcs = ["BusCommandProcessor.cc"],
   hdrs = ["BusCommandProcessor.h"],
   deps = [
+    ":bus_command_proxy",
     "//distributed_algorithm_node:bus_node",
     "//commons:commons_lib",
   ],

src/service_bus/BusCommandProcessor.h

@@ -1,19 +1,26 @@
-#ifndef _SERVICE_BUS_BUSCOMMANDPROCESSOR_H
-#define _SERVICE_BUS_BUSCOMMANDPROCESSOR_H
+#pragma once
 
 #include <set>
 #include <string>
 #include <vector>
 
+#include "BusCommandProxy.h"
+
 using namespace std;
 
 namespace service_bus {
 
 /**
  * Bus element responsible for processing of one or more bus commands.
  *
- * Subclasses must provide concrete implementation of run_command(), which
+ * Concrete subclasses must provide concrete implementation of run_command(), which
  * is the method called when one of the taken commands are issued in the bus.
+ *
+ * Whenever a new command processor is implemented, it's required to implement a
+ * concrete subclass of BusCommandProxy as well (or reuse some other concrete
+ * subclass). Anyway, the processor is required to define which subclass to use as
+ * it must implement the virtual method factory_empty_proxy() which is supposed to
+ * return an empty object of the proper BusCommandProxy.
  */
 class BusCommandProcessor {
     friend class ServiceBus;
@@ -25,12 +32,26 @@ class BusCommandProcessor {
      * @param commands Set of commands owned by the BusCommandProcessor
      */
     BusCommandProcessor(const set<string>& commands);
+
+    /**
+     * Desctructor.
+     */
     virtual ~BusCommandProcessor() {}
 
     // ---------------------------------------------------------------------------------------------
-    // Virtual API which need to be iomplemented in concrete subclasses.
+    // Virtual API which need to be implemented in concrete subclasses.
+
+    /**
+     * Returns an empty instance of the BusCommandProxy required to issue the command.
+     *
+     * @return An empty instance of the BusCommandProxy required to issue the command.
+     */
+    virtual shared_ptr<BusCommandProxy> factory_empty_proxy() = 0;
 
-    virtual void run_command(const string& command, const vector<string>& args) = 0;
+    /**
+     * Method which is called when a command owned by the processor is issued in the bus.
+     */
+    virtual void run_command(shared_ptr<BusCommandProxy> proxy) = 0;
 
    private:
     bool check_command(const string& command);
@@ -39,5 +60,3 @@ class BusCommandProcessor {
 };
 
 }  // namespace service_bus
-
-#endif  // _SERVICE_BUS_BUSCOMMANDPROCESSOR_H

src/service_bus/BusCommandProxy.cc

@@ -0,0 +1,116 @@
+#include "BusCommandProxy.h"
+
+#include "Utils.h"
+
+using namespace service_bus;
+
+string ProxyNode::PROXY_COMMAND = "bus_command_proxy";
+
+// -------------------------------------------------------------------------------------------------
+// Constructors and destructors
+
+BusCommandProxy::BusCommandProxy() {
+    this->proxy_port = 0;
+    this->port_pool = NULL;
+}
+
+BusCommandProxy::BusCommandProxy(const string& command, const vector<string>& args)
+    : command(command), args(args) {}
+
+BusCommandProxy::~BusCommandProxy() {
+    // port_pool is initialized by ServiceBus. It's an static member of a singleton so there's no
+    // need to manage its deletion
+    if ((this->port_pool != NULL) && (this->proxy_port != 0)) {
+        // Return the port to the pool of available ports
+        this->port_pool->enqueue((void*) this->proxy_port);
+    }
+}
+
+ProxyNode::ProxyNode(BusCommandProxy* proxy, const string& node_id) : StarNode(node_id) {
+    // SERVER running in BUS command requestor
+    this->proxy = proxy;
+}
+
+ProxyNode::ProxyNode(BusCommandProxy* proxy, const string& node_id, const string& server_id)
+    : StarNode(node_id, server_id) {
+    // CLIENT running in BUS command processor
+    this->proxy = proxy;
+}
+
+ProxyNode::~ProxyNode() {}
+
+// -------------------------------------------------------------------------------------------------
+// Proxy API
+
+void BusCommandProxy::setup_proxy_node(const string& client_id, const string& server_id) {
+    if ((this->port_pool == NULL) || (this->proxy_port == 0)) {
+        Utils::error("Proxy node can't be set up");
+    } else {
+        if (client_id == "") {
+            // This proxy is running in the requestor
+            string id = this->requestor_id;
+            string requestor_host = id.substr(0, id.find(":"));
+            string requestor_id = requestor_host + ":" + to_string(this->proxy_port);
+            this->proxy_node = new ProxyNode(this, requestor_id);
+        } else {
+            // This proxy is running in the processor
+            this->proxy_node = new ProxyNode(this, client_id, server_id);
+            this->proxy_node->peer_id = server_id;
+        }
+    }
+}
+
+void BusCommandProxy::to_remote_peer(const string& command, const vector<string>& args) {
+    this->proxy_node->to_remote_peer(command, args);
+}
+
+const string& BusCommandProxy::get_command() { return this->command; }
+
+const vector<string>& BusCommandProxy::get_args() { return this->args; }
+
+// -------------------------------------------------------------------------------------------------
+// ProxyNode API
+
+void ProxyNode::remote_call(const string& command, const vector<string>& args) {
+    this->proxy->from_remote_peer(command, args);
+}
+
+void ProxyNode::to_remote_peer(const string& command, const vector<string>& args) {
+    if (this->peer_id == "") {
+        Utils::error("Unknown peer");
+    }
+    vector<string> new_args(args);
+    new_args.push_back(command);
+    send(PROXY_COMMAND, new_args, this->peer_id);
+}
+
+void ProxyNode::node_joined_network(const string& node_id) {
+    StarNode::node_joined_network(node_id);
+    this->peer_id = node_id;
+}
+
+// -------------------------------------------------------------------------------------------------
+// Messages
+
+shared_ptr<Message> ProxyNode::message_factory(string& command, vector<string>& args) {
+    std::shared_ptr<Message> message = DistributedAlgorithmNode::message_factory(command, args);
+    if (message) {
+        return message;
+    }
+    if (command == ProxyNode::PROXY_COMMAND) {
+        vector<string> new_args(args);
+        new_args.pop_back();
+        return std::shared_ptr<Message>(new ProxyMessage(args.back(), new_args));
+    }
+    return std::shared_ptr<Message>{};
+}
+
+ProxyMessage::ProxyMessage(const string& command, const vector<string>& args) {
+    this->command = command;
+    this->args = args;
+}
+
+void ProxyMessage::act(shared_ptr<MessageFactory> node) {
+    auto proxy_node = dynamic_pointer_cast<ProxyNode>(node);
+    proxy_node->remote_call(this->command, this->args);
+}