Refactoring streams, will be an ongoing process.

common/net/daybreak_connection.cpp

@@ -152,6 +152,9 @@ void EQ::Net::DaybreakConnectionManager::Process()
 						connection->SendKeepAlive();
 					}
 				}
+
+				connection->Process();
+				break;
 			}
 			case StatusDisconnecting:
 				connection->Process();
@@ -435,7 +438,7 @@ void EQ::Net::DaybreakConnection::ProcessPacket(Packet &p)
 			return;
 		}
 
-		if (m_encode_passes[0] == EncodeCompression || m_encode_passes[1] == EncodeCompression)
+		if (m_encode_passes[0] & EncodeCompression || m_encode_passes[1] & EncodeCompression)
 		{
 			EQ::Net::DynamicPacket temp;
 			temp.PutPacket(0, p);
@@ -492,8 +495,10 @@ void EQ::Net::DaybreakConnection::ProcessPacket(Packet &p)
 void EQ::Net::DaybreakConnection::ProcessQueue()
 {
 	for (int i = 0; i < 4; ++i) {
-		auto stream = &m_streams[i];
+		auto &stream = m_streams[i];
 		for (;;) {
+			if (stream == nullptr)
+				break;
 
 			auto iter = stream->packet_queue.find(stream->sequence_in);
 			if (iter == stream->packet_queue.end()) {
@@ -510,7 +515,11 @@ void EQ::Net::DaybreakConnection::ProcessQueue()
 
 void EQ::Net::DaybreakConnection::RemoveFromQueue(int stream, uint16_t seq)
 {
-	auto s = &m_streams[stream];
+	auto &s = m_streams[stream];
+	if (s == nullptr) {
+		return;
+	}
+
 	auto iter = s->packet_queue.find(seq);
 	if (iter != s->packet_queue.end()) {
 		auto packet = iter->second;
@@ -521,7 +530,11 @@ void EQ::Net::DaybreakConnection::RemoveFromQueue(int stream, uint16_t seq)
 
 void EQ::Net::DaybreakConnection::AddToQueue(int stream, uint16_t seq, const Packet &p)
 {
-	auto s = &m_streams[stream];
+	auto &s = m_streams[stream];
+	if (s == nullptr) {
+		return;
+	}
+
 	auto iter = s->packet_queue.find(seq);
 	if (iter == s->packet_queue.end()) {
 		DynamicPacket *out = new DynamicPacket();
@@ -676,7 +689,10 @@ void EQ::Net::DaybreakConnection::ProcessDecodedPacket(const Packet &p)
 				auto header = p.GetSerialize<DaybreakReliableHeader>(0);
 				auto sequence = NetworkToHost(header.sequence);
 				auto stream_id = header.opcode - OP_Packet;
-				auto stream = &m_streams[stream_id];
+				auto &stream = m_streams[stream_id];
+				if (stream == nullptr) {
+					stream.reset(new DaybreakReliableStream());
+				}
 
 				auto order = CompareSequence(stream->sequence_in, sequence);
 				if (order == SequenceFuture) {
@@ -705,7 +721,10 @@ void EQ::Net::DaybreakConnection::ProcessDecodedPacket(const Packet &p)
 				auto header = p.GetSerialize<DaybreakReliableHeader>(0);
 				auto sequence = NetworkToHost(header.sequence);
 				auto stream_id = header.opcode - OP_Fragment;
-				auto stream = &m_streams[stream_id];
+				auto &stream = m_streams[stream_id];
+				if (stream == nullptr) {
+					stream.reset(new DaybreakReliableStream());
+				}
 
 				auto order = CompareSequence(stream->sequence_in, sequence);
 
@@ -1099,7 +1118,11 @@ void EQ::Net::DaybreakConnection::ProcessResend(int stream)
 
 	auto resends = 0;
 	auto now = Clock::now();
-	auto s = &m_streams[stream];
+	auto &s = m_streams[stream];
+	if (s == nullptr) {
+		return;
+	}
+
 	for (auto &entry : s->sent_packets) {
 		auto time_since_last_send = std::chrono::duration_cast<std::chrono::milliseconds>(now - entry.second.last_sent);
 		if (entry.second.times_resent == 0) {
@@ -1161,7 +1184,11 @@ void EQ::Net::DaybreakConnection::Ack(int stream, uint16_t seq)
 {
 
 	auto now = Clock::now();
-	auto s = &m_streams[stream];
+	auto &s = m_streams[stream];
+	if (s == nullptr) {
+		return;
+	}
+
 	auto iter = s->sent_packets.begin();
 	while (iter != s->sent_packets.end()) {
 		auto order = CompareSequence(seq, iter->first);
@@ -1185,7 +1212,11 @@ void EQ::Net::DaybreakConnection::Ack(int stream, uint16_t seq)
 void EQ::Net::DaybreakConnection::OutOfOrderAck(int stream, uint16_t seq)
 {
 	auto now = Clock::now();
-	auto s = &m_streams[stream];
+	auto &s = m_streams[stream];
+	if (s == nullptr) {
+		return;
+	}
+
 	auto iter = s->sent_packets.find(seq);
 	if (iter != s->sent_packets.end()) {
 		uint64_t round_time = (uint64_t)std::chrono::duration_cast<std::chrono::milliseconds>(now - iter->second.last_sent).count();
@@ -1401,7 +1432,11 @@ void EQ::Net::DaybreakConnection::InternalQueuePacket(Packet &p, int stream_id,
 		return;
 	}
 
-	auto stream = &m_streams[stream_id];
+	auto &stream = m_streams[stream_id];
+	if (stream == nullptr) {
+		stream.reset(new DaybreakReliableStream());
+	}
+
 	auto max_raw_size = m_max_packet_size - m_crc_bytes - DaybreakReliableHeader::size() - 1; // -1 for compress flag
 	size_t length = p.Length();
 	if (length > max_raw_size) {

common/net/daybreak_connection.h

@@ -133,6 +133,38 @@ namespace EQ
 			uint64_t bytes_before_encode;
 		};
 
+		//Refactoring this stuff
+		struct DaybreakSentPacket
+		{
+			DynamicPacket packet;
+			Timestamp last_sent;
+			Timestamp first_sent;
+			size_t times_resent;
+			size_t resend_delay;
+		};
+
+		class DaybreakReliableStream
+		{
+		public:
+			DaybreakReliableStream() {
+				sequence_in = 0;
+				sequence_out = 0;
+				fragment_current_bytes = 0;
+				fragment_total_bytes = 0;
+			}
+			//private:
+
+			uint16_t sequence_in;
+			uint16_t sequence_out;
+			std::map<uint16_t, Packet*> packet_queue;
+
+			DynamicPacket fragment_packet;
+			uint32_t fragment_current_bytes;
+			uint32_t fragment_total_bytes;
+
+			std::map<uint16_t, DaybreakSentPacket> sent_packets;
+		};
+
 		class DaybreakConnectionManager;
 		class DaybreakConnection;
 		class DaybreakConnection
@@ -181,36 +213,7 @@ namespace EQ
 			Timestamp m_close_time;
 			double m_outgoing_budget;
 
-			struct DaybreakSentPacket
-			{
-				DynamicPacket packet;
-				Timestamp last_sent;
-				Timestamp first_sent;
-				size_t times_resent;
-				size_t resend_delay;
-			};
-
-			struct DaybreakStream
-			{
-				DaybreakStream() {
-					sequence_in = 0;
-					sequence_out = 0;
-					fragment_current_bytes = 0;
-					fragment_total_bytes = 0;
-				}
-
-				uint16_t sequence_in;
-				uint16_t sequence_out;
-				std::map<uint16_t, Packet*> packet_queue;
-
-				DynamicPacket fragment_packet;
-				uint32_t fragment_current_bytes;
-				uint32_t fragment_total_bytes;
-
-				std::map<uint16_t, DaybreakSentPacket> sent_packets;
-			};
-
-			DaybreakStream m_streams[4];
+			std::unique_ptr<DaybreakReliableStream> m_streams[4];
 			std::weak_ptr<DaybreakConnection> m_self;
 
 			void Process();
@@ -272,6 +275,14 @@ namespace EQ
 				resend_timeout = 30000;
 				connection_close_time = 2000;
 				outgoing_data_rate = 0.0;
+				//this is emperically based on what the client seems to set for theirs
+				//this is the max number of packets that can be sent/recv per channel before acks come in
+				//we'll use this to preallocate some buffers
+				max_outgoing_packets_per_channel = 400;
+				max_incoming_packets_per_channel = 400;
+
+				//This is the max size of a packet that can be sent or received
+				max_total_packet_size = 1024 * 1024 * 16;
 			}
 
 			size_t max_packet_size;
@@ -295,6 +306,9 @@ namespace EQ
 			DaybreakEncodeType encode_passes[2];
 			int port;
 			double outgoing_data_rate;
+			int max_outgoing_packets_per_channel;
+			int max_incoming_packets_per_channel;
+			int max_total_packet_size;
 		};
 
 		class DaybreakConnectionManager