improvement of udp punching hole

 - RTMFP : added function parameter to be called when a new address is discovered
 - RTMFPHandshaker : now we send the handshake 30 to peers (if addresses are known) before receiving the list of addresses to grow udp hole punching result
 - NetGroup : update of manageBestConnections to not have more peers than wanted

Author: thomasjammet

include/FlowManager.h

@@ -125,8 +125,8 @@ class FlowManager : public BandWriter {
 	// Close the session properly or abruptly if parameter is true
 	virtual void					close(bool abrupt);
 
-	// Set the host and peer addresses when receiving redirection address (only for P2P)
-	virtual void					setAddresses(const Mona::SocketAddress& address, const PEER_LIST_ADDRESS_TYPE& addresses) {}
+	// Set the host and peer addresses when receiving redirection request (only for P2P)
+	virtual void					addAddress(const Mona::SocketAddress& address, RTMFP::AddressType type) {}
 
 protected:
 

include/P2PSession.h

@@ -126,8 +126,8 @@ class P2PSession : public FlowManager,
 	// Return the diffie hellman object (related to main session)
 	virtual bool					diffieHellman(Mona::DiffieHellman* &pDh);
 
-	// Set the host and peer addresses when receiving redirection address (only for P2P)
-	virtual void					setAddresses(const Mona::SocketAddress& address, const PEER_LIST_ADDRESS_TYPE& addresses) { hostAddress = address; _knownAddresses.insert(addresses.begin(), addresses.end()); }
+	// Set the host and peer addresses when receiving redirection request (only for P2P)
+	virtual void					addAddress(const Mona::SocketAddress& address, RTMFP::AddressType type);
 
 	/*** Public members ***/
 

include/RTMFP.h

@@ -32,7 +32,7 @@ along with Librtmfp.  If not, see <http://www.gnu.org/licenses/>.
 
 #include "Mona/Logs.h"
 
-#define RTMFP_LIB_VERSION	0x0102000C	// (1.2.12)
+#define RTMFP_LIB_VERSION	0x0102000D	// (1.2.13)
 
 #define RTMFP_DEFAULT_KEY	(UInt8*)"Adobe Systems 02"
 #define RTMFP_KEY_SIZE		0x10
@@ -101,7 +101,7 @@ class RTMFP : virtual Mona::Static {
 		FAILED
 	};
 
-	static bool						ReadAddress(Mona::BinaryReader& reader, Mona::SocketAddress& address, Mona::UInt8& addressType);
+	static bool						ReadAddress(Mona::BinaryReader& reader, Mona::SocketAddress& address, AddressType& addressType);
 	static Mona::BinaryWriter&		WriteAddress(Mona::BinaryWriter& writer, const Mona::SocketAddress& address, AddressType type=ADDRESS_UNSPECIFIED);
 
 	static Mona::UInt32				Unpack(Mona::BinaryReader& reader);
@@ -126,7 +126,7 @@ class RTMFP : virtual Mona::Static {
 
 	// Read addresses from the buffer reader
 	// return : True if at least an address has been read
-	static bool	ReadAddresses(Mona::BinaryReader& reader, PEER_LIST_ADDRESS_TYPE& addresses, Mona::SocketAddress& hostAddress);
+	static bool	ReadAddresses(Mona::BinaryReader& reader, PEER_LIST_ADDRESS_TYPE& addresses, Mona::SocketAddress& hostAddress, std::function<void(const Mona::SocketAddress&, AddressType)> onNewAddress);
 
 	// Return a random iterator which respect the isAllowed condition
 	template<class ContainerType, typename Iterator>

sources/NetGroup.cpp

@@ -517,16 +517,20 @@ void NetGroup::manageBestConnections() {
 	}
 
 	// Connect to new peers
-	for (auto it : _bestList) {
-		if (_mapPeers.find(it) == _mapPeers.end()) {
-			auto itNode = _mapHeardList.find(it);
+	int nbConnect = _bestList.size() - _mapPeers.size(); // trick to keep the target count of peers
+	auto it2Connect = _bestList.begin();
+	while (nbConnect > 0 && it2Connect != _bestList.end()) {
+		if (_mapPeers.find(*it2Connect) == _mapPeers.end()) {
+			auto itNode = _mapHeardList.find(*it2Connect);
 			if (itNode == _mapHeardList.end())
-				WARN("Unable to find the peer ", it) // implementation error, should not happen
+				WARN("Unable to find the peer ", *it2Connect) // implementation error, should not happen
 			else {
-				DEBUG("Best Peer - Connecting to peer ", it, "...")
-				_conn.connect2Peer(it.c_str(), stream.c_str(), itNode->second.addresses, itNode->second.hostAddress);
+				DEBUG("Best Peer - Connecting to peer ", *it2Connect, "...")
+				_conn.connect2Peer(it2Connect->c_str(), stream.c_str(), itNode->second.addresses, itNode->second.hostAddress);
+				--nbConnect;
 			}
 		}
+		++it2Connect;
 	}
 }
 
@@ -581,7 +585,7 @@ void NetGroup::ReadGroupConfig(shared_ptr<RTMFPGroupConfig>& parameters, PacketR
 bool NetGroup::readGroupReport(PacketReader& packet) {
 	string tmp, newPeerId, rawId;
 	SocketAddress myAddress, serverAddress;
-	UInt8 addressType;
+	RTMFP::AddressType addressType;
 	PEER_LIST_ADDRESS_TYPE listAddresses;
 	SocketAddress hostAddress(_conn.address());
 
@@ -607,7 +611,7 @@ bool NetGroup::readGroupReport(PacketReader& packet) {
 		return false;
 	}
 	BinaryReader peerAddressReader(packet.current(), size - 1);
-	RTMFP::ReadAddresses(peerAddressReader, listAddresses, hostAddress);
+	RTMFP::ReadAddresses(peerAddressReader, listAddresses, hostAddress, [](const SocketAddress&, RTMFP::AddressType) {});
 	packet.next(size - 1);
 
 	// Loop on each peer of the NetGroup
@@ -645,13 +649,13 @@ bool NetGroup::readGroupReport(PacketReader& packet) {
 			if (itHeardList == _mapHeardList.end()) {
 				hostAddress.clear();
 				listAddresses.clear();
-				RTMFP::ReadAddresses(addressReader, listAddresses, hostAddress);
+				RTMFP::ReadAddresses(addressReader, listAddresses, hostAddress, [](const SocketAddress&, RTMFP::AddressType) {});
 				newPeers = true;
 				addPeer2HeardList(newPeerId.c_str(), rawId.data(), listAddresses, hostAddress, time);  // To avoid memory sharing we use c_str() (copy-on-write implementation on linux)
 			} 
-			// Else if no host is set update the addresses
-			else if (!itHeardList->second.hostAddress)
-				RTMFP::ReadAddresses(addressReader, itHeardList->second.addresses, itHeardList->second.hostAddress);
+			// Else update the addresses
+			else
+				RTMFP::ReadAddresses(addressReader, itHeardList->second.addresses, itHeardList->second.hostAddress, [](const SocketAddress&, RTMFP::AddressType) {});
 
 		}
 		packet.next(size);

sources/P2PSession.cpp

@@ -504,18 +504,25 @@ bool P2PSession::onHandshake38(const SocketAddress& address, shared_ptr<Handshak
 	return true;
 }
 
-UDPSocket& P2PSession::socket(Mona::IPAddress::Family family) { 
+UDPSocket& P2PSession::socket(IPAddress::Family family) { 
 	return _parent->socket(family); 
 }
 
-void P2PSession::removeHandshake(std::shared_ptr<Handshake>& pHandshake) {
+void P2PSession::removeHandshake(shared_ptr<Handshake>& pHandshake) {
 	_parent->removeHandshake(pHandshake);
 }
 
 bool P2PSession::diffieHellman(DiffieHellman* &pDh) {
 	return _parent->diffieHellman(pDh);
 }
 
+void P2PSession::addAddress(const SocketAddress& address, RTMFP::AddressType type) {
+	if (type == RTMFP::ADDRESS_REDIRECTION)
+		hostAddress == address;
+	else
+		_knownAddresses.emplace(address, type);
+}
+
 void P2PSession::buildGroupKey() {
 
 	// Compile encrypted keys

sources/RTMFP.cpp

@@ -25,9 +25,9 @@ along with Librtmfp.  If not, see <http://www.gnu.org/licenses/>.
 using namespace std;
 using namespace Mona;
 
-bool RTMFP::ReadAddress(BinaryReader& reader, SocketAddress& address, UInt8& addressType) {
+bool RTMFP::ReadAddress(BinaryReader& reader, SocketAddress& address, AddressType& addressType) {
 	string data;
-	addressType = reader.read8();
+	addressType = (AddressType)reader.read8();
 	reader.read<string>((addressType & 0x80) ? sizeof(in6_addr) : sizeof(in_addr), data);
 	in_addr addrV4;
 	in6_addr addrV6;
@@ -104,21 +104,30 @@ void RTMFP::Write7BitValue(string& buff,UInt64 value) {
 	String::Append(buff, (char)(max ? value&0xFF : value&0x7F));
 }
 
-bool RTMFP::ReadAddresses(BinaryReader& reader, PEER_LIST_ADDRESS_TYPE& addresses, SocketAddress& hostAddress) {
+bool RTMFP::ReadAddresses(BinaryReader& reader, PEER_LIST_ADDRESS_TYPE& addresses, SocketAddress& hostAddress, function<void(const SocketAddress&, AddressType)> onNewAddress) {
 
 	// Read all addresses
 	SocketAddress address;
-	UInt8 addressType;
+	AddressType addressType;
 	while (reader.available()) {
 
 		RTMFP::ReadAddress(reader, address, addressType);
 		switch (addressType & 0x0F) {
 		case RTMFP::ADDRESS_LOCAL:
-		case RTMFP::ADDRESS_PUBLIC:
-			addresses.emplace(address, (RTMFP::AddressType)addressType);
+		case RTMFP::ADDRESS_PUBLIC: {
+			auto itAddress = addresses.lower_bound(address);
+			if (itAddress == addresses.end() || itAddress->first != address) { // new address?
+				addresses.emplace_hint(itAddress, address, addressType);
+				onNewAddress(address, addressType);
+			}
 			break;
+		}
 		case RTMFP::ADDRESS_REDIRECTION:
-			hostAddress = address; break;
+			if (hostAddress != address) { // new address?
+				hostAddress = address;
+				onNewAddress(address, addressType);
+			}
+			break;
 		}
 		TRACE("IP Address : ", address.toString(), " - type : ", addressType)
 	}

sources/RTMFPHandshaker.cpp

@@ -142,12 +142,9 @@ void RTMFPHandshaker::manage() {
 					_address.set(pHandshake->hostAddress);
 					sendHandshake30(pHandshake->pSession->epd(), itHandshake->first);
 				}
-				// If we are not in p2p mode we must send to all known addresses
-				if (!pHandshake->isP2P) {
-					for (auto itAddresses : pHandshake->listAddresses) {
-						_address.set(itAddresses.first);
-						sendHandshake30(pHandshake->pSession->epd(), itHandshake->first);
-					}
+				for (auto itAddresses : pHandshake->listAddresses) {
+					_address.set(itAddresses.first);
+					sendHandshake30(pHandshake->pSession->epd(), itHandshake->first);
 				}
 				if (pHandshake->status == RTMFP::STOPPED)
 					pHandshake->status = RTMFP::HANDSHAKE30;
@@ -450,10 +447,10 @@ void RTMFPHandshaker::handleRedirection(BinaryReader& reader) {
 		ERROR("Unexpected tag size : ", tagSize)
 			return;
 	}
-	string tagReceived;
-	reader.read(16, tagReceived);
+	string tag;
+	reader.read(16, tag);
 
-	auto itTag = _mapTags.find(tagReceived);
+	auto itTag = _mapTags.find(tag);
 	if (itTag == _mapTags.end()) {
 		DEBUG("Unexpected tag received from ", _address.toString(), ", possible old request")
 		return;
@@ -467,27 +464,20 @@ void RTMFPHandshaker::handleRedirection(BinaryReader& reader) {
 		DEBUG("Redirection message ignored, we have already received handshake 70")
 		return;
 	}
+	DEBUG(pHandshake->isP2P ? "Server has sent to us the peer addresses of responders" : "Server redirection messsage, sending back the handshake 30")
 
 	// Read addresses
 	SocketAddress hostAddress;
-	RTMFP::ReadAddresses(reader, pHandshake->listAddresses, hostAddress);
-
-	if (pHandshake->isP2P) {
-		DEBUG("Server has sent to us the peer addresses of responders") // (we are the initiator)
-
-		// Reset the host address if it changes
-		if (hostAddress && hostAddress != pHandshake->hostAddress) {
-			pHandshake->pSession->setAddresses(hostAddress, pHandshake->listAddresses);
-			pHandshake->hostAddress = hostAddress;
-		}
-
-		// Send the handshake 30 to all addresses
-		for (auto itAddresses : pHandshake->listAddresses) {
-			_address.set(itAddresses.first);
-			sendHandshake30(pHandshake->pSession->epd(), tagReceived);
+	RTMFP::ReadAddresses(reader, pHandshake->listAddresses, pHandshake->hostAddress, [this, pHandshake, hostAddress, tag](const SocketAddress& address, RTMFP::AddressType type) {
+		if (pHandshake->isP2P)
+			pHandshake->pSession->addAddress(address, type);
+
+		// Send the handshake 30 to new address
+		if (type != RTMFP::ADDRESS_REDIRECTION) {
+			_address.set(address);
+			sendHandshake30(pHandshake->pSession->epd(), tag);
 		}
-	} else
-		DEBUG("Server redirection messsage, sending back the handshake 30")
+	});
 }
 
 void RTMFPHandshaker::flush(UInt8 marker, UInt32 size) {

sources/RTMFPSession.cpp

@@ -617,7 +617,7 @@ void RTMFPSession::handleP2PAddressExchange(PacketReader& reader) {
 	string buff;
 	reader.read(PEER_ID_SIZE, buff);
 	SocketAddress address;
-	UInt8 addressType;
+	RTMFP::AddressType addressType;
 	RTMFP::ReadAddress(reader, address, addressType);
 
 	string tag;