adding a port manager to track client ports and allow clients to disc…

…onnect/reconnect to the server

libXplane-udp-client/ClientManager.cpp

@@ -8,6 +8,7 @@ std::shared_mutex mutex_;
 
 
 std::map<std::string, std::string> dataRefValuesMap;
+PortManager portManager(5557, 20);
 
 void receiverCallbackFloat(std::string dataref, float value)
 {
@@ -123,12 +124,20 @@ void attachToClient(zmq::context_t* ctx, std::string topic, int thread_number, i
 
         }
 
+        if (command == "Disconnection")
+        {
+            publisher.send(zmq::buffer(topic), zmq::send_flags::sndmore);
+            publisher.send(zmq::buffer("Received"));
+            break;
+        }
+
 
         std::cout << "Thread topic - " << topic << ": [" << recv_msgs[0].to_string() << "] " << recv_msgs[1].to_string() << std::endl;
 
 
     }
 
+    portManager.removeClient(topic);
     std::cout << "Terminating connection for topic: " << topic << std::endl;
 }
 
@@ -142,7 +151,6 @@ void listenForClients(XPlaneUDPClient& xp)
     std::vector<std::future<void>> threads;
     std::map<std::string, int> topics;
     int thread_number = 0;
-    int client_port_number = 5557;
 
 
     // Prepare our context
@@ -172,18 +180,20 @@ void listenForClients(XPlaneUDPClient& xp)
         //assert(*result == 2);
         std::string topic = recv_msgs[0].to_string();
 
-        if (topics[topic] == 0)
+        if (!portManager.isConnected(topic))
         {
-            std::cout << "New Client Connected : " << topic << std::endl;
-            //Sleep(1000); // Give client a second to prepare for message
+
+            auto [new_subscribing_port, new_publishing_port] = portManager.storeClient(topic);
+
+            std::cout << "New Client Connected : " << topic << " - on port " << new_subscribing_port << " and " << new_publishing_port << std::endl;
+
             // Send back the new port they should connect to for their personal connection
             publisher.send(zmq::buffer(topic), zmq::send_flags::sndmore);
-            publisher.send(zmq::buffer(std::to_string(client_port_number)));
+            publisher.send(zmq::buffer(std::to_string(new_subscribing_port)));
 
             topics[topic] = 1; // Store it in the map
-            threads.push_back(std::async(std::launch::async, attachToClient, &ctx, topic, thread_number, client_port_number, &xp)); // push it back into the thread vector
+            threads.push_back(std::async(std::launch::async, attachToClient, &ctx, topic, thread_number, new_subscribing_port, &xp)); // push it back into the thread vector
             thread_number++;
-            client_port_number += 2;
         }
 
     }

libXplane-udp-client/ClientManager.h

@@ -20,6 +20,7 @@
 #include "zmq.hpp"
 #include "zmq_addon.hpp"
 
+#include "PortManager.h"
 
 #include "XPlaneBeaconListener.h"
 #include "XPlaneUDPClient.h"

libXplane-udp-client/PortManager.cpp

@@ -0,0 +1,132 @@
+
+#include "PortManager.h"
+
+
+void PortManager::liberatePorts(int port1, int port2)
+{
+	for (auto const& [key, val] : m_map_taken_ports)
+	{
+		if (key == port1 || key == port2)
+		{
+			m_map_taken_ports[key] = 0;
+		}
+	}
+
+}
+
+void PortManager::occupyPorts(int port1, int port2)
+{
+	for (auto const& [key, val] : m_map_taken_ports)
+	{
+		if (key == port1 || key == port2)
+		{
+			m_map_taken_ports[key] = 1;
+		}
+	}
+
+}
+
+int PortManager::returnAvailablePort()
+{
+	for (auto const& [key, val] : m_map_taken_ports)
+	{
+		if (val == 0)
+		{
+			return key;
+		}
+	}
+
+	return -1;
+}
+
+bool PortManager::removeClient(std::string client_name)
+{
+
+	for (unsigned int i = 0; i < m_clients.size(); i++)
+	{
+		if (m_clients[i].name.find(client_name) != std::string::npos)
+		{
+			// Free the ports
+			liberatePorts(m_clients[i].publishing_port, m_clients[i].subscribing_port);
+
+			// Found client object
+			m_clients.erase(m_clients.begin() + i);
+		}
+	}
+
+	return true;
+}
+
+
+std::pair<int, int> PortManager::storeClient(std::string client_name)
+{
+	// Store values into vector of client objects
+	ClientObj new_client;
+
+	int subscriber_port = returnAvailablePort();
+	int publisher_port = subscriber_port + 1;
+
+	new_client.name = client_name;
+	new_client.publishing_port = subscriber_port;
+	new_client.subscribing_port = publisher_port;
+
+	m_clients.push_back(new_client);
+	std::pair<int, int> ports;
+	ports.first = subscriber_port;
+	ports.second = publisher_port;
+
+	return ports;
+
+}
+
+
+ClientObj PortManager::getClient(std::string client_name)
+{
+	// Store values into vector of client objects
+	ClientObj empty_client;
+
+	for (unsigned int i = 0; i < m_clients.size(); i++)
+	{
+		if (m_clients[i].name.find(client_name) != std::string::npos) 
+		{
+			// Found client object
+			return m_clients[i];
+		}
+	}
+
+	return empty_client;
+}
+
+bool PortManager::isConnected(std::string client_name)
+{
+
+	for (unsigned int i = 0; i < m_clients.size(); i++)
+	{
+		if (m_clients[i].name.find(client_name) != std::string::npos)
+		{
+			// Found client object
+			return true;
+		}
+	}
+
+	return false;
+}
+
+
+PortManager::PortManager(int starting_port, int max_number_clients)
+{
+	m_starting_port = starting_port;
+	m_max_clients = max_number_clients;
+
+
+	for (unsigned int port_number = starting_port; port_number < (starting_port + (max_number_clients * 2)); port_number++)
+	{
+		m_map_taken_ports[port_number] = 0;
+	}
+
+}
+
+PortManager::~PortManager()
+{
+
+}

libXplane-udp-client/PortManager.h

@@ -0,0 +1,48 @@
+#pragma once
+#pragma once
+
+
+#ifndef PORTMANAGER_SRC_PORTMANAGER_H_
+#define PORTMANAGER_SRC_PORTMANAGER_H_
+
+
+#include <future>
+#include <iostream>
+#include <string>
+#include <vector>
+#include <map>
+
+
+struct ClientObj
+{
+	std::string name = "";
+	unsigned int publishing_port = 0;
+	unsigned int subscribing_port = 0;
+};
+
+class PortManager
+{
+private:
+
+	int m_starting_port;						// Starting port in which to store and search for ports
+	int m_max_clients;							// Max number of clients to consider
+	std::vector<ClientObj> m_clients;			// Clients to be managing
+	std::map<int, int> m_map_taken_ports;		// Store ports
+
+	int returnAvailablePort();					// Return a port that is available
+	void occupyPorts(int port1, int port2);		// Label the ports specified as occupied
+	void liberatePorts(int port1, int port2);	// Label the ports specified as unoccupied
+
+public:
+	PortManager(int starting_port, int max_number_clients);
+	~PortManager();
+
+
+	std::pair<int, int> storeClient(std::string client_name);
+	bool removeClient(std::string client_name);
+	ClientObj getClient(std::string client_name);
+	bool isConnected(std::string client_name);
+
+
+};
+#endif /* CLIENTMANAGER_SRC_CLIENTMANAGER_H_ */

libXplane-udp-client/libXplane-udp-client.vcxproj

@@ -20,6 +20,7 @@
   </ItemGroup>
   <ItemGroup>
     <ClCompile Include="ClientManager.cpp" />
+    <ClCompile Include="PortManager.cpp" />
     <ClCompile Include="TestXPlaneUDPClient.cpp" />
     <ClCompile Include="XPlaneBeaconListener.cpp" />
     <ClCompile Include="XPlaneUDPClient.cpp" />
@@ -28,6 +29,7 @@
   <ItemGroup>
     <ClInclude Include="ClientManager.h" />
     <ClInclude Include="Logging.h" />
+    <ClInclude Include="PortManager.h" />
     <ClInclude Include="XPlaneBeaconListener.h" />
     <ClInclude Include="XPlaneUDPClient.h" />
     <ClInclude Include="XPUtils.h" />

libXplane-udp-client/libXplane-udp-client.vcxproj.filters

@@ -30,6 +30,9 @@
     <ClCompile Include="ClientManager.cpp">
       <Filter>Source Files</Filter>
     </ClCompile>
+    <ClCompile Include="PortManager.cpp">
+      <Filter>Source Files</Filter>
+    </ClCompile>
   </ItemGroup>
   <ItemGroup>
     <ClInclude Include="XPlaneBeaconListener.h">
@@ -47,5 +50,8 @@
     <ClInclude Include="Logging.h">
       <Filter>Source Files</Filter>
     </ClInclude>
+    <ClInclude Include="PortManager.h">
+      <Filter>Source Files</Filter>
+    </ClInclude>
   </ItemGroup>
 </Project>