identify protocol signals (#69)

* identify protocol signals

* clang-patch

Co-authored-by: turuslan <[email protected]>

Author: art-gor

.github/workflows/ci.yml

@@ -53,7 +53,7 @@ jobs:
         GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} # has to be included to access other secrets
         GITHUB_HUNTER_USERNAME: ${{ secrets.GITHUB_HUNTER_USERNAME }}
         GITHUB_HUNTER_TOKEN: ${{ secrets.GITHUB_HUNTER_TOKEN }}
-      run: cmake . -GNinja -Bbuild
+      run: cmake . -GNinja -Bbuild `housekeeping/clang-patch/apply.sh`
     - name: build
       run: cmake --build build -- -j4
     - name: test

.github/workflows/clang-tidy.yml

@@ -34,7 +34,7 @@ jobs:
         LLVM_VER=$(ls -1 ${LLVM_DIR})
         export LLVM_ROOT=${LLVM_DIR}/${LLVM_VER}
         export PATH=${LLVM_ROOT}/bin:${LLVM_ROOT}/share/clang:${PATH}
-        cmake . -GNinja -Bbuild
+        cmake . -GNinja -Bbuild `housekeeping/clang-patch/apply.sh`
         cmake --build build --target generated
         housekeeping/clang-tidy.sh build
 

CMakeLists.txt

@@ -32,6 +32,7 @@ include(cmake/san.cmake)
 set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 
 option(TESTING "Build tests" ON)
+option(EXAMPLES "Build examples" ON)
 option(CLANG_FORMAT "Enable clang-format target" ON)
 option(CLANG_TIDY "Enable clang-tidy checks during compilation" OFF)
 option(COVERAGE "Enable generation of coverage info" OFF)
@@ -93,7 +94,10 @@ include_directories(
 )
 
 add_subdirectory(src)
-add_subdirectory(example)
+
+if(EXAMPLES)
+  add_subdirectory(example)
+endif()
 
 if(TESTING)
   enable_testing()

housekeeping/clang-patch/apply.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+set -e
+if ([ "$RUNNER_OS" = "macOS" ] && which clang++ && ! clang++ -fcoalesce-templates --version) >> null; then
+  dir=$(cd `dirname $0` && pwd)
+  exe="$dir/clang-patch"
+  cc -o $exe "$dir/clang-patch.c"
+  echo "-DCMAKE_CXX_COMPILER=$exe"
+fi

housekeeping/clang-patch/clang-patch.c

@@ -0,0 +1,14 @@
+#include <unistd.h>
+#include <string.h>
+
+int main(int argc, char *argv[]) {
+  char **oa = argv;
+  for (char **ia = argv; *ia; ++ia) {
+    if (strcmp(*ia, "-fcoalesce-templates") != 0) {
+      *oa = *ia;
+      ++oa;
+    }
+  }
+  *oa = NULL;
+  execv("/usr/bin/clang++", argv);
+}

include/libp2p/protocol/identify/identify.hpp

@@ -37,6 +37,9 @@ namespace libp2p::protocol {
 
     ~Identify() override = default;
 
+    boost::signals2::connection onIdentifyReceived(
+        const std::function<IdentifyMessageProcessor::IdentifyCallback> &cb);
+
     /**
      * Get addresses other peers reported we have dialed from
      * @return set of addresses

include/libp2p/protocol/identify/identify_msg_processor.hpp

@@ -17,10 +17,10 @@
 #include <libp2p/host/host.hpp>
 #include <libp2p/multi/multiaddress.hpp>
 #include <libp2p/network/connection_manager.hpp>
+#include <libp2p/outcome/outcome.hpp>
 #include <libp2p/peer/identity_manager.hpp>
 #include <libp2p/peer/peer_id.hpp>
 #include <libp2p/protocol/identify/observed_addresses.hpp>
-#include <libp2p/outcome/outcome.hpp>
 
 namespace identify::pb {
   class Identify;
@@ -35,11 +35,16 @@ namespace libp2p::protocol {
     using StreamSPtr = std::shared_ptr<connection::Stream>;
 
    public:
+    using IdentifyCallback = void(const peer::PeerId &);
+
     IdentifyMessageProcessor(
         Host &host, network::ConnectionManager &conn_manager,
         peer::IdentityManager &identity_manager,
         std::shared_ptr<crypto::marshaller::KeyMarshaller> key_marshaller);
 
+    boost::signals2::connection onIdentifyReceived(
+        const std::function<IdentifyCallback> &cb);
+
     /**
      * Send an Identify message over the provided stream
      * @param stream to be identified over
@@ -131,6 +136,7 @@ namespace libp2p::protocol {
     peer::IdentityManager &identity_manager_;
     std::shared_ptr<crypto::marshaller::KeyMarshaller> key_marshaller_;
     ObservedAddresses observed_addresses_;
+    boost::signals2::signal<IdentifyCallback> signal_identify_received_;
 
     common::Logger log_ = common::createLogger("IdentifyMsgProcessor");
   };

src/peer/address_repository/inmem_address_repository.cpp

@@ -19,7 +19,7 @@ namespace libp2p::peer {
       it = db_.find(p);
     }
 
-    for (auto &m : ma) {
+    for (const auto &m : ma) {
       it->second->insert({m, Clock::now() + ttl});
       signal_added_(p, m);
     }
@@ -38,7 +38,13 @@ namespace libp2p::peer {
 
     auto expires_in = Clock::now() + ttl;
     for (const auto &m : ma) {
-      (*it->second)[m] = expires_in;
+      auto item_it = it->second->find(m);
+      if (item_it == it->second->end()) {
+        it->second->insert({m, expires_in});
+        signal_added_(p, m);
+      } else {
+        item_it->second = expires_in;
+      }
     }
 
     return outcome::success();

src/protocol/identify/identify.cpp

@@ -21,6 +21,11 @@ namespace libp2p::protocol {
     BOOST_ASSERT(msg_processor_);
   }
 
+  boost::signals2::connection Identify::onIdentifyReceived(
+      const std::function<IdentifyMessageProcessor::IdentifyCallback> &cb) {
+    return msg_processor_->onIdentifyReceived(cb);
+  }
+
   std::vector<multi::Multiaddress> Identify::getAllObservedAddresses() const {
     return msg_processor_->getObservedAddresses().getAllAddresses();
   }

src/protocol/identify/identify_msg_processor.cpp

@@ -26,6 +26,11 @@ namespace libp2p::protocol {
     BOOST_ASSERT(key_marshaller_);
   }
 
+  boost::signals2::connection IdentifyMessageProcessor::onIdentifyReceived(
+      const std::function<IdentifyCallback> &cb) {
+    return signal_identify_received_.connect(cb);
+  }
+
   void IdentifyMessageProcessor::sendIdentify(StreamSPtr stream) {
     identify::pb::Identify msg;
 
@@ -170,6 +175,8 @@ namespace libp2p::protocol {
       addresses.push_back(addr);
     }
     consumeListenAddresses(addresses, peer_id);
+
+    signal_identify_received_(peer_id);
   }
 
   boost::optional<peer::PeerId> IdentifyMessageProcessor::consumePublicKey(
@@ -326,15 +333,14 @@ namespace libp2p::protocol {
     // invalidate previously known addresses of that peer
     auto add_res = addr_repo.updateAddresses(peer_id, peer::ttl::kTransient);
     if (!add_res) {
-      log_->error("cannot update listen addresses of the peer {}: {}",
+      log_->debug("cannot update listen addresses of the peer {}: {}",
                   peer_id.toBase58(), add_res.error().message());
     }
 
     // memorize the addresses
     auto addresses = addr_repo.getAddresses(peer_id);
     if (!addresses) {
-      log_->error("can not get addresses for peer {}", peer_id.toBase58());
-      return;
+      log_->debug("can not get addresses for peer {}", peer_id.toBase58());
     }
 
     switch (conn_manager_.connectedness({peer_id, addresses.value()})) {