Add bootstrap node, listen method

5 files changed, 94 insertions, 14 deletions

diff --git a/include/sibs/BootstrapConnection.hpp b/include/sibs/BootstrapConnection.hpp
new file mode 100644
index 0000000..964d777
--- /dev/null
+++ b/include/sibs/BootstrapConnection.hpp
@@ -0,0 +1,33 @@
+#pragma once
+
+#include "../utils.hpp"
+#include "DirectConnection.hpp"
+#include "PubsubKey.hpp"
+
+namespace sibs
+{
+    class PubsubKeyAlreadyListeningException : public std::runtime_error
+    {
+    public:
+        PubsubKeyAlreadyListeningException(const std::string &errMsg) : std::runtime_error(errMsg) {}
+    };
+    
+    using BoostrapConnectionListenCallbackFunc = std::function<void(const void *data, const usize size)>;
+    
+    class BootstrapConnection
+    {
+        DISABLE_COPY(BootstrapConnection)
+    public:
+        BootstrapConnection(const Ipv4 &bootstrapAddress);
+        
+        // Throws PubsubKeyAlreadyListeningException if we are already listening on the key @pubsubKey
+        void listen(const PubsubKey &pubsubKey, BoostrapConnectionListenCallbackFunc callbackFunc);
+    private:
+        void receiveDataFromServer(std::shared_ptr<DirectConnectionPeer> peer, const void *data, const usize size);
+        void receiveDataFromPeer(BoostrapConnectionListenCallbackFunc listenCallbackFunc, std::shared_ptr<DirectConnectionPeer> peer, const void *data, const usize size);
+    private:
+        DirectConnections connections;
+        std::shared_ptr<DirectConnectionPeer> serverPeer;
+        PubsubKeyMap<BoostrapConnectionListenCallbackFunc> listenCallbackFuncs;
+    };
+}
diff --git a/include/sibs/BootstrapNode.hpp b/include/sibs/BootstrapNode.hpp
index eaf48eb..13a62e1 100644
--- a/include/sibs/BootstrapNode.hpp
+++ b/include/sibs/BootstrapNode.hpp
@@ -2,6 +2,8 @@
 
 #include "DirectConnection.hpp"
 #include "IpAddress.hpp"
+#include "PubsubKey.hpp"
+#include <vector>
 
 namespace sibs
 {
@@ -20,9 +22,11 @@ namespace sibs
         ~BootstrapNode();
     private:
         void acceptConnections();
+        void peerSubscribe(std::shared_ptr<DirectConnectionPeer> peer, const void *data, const usize size);
     private:
         DirectConnections connections;
         int socket;
         std::thread acceptConnectionsThread;
+        PubsubKeyMap<std::vector<std::shared_ptr<DirectConnectionPeer>>> subscribedPeers;
     };
 }
diff --git a/include/sibs/DirectConnection.hpp b/include/sibs/DirectConnection.hpp
index c4431a5..0e8b961 100644
--- a/include/sibs/DirectConnection.hpp
+++ b/include/sibs/DirectConnection.hpp
@@ -7,6 +7,7 @@
 #include <memory>
 #include <thread>
 #include <mutex>
+#include <vector>
 #include "IpAddress.hpp"
 #include "../types.hpp"
 #include "../utils.hpp"
@@ -19,25 +20,22 @@ namespace sibs
         ConnectionException(const std::string &errMsg) : std::runtime_error(errMsg) {}
     };
     
-    class SendException : public std::runtime_error
-    {
-    public:
-        SendException(const std::string &errMsg) : std::runtime_error(errMsg) {}
-    };
-    
-    
-    enum class PubSubConnectResult
+    enum class PubSubResult
     {
         OK,
         ERROR
     };
     
-    using PubSubConnectCallback = std::function<void(PubSubConnectResult result, const std::string &resultStr)>;
-    using PubSubReceiveDataCallback = std::function<void(const void *data, const usize size)>;
+    struct DirectConnectionPeer;
+    
+    using PubSubConnectCallback = std::function<void(PubSubResult result, const std::string &resultStr)>;
+    using PubSubReceiveDataCallback = std::function<void(std::shared_ptr<DirectConnectionPeer> peer, const void *data, const usize size)>;
+    using PubSubSendDataCallback = std::function<void(PubSubResult result, const std::string &resultStr)>;
     
     struct DirectConnectionPeer
     {
         int socket;
+        Ipv4 address;
         PubSubReceiveDataCallback receiveDataCallbackFunc;
     };
     
@@ -50,12 +48,14 @@ namespace sibs
         ~DirectConnections();
         
         // Throws ConnectionException on error
+        std::shared_ptr<DirectConnectionPeer> connectServer(const Ipv4 &address, PubSubReceiveDataCallback receiveDataCallbackFunc);
+        // Throws ConnectionException on error
         std::shared_ptr<DirectConnectionPeer> connect(const Ipv4 &address, PubSubReceiveDataCallback receiveDataCallbackFunc);
-        // Throws SendException on error
-        void send(const std::shared_ptr<DirectConnectionPeer> &peer, const void *data, const usize size);
+        void send(const std::shared_ptr<DirectConnectionPeer> &peer, std::shared_ptr<std::vector<u8>> data, PubSubSendDataCallback sendDataCallbackFunc = nullptr);
     protected:
         int createSocket(const Ipv4 &addressToBind, bool rendezvous, bool reuseAddr);
     private:
+        std::shared_ptr<DirectConnectionPeer> connect(const Ipv4 &address, bool rendezvous, bool reuseAddr, PubSubReceiveDataCallback receiveDataCallbackFunc);
         void receiveData();
         bool receiveDataFromPeer(const int socket, char *output);
     private:
diff --git a/include/sibs/IpAddress.hpp b/include/sibs/IpAddress.hpp
index a0fad75..7b757f4 100644
--- a/include/sibs/IpAddress.hpp
+++ b/include/sibs/IpAddress.hpp
@@ -1,6 +1,5 @@
 #pragma once
 
-#include "../utils.hpp"
 #include <stdexcept>
 #include <string>
 #ifndef WIN32
@@ -21,11 +20,13 @@ namespace sibs
     
     class Ipv4
     {
-        DISABLE_COPY(Ipv4)
     public:
+        Ipv4();
         // If @ip is nullptr, then bind to all available sockets (typical for servers)
         // Throws InvalidAddressException on error.
         Ipv4(const char *ip, unsigned short port);
+        Ipv4(const Ipv4 &other);
+        Ipv4& operator = (const Ipv4 &other);
         
         std::string getAddress() const;
         unsigned short getPort() const;
diff --git a/include/sibs/PubsubKey.hpp b/include/sibs/PubsubKey.hpp
new file mode 100644
index 0000000..d123331
--- /dev/null
+++ b/include/sibs/PubsubKey.hpp
@@ -0,0 +1,42 @@
+#pragma once
+
+#include "../types.hpp"
+#include <array>
+#include <unordered_map>
+
+namespace sibs
+{
+    const usize PUBSUB_KEY_LENGTH = 32;
+    
+    // Source: https://stackoverflow.com/a/11414104 (public license)
+    static size_t fnvHash(const unsigned char *key, int len)
+    {
+        size_t h = 2166136261;
+        for (int i = 0; i < len; i++)
+            h = (h * 16777619) ^ key[i];
+        return h;
+    }
+    
+    class PubsubKey
+    {
+    public:
+        PubsubKey();
+        // Create pubsub key from existing data, data will be cutoff at PUBSUB_KEY_LENGTH. If @size is less than PUBSUB_KEY_LENGTH then data is appended with 0
+        PubsubKey(const void *data, const usize size);
+        bool operator == (const PubsubKey &other) const;
+        bool operator != (const PubsubKey &other) const;
+
+        std::array<u8, PUBSUB_KEY_LENGTH> data;
+    };
+    
+    struct PubsubKeyHasher
+    {
+        size_t operator()(const PubsubKey &pubsubKey) const
+        {
+            return fnvHash((const unsigned char*)pubsubKey.data.data(), PUBSUB_KEY_LENGTH);
+        }
+    };
+    
+    template <typename ValueType>
+    using PubsubKeyMap = std::unordered_map<PubsubKey, ValueType, PubsubKeyHasher>;
+}