fixed #1393. store streams by recvStreamID

libi2pd/Streaming.cpp

@@ -964,7 +964,6 @@ namespace stream
 
 	StreamingDestination::StreamingDestination (std::shared_ptr<i2p::client::ClientDestination> owner, uint16_t localPort, bool gzip):
 		m_Owner (owner), m_LocalPort (localPort), m_Gzip (gzip),
-		m_LastIncomingReceiveStreamID (0),
 		m_PendingIncomingTimer (m_Owner->GetService ())
 	{
 	}
@@ -1013,19 +1012,18 @@ namespace stream
 			if (packet->IsSYN () && !packet->GetSeqn ()) // new incoming stream
 			{
 				uint32_t receiveStreamID = packet->GetReceiveStreamID ();
-				if (receiveStreamID == m_LastIncomingReceiveStreamID)
+				auto it1 = m_IncomingStreams.find (receiveStreamID);
+				if (it1 != m_IncomingStreams.end ())
 				{
 					// already pending
 					LogPrint(eLogWarning, "Streaming: Incoming streaming with rSID=", receiveStreamID, " already exists");
 					DeletePacket (packet); // drop it, because previous should be connected
 					return;
 				}
-				auto incomingStream = CreateNewIncomingStream ();
+				auto incomingStream = CreateNewIncomingStream (receiveStreamID);
 				incomingStream->HandleNextPacket (packet); // SYN
 				auto ident = incomingStream->GetRemoteIdentity();
 
-				m_LastIncomingReceiveStreamID = receiveStreamID;
-
 				// handle saved packets if any
 				{
 					auto it = m_SavedPackets.find (receiveStreamID);
@@ -1062,11 +1060,11 @@ namespace stream
 			else // follow on packet without SYN
 			{
 				uint32_t receiveStreamID = packet->GetReceiveStreamID ();
-				for (auto& it: m_Streams)
+				auto it1 = m_IncomingStreams.find (receiveStreamID);
-					if (it.second->GetSendStreamID () == receiveStreamID)
+				if (it1 != m_IncomingStreams.end ())
 				{
 					// found
-						it.second->HandleNextPacket (packet);
+					it1->second->HandleNextPacket (packet);
 					return;
 				}
 				// save follow on packet
@@ -1105,11 +1103,12 @@ namespace stream
 		return s;
 	}
 
-	std::shared_ptr<Stream> StreamingDestination::CreateNewIncomingStream ()
+	std::shared_ptr<Stream> StreamingDestination::CreateNewIncomingStream (uint32_t receiveStreamID)
 	{
 		auto s = std::make_shared<Stream> (m_Owner->GetService (), *this);
 		std::unique_lock<std::mutex> l(m_StreamsMutex);
 		m_Streams[s->GetRecvStreamID ()] = s;
+		m_IncomingStreams[receiveStreamID] = s;
 		return s;
 	}
 
@@ -1118,9 +1117,8 @@ namespace stream
 		if (stream)
 		{
 			std::unique_lock<std::mutex> l(m_StreamsMutex);
-			auto it = m_Streams.find (stream->GetRecvStreamID ());
+			m_Streams.erase (stream->GetRecvStreamID ());
-			if (it != m_Streams.end ())
+			m_IncomingStreams.erase (stream->GetSendStreamID ());
-				m_Streams.erase (it);
 		}
 	}
 

libi2pd/Streaming.h

@@ -269,8 +269,10 @@ namespace stream
 
 			void AcceptOnceAcceptor (std::shared_ptr<Stream> stream, Acceptor acceptor, Acceptor prev);
 
+		private:
+
 			void HandleNextPacket (Packet * packet);
-			std::shared_ptr<Stream> CreateNewIncomingStream ();
+			std::shared_ptr<Stream> CreateNewIncomingStream (uint32_t receiveStreamID);
 			void HandlePendingIncomingTimer (const boost::system::error_code& ecode);
 
 		private:
@@ -280,8 +282,8 @@ namespace stream
 			bool m_Gzip; // gzip compression of data messages
 			std::mutex m_StreamsMutex;
 			std::map<uint32_t, std::shared_ptr<Stream> > m_Streams; // sendStreamID->stream
+			std::map<uint32_t, std::shared_ptr<Stream> > m_IncomingStreams; // receiveStreamID->stream
 			Acceptor m_Acceptor;
-			uint32_t m_LastIncomingReceiveStreamID;
 			std::list<std::shared_ptr<Stream> > m_PendingIncomingStreams;
 			boost::asio::deadline_timer m_PendingIncomingTimer;
 			std::map<uint32_t, std::list<Packet *> > m_SavedPackets; // receiveStreamID->packets, arrived before SYN