Implements #3 net auto reconnect

CMakeLists.txt

@@ -134,7 +134,7 @@ check_include_file_cxx("opencv2/cudastereo.hpp" HAVE_OPENCVCUDA)
 find_program(CPPCHECK_FOUND cppcheck)
 if (CPPCHECK_FOUND)
 	message(STATUS "Found cppcheck: will perform source checks")
-	set(CMAKE_CXX_CPPCHECK "cppcheck" "--enable=warning,performance,portability,style" "--inline-suppr" "--std=c++11" "--suppress=*:*catch.hpp" "--suppress=*:*elas*" "--suppress=*:*json.hpp" "--quiet")
+	set(CMAKE_CXX_CPPCHECK "cppcheck" "--enable=warning,performance,portability,style" "--inline-suppr" "--std=c++11" "--suppress=*:*catch.hpp" "--suppress=*:*elas*" "--suppress=*:*nanogui*" "--suppress=*:*json.hpp" "--quiet")
 endif()
 
 # include_directories(${PROJECT_SOURCE_DIR}/common/cpp/include)

applications/gui/src/main.cpp

@@ -59,7 +59,8 @@ int main(int argc, char **argv) {
 	auto root = ftl::configure(argc, argv, "gui_default");
 	ftl::net::Universe *net = ftl::create<ftl::net::Universe>(root, "net");
 
-	net->waitConnections();
+	net->start();
+	//net->waitConnections();
 
 	ftl::ctrl::Master controller(root, net);
 	controller.onLog([](const ftl::ctrl::LogEvent &e){

applications/gui/src/src_window.cpp

@@ -166,11 +166,16 @@ SourceWindow::SourceWindow(nanogui::Widget *parent, ftl::ctrl::Master *ctrl)
                                        Alignment::Middle, 0, 6));
 
     new Label(tools, "Select source","sans-bold");
-    auto available = ctrl->getNet()->findAll<string>("list_streams");
-    auto select = new ComboBox(tools, available);
-    select->setCallback([this,available](int ix) {
+    available_ = ctrl->getNet()->findAll<string>("list_streams");
+    auto select = new ComboBox(tools, available_);
+    select->setCallback([this,select](int ix) {
         LOG(INFO) << "Change source: " << ix;
-        src_->set("uri", available[ix]);
+        src_->set("uri", available_[ix]);
+    });
+
+	ctrl->getNet()->onConnect([this,select](ftl::net::Peer *p) {
+		 available_ = ctrl_->getNet()->findAll<string>("list_streams");
+		 select->setItems(available_);
 	});
 
 	auto depth = new Button(tools, "Depth");

applications/gui/src/src_window.hpp

@@ -5,6 +5,8 @@
 #include <ftl/master.hpp>
 #include <ftl/uuid.hpp>
 #include <ftl/net_source.hpp>
+#include <vector>
+#include <string>
 
 class VirtualCameraView;
 
@@ -23,6 +25,7 @@ class SourceWindow : public nanogui::Window {
 	ftl::ctrl::Master *ctrl_;
 	ftl::rgbd::NetSource *src_;
 	VirtualCameraView *image_;
+	std::vector<std::string> available_;
 
 };
 

applications/reconstruct/include/ftl/matrix_conversion.hpp

@@ -30,7 +30,7 @@ namespace MatrixConversion
 		return Eigen::Matrix4f(mat.ptr()).transpose();
 	}*/
 
-	static Eigen::Vector4f VecH(const Eigen::Vector3f& v)
+	/*static Eigen::Vector4f VecH(const Eigen::Vector3f& v)
 	{
 		return Eigen::Vector4f(v[0], v[1], v[2], 1.0);
 	}
@@ -38,7 +38,7 @@ namespace MatrixConversion
 	static Eigen::Vector3f VecDH(const Eigen::Vector4f& v)
 	{
 		return Eigen::Vector3f(v[0]/v[3], v[1]/v[3], v[2]/v[3]);
-	}
+	}*/
 
 	/*static Eigen::Vector3f VecToEig(const vec3f& v)
 	{

applications/reconstruct/include/ftl/registration.hpp

@@ -44,7 +44,7 @@ bool findChessboardCorners(cv::Mat &rgb, const cv::Mat &depth, const ftl::rgbd::
  */
 class Registration : public ftl::Configurable {
 public:
-	Registration(nlohmann::json &config);
+	explicit Registration(nlohmann::json &config);
 	void addSource(ftl::rgbd::RGBDSource* source);
 	size_t getSourcesCount() { return sources_.size(); }
 
@@ -161,7 +161,7 @@ private:
  */
 class ChessboardRegistration : public Registration {
 public:
-	ChessboardRegistration(nlohmann::json &config);
+	explicit ChessboardRegistration(nlohmann::json &config);
 	/** 
 	 * @brief	Creates new ChessboardRegistration or ChessboardRegistrationChain
 	 * 			object depending on chain option in config. User of the method
@@ -189,7 +189,7 @@ protected:
  */
 class ChessboardRegistrationChain : public ChessboardRegistration {
 public:
-	ChessboardRegistrationChain(nlohmann::json &config);
+	explicit ChessboardRegistrationChain(nlohmann::json &config);
 	
 	bool findTransformations(std::vector<Eigen::Matrix4f> &data) override;
 

applications/reconstruct/include/ftl/scene_rep_hash_sdf.hpp

@@ -195,7 +195,7 @@ class SceneRep : public ftl::Configurable {
 
 
 		std::unordered_set<unsigned int> pointersFreeHash;
-		std::vector<unsigned int> pointersFreeVec(m_hashParams.m_numSDFBlocks, 0);
+		std::vector<int> pointersFreeVec(m_hashParams.m_numSDFBlocks, 0);  // CHECK Nick Changed to int from unsigned in
 		for (unsigned int i = 0; i < heapCounterCPU; i++) {
 			pointersFreeHash.insert(heapCPU[i]);
 			pointersFreeVec[heapCPU[i]] = FREE_ENTRY;
@@ -207,7 +207,7 @@ class SceneRep : public ftl::Configurable {
 
 		unsigned int numOccupied = 0;
 		unsigned int numMinusOne = 0;
-		unsigned int listOverallFound = 0;
+		//unsigned int listOverallFound = 0;
 
 		std::list<myint3Voxel> l;
 		//std::vector<myint3Voxel> v;

applications/reconstruct/src/main.cpp

@@ -75,9 +75,7 @@ struct Cameras {
 static void run(ftl::Configurable *root) {
 	Universe *net = ftl::create<Universe>(root, "net");
 	
-	// Make sure connections are complete
-	// sleep_for(milliseconds(500));
-	// TODO: possible to do more reliably?
+	net->start();
 	net->waitConnections();
 	
 	std::vector<Cameras> inputs;

applications/reconstruct/src/virtual_source.cpp

@@ -58,5 +58,5 @@ void VirtualSource::grab() {
 }
 
 bool VirtualSource::isReady() {
-
+	return true;
 }

applications/vision/CMakeLists.txt

@@ -6,7 +6,6 @@
 
 set(CVNODESRC
 	src/main.cpp
-	src/streamer.cpp
 	src/middlebury.cpp
 )
 

applications/vision/src/main.cpp

@@ -96,6 +96,7 @@ static void run(ftl::Configurable *root) {
 	stream->add(source);
 	stream->run();
 
+	net->start();
 	LOG(INFO) << "Running...";
 
 	while (ftl::running && display->active()) {

components/common/cpp/include/ftl/configuration.hpp

@@ -68,12 +68,25 @@ Configurable *find(const std::string &uri);
  */
 void registerConfigurable(Configurable *cfg);
 
+/**
+ * Create a new configurable directly from a raw object. This should not be used.
+ */
 template <typename T, typename... ARGS>
 T *create(json_t &link, ARGS ...args);
 
+/**
+ * Create a configurable from an attribute of a parent configurable.
+ */
 template <typename T, typename... ARGS>
 T *create(ftl::Configurable *parent, const std::string &name, ARGS ...args);
 
+/**
+ * Create a configurable rooted on a parent but with a specific object
+ * that is not directly a child of the parent. Used by RGB-D Factory.
+ */
+template <typename T, typename... ARGS>
+T *create(ftl::Configurable *parent, json_t &obj, const std::string &name, ARGS ...args);
+
 void set(const std::string &uri, const nlohmann::json &);
 
 }  // namespace config
@@ -138,12 +151,38 @@ T *ftl::config::create(ftl::Configurable *parent, const std::string &name, ARGS
             id_str = id_str + std::string("#") + name;
         }
         parent->getConfig()[name] = {
+			// cppcheck-suppress constStatement
             {"$id", id_str}
         };
 
         nlohmann::json &entity2 = parent->getConfig()[name];
         return create<T>(entity2, args...);
     }
+
+	LOG(ERROR) << "Unable to create Configurable entity '" << name << "'";
+	return nullptr;
+}
+
+template <typename T, typename... ARGS>
+T *ftl::config::create(ftl::Configurable *parent, json_t &obj, const std::string &name, ARGS ...args) {
+    //nlohmann::json &entity = ftl::config::resolve(parent->getConfig()[name]);
+    nlohmann::json &entity = obj;
+
+    if (entity.is_object()) {
+        if (!entity["$id"].is_string()) {
+            std::string id_str = *parent->get<std::string>("$id");
+            if (id_str.find('#') != std::string::npos) {
+                entity["$id"] = id_str + std::string("/") + name;
+            } else {
+                entity["$id"] = id_str + std::string("#") + name;
+            }
+        }
+
+        return create<T>(entity, args...);
+    } else if (entity.is_null()) {
+        LOG(FATAL) << "Invalid raw object in Configurable construction";
+        return nullptr;
+    }
 }
 
 #endif  // _FTL_COMMON_CONFIGURATION_HPP_

components/common/cpp/src/opencv_to_pcl.cpp

@@ -24,6 +24,7 @@ pcl::PointCloud<pcl::PointXYZRGB>::Ptr ftl::utility::matToPointXYZ(const cv::Mat
 
 		// when color needs to be added:
 		uint32_t rgb = (static_cast<uint32_t>(prgb.z) << 16 | static_cast<uint32_t>(prgb.y) << 8 | static_cast<uint32_t>(prgb.x));
+		// cppcheck-suppress invalidPointerCast
 		point.rgb = *reinterpret_cast<float*>(&rgb);
 
 		point_cloud_ptr -> points.push_back(point);

components/common/cpp/test/configurable_unit.cpp

@@ -8,6 +8,7 @@ using std::string;
 
 SCENARIO( "Configurable::get()" ) {
 	GIVEN( "a non-existent property" ) {
+		// cppcheck-suppress constStatement
 		nlohmann::json json = {{"test",5}};
 		Configurable cfg(json);
 
@@ -16,6 +17,7 @@ SCENARIO( "Configurable::get()" ) {
 	}
 
 	GIVEN( "a valid property" ) {
+		// cppcheck-suppress constStatement
 		nlohmann::json json = {{"test",5}};
 		Configurable cfg(json);
 
@@ -27,6 +29,7 @@ SCENARIO( "Configurable::get()" ) {
 
 SCENARIO( "Configurable::on()" ) {
 	GIVEN( "a changed property with no handlers" ) {
+		// cppcheck-suppress constStatement
 		nlohmann::json json = {{"test",5}};
 		Configurable cfg(json);
 
@@ -36,6 +39,7 @@ SCENARIO( "Configurable::on()" ) {
 	}
 
 	GIVEN( "a changed property one handler" ) {
+		// cppcheck-suppress constStatement
 		nlohmann::json json = {{"test",5}};
 		Configurable cfg(json);
 		bool trig = false;
@@ -50,6 +54,7 @@ SCENARIO( "Configurable::on()" ) {
 	}
 
 	GIVEN( "a changed property two handlers" ) {
+		// cppcheck-suppress constStatement
 		nlohmann::json json = {{"test",5}};
 		Configurable cfg(json);
 		bool trig1 = false;

components/net/cpp/include/ftl/net/peer.hpp

@@ -68,8 +68,8 @@ class Peer {
 	};
 
 	public:
-	explicit Peer(const char *uri, ftl::net::Dispatcher *d=nullptr);
-	explicit Peer(SOCKET s, ftl::net::Dispatcher *d=nullptr);
+	explicit Peer(const char *uri, ftl::net::Universe *, ftl::net::Dispatcher *d=nullptr);
+	explicit Peer(SOCKET s, ftl::net::Universe *, ftl::net::Dispatcher *d=nullptr);
 	~Peer();
 	
 	/**
@@ -94,6 +94,11 @@ class Peer {
 	 */
 	bool waitConnection();
 
+	/**
+	 * Make a reconnect attempt. Called internally by Universe object.
+	 */
+	bool reconnect();
+	
 	/**
 	 * Test if the connection is valid. This returns true in all conditions
 	 * except where the socket has been disconnected permenantly or was never
@@ -176,8 +181,8 @@ class Peer {
 	void bind(const std::string &name, F func);
 
 	// void onError(std::function<void(Peer &, int err, const char *msg)> &f) {}
-	void onConnect(const std::function<void(Peer &)> &f);
-	void onDisconnect(std::function<void(Peer &)> &f) {}
+	//void onConnect(const std::function<void(Peer &)> &f);
+	//void onDisconnect(std::function<void(Peer &)> &f) {}
 	
 	bool isWaiting() const { return is_waiting_; }
 
@@ -226,11 +231,13 @@ class Peer {
 		}
 	}
 
-	private: // Data
-	Status status_;
-	SOCKET sock_;
-	ftl::URI::scheme_t scheme_;
-	uint32_t version_;
+	private:
+	Status status_;					// Connected, errored, reconnecting...
+	SOCKET sock_;					// Raw OS socket
+	ftl::URI::scheme_t scheme_;		// TCP / WS / UDP...
+	uint32_t version_;				// Received protocol version in handshake
+	bool can_reconnect_;			// Client connections can retry
+	ftl::net::Universe *universe_;	// Origin net universe
 	
 	// Receive buffers
 	bool is_waiting_;
@@ -241,16 +248,16 @@ class Peer {
 	msgpack::vrefbuffer send_buf_;
 	std::mutex send_mtx_;
 	
-	std::string uri_;
-	ftl::UUID peerid_;
+	std::string uri_;				// Original connection URI, or assumed URI
+	ftl::UUID peerid_;				// Received in handshake or allocated
 	
-	ftl::net::Dispatcher *disp_;
-	std::vector<std::function<void(Peer &)>> open_handlers_;
+	ftl::net::Dispatcher *disp_;	// For RPC call dispatch
+	//std::vector<std::function<void(Peer &)>> open_handlers_;
 	//std::vector<std::function<void(const ftl::net::Error &)>> error_handlers_
-	std::vector<std::function<void(Peer &)>> close_handlers_;
+	//std::vector<std::function<void(Peer &)>> close_handlers_;
 	std::map<int, std::unique_ptr<virtual_caller>> callbacks_;
 	
-	static int rpcid__;
+	static int rpcid__;				// Return ID for RPC calls
 };
 
 // --- Inline Template Implementations -----------------------------------------

components/net/cpp/include/ftl/net/universe.hpp

@@ -12,9 +12,10 @@
 #include <ftl/uuid.hpp>
 #include <nlohmann/json.hpp>
 #include <vector>
+#include <list>
 #include <string>
 #include <thread>
-#include <mutex>
+#include <shared_mutex>
 #include <map>
 
 namespace ftl {
@@ -24,6 +25,14 @@ struct Error {
 	int errno;
 };
 
+struct ReconnectInfo {
+	int tries;
+	float delay;
+	Peer *peer;
+};
+
+typedef unsigned int callback_t;
+
 /**
  * Represents a group of network peers and their resources, managing the
  * searching of and sharing of resources across peers. Each universe can
@@ -34,6 +43,9 @@ struct Error {
  * their actions.
  */
 class Universe : public ftl::Configurable {
+	public:
+	friend class Peer;
+
 	public:
 	Universe();
 	/**
@@ -49,6 +61,8 @@ class Universe : public ftl::Configurable {
 	 */
 	~Universe();
 
+	void start();
+	
 	/**
 	 * Open a new listening port on a given interfaces.
 	 *   eg. "tcp://localhost:9000"
@@ -82,10 +96,10 @@ class Universe : public ftl::Configurable {
 	
 	Peer *getPeer(const ftl::UUID &pid) const;
 	
-	int numberOfSubscribers(const std::string &res) const;
+	//int numberOfSubscribers(const std::string &res) const;
 
-	bool hasSubscribers(const std::string &res) const;
-	bool hasSubscribers(const ftl::URI &res) const;
+	//bool hasSubscribers(const std::string &res) const;
+	//bool hasSubscribers(const ftl::URI &res) const;
 	
 	/**
 	 * Bind a function to an RPC or service call name. This will implicitely
@@ -101,16 +115,18 @@ class Universe : public ftl::Configurable {
 	 * triggered whenever that resource is published to. It is akin to
 	 * RPC broadcast (no return value) to a subgroup of peers.
 	 */
-	template <typename F>
-	bool subscribe(const std::string &res, F func);
+	//template <typename F>
+	//[[deprecated("Pub sub no longer to be used")]]
+	//bool subscribe(const std::string &res, F func);
 
 	/**
 	 * Subscribe a function to a resource. The subscribed function is
 	 * triggered whenever that resource is published to. It is akin to
 	 * RPC broadcast (no return value) to a subgroup of peers.
 	 */
-	template <typename F>
-	bool subscribe(const ftl::URI &res, F func);
+	//template <typename F>
+	//[[deprecated("Pub sub no longer to be used")]]
+	//bool subscribe(const ftl::URI &res, F func);
 	
 	/**
 	 * Send a non-blocking RPC call with no return value to all connected
@@ -136,75 +152,78 @@ class Universe : public ftl::Configurable {
 	 * of a resource. There may be no subscribers. Note that query parameter
 	 * order in the URI string is not important.
 	 */
+	//template <typename... ARGS>
 	//[[deprecated("Pub sub no longer to be used")]]
-	template <typename... ARGS>
-	void publish(const std::string &res, ARGS... args);
+	//void publish(const std::string &res, ARGS... args);
 
 	/**
 	 * Send a non-blocking RPC call with no return value to all subscribers
 	 * of a resource. There may be no subscribers. This overload accepts a
 	 * URI object directly to enable more efficient modification of parameters.
 	 */
+	//template <typename... ARGS>
 	//[[deprecated("Pub sub no longer to be used")]]
-	template <typename... ARGS>
-	void publish(const ftl::URI &res, ARGS... args);
+	//void publish(const ftl::URI &res, ARGS... args);
 	
 	/**
 	 * Register your ownership of a new resource. This must be called before
 	 * publishing to this resource and before any peers attempt to subscribe.
 	 */
 	//[[deprecated("Pub sub no longer to be used")]]
-	bool createResource(const std::string &uri);
+	//bool createResource(const std::string &uri);
 
 	//[[deprecated("Pub sub no longer to be used")]]
-	std::optional<ftl::UUID> findOwner(const std::string &res);
+	//std::optional<ftl::UUID> findOwner(const std::string &res);
 
 	void setLocalID(const ftl::UUID &u) { this_peer = u; };
 	const ftl::UUID &id() const { return this_peer; }
 
 	// --- Event Handlers ------------------------------------------------------
 
-	void onConnect(const std::string &, std::function<void(ftl::net::Peer*)>);
-	void onDisconnect(const std::string &, std::function<void(ftl::net::Peer*)>);
-	void onError(const std::string &, std::function<void(ftl::net::Peer*, const ftl::net::Error &)>);
+	ftl::net::callback_t onConnect(const std::function<void(ftl::net::Peer*)>&);
+	ftl::net::callback_t onDisconnect(const std::function<void(ftl::net::Peer*)>&);
+	ftl::net::callback_t onError(const std::function<void(ftl::net::Peer*, const ftl::net::Error &)>&);
 
-	void removeCallbacks(const std::string &);
+	void removeCallback(ftl::net::callback_t cbid);
 	
 	private:
 	void _run();
 	int _setDescriptors();
 	void _installBindings();
 	void _installBindings(Peer *);
-	bool _subscribe(const std::string &res);
+	//bool _subscribe(const std::string &res);
 	void _cleanupPeers();
 	void _notifyConnect(Peer *);
 	void _notifyDisconnect(Peer *);
 	void _notifyError(Peer *, const ftl::net::Error &);
+	void _periodic();
 	
 	static void __start(Universe *u);
 	
 	private:
 	bool active_;
 	ftl::UUID this_peer;
-	std::mutex net_mutex_;
+	std::shared_mutex net_mutex_;
+	std::shared_mutex handler_mutex_;
 	fd_set sfderror_;
 	fd_set sfdread_;
 	std::vector<ftl::net::Listener*> listeners_;
 	std::vector<ftl::net::Peer*> peers_;
-	std::map<std::string, std::vector<ftl::UUID>> subscribers_;
-	std::unordered_set<std::string> owned_;
+	//std::map<std::string, std::vector<ftl::UUID>> subscribers_;
+	//std::unordered_set<std::string> owned_;
 	std::map<ftl::UUID, ftl::net::Peer*> peer_ids_;
 	ftl::UUID id_;
 	ftl::net::Dispatcher disp_;
 	std::thread thread_;
+	std::list<ReconnectInfo> reconnects_;
 
 	struct ConnHandler {
-		std::string name;
+		callback_t id;
 		std::function<void(ftl::net::Peer*)> h;
 	};
 
 	struct ErrHandler {
-		std::string name;
+		callback_t id;
 		std::function<void(ftl::net::Peer*, const ftl::net::Error &)> h;
 	};
 
@@ -213,6 +232,8 @@ class Universe : public ftl::Configurable {
 	std::list<ConnHandler> on_disconnect_;
 	std::list<ErrHandler> on_error_;
 
+	static callback_t cbid__;
+
 	// std::map<std::string, std::vector<ftl::net::Peer*>> subscriptions_;
 };
 
@@ -220,13 +241,13 @@ class Universe : public ftl::Configurable {
 
 template <typename F>
 void Universe::bind(const std::string &name, F func) {
-	// CHECK Need mutex?
+	std::unique_lock<std::shared_mutex> lk(net_mutex_);
 	disp_.bind(name, func,
 		typename ftl::internal::func_kind_info<F>::result_kind(),
 	    typename ftl::internal::func_kind_info<F>::args_kind());
 }
 
-template <typename F>
+/*template <typename F>
 bool Universe::subscribe(const std::string &res, F func) {
 	return subscribe(ftl::URI(res), func);
 }
@@ -235,12 +256,13 @@ template <typename F>
 bool Universe::subscribe(const ftl::URI &res, F func) {
 	bind(res.to_string(), func);
 	return _subscribe(res.to_string());
-}
+}*/
 
 template <typename... ARGS>
 void Universe::broadcast(const std::string &name, ARGS... args) {
+	std::shared_lock<std::shared_mutex> lk(net_mutex_);
 	for (auto p : peers_) {
-		p->send(name, args...);
+		if (p->isConnected()) p->send(name, args...);
 	}
 }
 
@@ -262,15 +284,18 @@ std::optional<R> Universe::findOne(const std::string &name, ARGS... args) {
 	};
 
 	std::map<Peer*, int> record;
+	std::shared_lock<std::shared_mutex> lk(net_mutex_);
 	for (auto p : peers_) {
-		record[p] = p->asyncCall<std::optional<R>>(name, handler, args...);
+		if (p->isConnected()) record[p] = p->asyncCall<std::optional<R>>(name, handler, args...);
 	}
+	lk.unlock();
 	
 	{  // Block thread until async callback notifies us
-		std::unique_lock<std::mutex> lk(m);
-		cv.wait_for(lk, std::chrono::seconds(1), [&hasreturned]{return hasreturned;});
+		std::unique_lock<std::mutex> llk(m);
+		cv.wait_for(llk, std::chrono::seconds(1), [&hasreturned]{return hasreturned;});
 
 		// Cancel any further results
+		lk.lock();
 		for (auto p : peers_) {
 			auto m = record.find(p);
 			if (m != record.end()) {
@@ -300,16 +325,20 @@ std::vector<R> Universe::findAll(const std::string &name, ARGS... args) {
 	};
 
 	std::map<Peer*, int> record;
+	std::shared_lock<std::shared_mutex> lk(net_mutex_);
 	for (auto p : peers_) {
+		if (!p->isConnected()) continue;
 		sentcount++;
 		record[p] = p->asyncCall<std::vector<R>>(name, handler, args...);
 	}
+	lk.unlock();
 	
 	{  // Block thread until async callback notifies us
-		std::unique_lock<std::mutex> lk(m);
-		cv.wait_for(lk, std::chrono::seconds(1), [&returncount,&sentcount]{return returncount == sentcount;});
+		std::unique_lock<std::mutex> llk(m);
+		cv.wait_for(llk, std::chrono::seconds(1), [&returncount,&sentcount]{return returncount == sentcount;});
 
 		// Cancel any further results
+		lk.lock();
 		for (auto p : peers_) {
 			auto m = record.find(p);
 			if (m != record.end()) {
@@ -324,7 +353,7 @@ std::vector<R> Universe::findAll(const std::string &name, ARGS... args) {
 template <typename R, typename... ARGS>
 R Universe::call(const ftl::UUID &pid, const std::string &name, ARGS... args) {
 	Peer *p = getPeer(pid);
-	if (p == nullptr) {
+	if (p == nullptr || !p->isConnected()) {
 		DLOG(WARNING) << "Attempting to call an unknown peer : " << pid.to_string();
 		throw -1;
 	}
@@ -336,12 +365,12 @@ bool Universe::send(const ftl::UUID &pid, const std::string &name, ARGS... args)
 	Peer *p = getPeer(pid);
 	if (p == nullptr) {
 		DLOG(WARNING) << "Attempting to call an unknown peer : " << pid.to_string();
-		throw -1;
+		return false;
 	}
-	return p->send(name, args...) > 0;
+	return p->isConnected() && p->send(name, args...) > 0;
 }
 
-template <typename... ARGS>
+/*template <typename... ARGS>
 void Universe::publish(const std::string &res, ARGS... args) {
 	ftl::URI uri(res);
 	publish(uri, args...);
@@ -349,7 +378,7 @@ void Universe::publish(const std::string &res, ARGS... args) {
 
 template <typename... ARGS>
 void Universe::publish(const ftl::URI &res, ARGS... args) {
-	std::unique_lock<std::mutex> lk(net_mutex_);
+	std::unique_lock<std::shared_mutex> lk(net_mutex_);
 	auto subs = subscribers_[res.getBaseURI()];
 	lk.unlock();
 	for (auto p : subs) {
@@ -358,7 +387,7 @@ void Universe::publish(const ftl::URI &res, ARGS... args) {
 			peer->send(res.getBaseURI(), args...);
 		}
 	}
-}
+}*/
 
 };  // namespace net
 };  // namespace ftl

components/net/cpp/src/dispatcher.cpp

@@ -140,7 +140,7 @@ void ftl::net::Dispatcher::dispatch_notification(Peer &s, msgpack::object const
 			throw &e;
 		}
     } else {
-    	LOG(ERROR) << "Missing handler for incoming message";
+    	LOG(ERROR) << "Missing handler for incoming message (" << name << ")";
     }
 }
 

components/net/cpp/src/peer.cpp

@@ -23,6 +23,7 @@
 #include <ftl/net/ws_internal.hpp>
 #include <ftl/config.h>
 #include "net_internal.hpp"
+#include <ftl/net/universe.hpp>
 
 #include <iostream>
 #include <memory>
@@ -37,6 +38,10 @@ using ftl::URI;
 using ftl::net::ws_connect;
 using ftl::net::Dispatcher;
 using std::chrono::seconds;
+using ftl::net::Universe;
+using ftl::net::callback_t;
+using std::mutex;
+using std::unique_lock;
 
 /*static std::string hexStr(const std::string &s)
 {
@@ -127,7 +132,7 @@ static SOCKET tcpConnect(URI &uri) {
 	return csocket;
 }
 
-Peer::Peer(SOCKET s, Dispatcher *d) : sock_(s) {
+Peer::Peer(SOCKET s, Universe *u, Dispatcher *d) : sock_(s), can_reconnect_(false), universe_(u) {
 	status_ = (s == INVALID_SOCKET) ? kInvalid : kConnecting;
 	_updateURI();
 	
@@ -149,7 +154,10 @@ Peer::Peer(SOCKET s, Dispatcher *d) : sock_(s) {
 				peerid_ = pid;
 				if (version != ftl::net::kVersion) LOG(WARNING) << "Net protocol using different versions!";
 				
-				_trigger(open_handlers_);
+				// Ensure handlers called later or in new thread
+				pool.push([this](int id) {
+					universe_->_notifyConnect(this);
+				});
 			}
 		});
 
@@ -158,11 +166,15 @@ Peer::Peer(SOCKET s, Dispatcher *d) : sock_(s) {
 			LOG(INFO) << "Peer elected to disconnect: " << id().to_string();
 		});
 
+		bind("__ping__", [this](unsigned long long timestamp) {
+			return timestamp;
+		});
+
 		send("__handshake__", ftl::net::kMagic, ftl::net::kVersion, ftl::net::this_peer); 
 	}
 }
 
-Peer::Peer(const char *pUri, Dispatcher *d) : uri_(pUri) {	
+Peer::Peer(const char *pUri, Universe *u, Dispatcher *d) : can_reconnect_(true), universe_(u), uri_(pUri) {	
 	URI uri(pUri);
 	
 	status_ = kInvalid;
@@ -170,6 +182,9 @@ Peer::Peer(const char *pUri, Dispatcher *d) : uri_(pUri) {
 	
 	disp_ = new Dispatcher(d);
 
+	// Must to to prevent receiving message before handlers are installed
+	unique_lock<mutex> lk(recv_mtx_);
+
 	scheme_ = uri.getProtocol();
 	if (uri.getProtocol() == URI::SCHEME_TCP) {
 		sock_ = tcpConnect(uri);
@@ -194,7 +209,7 @@ Peer::Peer(const char *pUri, Dispatcher *d) : uri_(pUri) {
 	
 	is_waiting_ = true;
 	
-	if (status_ == kConnecting) {
+	if (status_ == kConnecting || status_ == kReconnecting) {
 		// Install return handshake handler.
 		bind("__handshake__", [this](uint64_t magic, uint32_t version, UUID pid) {
 			LOG(INFO) << "Handshake 1 received";
@@ -208,12 +223,46 @@ Peer::Peer(const char *pUri, Dispatcher *d) : uri_(pUri) {
 				if (version != ftl::net::kVersion) LOG(WARNING) << "Net protocol using different versions!";
 				send("__handshake__", ftl::net::kMagic, ftl::net::kVersion, ftl::net::this_peer);
 				
-				_trigger(open_handlers_);
+				// Ensure handlers called later or in new thread
+				pool.push([this](int id) {
+					universe_->_notifyConnect(this);
+				});
 			}
 		}); 
+
+		bind("__disconnect__", [this]() {
+			_badClose(false);
+			LOG(INFO) << "Peer elected to disconnect: " << id().to_string();
+		});
+
+		bind("__ping__", [this](unsigned long long timestamp) {
+			return timestamp;
+		});
 	}
 }
 
+bool Peer::reconnect() {
+	if (status_ != kReconnecting || !can_reconnect_) return false;
+
+	URI uri(uri_);
+
+	LOG(INFO) << "Reconnecting to " << uri_ << " ...";
+
+	if (scheme_ == URI::SCHEME_TCP) {
+		sock_ = tcpConnect(uri);
+		if (sock_ != INVALID_SOCKET) {
+			status_ = kConnecting;
+			is_waiting_ = true;
+			return true;
+		} else {
+			return false;
+		}
+	}
+
+	// TODO(Nick) allow for other protocols in reconnect
+	return false;
+}
+
 void Peer::_updateURI() {
 	sockaddr_storage addr;
 	int rsize = sizeof(sockaddr_storage);
@@ -265,10 +314,12 @@ void Peer::_badClose(bool retry) {
 		//auto i = find(sockets.begin(),sockets.end(),this);
 		//sockets.erase(i);
 
-		_trigger(close_handlers_);
+		universe_->_notifyDisconnect(this);
 
 		// Attempt auto reconnect?
-		if (retry) LOG(INFO) << "Should attempt reconnect...";
+		if (retry && can_reconnect_) {
+			status_ = kReconnecting;
+		}
 	}
 }
 
@@ -324,7 +375,7 @@ bool Peer::_data() {
 				
 				if (get<1>(hs) != "__handshake__") {
 					_badClose(false);
-					LOG(ERROR) << "Missing handshake";
+					LOG(ERROR) << "Missing handshake - got '" << get<1>(hs) << "'";
 					return false;
 				}
 			} catch(...) {
@@ -368,24 +419,28 @@ void Peer::_sendResponse(uint32_t id, const msgpack::object &res) {
 bool Peer::waitConnection() {
 	if (status_ == kConnected) return true;
 	
-	std::unique_lock<std::mutex> lk(send_mtx_);
+	std::mutex m;
+	std::unique_lock<std::mutex> lk(m);
 	std::condition_variable cv;
 
-	onConnect([&](Peer &p) {
-		cv.notify_all();
+	callback_t h = universe_->onConnect([this,&cv](Peer *p) {
+		if (p == this) {
+			cv.notify_one();
+		}
 	});
 
 	cv.wait_for(lk, seconds(5));
+	universe_->removeCallback(h);
 	return status_ == kConnected;
 }
 
-void Peer::onConnect(const std::function<void(Peer&)> &f) {
+/*void Peer::onConnect(const std::function<void(Peer&)> &f) {
 	if (status_ == kConnected) {
 		f(*this);
 	} else {
 		open_handlers_.push_back(f);
 	}
-}
+}*/
 
 void Peer::_connected() {
 	status_ = kConnected;

components/net/cpp/src/universe.cpp

@@ -20,8 +20,13 @@ using nlohmann::json;
 using ftl::UUID;
 using std::optional;
 using std::unique_lock;
+using std::shared_lock;
 using std::mutex;
+using std::shared_mutex;
 using ftl::config::json_t;
+using ftl::net::callback_t;
+
+callback_t ftl::net::Universe::cbid__ = 0;
 
 Universe::Universe() : Configurable(), active_(true), this_peer(ftl::net::this_peer), thread_(Universe::__start, this) {
 	_installBindings();
@@ -30,6 +35,14 @@ Universe::Universe() : Configurable(), active_(true), this_peer(ftl::net::this_p
 Universe::Universe(nlohmann::json &config) :
 		Configurable(config), active_(true), this_peer(ftl::net::this_peer), thread_(Universe::__start, this) {
 
+	_installBindings();
+}
+
+Universe::~Universe() {
+	shutdown();
+}
+
+void Universe::start() {
 	auto l = get<json_t>("listen");
 
 	if (l && (*l).is_array()) {
@@ -46,12 +59,6 @@ Universe::Universe(nlohmann::json &config) :
 			connect(pp);
 		}
 	}
-	
-	_installBindings();
-}
-
-Universe::~Universe() {
-	shutdown();
 }
 
 void Universe::shutdown() {
@@ -77,32 +84,26 @@ void Universe::shutdown() {
 bool Universe::listen(const string &addr) {
 	auto l = new Listener(addr.c_str());
 	if (!l) return false;
-	unique_lock<mutex> lk(net_mutex_);
+	unique_lock<shared_mutex> lk(net_mutex_);
 	listeners_.push_back(l);
 	return l->isListening();
 }
 
 Peer *Universe::connect(const string &addr) {
-	auto p = new Peer(addr.c_str(), &disp_);
+	auto p = new Peer(addr.c_str(), this, &disp_);
 	if (!p) return nullptr;
 	
 	if (p->status() != Peer::kInvalid) {
-		unique_lock<mutex> lk(net_mutex_);
+		unique_lock<shared_mutex> lk(net_mutex_);
 		peers_.push_back(p);
 	}
 	
 	_installBindings(p);
-	
-	p->onConnect([this](Peer &p) {
-		peer_ids_[p.id()] = &p;
-		_notifyConnect(&p);
-	});
-	
 	return p;
 }
 
 void Universe::unbind(const std::string &name) {
-	unique_lock<mutex> lk(net_mutex_);
+	unique_lock<shared_mutex> lk(net_mutex_);
 	disp_.unbind(name);
 }
 
@@ -121,7 +122,7 @@ int Universe::_setDescriptors() {
 
 	SOCKET n = 0;
 
-	unique_lock<mutex> lk(net_mutex_);
+	unique_lock<shared_mutex> lk(net_mutex_);
 
 	//Set file descriptor for the listening sockets.
 	for (auto l : listeners_) {
@@ -156,18 +157,17 @@ void Universe::_installBindings(Peer *p) {
 }
 
 void Universe::_installBindings() {
-	bind("__subscribe__", [this](const UUID &id, const string &uri) -> bool {
+	/*bind("__subscribe__", [this](const UUID &id, const string &uri) -> bool {
 		LOG(INFO) << "Subscription to " << uri << " by " << id.to_string();
-		unique_lock<mutex> lk(net_mutex_);
+		unique_lock<shared_mutex> lk(net_mutex_);
 		subscribers_[ftl::URI(uri).to_string()].push_back(id);
 		return true;
 	});
 	
 	bind("__owner__", [this](const std::string &res) -> optional<UUID> {
-		LOG(INFO) << "SOMEONE ASKS FOR " << res;
 		if (owned_.count(res) > 0) return this_peer;
 		else return {};
-	});
+	});*/
 }
 
 // Note: should be called inside a net lock
@@ -178,13 +178,18 @@ void Universe::_cleanupPeers() {
 		if (!(*i)->isValid()) {
 			Peer *p = *i;
 			LOG(INFO) << "Removing disconnected peer: " << p->id().to_string();
-			_notifyDisconnect(p);
+			//_notifyDisconnect(p);
 
 			auto ix = peer_ids_.find(p->id());
 			if (ix != peer_ids_.end()) peer_ids_.erase(ix);
-			delete p;
 
 			i = peers_.erase(i);
+
+			if (p->status() == ftl::net::Peer::kReconnecting) {
+				reconnects_.push_back({50, 1.0f, p});
+			} else {
+				delete p;
+			}
 		} else {
 			i++;
 		}
@@ -197,7 +202,7 @@ Peer *Universe::getPeer(const UUID &id) const {
 	else return ix->second;
 }
 
-optional<UUID> Universe::findOwner(const string &res) {
+/*optional<UUID> Universe::findOwner(const string &res) {
 	// TODO(nick) cache this information
 	return findOne<UUID>("__owner__", res);
 }
@@ -238,6 +243,24 @@ bool Universe::_subscribe(const std::string &res) {
 		LOG(WARNING) << "Subscribe to unknown resource: " << res;
 		return false;
 	}
+}*/
+
+void Universe::_periodic() {
+	auto i = reconnects_.begin();
+	while (i != reconnects_.end()) {
+		if ((*i).peer->reconnect()) {
+			unique_lock<shared_mutex> lk(net_mutex_);
+			peers_.push_back((*i).peer);
+			i = reconnects_.erase(i);
+		} else if ((*i).tries > 0) {
+			(*i).tries--;
+			i++;
+		} else {
+			delete (*i).peer;
+			i = reconnects_.erase(i);
+			LOG(WARNING) << "Reconnection to peer failed";
+		}
+	}
 }
 
 void Universe::__start(Universe * u) {
@@ -259,10 +282,20 @@ void Universe::_run() {
 	}
 #endif
 
+	auto start = std::chrono::high_resolution_clock::now();
+
 	while (active_) {
 		int n = _setDescriptors();
 		int selres = 1;
 
+		// Do periodics
+		auto now = std::chrono::high_resolution_clock::now();
+		std::chrono::duration<double> elapsed = now - start;
+		if (elapsed.count() >= 1.0) {
+			start = now;
+			_periodic();
+		}
+
 		// It is an error to use "select" with no sockets ... so just sleep
 		if (n == 0) {
 			std::this_thread::sleep_for(std::chrono::milliseconds(300));
@@ -287,7 +320,7 @@ void Universe::_run() {
 			continue;
 		}
 
-		unique_lock<mutex> lk(net_mutex_);
+		unique_lock<shared_mutex> lk(net_mutex_);
 
 		//If connection request is waiting
 		for (auto l : listeners_) {
@@ -300,20 +333,16 @@ void Universe::_run() {
 					SOCKET csock = accept(l->_socket(), (sockaddr*)&addr, (socklen_t*)&rsize);
 
 					if (csock != INVALID_SOCKET) {
-						auto p = new Peer(csock, &disp_);
+						auto p = new Peer(csock, this, &disp_);
 						peers_.push_back(p);
 						_installBindings(p);
-						p->onConnect([this](Peer &p) {
-							peer_ids_[p.id()] = &p;
-							// Note, called in another thread so above lock
-							// does not apply.
-							_notifyConnect(&p);
-						});
 					}
 				}
 			}
 		}
 
+		// TODO(Nick) Might switch to shared lock here?
+
 		//Also check each clients socket to see if any messages or errors are waiting
 		for (auto s : peers_) {
 			if (s != NULL && s->isValid()) {
@@ -332,27 +361,33 @@ void Universe::_run() {
 	}
 }
 
-void Universe::onConnect(const std::string &name, std::function<void(ftl::net::Peer*)> cb) {
-	unique_lock<mutex> lk(net_mutex_);
-	on_connect_.push_back({name, cb});
+callback_t Universe::onConnect(const std::function<void(ftl::net::Peer*)> &cb) {
+	unique_lock<shared_mutex> lk(handler_mutex_);
+	callback_t id = cbid__++;
+	on_connect_.push_back({id, cb});
+	return id;
 }
 
-void Universe::onDisconnect(const std::string &name, std::function<void(ftl::net::Peer*)> cb) {
-	unique_lock<mutex> lk(net_mutex_);
-	on_disconnect_.push_back({name, cb});
+callback_t Universe::onDisconnect(const std::function<void(ftl::net::Peer*)> &cb) {
+	unique_lock<shared_mutex> lk(handler_mutex_);
+	callback_t id = cbid__++;
+	on_disconnect_.push_back({id, cb});
+	return id;
 }
 
-void Universe::onError(const std::string &name, std::function<void(ftl::net::Peer*, const ftl::net::Error &)> cb) {
-	unique_lock<mutex> lk(net_mutex_);
-	on_error_.push_back({name, cb});
+callback_t Universe::onError(const std::function<void(ftl::net::Peer*, const ftl::net::Error &)> &cb) {
+	unique_lock<shared_mutex> lk(handler_mutex_);
+	callback_t id = cbid__++;
+	on_error_.push_back({id, cb});
+	return id;
 }
 
-void Universe::removeCallbacks(const std::string &name) {
-	unique_lock<mutex> lk(net_mutex_);
+void Universe::removeCallback(callback_t cbid) {
+	unique_lock<shared_mutex> lk(handler_mutex_);
 	{
 		auto i = on_connect_.begin();
 		while (i != on_connect_.end()) {
-			if ((*i).name == name) {
+			if ((*i).id == cbid) {
 				i = on_connect_.erase(i);
 			} else {
 				i++;
@@ -363,7 +398,7 @@ void Universe::removeCallbacks(const std::string &name) {
 	{
 		auto i = on_disconnect_.begin();
 		while (i != on_disconnect_.end()) {
-			if ((*i).name == name) {
+			if ((*i).id == cbid) {
 				i = on_disconnect_.erase(i);
 			} else {
 				i++;
@@ -374,7 +409,7 @@ void Universe::removeCallbacks(const std::string &name) {
 	{
 		auto i = on_error_.begin();
 		while (i != on_error_.end()) {
-			if ((*i).name == name) {
+			if ((*i).id == cbid) {
 				i = on_error_.erase(i);
 			} else {
 				i++;
@@ -384,12 +419,14 @@ void Universe::removeCallbacks(const std::string &name) {
 }
 
 void Universe::_notifyConnect(Peer *p) {
-	unique_lock<mutex> lk(net_mutex_);
+	shared_lock<shared_mutex> lk(handler_mutex_);
+	peer_ids_[p->id()] = p;
+
 	for (auto &i : on_connect_) {
 		try {
 			i.h(p);
 		} catch(...) {
-			LOG(ERROR) << "Exception inside OnConnect hander: " << i.name;
+			LOG(ERROR) << "Exception inside OnConnect hander: " << i.id;
 		}
 	}
 }
@@ -397,11 +434,12 @@ void Universe::_notifyConnect(Peer *p) {
 void Universe::_notifyDisconnect(Peer *p) {
 	// In all cases, should already be locked outside this function call
 	//unique_lock<mutex> lk(net_mutex_);
+	shared_lock<shared_mutex> lk(handler_mutex_);
 	for (auto &i : on_disconnect_) {
 		try {
 			i.h(p);
 		} catch(...) {
-			LOG(ERROR) << "Exception inside OnDisconnect hander: " << i.name;
+			LOG(ERROR) << "Exception inside OnDisconnect hander: " << i.id;
 		}
 	}
 }

components/net/cpp/test/net_integration.cpp

@@ -18,7 +18,6 @@ TEST_CASE("Universe::connect()", "[net]") {
 	Universe b;
 	
 	a.listen("tcp://localhost:7077");
-	//sleep_for(milliseconds(100));
 
 	SECTION("valid tcp connection using ipv4") {
 		auto p = b.connect("tcp://127.0.0.1:7077");
@@ -86,7 +85,7 @@ TEST_CASE("Universe::onConnect()", "[net]") {
 	SECTION("single valid remote init connection") {
 		bool done = false;
 
-		a.onConnect("test", [&done](Peer *p) {
+		a.onConnect([&done](Peer *p) {
 			done = true;
 		});
 
@@ -98,7 +97,7 @@ TEST_CASE("Universe::onConnect()", "[net]") {
 	SECTION("single valid init connection") {
 		bool done = false;
 
-		b.onConnect("test", [&done](Peer *p) {
+		b.onConnect([&done](Peer *p) {
 			done = true;
 		});
 
@@ -117,7 +116,7 @@ TEST_CASE("Universe::onDisconnect()", "[net]") {
 	SECTION("single valid remote close") {
 		bool done = false;
 
-		a.onDisconnect("test", [&done](Peer *p) {
+		a.onDisconnect([&done](Peer *p) {
 			done = true;
 		});
 
@@ -132,7 +131,7 @@ TEST_CASE("Universe::onDisconnect()", "[net]") {
 	SECTION("single valid close") {
 		bool done = false;
 
-		b.onDisconnect("test", [&done](Peer *p) {
+		b.onDisconnect([&done](Peer *p) {
 			done = true;
 		});
 
@@ -209,6 +208,9 @@ TEST_CASE("Universe::broadcast()", "[net]") {
 			done2 = v;
 		});
 
+		REQUIRE( a.numberOfPeers() == 2 );
+		sleep_for(milliseconds(100)); // NOTE: Binding might not be ready
+		
 		a.broadcast("hello", 676);
 		
 		sleep_for(milliseconds(100));
@@ -222,6 +224,7 @@ TEST_CASE("Universe::findAll()", "") {
 	Universe a;
 	Universe b;
 	Universe c;
+
 	a.listen("tcp://localhost:7077");
 	b.connect("tcp://localhost:7077")->waitConnection();
 	c.connect("tcp://localhost:7077")->waitConnection();
@@ -248,107 +251,10 @@ TEST_CASE("Universe::findAll()", "") {
 			return {6,7,8};
 		});
 
+		sleep_for(milliseconds(100)); // NOTE: Binding might not be ready
+
 		auto res = a.findAll<int>("test_all");
 		REQUIRE( (res.size() == 6) );
 		REQUIRE( (res[0] == 3 || res[0] == 6) );
 	}
 }
-
-TEST_CASE("Universe::findOwner()", "") {
-	Universe a;
-	Universe b;
-	a.listen("tcp://localhost:7077");
-	b.connect("tcp://localhost:7077")->waitConnection();
-
-	SECTION("no owners exist") {
-		REQUIRE( !b.findOwner("ftl://test") );
-	}
-
-	SECTION("one owner exists") {
-		a.createResource("ftl://test");
-		REQUIRE( *(b.findOwner("ftl://test")) == ftl::net::this_peer );
-	}
-	
-	SECTION("three peers and one owner") {
-		Universe c;
-		c.connect("tcp://localhost:7077")->waitConnection();
-		b.setLocalID(ftl::UUID(7));
-
-		b.createResource("ftl://test");
-		REQUIRE( *(a.findOwner("ftl://test")) == ftl::UUID(7) );
-	}
-
-	SECTION("three peers and one owner (2)") {
-		Universe c;
-		c.connect("tcp://localhost:7077")->waitConnection();
-		c.setLocalID(ftl::UUID(7));
-
-		c.createResource("ftl://test");
-		auto r = a.findOwner("ftl://test");
-		REQUIRE( r );
-		REQUIRE( *r == ftl::UUID(7) );
-	}
-}
-
-TEST_CASE("Universe::subscribe()", "") {
-	Universe a;
-	Universe b;
-	a.listen("tcp://localhost:7077");
-	b.connect("tcp://localhost:7077")->waitConnection();
-
-	SECTION("no resource exists") {
-		REQUIRE( !b.subscribe("ftl://test", []() {}) );
-	}
-
-	SECTION("one resource exists") {
-		a.createResource("ftl://test");
-		REQUIRE( b.subscribe("ftl://test", []() {}) );
-		sleep_for(milliseconds(50));
-		REQUIRE( a.numberOfSubscribers("ftl://test") == 1);
-	}
-}
-
-TEST_CASE("Universe::publish()", "") {
-	Universe a;
-	Universe b;
-	a.listen("tcp://localhost:7077");
-	ftl::net::Peer *p = b.connect("tcp://localhost:7077");
-	p->waitConnection();
-
-	SECTION("no subscribers") {
-		a.createResource("ftl://test");
-		a.publish("ftl://test", 55);
-	}
-
-	SECTION("one subscriber") {
-		int done = 0;
-		a.createResource("ftl://test");
-		REQUIRE( b.subscribe("ftl://test", [&done](int a) {
-			done = a;
-		}) );
-		sleep_for(milliseconds(50));
-
-		a.publish("ftl://test", 56);
-		sleep_for(milliseconds(50));
-		
-		REQUIRE( done == 56 );
-	}
-
-	SECTION("publish to disconnected subscriber") {
-		int done = 0;
-		a.createResource("ftl://test2");
-		REQUIRE( b.subscribe("ftl://test2", [&done](int a) {
-			done = a;
-		}) );
-		sleep_for(milliseconds(50));
-
-		p->close();
-		sleep_for(milliseconds(100));
-
-		a.publish("ftl://test2", 56);
-		sleep_for(milliseconds(50));
-		
-		REQUIRE( done == 0 );
-	}
-}
-