ocrm.bsky.social
the game where you go into mines and start crafting! but for consoles (forked directly from smartcmd's github)
1#include "stdafx.h"
2#include "InputOutputStream.h"
3#include "SocketAddress.h"
4#include "Socket.h"
5#include "ThreadName.h"
6#include "..\Minecraft.Client\ServerConnection.h"
7#include <algorithm>
8#include "..\Minecraft.Client\PS3\PS3Extras\ShutdownManager.h"
9
10// This current socket implementation is for the creation of a single local link. 2 sockets can be created, one for either end of this local
11// link, the end (0 or 1) is passed as a parameter to the ctor.
12
13CRITICAL_SECTION Socket::s_hostQueueLock[2];
14std::queue<byte> Socket::s_hostQueue[2];
15Socket::SocketOutputStreamLocal *Socket::s_hostOutStream[2];
16Socket::SocketInputStreamLocal *Socket::s_hostInStream[2];
17ServerConnection *Socket::s_serverConnection = NULL;
18
19void Socket::Initialise(ServerConnection *serverConnection)
20{
21 s_serverConnection = serverConnection;
22
23 // Only initialise everything else once - just setting up static data, one time xrnm things, thread for ticking sockets
24 static bool init = false;
25 if( init )
26 {
27 for( int i = 0; i < 2; i++ )
28 {
29 if(TryEnterCriticalSection(&s_hostQueueLock[i]))
30 {
31 // Clear the queue
32 std::queue<byte> empty;
33 std::swap( s_hostQueue[i], empty );
34 LeaveCriticalSection(&s_hostQueueLock[i]);
35 }
36 s_hostOutStream[i]->m_streamOpen = true;
37 s_hostInStream[i]->m_streamOpen = true;
38 }
39 return;
40 }
41 init = true;
42
43 for( int i = 0; i < 2; i++ )
44 {
45 InitializeCriticalSection(&Socket::s_hostQueueLock[i]);
46 s_hostOutStream[i] = new SocketOutputStreamLocal(i);
47 s_hostInStream[i] = new SocketInputStreamLocal(i);
48 }
49}
50
51Socket::Socket(bool response)
52{
53 m_hostServerConnection = true;
54 m_hostLocal = true;
55 if( response )
56 {
57 m_end = SOCKET_SERVER_END;
58 }
59 else
60 {
61 m_end = SOCKET_CLIENT_END;
62 Socket *socket = new Socket(1);
63 s_serverConnection->NewIncomingSocket(socket);
64 }
65
66 for( int i = 0; i < 2; i++ )
67 {
68 m_endClosed[i] = false;
69 }
70 m_socketClosedEvent = NULL;
71 createdOk = true;
72 networkPlayerSmallId = g_NetworkManager.GetHostPlayer()->GetSmallId();
73}
74
75Socket::Socket(INetworkPlayer *player, bool response /* = false*/, bool hostLocal /*= false*/)
76{
77 m_hostServerConnection = false;
78 m_hostLocal = hostLocal;
79
80 for( int i = 0; i < 2; i++ )
81 {
82 InitializeCriticalSection(&m_queueLockNetwork[i]);
83 m_inputStream[i] = NULL;
84 m_outputStream[i] = NULL;
85 m_endClosed[i] = false;
86 }
87
88 if(!response || hostLocal)
89 {
90 m_inputStream[0] = new SocketInputStreamNetwork(this,0);
91 m_outputStream[0] = new SocketOutputStreamNetwork(this,0);
92 m_end = SOCKET_CLIENT_END;
93 }
94 if(response || hostLocal)
95 {
96 m_inputStream[1] = new SocketInputStreamNetwork(this,1);
97 m_outputStream[1] = new SocketOutputStreamNetwork(this,1);
98 m_end = SOCKET_SERVER_END;
99 }
100 m_socketClosedEvent = new C4JThread::Event;
101 //printf("New socket made %s\n", player->GetGamertag() );
102 networkPlayerSmallId = player->GetSmallId();
103 createdOk = true;
104}
105
106SocketAddress *Socket::getRemoteSocketAddress()
107{
108 return NULL;
109}
110
111INetworkPlayer *Socket::getPlayer()
112{
113 return g_NetworkManager.GetPlayerBySmallId(networkPlayerSmallId);
114}
115
116void Socket::setPlayer(INetworkPlayer *player)
117{
118 if(player!=NULL)
119 {
120 networkPlayerSmallId = player->GetSmallId();
121 }
122 else
123 {
124 networkPlayerSmallId = 0;
125 }
126}
127
128void Socket::pushDataToQueue(const BYTE * pbData, DWORD dwDataSize, bool fromHost /*= true*/)
129{
130 int queueIdx = SOCKET_CLIENT_END;
131 if(!fromHost)
132 queueIdx = SOCKET_SERVER_END;
133
134 if( queueIdx != m_end && !m_hostLocal )
135 {
136 app.DebugPrintf("SOCKET: Error pushing data to queue. End is %d but queue idx id %d\n", m_end, queueIdx);
137 return;
138 }
139
140 EnterCriticalSection(&m_queueLockNetwork[queueIdx]);
141 for( unsigned int i = 0; i < dwDataSize; i++ )
142 {
143 m_queueNetwork[queueIdx].push(*pbData++);
144 }
145 LeaveCriticalSection(&m_queueLockNetwork[queueIdx]);
146}
147
148void Socket::addIncomingSocket(Socket *socket)
149{
150 if( s_serverConnection != NULL )
151 {
152 s_serverConnection->NewIncomingSocket(socket);
153 }
154}
155
156InputStream *Socket::getInputStream(bool isServerConnection)
157{
158 if( !m_hostServerConnection )
159 {
160 if( m_hostLocal )
161 {
162 if( isServerConnection )
163 {
164 return m_inputStream[SOCKET_SERVER_END];
165 }
166 else
167 {
168 return m_inputStream[SOCKET_CLIENT_END];
169 }
170 }
171 else
172 {
173 return m_inputStream[m_end];
174 }
175 }
176 else
177 {
178 return s_hostInStream[m_end];
179 }
180}
181
182void Socket::setSoTimeout(int a )
183{
184}
185
186void Socket::setTrafficClass( int a )
187{
188}
189
190Socket::SocketOutputStream *Socket::getOutputStream(bool isServerConnection)
191{
192 if( !m_hostServerConnection )
193 {
194 if( m_hostLocal )
195 {
196 if( isServerConnection )
197 {
198 return m_outputStream[SOCKET_SERVER_END];
199 }
200 else
201 {
202 return m_outputStream[SOCKET_CLIENT_END];
203 }
204 }
205 else
206 {
207 return m_outputStream[m_end];
208 }
209 }
210 else
211 {
212 return s_hostOutStream[ 1 - m_end ];
213 }
214}
215
216bool Socket::close(bool isServerConnection)
217{
218 bool allClosed = false;
219 if( m_hostLocal )
220 {
221 if( isServerConnection )
222 {
223 m_endClosed[SOCKET_SERVER_END] = true;
224 if(m_endClosed[SOCKET_CLIENT_END])
225 {
226 allClosed = true;
227 }
228 }
229 else
230 {
231 m_endClosed[SOCKET_CLIENT_END] = true;
232 if(m_endClosed[SOCKET_SERVER_END])
233 {
234 allClosed = true;
235 }
236 }
237 }
238 else
239 {
240 allClosed = true;
241 m_endClosed[m_end] = true;
242 }
243 if( allClosed && m_socketClosedEvent != NULL )
244 {
245 m_socketClosedEvent->Set();
246 }
247 if(allClosed) createdOk = false;
248 return allClosed;
249}
250
251/////////////////////////////////// Socket for input, on local connection ////////////////////
252
253Socket::SocketInputStreamLocal::SocketInputStreamLocal(int queueIdx)
254{
255 m_streamOpen = true;
256 m_queueIdx = queueIdx;
257}
258
259// Try and get an input byte, blocking until one is available
260int Socket::SocketInputStreamLocal::read()
261{
262 while(m_streamOpen && ShutdownManager::ShouldRun(ShutdownManager::eConnectionReadThreads))
263 {
264 if(TryEnterCriticalSection(&s_hostQueueLock[m_queueIdx]))
265 {
266 if( s_hostQueue[m_queueIdx].size() )
267 {
268 byte retval = s_hostQueue[m_queueIdx].front();
269 s_hostQueue[m_queueIdx].pop();
270 LeaveCriticalSection(&s_hostQueueLock[m_queueIdx]);
271 return retval;
272 }
273 LeaveCriticalSection(&s_hostQueueLock[m_queueIdx]);
274 }
275 Sleep(1);
276 }
277 return -1;
278}
279
280// Try and get an input array of bytes, blocking until enough bytes are available
281int Socket::SocketInputStreamLocal::read(byteArray b)
282{
283 return read(b, 0, b.length);
284}
285
286// Try and get an input range of bytes, blocking until enough bytes are available
287int Socket::SocketInputStreamLocal::read(byteArray b, unsigned int offset, unsigned int length)
288{
289 while(m_streamOpen)
290 {
291 if(TryEnterCriticalSection(&s_hostQueueLock[m_queueIdx]))
292 {
293 if( s_hostQueue[m_queueIdx].size() >= length )
294 {
295 for( unsigned int i = 0; i < length; i++ )
296 {
297 b[i+offset] = s_hostQueue[m_queueIdx].front();
298 s_hostQueue[m_queueIdx].pop();
299 }
300 LeaveCriticalSection(&s_hostQueueLock[m_queueIdx]);
301 return length;
302 }
303 LeaveCriticalSection(&s_hostQueueLock[m_queueIdx]);
304 }
305 Sleep(1);
306 }
307 return -1;
308}
309
310void Socket::SocketInputStreamLocal::close()
311{
312 m_streamOpen = false;
313 EnterCriticalSection(&s_hostQueueLock[m_queueIdx]);
314 s_hostQueue[m_queueIdx].empty();
315 LeaveCriticalSection(&s_hostQueueLock[m_queueIdx]);
316}
317
318/////////////////////////////////// Socket for output, on local connection ////////////////////
319
320Socket::SocketOutputStreamLocal::SocketOutputStreamLocal(int queueIdx)
321{
322 m_streamOpen = true;
323 m_queueIdx = queueIdx;
324}
325
326void Socket::SocketOutputStreamLocal::write(unsigned int b)
327{
328 if( m_streamOpen != true )
329 {
330 return;
331 }
332 EnterCriticalSection(&s_hostQueueLock[m_queueIdx]);
333 s_hostQueue[m_queueIdx].push((byte)b);
334 LeaveCriticalSection(&s_hostQueueLock[m_queueIdx]);
335}
336
337void Socket::SocketOutputStreamLocal::write(byteArray b)
338{
339 write(b, 0, b.length);
340}
341
342void Socket::SocketOutputStreamLocal::write(byteArray b, unsigned int offset, unsigned int length)
343{
344 if( m_streamOpen != true )
345 {
346 return;
347 }
348 MemSect(12);
349 EnterCriticalSection(&s_hostQueueLock[m_queueIdx]);
350 for( unsigned int i = 0; i < length; i++ )
351 {
352 s_hostQueue[m_queueIdx].push(b[offset+i]);
353 }
354 LeaveCriticalSection(&s_hostQueueLock[m_queueIdx]);
355 MemSect(0);
356}
357
358void Socket::SocketOutputStreamLocal::close()
359{
360 m_streamOpen = false;
361 EnterCriticalSection(&s_hostQueueLock[m_queueIdx]);
362 s_hostQueue[m_queueIdx].empty();
363 LeaveCriticalSection(&s_hostQueueLock[m_queueIdx]);
364}
365
366/////////////////////////////////// Socket for input, on network connection ////////////////////
367
368Socket::SocketInputStreamNetwork::SocketInputStreamNetwork(Socket *socket, int queueIdx)
369{
370 m_streamOpen = true;
371 m_queueIdx = queueIdx;
372 m_socket = socket;
373}
374
375// Try and get an input byte, blocking until one is available
376int Socket::SocketInputStreamNetwork::read()
377{
378 while(m_streamOpen && ShutdownManager::ShouldRun(ShutdownManager::eConnectionReadThreads))
379 {
380 if(TryEnterCriticalSection(&m_socket->m_queueLockNetwork[m_queueIdx]))
381 {
382 if( m_socket->m_queueNetwork[m_queueIdx].size() )
383 {
384 byte retval = m_socket->m_queueNetwork[m_queueIdx].front();
385 m_socket->m_queueNetwork[m_queueIdx].pop();
386 LeaveCriticalSection(&m_socket->m_queueLockNetwork[m_queueIdx]);
387 return retval;
388 }
389 LeaveCriticalSection(&m_socket->m_queueLockNetwork[m_queueIdx]);
390 }
391 Sleep(1);
392 }
393 return -1;
394}
395
396// Try and get an input array of bytes, blocking until enough bytes are available
397int Socket::SocketInputStreamNetwork::read(byteArray b)
398{
399 return read(b, 0, b.length);
400}
401
402// Try and get an input range of bytes, blocking until enough bytes are available
403int Socket::SocketInputStreamNetwork::read(byteArray b, unsigned int offset, unsigned int length)
404{
405 while(m_streamOpen)
406 {
407 if(TryEnterCriticalSection(&m_socket->m_queueLockNetwork[m_queueIdx]))
408 {
409 if( m_socket->m_queueNetwork[m_queueIdx].size() >= length )
410 {
411 for( unsigned int i = 0; i < length; i++ )
412 {
413 b[i+offset] = m_socket->m_queueNetwork[m_queueIdx].front();
414 m_socket->m_queueNetwork[m_queueIdx].pop();
415 }
416 LeaveCriticalSection(&m_socket->m_queueLockNetwork[m_queueIdx]);
417 return length;
418 }
419 LeaveCriticalSection(&m_socket->m_queueLockNetwork[m_queueIdx]);
420 }
421 Sleep(1);
422 }
423 return -1;
424}
425
426void Socket::SocketInputStreamNetwork::close()
427{
428 m_streamOpen = false;
429}
430
431/////////////////////////////////// Socket for output, on network connection ////////////////////
432
433Socket::SocketOutputStreamNetwork::SocketOutputStreamNetwork(Socket *socket, int queueIdx)
434{
435 m_queueIdx = queueIdx;
436 m_socket = socket;
437 m_streamOpen = true;
438}
439
440void Socket::SocketOutputStreamNetwork::write(unsigned int b)
441{
442 if( m_streamOpen != true ) return;
443 byteArray barray;
444 byte bb;
445 bb = (byte)b;
446 barray.data = &bb;
447 barray.length = 1;
448 write(barray, 0, 1);
449
450}
451
452void Socket::SocketOutputStreamNetwork::write(byteArray b)
453{
454 write(b, 0, b.length);
455}
456
457void Socket::SocketOutputStreamNetwork::write(byteArray b, unsigned int offset, unsigned int length)
458{
459 writeWithFlags(b, offset, length, 0);
460}
461
462void Socket::SocketOutputStreamNetwork::writeWithFlags(byteArray b, unsigned int offset, unsigned int length, int flags)
463{
464 if( m_streamOpen != true ) return;
465 if( length == 0 ) return;
466
467 // If this is a local connection, don't bother going through QNet as it just delivers it straight anyway
468 if( m_socket->m_hostLocal )
469 {
470 // We want to write to the queue for the other end of this socket stream
471 int queueIdx = m_queueIdx;
472 if(queueIdx == SOCKET_CLIENT_END)
473 queueIdx = SOCKET_SERVER_END;
474 else
475 queueIdx = SOCKET_CLIENT_END;
476
477 EnterCriticalSection(&m_socket->m_queueLockNetwork[queueIdx]);
478 for( unsigned int i = 0; i < length; i++ )
479 {
480 m_socket->m_queueNetwork[queueIdx].push(b[offset+i]);
481 }
482 LeaveCriticalSection(&m_socket->m_queueLockNetwork[queueIdx]);
483 }
484 else
485 {
486 XRNM_SEND_BUFFER buffer;
487 buffer.pbyData = &b[offset];
488 buffer.dwDataSize = length;
489
490 INetworkPlayer *hostPlayer = g_NetworkManager.GetHostPlayer();
491 if(hostPlayer == NULL)
492 {
493 app.DebugPrintf("Trying to write to network, but the hostPlayer is NULL\n");
494 return;
495 }
496 INetworkPlayer *socketPlayer = m_socket->getPlayer();
497 if(socketPlayer == NULL)
498 {
499 app.DebugPrintf("Trying to write to network, but the socketPlayer is NULL\n");
500 return;
501 }
502
503#ifdef _XBOX
504 bool lowPriority = ( ( flags & QNET_SENDDATA_LOW_PRIORITY ) == QNET_SENDDATA_LOW_PRIORITY );
505 bool requireAck = lowPriority;
506#else
507 bool lowPriority = false;
508 bool requireAck = ( ( flags & NON_QNET_SENDDATA_ACK_REQUIRED ) == NON_QNET_SENDDATA_ACK_REQUIRED );
509#endif
510
511 if( m_queueIdx == SOCKET_SERVER_END )
512 {
513 //printf( "Sent %u bytes of data from \"%ls\" to \"%ls\"\n",
514 //buffer.dwDataSize,
515 //hostPlayer->GetGamertag(),
516 //m_socket->networkPlayer->GetGamertag());
517
518 hostPlayer->SendData(socketPlayer, buffer.pbyData, buffer.dwDataSize, lowPriority, requireAck);
519
520 // DWORD queueSize = hostPlayer->GetSendQueueSize( NULL, QNET_GETSENDQUEUESIZE_BYTES );
521 // if( queueSize > 24000 )
522 // {
523 // //printf("Queue size is: %d, forcing doWork()\n",queueSize);
524 // g_NetworkManager.DoWork();
525 // }
526 }
527 else
528 {
529 //printf( "Sent %u bytes of data from \"%ls\" to \"%ls\"\n",
530 //buffer.dwDataSize,
531 //m_socket->networkPlayer->GetGamertag(),
532 //hostPlayer->GetGamertag());
533
534 socketPlayer->SendData(hostPlayer, buffer.pbyData, buffer.dwDataSize, lowPriority, requireAck);
535 }
536 }
537}
538
539void Socket::SocketOutputStreamNetwork::close()
540{
541 m_streamOpen = false;
542}