31 public static int $WINDOW_SIZE = 2048;
33 private int $windowStart;
34 private int $windowEnd;
35 private int $highestSeqNumber = -1;
38 private array $ACKQueue = [];
40 private array $NACKQueue = [];
42 private int $reliableWindowStart;
43 private int $reliableWindowEnd;
45 private array $reliableWindow = [];
48 private array $receiveOrderedIndex;
50 private array $receiveSequencedHighestIndex;
52 private array $receiveOrderedPackets;
55 private array $splitPackets = [];
64 private \Closure $onRecv,
65 private \Closure $sendPacket,
66 private int $maxSplitPacketPartCount = PHP_INT_MAX,
67 private int $maxConcurrentSplitPackets = PHP_INT_MAX
69 $this->windowStart = 0;
70 $this->windowEnd = self::$WINDOW_SIZE;
72 $this->reliableWindowStart = 0;
73 $this->reliableWindowEnd = self::$WINDOW_SIZE;
75 $this->receiveOrderedIndex = array_fill(0, PacketReliability::MAX_ORDER_CHANNELS, 0);
76 $this->receiveSequencedHighestIndex = array_fill(0, PacketReliability::MAX_ORDER_CHANNELS, 0);
78 $this->receiveOrderedPackets = array_fill(0, PacketReliability::MAX_ORDER_CHANNELS, []);
96 private function handleSplit(EncapsulatedPacket $packet) : ?EncapsulatedPacket{
97 if($packet->splitInfo === null){
100 $totalParts = $packet->splitInfo->getTotalPartCount();
101 $partIndex = $packet->splitInfo->getPartIndex();
102 if($totalParts >= $this->maxSplitPacketPartCount || $totalParts < 0){
103 throw new PacketHandlingException(
"Invalid split packet part count ($totalParts)", DisconnectReason::SPLIT_PACKET_TOO_LARGE);
105 if($partIndex >= $totalParts || $partIndex < 0){
106 throw new PacketHandlingException(
"Invalid split packet part index (part index $partIndex, part count $totalParts)", DisconnectReason::SPLIT_PACKET_INVALID_PART_INDEX);
109 $splitId = $packet->splitInfo->getId();
110 if(!isset($this->splitPackets[$splitId])){
111 if(count($this->splitPackets) >= $this->maxConcurrentSplitPackets){
112 throw new PacketHandlingException(
"Exceeded concurrent split packet reassembly limit of $this->maxConcurrentSplitPackets", DisconnectReason::SPLIT_PACKET_TOO_MANY_CONCURRENT);
114 $this->splitPackets[$splitId] = array_fill(0, $totalParts,
null);
115 }elseif(count($this->splitPackets[$splitId]) !== $totalParts){
116 throw new PacketHandlingException(
"Wrong split count $totalParts for split packet $splitId, expected " . count($this->splitPackets[$splitId]), DisconnectReason::SPLIT_PACKET_INCONSISTENT_HEADER);
119 $this->splitPackets[$splitId][$partIndex] = $packet;
122 foreach($this->splitPackets[$splitId] as $splitIndex => $part){
126 $parts[$splitIndex] = $part;
130 $pk =
new EncapsulatedPacket();
133 $pk->reliability = $packet->reliability;
134 $pk->messageIndex = $packet->messageIndex;
135 $pk->sequenceIndex = $packet->sequenceIndex;
136 $pk->orderIndex = $packet->orderIndex;
137 $pk->orderChannel = $packet->orderChannel;
139 for($i = 0; $i < $totalParts; ++$i){
140 $pk->buffer .= $parts[$i]->buffer;
143 unset($this->splitPackets[$splitId]);
151 private function handleEncapsulatedPacket(EncapsulatedPacket $packet) : void{
152 if($packet->messageIndex !== null){
154 if($packet->messageIndex < $this->reliableWindowStart or $packet->messageIndex > $this->reliableWindowEnd or isset($this->reliableWindow[$packet->messageIndex])){
158 $this->reliableWindow[$packet->messageIndex] =
true;
160 if($packet->messageIndex === $this->reliableWindowStart){
161 for(; isset($this->reliableWindow[$this->reliableWindowStart]); ++$this->reliableWindowStart){
162 unset($this->reliableWindow[$this->reliableWindowStart]);
163 ++$this->reliableWindowEnd;
168 if(($packet = $this->handleSplit($packet)) ===
null){
172 if(PacketReliability::isSequencedOrOrdered($packet->reliability) and ($packet->orderChannel < 0 or $packet->orderChannel >= PacketReliability::MAX_ORDER_CHANNELS)){
174 $this->logger->debug(
"Invalid packet, bad order channel ($packet->orderChannel)");
178 if(PacketReliability::isSequenced($packet->reliability)){
179 if($packet->sequenceIndex < $this->receiveSequencedHighestIndex[$packet->orderChannel] or $packet->orderIndex < $this->receiveOrderedIndex[$packet->orderChannel]){
184 $this->receiveSequencedHighestIndex[$packet->orderChannel] = $packet->sequenceIndex + 1;
185 $this->handleEncapsulatedPacketRoute($packet);
186 }elseif(PacketReliability::isOrdered($packet->reliability)){
187 if($packet->orderIndex === $this->receiveOrderedIndex[$packet->orderChannel]){
192 $this->receiveSequencedHighestIndex[$packet->orderChannel] = 0;
193 $this->receiveOrderedIndex[$packet->orderChannel] = $packet->orderIndex + 1;
195 $this->handleEncapsulatedPacketRoute($packet);
196 $i = $this->receiveOrderedIndex[$packet->orderChannel];
197 for(; isset($this->receiveOrderedPackets[$packet->orderChannel][$i]); ++$i){
198 $this->handleEncapsulatedPacketRoute($this->receiveOrderedPackets[$packet->orderChannel][$i]);
199 unset($this->receiveOrderedPackets[$packet->orderChannel][$i]);
202 $this->receiveOrderedIndex[$packet->orderChannel] = $i;
203 }elseif($packet->orderIndex > $this->receiveOrderedIndex[$packet->orderChannel]){
204 if(count($this->receiveOrderedPackets[$packet->orderChannel]) >= self::$WINDOW_SIZE){
208 $this->receiveOrderedPackets[$packet->orderChannel][$packet->orderIndex] = $packet;
214 $this->handleEncapsulatedPacketRoute($packet);
222 if($packet->seqNumber < $this->windowStart or $packet->seqNumber > $this->windowEnd or isset($this->ACKQueue[$packet->seqNumber])){
223 $this->logger->debug(
"Received duplicate or out-of-window packet (sequence number $packet->seqNumber, window " . $this->windowStart .
"-" . $this->windowEnd .
")");
227 unset($this->NACKQueue[$packet->seqNumber]);
228 $this->ACKQueue[$packet->seqNumber] = $packet->seqNumber;
229 if($this->highestSeqNumber < $packet->seqNumber){
230 $this->highestSeqNumber = $packet->seqNumber;
233 if($packet->seqNumber === $this->windowStart){
237 for(; isset($this->ACKQueue[$this->windowStart]); ++$this->windowStart){
240 }elseif($packet->seqNumber > $this->windowStart){
244 for($i = $this->windowStart; $i < $packet->seqNumber; ++$i){
245 if(!isset($this->ACKQueue[$i])){
246 $this->NACKQueue[$i] = $i;
250 assert(
false,
"received packet before window start");
253 foreach($packet->packets as $pk){
254 $this->handleEncapsulatedPacket($pk);
258 public function update() : void{
259 $diff = $this->highestSeqNumber - $this->windowStart + 1;
266 $this->windowStart += $diff;
267 $this->windowEnd += $diff;
270 if(count($this->ACKQueue) > 0){
272 $pk->packets = $this->ACKQueue;
273 ($this->sendPacket)($pk);
274 $this->ACKQueue = [];
277 if(count($this->NACKQueue) > 0){
279 $pk->packets = $this->NACKQueue;
280 ($this->sendPacket)($pk);
281 $this->NACKQueue = [];
285 public function needsUpdate() : bool{
286 return count($this->ACKQueue) !== 0 or count($this->NACKQueue) !== 0;