| 102 | } |
| 103 | |
| 104 | bool Processor::addConnection(std::shared_ptr<Connectable> conn) { |
| 105 | enum class SetAs{ |
| 106 | NONE, |
| 107 | OUTPUT, |
| 108 | INPUT, |
| 109 | }; |
| 110 | SetAs result = SetAs::NONE; |
| 111 | |
| 112 | if (isRunning()) { |
| 113 | logger_->log_warn("Can not add connection while the process %s is running", name_); |
| 114 | return false; |
| 115 | } |
| 116 | std::shared_ptr<Connection> connection = std::static_pointer_cast<Connection>(conn); |
| 117 | std::lock_guard<std::mutex> lock(getGraphMutex()); |
| 118 | |
| 119 | auto updateGraph = gsl::finally([&] { |
| 120 | if (result == SetAs::INPUT) { |
| 121 | updateReachability(lock); |
| 122 | } else if (result == SetAs::OUTPUT) { |
| 123 | updateReachability(lock, true); |
| 124 | } |
| 125 | }); |
| 126 | |
| 127 | utils::Identifier srcUUID = connection->getSourceUUID(); |
| 128 | utils::Identifier destUUID = connection->getDestinationUUID(); |
| 129 | |
| 130 | if (uuid_ == destUUID) { |
| 131 | // Connection is destination to the current processor |
| 132 | if (_incomingConnections.find(connection) == _incomingConnections.end()) { |
| 133 | _incomingConnections.insert(connection); |
| 134 | connection->setDestination(shared_from_this()); |
| 135 | logger_->log_debug("Add connection %s into Processor %s incoming connection", connection->getName(), name_); |
| 136 | incoming_connections_Iter = this->_incomingConnections.begin(); |
| 137 | result = SetAs::OUTPUT; |
| 138 | } |
| 139 | } |
| 140 | if (uuid_ == srcUUID) { |
| 141 | const auto &rels = connection->getRelationships(); |
| 142 | for (auto i = rels.begin(); i != rels.end(); i++) { |
| 143 | const auto relationship = (*i).getName(); |
| 144 | // Connection is source from the current processor |
| 145 | auto &&it = out_going_connections_.find(relationship); |
| 146 | if (it != out_going_connections_.end()) { |
| 147 | // We already has connection for this relationship |
| 148 | std::set<std::shared_ptr<Connectable>> existedConnection = it->second; |
| 149 | if (existedConnection.find(connection) == existedConnection.end()) { |
| 150 | // We do not have the same connection for this relationship yet |
| 151 | existedConnection.insert(connection); |
| 152 | connection->setSource(shared_from_this()); |
| 153 | out_going_connections_[relationship] = existedConnection; |
| 154 | logger_->log_debug("Add connection %s into Processor %s outgoing connection for relationship %s", connection->getName(), name_, relationship); |
| 155 | result = SetAs::INPUT; |
| 156 | } |
| 157 | } else { |
| 158 | // We do not have any outgoing connection for this relationship yet |
| 159 | std::set<std::shared_ptr<Connectable>> newConnection; |
| 160 | newConnection.insert(connection); |
| 161 | connection->setSource(shared_from_this()); |
no test coverage detected