Finished video depacketizing. Starting tests

limelight-common/ByteBuffer.cpp

@@ -31,7 +31,8 @@ int BbGet(PBYTE_BUFFER buff, char *c) {
 		return 0;
 	}
 
-	memcpy(c, &buff[buff->position], sizeof(*c));
+	memcpy(c, &buff->buffer[buff->position], sizeof(*c));
+	buff->position += sizeof(c);
 
 	return 1;
 }
@@ -41,7 +42,8 @@ int BbGetShort(PBYTE_BUFFER buff, short *s) {
 		return 0;
 	}
 
-	memcpy(s, &buff[buff->position], sizeof(*s));
+	memcpy(s, &buff->buffer[buff->position], sizeof(*s));
+	buff->position += sizeof(s);
 
 	*s = byteSwapShort(buff, *s);
 
@@ -53,7 +55,8 @@ int BbGetInt(PBYTE_BUFFER buff, int *i) {
 		return 0;
 	}
 
-	memcpy(i, &buff[buff->position], sizeof(*i));
+	memcpy(i, &buff->buffer[buff->position], sizeof(*i));
+	buff->position += sizeof(i);
 
 	*i = byteSwapInt(buff, *i);
 
@@ -67,7 +70,7 @@ int BbPutInt(PBYTE_BUFFER buff, int i) {
 
 	i = byteSwapInt(buff, i);
 
-	memcpy(&buff[buff->position], &i, sizeof(i));
+	memcpy(&buff->buffer[buff->position], &i, sizeof(i));
 	buff->position += sizeof(i);
 
 	return 1;
@@ -80,7 +83,7 @@ int BbPutShort(PBYTE_BUFFER buff, short s) {
 
 	s = byteSwapShort(buff, s);
 
-	memcpy(&buff[buff->position], &s, sizeof(s));
+	memcpy(&buff->buffer[buff->position], &s, sizeof(s));
 	buff->position += sizeof(s);
 
 	return 1;
@@ -91,7 +94,7 @@ int BbPut(PBYTE_BUFFER buff, char c) {
 		return 0;
 	}
 
-	memcpy(&buff[buff->position], &c, sizeof(c));
+	memcpy(&buff->buffer[buff->position], &c, sizeof(c));
 	buff->position += sizeof(c);
 	
 	return 1;

limelight-common/ControlStream.cpp

@@ -12,6 +12,7 @@ STREAM_CONFIGURATION streamConfig;
 PLT_THREAD heartbeatThread;
 PLT_THREAD jitterThread;
 PLT_THREAD resyncThread;
+PLT_EVENT resyncEvent;
 
 const short PTYPE_KEEPALIVE = 0x13ff;
 const short PPAYLEN_KEEPALIVE = 0x0000;
@@ -42,7 +43,7 @@ int initializeControlStream(IP_ADDRESS host, PSTREAM_CONFIGURATION streamConfigP
 }
 
 void requestIdrFrame(void) {
-
+	PltSetEvent(resyncEvent);
 }
 
 static PNVCTL_PACKET_HEADER readNvctlPacket(void) {
@@ -97,7 +98,7 @@ static void heartbeatThreadFunc(void* context) {
 			return;
 		}
 
-		Sleep(3000);
+		PltSleepMs(3000);
 	}
 }
 
@@ -126,11 +127,37 @@ static void jitterThreadFunc(void* context) {
 			return;
 		}
 
-		Sleep(100);
+		PltSleepMs(100);
 	}
 }
 
 static void resyncThreadFunc(void* context) {
+	long payload[2];
+	NVCTL_PACKET_HEADER header;
+	int err;
+
+	header.type = PTYPE_RESYNC;
+	header.payloadLength = PPAYLEN_RESYNC;
+	for (;;) {
+		PltWaitForEvent(resyncEvent);
+
+		err = send(ctlSock, (char*) &header, sizeof(header), 0);
+		if (err != sizeof(header)) {
+			Limelog("Resync thread terminating #1\n");
+			return;
+		}
+
+		payload[0] = 0;
+		payload[1] = 0xFFFFF;
+
+		err = send(ctlSock, (char*) payload, sizeof(payload), 0);
+		if (err != sizeof(payload)) {
+			Limelog("Resync thread terminating #2\n");
+			return;
+		}
+
+		PltClearEvent(resyncEvent);
+	}
 }
 
 int stopControlStream(void) {

limelight-common/Limelight.h

@@ -33,6 +33,7 @@ int getConfigDataSize(PSTREAM_CONFIGURATION streamConfig);
 int initializeControlStream(IP_ADDRESS host, PSTREAM_CONFIGURATION streamConfig);
 int startControlStream(void);
 int stopControlStream(void);
+void requestIdrFrame(void);
 
 int performHandshake(IP_ADDRESS host);
 
@@ -40,4 +41,7 @@ void initializeVideoDepacketizer(void);
 void processRtpPayload(PNV_VIDEO_PACKET videoPacket, int length);
 PDECODE_UNIT getNextDecodeUnit(void);
 void freeDecodeUnit(PDECODE_UNIT decodeUnit);
-void queueRtpPacket(PRTP_PACKET rtpPacket, int length);
+void queueRtpPacket(PRTP_PACKET rtpPacket, int length);
+
+void initializeVideoStream(IP_ADDRESS host, PSTREAM_CONFIGURATION streamConfig, PDECODER_RENDERER_CALLBACKS drCallbacks);
+int startVideoStream(void* rendererContext, int drFlags);

limelight-common/PlatformSockets.cpp

@@ -1,5 +1,25 @@
 #include "PlatformSockets.h"
 
+SOCKET bindUdpSocket(unsigned short port) {
+	SOCKET s;
+	struct sockaddr_in addr;
+
+	s = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+	if (s == INVALID_SOCKET) {
+		return INVALID_SOCKET;
+	}
+
+	memset(&addr, 0, sizeof(addr));
+	addr.sin_family = AF_INET;
+	addr.sin_port = htons(port);
+	if (bind(s, (struct sockaddr*) &addr, sizeof(addr)) == SOCKET_ERROR) {
+		closesocket(s);
+		return INVALID_SOCKET;
+	}
+
+	return s;
+}
+
 SOCKET connectTcpSocket(IP_ADDRESS dstaddr, unsigned short port) {
 	SOCKET s;
 	struct sockaddr_in addr;

limelight-common/PlatformSockets.h

@@ -14,4 +14,5 @@
 #define IP_ADDRESS unsigned int
 
 SOCKET connectTcpSocket(IP_ADDRESS dstaddr, unsigned short port);
+SOCKET bindUdpSocket(unsigned short port);
 int enableNoDelay(SOCKET s);

limelight-common/PlatformThreads.cpp

@@ -19,6 +19,13 @@ DWORD WINAPI ThreadProc(LPVOID lpParameter) {
 #error POSIX threads not implemented
 #endif
 
+void PltSleepMs(int ms) {
+#ifdef _WIN32
+	Sleep(ms);
+#else
+#endif
+}
+
 int PltCreateMutex(PLT_MUTEX *mutex) {
 #ifdef _WIN32
 	*mutex = CreateMutex(NULL, FALSE, NULL);

limelight-common/PlatformThreads.h

@@ -25,4 +25,6 @@ void PltCloseEvent(PLT_EVENT event);
 void PltSetEvent(PLT_EVENT event);
 void PltClearEvent(PLT_EVENT event);
 void PltPulseEvent(PLT_EVENT event);
-void PltWaitForEvent(PLT_EVENT event);
+void PltWaitForEvent(PLT_EVENT event);
+
+void PltSleepMs(int ms);

limelight-common/Video.h

@@ -1,11 +1,5 @@
 #pragma once
 
-typedef void(*RequestIdrFrame)(void);
-
-typedef struct _CONTROL_STATUS_LISTENER {
-	RequestIdrFrame requestIdrFrame;
-} CONTROL_STATUS_LISTENER, *PCONTROL_STATUS_LISTENER;
-
 typedef struct _LENTRY {
 	struct _LENTRY *next;
 	char* data;

limelight-common/VideoDepacketizer.cpp

@@ -99,7 +99,7 @@ static void reassembleFrame(void) {
 
 				clearAvcNalState();
 
-				// FIXME: IDR frame!!!
+				requestIdrFrame();
 			}
 		}
 	}
@@ -213,7 +213,7 @@ void queueRtpPacket(PRTP_PACKET rtpPacket, int length) {
 
 		clearAvcNalState();
 
-		// FIXME: IDR frame here!!!
+		requestIdrFrame();
 	}
 
 	lastSequenceNumber = rtpPacket->sequenceNumber;

limelight-common/VideoStream.cpp

@@ -1,5 +1,162 @@
 #include "Limelight.h"
+#include "PlatformSockets.h"
+#include "PlatformThreads.h"
+#include "LinkedBlockingQueue.h"
+
+PDECODER_RENDERER_CALLBACKS callbacks;
+PSTREAM_CONFIGURATION configuration;
+IP_ADDRESS remoteHost;
+
+SOCKET rtpSocket;
+
+LINKED_BLOCKING_QUEUE packetQueue;
+
+PLT_THREAD udpPingThread;
+PLT_THREAD receiveThread;
+PLT_THREAD depacketizerThread;
+PLT_THREAD decoderThread;
+
+void initializeVideoStream(IP_ADDRESS host, PSTREAM_CONFIGURATION streamConfig, PDECODER_RENDERER_CALLBACKS drCallbacks) {
+	callbacks = drCallbacks;
+	configuration = streamConfig;
+	remoteHost = host;
+
+	initializeLinkedBlockingQueue(&packetQueue, 15);
+}
+
+static void UdpPingThreadProc(void *context) {
+	char pingData [] = { 0x50, 0x49, 0x4E, 0x47 };
+	struct sockaddr_in saddr;
+	int err;
+
+	memset(&saddr, 0, sizeof(saddr));
+	saddr.sin_family = AF_INET;
+	saddr.sin_port = htons(47998);
+	memcpy(&saddr.sin_addr, &remoteHost, sizeof(remoteHost));
+
+	for (;;) {
+		err = sendto(rtpSocket, pingData, sizeof(pingData), 0, (struct sockaddr*)&saddr, sizeof(saddr));
+		if (err != sizeof(pingData)) {
+			Limelog("UDP ping thread terminating\n");
+			return;
+		}
+
+		PltSleepMs(100);
+	}
+}
+
+static void ReceiveThreadProc(void* context) {
+	int err;
+
+	for (;;) {
+		char* buffer = (char*) malloc(1500 + sizeof(int));
+		if (buffer == NULL) {
+			Limelog("Receive thread terminating\n");
+			return;
+		}
+
+		err = recv(rtpSocket, &buffer[sizeof(int)], 1500, 0);
+		if (err <= 0) {
+			Limelog("Receive thread terminating #2\n");
+			return;
+		}
+		
+		memcpy(buffer, &err, sizeof(err));
+
+		if (!offerQueueItem(&packetQueue, buffer)) {
+			free(buffer);
+		}
+	}
+}
+
+static void DepacketizerThreadProc(void* context) {
+	int length;
+
+	for (;;) {
+		char* data = (char*) waitForQueueElement(&packetQueue);
+
+		memcpy(&length, data, sizeof(int));
+		queueRtpPacket((PRTP_PACKET) &data[sizeof(int)], length);
+	}
+}
+
+static void DecoderThreadProc(void* context) {
+	for (;;) {
+		PDECODE_UNIT du = getNextDecodeUnit();
+
+		callbacks->submitDecodeUnit(du);
+
+		freeDecodeUnit(du);
+	}
+}
+
+int readFirstFrame(void) {
+	char firstFrame[1000];
+	int err;
+	int offset = 0;
+	SOCKET s;
+
+	s = connectTcpSocket(remoteHost, 47996);
+	if (s == INVALID_SOCKET) {
+		return LastSocketError();
+	}
+
+	for (;;) {
+		err = recv(s, &firstFrame[offset], sizeof(firstFrame) - offset, 0);
+		if (err <= 0) {
+			break;
+		}
+
+		offset += err;
+	}
+
+	processRtpPayload((PNV_VIDEO_PACKET) firstFrame, offset);
+
+	return 0;
+}
+
+int startVideoStream(void* rendererContext, int drFlags) {
+	int err;
+
+	if (callbacks != NULL) {
+		callbacks->setup(configuration->width,
+			configuration->height, 60, rendererContext, drFlags);
+	}
+
+	initializeVideoDepacketizer();
+
+	// FIXME: Set socket options here
+	rtpSocket = bindUdpSocket(47998);
+
+	err = PltCreateThread(UdpPingThreadProc, NULL, &udpPingThread);
+	if (err != 0) {
+		return err;
+	}
+
+	err = readFirstFrame();
+	if (err != 0) {
+		return err;
+	}
+
+	if (callbacks != NULL) {
+		err = PltCreateThread(ReceiveThreadProc, NULL, &receiveThread);
+		if (err != 0) {
+			return err;
+		}
+
+		err = PltCreateThread(DepacketizerThreadProc, NULL, &depacketizerThread);
+		if (err != 0) {
+			return err;
+		}
+
+		err = PltCreateThread(DecoderThreadProc, NULL, &decoderThread);
+		if (err != 0) {
+			return err;
+		}
+
+		callbacks->start();
+	}
+
+	return 0;
+}
 
-void* initializeVideoStream(IP_ADDRESS host, PSTREAM_CONFIGURATION streamConfig) {
-	return NULL;
-}