RethinkDB direct connectivity integration.

1 files changed, 54 insertions, 5 deletions

diff --git a/controller/RethinkDB.hpp b/controller/RethinkDB.hpp
index 7ed0e0a8..5708bd65 100644
--- a/controller/RethinkDB.hpp
+++ b/controller/RethinkDB.hpp
@@ -1,3 +1,23 @@
+/*
+ * ZeroTier One - Network Virtualization Everywhere
+ * Copyright (C) 2011-2015  ZeroTier, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifdef ZT_CONTROLLER_USE_RETHINKDB
+
 #ifndef ZT_CONTROLLER_RETHINKDB_HPP
 #define ZT_CONTROLLER_RETHINKDB_HPP
 
@@ -5,6 +25,7 @@
 #include "../node/Address.hpp"
 #include "../node/InetAddress.hpp"
 #include "../osdep/OSUtils.hpp"
+#include "../osdep/BlockingQueue.hpp"
 
 #include <memory>
 #include <string>
@@ -15,9 +36,13 @@
 
 #include "../ext/json/json.hpp"
 
+#define ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS 2
+
 namespace ZeroTier
 {
 
+class EmbeddedNetworkController;
+
 class RethinkDB
 {
 public:
@@ -31,24 +56,41 @@ public:
 		int64_t mostRecentDeauthTime;
 	};
 
-	RethinkDB(const Address &myAddress,const char *host,const int port,const char *db,const char *auth);
+	RethinkDB(EmbeddedNetworkController *const nc,const Address &myAddress,const char *path);
 	~RethinkDB();
 
-	inline bool ready() const { return (_ready <= 0); }
-
 	inline void waitForReady() const
 	{
 		while (_ready > 0) {
+			if (!_waitNoticePrinted) {
+				_waitNoticePrinted = true;
+				fprintf(stderr,"NOTICE: controller RethinkDB waiting for initial data download..." ZT_EOL_S);
+			}
 			_readyLock.lock();
 			_readyLock.unlock();
 		}
 	}
 
+	inline bool hasNetwork(const uint64_t networkId) const
+	{
+		std::lock_guard<std::mutex> l(_networks_l);
+		return (_networks.find(networkId) != _networks.end());
+	}
+
 	bool get(const uint64_t networkId,nlohmann::json &network);
+	bool get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member);
 	bool get(const uint64_t networkId,nlohmann::json &network,const uint64_t memberId,nlohmann::json &member,NetworkSummaryInfo &info);
 	bool get(const uint64_t networkId,nlohmann::json &network,std::vector<nlohmann::json> &members);
+
 	bool summary(const uint64_t networkId,NetworkSummaryInfo &info);
 
+	void networks(std::vector<uint64_t> &networks);
+
+	void save(const nlohmann::json &record);
+
+	void eraseNetwork(const uint64_t networkId);
+	void eraseMember(const uint64_t networkId,const uint64_t memberId);
+
 private:
 	struct _Network
 	{
@@ -76,12 +118,13 @@ private:
 		info.mostRecentDeauthTime = nw->mostRecentDeauthTime;
 	}
 
+	EmbeddedNetworkController *const _controller;
 	const Address _myAddress;
 	std::string _myAddressStr;
 	std::string _host;
 	std::string _db;
 	std::string _auth;
-	const int _port;
+	int _port;
 
 	void *_networksDbWatcherConnection;
 	void *_membersDbWatcherConnection;
@@ -89,13 +132,19 @@ private:
 	std::thread _membersDbWatcher;
 
 	std::unordered_map< uint64_t,std::shared_ptr<_Network> > _networks;
-	std::mutex _networks_l;
+	mutable std::mutex _networks_l;
+
+	BlockingQueue< std::unique_ptr<nlohmann::json> > _commitQueue;
+	std::thread _commitThread[ZT_CONTROLLER_RETHINKDB_COMMIT_THREADS];
 
 	mutable std::mutex _readyLock; // locked until ready
 	std::atomic<int> _ready;
 	std::atomic<int> _run;
+	mutable volatile bool _waitNoticePrinted;
 };
 
 } // namespace ZeroTier
 
 #endif
+
+#endif // ZT_CONTROLLER_USE_RETHINKDB