RakNet: PacketizedTCP.cpp Source File

Go to the documentation of this file.
00001 #include "NativeFeatureIncludes.h"
00002 #if _RAKNET_SUPPORT_PacketizedTCP==1
00003 
00004 #include "PacketizedTCP.h"
00005 #include "NativeTypes.h"
00006 #include "BitStream.h"
00007 #include "MessageIdentifiers.h"
00008 #include "RakAlloca.h"
00009 
00010 typedef uint32_t PTCPHeader;
00011 
00012 PacketizedTCP::PacketizedTCP()
00013 {
00014 
00015 }
00016 PacketizedTCP::~PacketizedTCP()
00017 {
00018         ClearAllConnections();
00019 }
00020 
00021 bool PacketizedTCP::Start(unsigned short port, unsigned short maxIncomingConnections, int threadPriority)
00022 {
00023         bool success = TCPInterface::Start(port, maxIncomingConnections,0,threadPriority);
00024         if (success)
00025         {
00026                 unsigned int i;
00027                 for (i=0; i < messageHandlerList.Size(); i++)
00028                         messageHandlerList[i]->OnRakPeerStartup();
00029         }
00030         return success;
00031 }
00032 
00033 void PacketizedTCP::Stop(void)
00034 {
00035         unsigned int i;
00036         for (i=0; i < messageHandlerList.Size(); i++)
00037                 messageHandlerList[i]->OnRakPeerShutdown();
00038         for (i=0; i < waitingPackets.Size(); i++)
00039                 DeallocatePacket(waitingPackets[i]);
00040         TCPInterface::Stop();
00041         ClearAllConnections();
00042 }
00043 
00044 void PacketizedTCP::Send( const char *data, unsigned length, SystemAddress systemAddress, bool broadcast )
00045 {
00046         PTCPHeader dataLength;
00047         dataLength=length;
00048 #ifndef __BITSTREAM_NATIVE_END
00049         if (RakNet::BitStream::DoEndianSwap())
00050                 RakNet::BitStream::ReverseBytes((unsigned char*) &length,(unsigned char*) &dataLength,sizeof(dataLength));
00051 #else
00052                 dataLength=length;
00053 #endif
00054 
00055         unsigned int lengthsArray[2];
00056         const char *dataArray[2];
00057         dataArray[0]=(char*) &dataLength;
00058         dataArray[1]=data;
00059         lengthsArray[0]=sizeof(dataLength);
00060         lengthsArray[1]=length;
00061         TCPInterface::SendList(dataArray,lengthsArray,2,systemAddress,broadcast);
00062 }
00063 bool PacketizedTCP::SendList( const char **data, const int *lengths, const int numParameters, SystemAddress systemAddress, bool broadcast )
00064 {
00065         if (isStarted==false)
00066                 return false;
00067         if (data==0)
00068                 return false;
00069         if (systemAddress==UNASSIGNED_SYSTEM_ADDRESS && broadcast==false)
00070                 return false;
00071         PTCPHeader totalLengthOfUserData=0;
00072         int i;
00073         for (i=0; i < numParameters; i++)
00074         {
00075                 if (lengths[i]>0)
00076                         totalLengthOfUserData+=lengths[i];
00077         }
00078         if (totalLengthOfUserData==0)
00079                 return false;
00080 
00081         PTCPHeader dataLength;
00082 #ifndef __BITSTREAM_NATIVE_END
00083         if (RakNet::BitStream::DoEndianSwap())
00084                 RakNet::BitStream::ReverseBytes((unsigned char*) &totalLengthOfUserData,(unsigned char*) &dataLength,sizeof(dataLength));
00085 #else
00086         dataLength=totalLengthOfUserData;
00087 #endif
00088         
00089 
00090         unsigned int lengthsArray[512];
00091         const char *dataArray[512];
00092         dataArray[0]=(char*) &dataLength;
00093         lengthsArray[0]=sizeof(dataLength);
00094         for (int i=0; i < 512 && i < numParameters; i++)
00095         {
00096                 dataArray[i+1]=data[i];
00097                 lengthsArray[i+1]=lengths[i];
00098         }       
00099         return TCPInterface::SendList(dataArray,lengthsArray,numParameters+1,systemAddress,broadcast);
00100 }
00101 void PacketizedTCP::PushNotificationsToQueues(void)
00102 {
00103         SystemAddress sa;
00104         sa = TCPInterface::HasNewIncomingConnection();
00105         if (sa!=UNASSIGNED_SYSTEM_ADDRESS)
00106         {
00107                 _newIncomingConnections.Push(sa, __FILE__, __LINE__ );
00108                 AddToConnectionList(sa);
00109                 unsigned int i;
00110                 for (i=0; i < messageHandlerList.Size(); i++)
00111                         messageHandlerList[i]->OnNewConnection(sa, UNASSIGNED_RAKNET_GUID, true);
00112         }
00113 
00114         sa = TCPInterface::HasFailedConnectionAttempt();
00115         if (sa!=UNASSIGNED_SYSTEM_ADDRESS)
00116         {
00117                 _failedConnectionAttempts.Push(sa, __FILE__, __LINE__ );
00118                 unsigned int i;
00119                 for (i=0; i < messageHandlerList.Size(); i++)
00120                 {
00121                         Packet p;
00122                         p.systemAddress=sa;
00123                         p.data=0;
00124                         p.length=0;
00125                         p.bitSize=0;
00126                         messageHandlerList[i]->OnFailedConnectionAttempt(&p, FCAR_CONNECTION_ATTEMPT_FAILED);
00127                 }
00128         }
00129 
00130         sa = TCPInterface::HasLostConnection();
00131         if (sa!=UNASSIGNED_SYSTEM_ADDRESS)
00132         {
00133                 _lostConnections.Push(sa, __FILE__, __LINE__ );
00134                 RemoveFromConnectionList(sa);
00135                 unsigned int i;
00136                 for (i=0; i < messageHandlerList.Size(); i++)
00137                         messageHandlerList[i]->OnClosedConnection(sa, UNASSIGNED_RAKNET_GUID, LCR_DISCONNECTION_NOTIFICATION);
00138         }
00139 
00140         sa = TCPInterface::HasCompletedConnectionAttempt();
00141         if (sa!=UNASSIGNED_SYSTEM_ADDRESS)
00142         {
00143                 _completedConnectionAttempts.Push(sa, __FILE__, __LINE__ );
00144                 AddToConnectionList(sa);
00145                 unsigned int i;
00146                 for (i=0; i < messageHandlerList.Size(); i++)
00147                         messageHandlerList[i]->OnNewConnection(sa, UNASSIGNED_RAKNET_GUID, true);
00148         }
00149 }
00150 Packet* PacketizedTCP::Receive( void )
00151 {
00152         PushNotificationsToQueues();
00153 
00154         unsigned int i;
00155         for (i=0; i < messageHandlerList.Size(); i++)
00156                 messageHandlerList[i]->Update();
00157 
00158         Packet *outgoingPacket=ReturnOutgoingPacket();
00159         if (outgoingPacket)
00160                 return outgoingPacket;
00161 
00162         Packet *incomingPacket;
00163         incomingPacket = TCPInterface::Receive();
00164         unsigned int index;
00165 
00166         while (incomingPacket)
00167         {
00168                 if (connections.Has(incomingPacket->systemAddress))
00169                         index = connections.GetIndexAtKey(incomingPacket->systemAddress);
00170                 else
00171                         index=(unsigned int) -1;
00172                 if ((unsigned int)index==(unsigned int)-1)
00173                 {
00174                         DeallocatePacket(incomingPacket);
00175                         incomingPacket = TCPInterface::Receive();
00176                         continue;
00177                 }
00178 
00179 
00180                 if (incomingPacket->deleteData==true)
00181                 {
00182                         // Came from network
00183                         SystemAddress systemAddressFromPacket;
00184                         if (index < connections.Size())
00185                         {
00186                                 DataStructures::ByteQueue *bq = connections[index];
00187                                 // Buffer data
00188                                 bq->WriteBytes((const char*) incomingPacket->data,incomingPacket->length, __FILE__,__LINE__);
00189                                 systemAddressFromPacket=incomingPacket->systemAddress;
00190                                 PTCPHeader dataLength;
00191 
00192                                 // Peek the header to see if a full message is waiting
00193                                 bq->ReadBytes((char*) &dataLength,sizeof(PTCPHeader),true);
00194                                 if (RakNet::BitStream::DoEndianSwap())
00195                                         RakNet::BitStream::ReverseBytesInPlace((unsigned char*) &dataLength,sizeof(dataLength));
00196                                 // Header indicates packet length. If enough data is available, read out and return one packet
00197                                 if (bq->GetBytesWritten()>=dataLength+sizeof(PTCPHeader))
00198                                 {
00199                                         do 
00200                                         {
00201                                                 bq->IncrementReadOffset(sizeof(PTCPHeader));
00202                                                 outgoingPacket = RakNet::OP_NEW<Packet>(__FILE__, __LINE__);
00203                                                 outgoingPacket->length=dataLength;
00204                                                 outgoingPacket->bitSize=BYTES_TO_BITS(dataLength);
00205                                                 outgoingPacket->guid=UNASSIGNED_RAKNET_GUID;
00206                                                 outgoingPacket->systemAddress=systemAddressFromPacket;
00207                                                 outgoingPacket->deleteData=false; // Did not come from the network
00208                                                 outgoingPacket->data=(unsigned char*) rakMalloc_Ex(dataLength, __FILE__, __LINE__);
00209                                                 if (outgoingPacket->data==0)
00210                                                 {
00211                                                         notifyOutOfMemory(__FILE__, __LINE__);
00212                                                         RakNet::OP_DELETE(outgoingPacket,__FILE__,__LINE__);
00213                                                         return 0;
00214                                                 }
00215                                                 bq->ReadBytes((char*) outgoingPacket->data,dataLength,false);
00216 
00217                                                 waitingPackets.Push(outgoingPacket, __FILE__, __LINE__ );
00218 
00219                                                 // Peek the header to see if a full message is waiting
00220                                                 if (bq->ReadBytes((char*) &dataLength,sizeof(PTCPHeader),true))
00221                                                 {
00222                                                         if (RakNet::BitStream::DoEndianSwap())
00223                                                                 RakNet::BitStream::ReverseBytesInPlace((unsigned char*) &dataLength,sizeof(dataLength));
00224                                                 }
00225                                                 else
00226                                                         break;
00227                                         } while (bq->GetBytesWritten()>=dataLength+sizeof(PTCPHeader));
00228                                 }
00229                                 else
00230                                 {
00231 
00232                                         unsigned int oldWritten = bq->GetBytesWritten()-incomingPacket->length;
00233                                         unsigned int newWritten = bq->GetBytesWritten();
00234 
00235                                         // Return ID_DOWNLOAD_PROGRESS
00236                                         if (newWritten/65536!=oldWritten/65536)
00237                                         {
00238                                                 outgoingPacket = RakNet::OP_NEW<Packet>(__FILE__, __LINE__);
00239                                                 outgoingPacket->length=sizeof(MessageID) +
00240                                                         sizeof(unsigned int)*2 +
00241                                                         sizeof(unsigned int) +
00242                                                         65536;
00243                                                 outgoingPacket->bitSize=BYTES_TO_BITS(incomingPacket->length);
00244                                                 outgoingPacket->guid=UNASSIGNED_RAKNET_GUID;
00245                                                 outgoingPacket->systemAddress=incomingPacket->systemAddress;
00246                                                 outgoingPacket->deleteData=false;
00247                                                 outgoingPacket->data=(unsigned char*) rakMalloc_Ex(outgoingPacket->length, __FILE__, __LINE__);
00248                                                 if (outgoingPacket->data==0)
00249                                                 {
00250                                                         notifyOutOfMemory(__FILE__, __LINE__);
00251                                                         RakNet::OP_DELETE(outgoingPacket,__FILE__,__LINE__);
00252                                                         return 0;
00253                                                 }
00254 
00255                                                 outgoingPacket->data[0]=(MessageID)ID_DOWNLOAD_PROGRESS;
00256                                                 unsigned int totalParts=dataLength/65536;
00257                                                 unsigned int partIndex=newWritten/65536;
00258                                                 unsigned int oneChunkSize=65536;
00259                                                 memcpy(outgoingPacket->data+sizeof(MessageID), &partIndex, sizeof(unsigned int));
00260                                                 memcpy(outgoingPacket->data+sizeof(MessageID)+sizeof(unsigned int)*1, &totalParts, sizeof(unsigned int));
00261                                                 memcpy(outgoingPacket->data+sizeof(MessageID)+sizeof(unsigned int)*2, &oneChunkSize, sizeof(unsigned int));
00262                                                 bq->IncrementReadOffset(sizeof(PTCPHeader));
00263                                                 bq->ReadBytes((char*) outgoingPacket->data+sizeof(MessageID)+sizeof(unsigned int)*3,oneChunkSize,true);
00264                                                 bq->DecrementReadOffset(sizeof(PTCPHeader));
00265 
00266                                                 waitingPackets.Push(outgoingPacket, __FILE__, __LINE__ );
00267                                         }
00268                                 }
00269 
00270                         }
00271 
00272                         DeallocatePacket(incomingPacket);
00273                         incomingPacket=0;
00274                 }
00275                 else
00276                         waitingPackets.Push(incomingPacket, __FILE__, __LINE__ );
00277 
00278                 incomingPacket = TCPInterface::Receive();
00279         }
00280 
00281         return ReturnOutgoingPacket();
00282 }
00283 Packet *PacketizedTCP::ReturnOutgoingPacket(void)
00284 {
00285         Packet *outgoingPacket=0;
00286         unsigned int i;
00287         while (outgoingPacket==0 && waitingPackets.IsEmpty()==false)
00288         {
00289                 outgoingPacket=waitingPackets.Pop();
00290                 PluginReceiveResult pluginResult;
00291                 for (i=0; i < messageHandlerList.Size(); i++)
00292                 {
00293                         pluginResult=messageHandlerList[i]->OnReceive(outgoingPacket);
00294                         if (pluginResult==RR_STOP_PROCESSING_AND_DEALLOCATE)
00295                         {
00296                                 DeallocatePacket( outgoingPacket );
00297                                 outgoingPacket=0; // Will do the loop again and get another packet
00298                                 break; // break out of the enclosing for
00299                         }
00300                         else if (pluginResult==RR_STOP_PROCESSING)
00301                         {
00302                                 outgoingPacket=0;
00303                                 break;
00304                         }
00305                 }
00306         }
00307 
00308         return outgoingPacket;
00309 }
00310 
00311 void PacketizedTCP::AttachPlugin( PluginInterface2 *plugin )
00312 {
00313         if (messageHandlerList.GetIndexOf(plugin)==MAX_UNSIGNED_LONG)
00314         {
00315                 messageHandlerList.Insert(plugin, __FILE__, __LINE__);
00316                 plugin->SetPacketizedTCP(this);
00317                 plugin->OnAttach();
00318         }
00319 }
00320 void PacketizedTCP::DetachPlugin( PluginInterface2 *plugin )
00321 {
00322         if (plugin==0)
00323                 return;
00324 
00325         unsigned int index;
00326         index = messageHandlerList.GetIndexOf(plugin);
00327         if (index!=MAX_UNSIGNED_LONG)
00328         {
00329                 messageHandlerList[index]->OnDetach();
00330                 // Unordered list so delete from end for speed
00331                 messageHandlerList[index]=messageHandlerList[messageHandlerList.Size()-1];
00332                 messageHandlerList.RemoveFromEnd();
00333                 plugin->SetPacketizedTCP(0);
00334         }
00335 }
00336 void PacketizedTCP::CloseConnection( SystemAddress systemAddress )
00337 {
00338         RemoveFromConnectionList(systemAddress);
00339         unsigned int i;
00340         for (i=0; i < messageHandlerList.Size(); i++)
00341                 messageHandlerList[i]->OnClosedConnection(systemAddress, UNASSIGNED_RAKNET_GUID, LCR_CLOSED_BY_USER);
00342         TCPInterface::CloseConnection(systemAddress);
00343 }
00344 void PacketizedTCP::RemoveFromConnectionList(SystemAddress sa)
00345 {
00346         if (sa==UNASSIGNED_SYSTEM_ADDRESS)
00347                 return;
00348         if (connections.Has(sa))
00349         {
00350                 unsigned int index = connections.GetIndexAtKey(sa);
00351                 if (index!=(unsigned int)-1)
00352                 {
00353                         RakNet::OP_DELETE(connections[index],__FILE__,__LINE__);
00354                         connections.RemoveAtIndex(index);
00355                 }
00356         }
00357 }
00358 void PacketizedTCP::AddToConnectionList(SystemAddress sa)
00359 {
00360         if (sa==UNASSIGNED_SYSTEM_ADDRESS)
00361                 return;
00362         connections.SetNew(sa, RakNet::OP_NEW<DataStructures::ByteQueue>(__FILE__, __LINE__));
00363 }
00364 void PacketizedTCP::ClearAllConnections(void)
00365 {
00366         unsigned int i;
00367         for (i=0; i < connections.Size(); i++)
00368                 RakNet::OP_DELETE(connections[i],__FILE__,__LINE__);
00369         connections.Clear();
00370 }
00371 SystemAddress PacketizedTCP::HasCompletedConnectionAttempt(void)
00372 {
00373         PushNotificationsToQueues();
00374 
00375         if (_completedConnectionAttempts.IsEmpty()==false)
00376                 return _completedConnectionAttempts.Pop();
00377         return UNASSIGNED_SYSTEM_ADDRESS;
00378 }
00379 SystemAddress PacketizedTCP::HasFailedConnectionAttempt(void)
00380 {
00381         PushNotificationsToQueues();
00382 
00383         if (_failedConnectionAttempts.IsEmpty()==false)
00384                 return _failedConnectionAttempts.Pop();
00385         return UNASSIGNED_SYSTEM_ADDRESS;
00386 }
00387 SystemAddress PacketizedTCP::HasNewIncomingConnection(void)
00388 {
00389         PushNotificationsToQueues();
00390 
00391         if (_newIncomingConnections.IsEmpty()==false)
00392                 return _newIncomingConnections.Pop();
00393         return UNASSIGNED_SYSTEM_ADDRESS;
00394 }
00395 SystemAddress PacketizedTCP::HasLostConnection(void)
00396 {
00397         PushNotificationsToQueues();
00398 
00399         if (_lostConnections.IsEmpty()==false)
00400                 return _lostConnections.Pop();
00401         return UNASSIGNED_SYSTEM_ADDRESS;
00402 }
00403 
00404 #endif // _RAKNET_SUPPORT_*