- tests mit UDP-Streaming

git-svn-id: svn://svn.berlios.de/ffnetdev/trunk@24 1f4bef6d-8e0a-0410-8695-e467da8aaccf

10 files changed, 334 insertions, 40 deletions

diff --git a/netosd.c b/netosd.c
index 87fec69..08ab350 100644
--- a/netosd.c
+++ b/netosd.c
@@ -59,9 +59,9 @@ void cNetOSD::Flush(void)
        if (Bitmap->Dirty(x1, y1, x2, y2))   
        {  
 #ifdef DEBUG         
-	        fprintf(stderr, "[ffnetdev] NetOSD: Left: %d, Top: %d, X0: %d, Y0: %d, Width: %d, Height: %d\n", 
-               			 Left(), Top(), Bitmap->X0(), Bitmap->Y0(), Bitmap->Width(), Bitmap->Height()); 
-       		fprintf(stderr, "[ffnetdev] NetOSD: Dirty area x1: %d, y1: %d, x2: %d, y2: %d\n",x1,y1,x2,y2);
+	        //fprintf(stderr, "[ffnetdev] NetOSD: Left: %d, Top: %d, X0: %d, Y0: %d, Width: %d, Height: %d\n", 
+               	//		 Left(), Top(), Bitmap->X0(), Bitmap->Y0(), Bitmap->Width(), Bitmap->Height()); 
+       		//fprintf(stderr, "[ffnetdev] NetOSD: Dirty area x1: %d, y1: %d, x2: %d, y2: %d\n",x1,y1,x2,y2);
 #endif
 		cOSDWorker::DrawBitmap(Left() + Bitmap->X0(), Top() + Bitmap->Y0(), *Bitmap);
        		
diff --git a/osdworker.c b/osdworker.c
index cb939b3..0f3d746 100644
--- a/osdworker.c
+++ b/osdworker.c
@@ -253,15 +253,18 @@ bool cOSDWorker::SendScreen(int x1, int y1, int x2, int y2)
    	OSDWrite((unsigned char*)&fu, sz_rfbFramebufferUpdateMsg);
 	//int BufferSize = m_Instance->m_pEncoder->RequiredBuffSize(x2-x1, y2-y1);
 	//m_Instance->CreateSendBuffer(BufferSize);
-	int BufferSize = m_Instance->m_pEncoder->EncodeRect((BYTE*)(m_Instance->m_pOsdBitmap->Data(0, 0)), m_Instance->m_pSendBuffer, rect);
+	if (m_Instance->m_pEncoder != NULL)
+	{
+	    int BufferSize = m_Instance->m_pEncoder->EncodeRect((BYTE*)(m_Instance->m_pOsdBitmap->Data(0, 0)), m_Instance->m_pSendBuffer, rect);
 #ifdef DEBUG
-	fprintf(stderr, "[ffnetdev] VNC: Send OSD Data %d Bytes\n", BufferSize);
+	    fprintf(stderr, "[ffnetdev] VNC: Send OSD Data %d/%d/%d/%d x1/y1/x2/y2 %d Bytes\n", x1, y1, x2, y2, BufferSize);
 #endif
-	dsyslog("[ffnetdev] VNC: Send OSD Data %d Bytes\n", BufferSize);
-       	OSDWrite((unsigned char*)m_Instance->m_pSendBuffer, BufferSize);
-	m_Instance->m_pOsdBitmap->Clean();
+	    dsyslog("[ffnetdev] VNC: Send OSD Data %d/%d/%d/%d x1/y1/x2/y2 %d Bytes\n", x1, y1, x2, y2, BufferSize);
+       	    OSDWrite((unsigned char*)m_Instance->m_pSendBuffer, BufferSize);
+	    m_Instance->m_pOsdBitmap->Clean();
 	
-	m_Instance->m_bOSDisClear = false;
+	    m_Instance->m_bOSDisClear = false;
+	}
 	
    	return true;
    }
diff --git a/tools/socket.c b/tools/socket.c
index 2061dec..8bb7359 100644
--- a/tools/socket.c
+++ b/tools/socket.c
@@ -1,11 +1,15 @@
 #include "tools/socket.h"
 
+#include <vdr/tools.h>
+
 #include <string.h>
 #include <arpa/inet.h>
 #include <unistd.h>
 #include <errno.h>
 #include <fcntl.h>
 
+#define UDP_TX_BUF_SIZE ((188*7+3)*20)
+
 cTBSocket::cTBSocket(int Type) {
 	memset(&m_LocalAddr, 0, sizeof(m_LocalAddr));
 	memset(&m_RemoteAddr, 0, sizeof(m_RemoteAddr));
@@ -16,6 +20,46 @@ cTBSocket::~cTBSocket() {
 	if (IsOpen()) Close();
 }
 
+bool cTBSocket::OpenUDP(const std::string &Host, unsigned int Port) {
+	int socket, tmp;
+	struct sockaddr_in my_addr;
+
+	if (IsOpen()) Close();
+	
+	if ((socket = ::socket(AF_INET, m_Type, 0)) == -1)
+		return false;
+		
+	m_RemoteAddr.sin_family = AF_INET;
+	m_RemoteAddr.sin_port   = htons(Port);
+	m_RemoteAddr.sin_addr.s_addr = inet_addr(Host.c_str());
+	
+	my_addr.sin_family = AF_INET;
+	my_addr.sin_port = htons(Port);
+	my_addr.sin_addr.s_addr = htonl(INADDR_ANY);
+		    
+/*	tmp = 1;
+	if (setsockopt(socket, SOL_SOCKET, SO_REUSEADDR, (char*)&tmp, sizeof(tmp)) < 0)
+	    goto closefd;*/
+		    
+	/* limit the tx buf size to limit latency */
+	tmp = UDP_TX_BUF_SIZE;
+	if (setsockopt(socket, SOL_SOCKET, SO_SNDBUF, (char*)&tmp, sizeof(tmp)) < 0)
+	    goto closefd;
+
+	/* the bind is needed to give a port to the socket now */
+/*	if (bind(socket,(struct sockaddr *)&my_addr, sizeof(my_addr)) < 0) 
+	    goto closefd;*/
+
+	if (!cTBSource::Open(socket))
+	    goto closefd; // return false;
+
+	return true;
+
+closefd:
+    ::close(socket);
+    return false;
+}
+
 bool cTBSocket::Connect(const std::string &Host, unsigned int Port) {
 	socklen_t len;
 	int socket;
diff --git a/tools/socket.h b/tools/socket.h
index d1a7d62..ee9c632 100644
--- a/tools/socket.h
+++ b/tools/socket.h
@@ -44,6 +44,11 @@ public:
 	   otherwise, setting errno appropriately. */
 	virtual bool Shutdown(int How);
 
+	/* OpenUDP() Open the associated socket. 
+	   Returns true on success and false otherwise, setting errno 
+	   appropriately. */
+	virtual bool OpenUDP(const std::string &Ip, uint Port);
+	
 	/* Close() closes the associated socket and releases all structures. 
 	   Returns true on success and false otherwise, setting errno 
 	   appropriately. The object is in the closed state afterwards, regardless
@@ -102,7 +107,11 @@ inline ssize_t cTBSocket::SysRead(void *Buffer, size_t Length) const {
 }
 
 inline ssize_t cTBSocket::SysWrite(const void *Buffer, size_t Length) const {
-	return ::send(*this, Buffer, Length, 0);
+	if ((m_Type == SOCK_DGRAM)) {
+		socklen_t len = sizeof(m_RemoteAddr);
+		return ::sendto(*this, Buffer, Length, 0, (sockaddr*)&m_RemoteAddr, len);
+	} else
+		return ::send(*this, Buffer, Length, 0);
 }
 
 #endif // TOOLBOX_SOCKET_H
diff --git a/tsworker.c b/tsworker.c
index 98e893e..cf10b52 100644
--- a/tsworker.c
+++ b/tsworker.c
@@ -15,7 +15,18 @@
 #include "tsworker.h"
 #include "config.h"
 
-#define MINSENDBYTES KILOBYTE(500)
+#define TS_PACKET_SIZE (188)
+#define UDP_PACKET_SIZE (TS_PACKET_SIZE * 7)
+#define UDP_MAX_BITRATE 7000000
+#define UDP_SEND_INTERVALL 1000
+
+struct TSData
+{
+	char packNr;
+    char packsCount;
+    char tsHeaderCRC;
+    char data[UDP_PACKET_SIZE];
+};
 
 //////////////////////////////////////////////////////////////////////////////////////////////////
 //////////////////////////////////////////////////////////////////////////////////////////////////
@@ -25,7 +36,7 @@ cTSWorker *cTSWorker::m_Instance = NULL;
 cTSWorker::cTSWorker(void)
 		: cThread("[ffnetdev] TS streamer")
 {
-	m_Active = false;
+	m_Active = false;	
 	
 	m_StreamClient = NULL;
 	origPrimaryDevice = -1;
@@ -42,6 +53,7 @@ void cTSWorker::Init(cStreamDevice *StreamDevice, int tsport, cPluginFFNetDev *p
 		m_Instance->TSPort = tsport;		
 		m_Instance->Start();
 		m_Instance->m_pPlugin = pPlugin;
+		m_Instance->close_Streamclient_request = false;
 	}
 }
 
@@ -65,8 +77,13 @@ void cTSWorker::CloseStreamClient(void) {
 }
 
 void cTSWorker::Action(void) {
+    ActionTCP();
+    //ActionUDP();
+}
+
+
+void cTSWorker::ActionTCP(void) {
 	cTBSelect select;
-	//cTBSocket m_StreamListen(SOCK_DGRAM);
 	cTBSocket m_StreamListen;
 	struct timeval oldtime;
 	long bytessend = 0;
@@ -181,18 +198,17 @@ void cTSWorker::Action(void) {
 					    if (((written=m_StreamClient->Write(&buffer[done], available)) < 0) && 
 						(errno != EAGAIN)) 
 					    {
-						CloseStreamClient();
+							CloseStreamClient();
 					    }
 					    
 					    if (written > 0)
 					    {
-						
-						available -= written;
-						done      += written;
+							available -= written;
+							done      += written;
 					    }
 					    else
 					    {
-						cCondWait::SleepMs(5);
+							cCondWait::SleepMs(5);
 					    }
 					}
 					m_StreamDevice->Del(count);
@@ -209,17 +225,18 @@ void cTSWorker::Action(void) {
 					bytessend += count;
 					if (curtime.tv_sec > oldtime.tv_sec + 10)
 					{
-					    double secs = (curtime.tv_sec * 1000 + (curtime.tv_usec / 1000)) / 1000 
+						double secs = (curtime.tv_sec * 1000 + (curtime.tv_usec / 1000)) / 1000 
 							- (oldtime.tv_sec * 1000 + (oldtime.tv_usec / 1000)) / 1000;
+						double rate = (double)((bytessend - oldbytessend) / secs) * 8 / 1024 / 1024;
 #ifdef DEBUG
-					    fprintf(stderr, "[ffnetdev] Streamer: current TransferRate %d Byte/Sec, %d Bytes send\n",
-						(int)((bytessend - oldbytessend) / secs), bytessend - oldbytessend);
+						fprintf(stderr, "[ffnetdev] Streamer: current TransferRate %2.3f MBit/Sec, %d Bytes send\n",
+							rate, bytessend - oldbytessend);
 #endif						
-					    dsyslog("[ffnetdev] Streamer: current TransferRate %d Byte/Sec, %d Bytes send\n",
-						(int)((bytessend - oldbytessend) / secs), bytessend - oldbytessend);
-					    
-					    oldbytessend = bytessend;
-					    oldtime = curtime;
+						dsyslog("[ffnetdev] Streamer: current TransferRate %2.3f MBit/Sec, %d Bytes send\n",
+							rate, bytessend - oldbytessend);
+						
+						oldbytessend = bytessend;
+						oldtime = curtime;
 					}
 				}
 				m_StreamDevice->UnlockOutput();
@@ -246,3 +263,219 @@ void cTSWorker::Action(void) {
 
 }
 
+
+void cTSWorker::ActionUDP(void) 
+{
+	cTBSocket m_StreamClient(SOCK_DGRAM);
+	struct timeval oldtime, curtime;
+	u64 oldPacketTime = 0;
+	long bytessend = 0;
+	long oldbytessend = 0;
+	long toSend = 0;
+	int restData = 0;
+	TSData tsData;
+
+	const char* StreamIp = "192.168.0.61";
+	uint StreamPort = TSPort;
+
+	m_Active		= true;
+	have_Streamclient	= true;
+	
+	if (!m_StreamClient.OpenUDP(StreamIp, StreamPort)) 
+	{
+		isyslog("[ffnetdev] Streamer: Couldn't create UDP-Socket: %s", strerror(errno));
+		m_Active = false;
+	} 
+	else
+	    isyslog("[ffnetdev] Streamer: UDP-Socket create successful");
+
+	gettimeofday(&oldtime, 0);
+	tsData.packNr = 0;
+	tsData.packsCount = 0;
+	tsData.tsHeaderCRC = 0;
+
+	while (m_Active) 
+	{
+		/* Check for closed streaming client connection */		
+		if (have_Streamclient==true) 
+		{ 
+			if (close_Streamclient_request==true) 
+			{
+				close_Streamclient_request = false;
+				have_Streamclient = false;
+				
+				m_pPlugin->RestorePrimaryDevice();
+				
+				if ( m_StreamClient.Close() ) 
+				{
+#ifdef DEBUG
+					fprintf(stderr, "[ffnetdev] Streamer: Client socket closed successfully.\n");
+#endif
+					isyslog("[ffnetdev] Streamer: Connection closed: client %s:%d",
+						 m_StreamClient.RemoteIp().c_str(), m_StreamClient.RemotePort());
+				}
+				else
+				{
+#ifdef DEBUG
+					fprintf(stderr, "[ffnetdev] Streamer: Error closing client socket.\n");
+#endif
+					esyslog("[ffnetdev] Streamer: Error closing connection.");
+					m_Active=false;
+					continue;
+				} 
+					  
+			}
+			
+			int count=0;
+			
+			m_StreamDevice->LockOutput();
+			uchar *buffer = m_StreamDevice->Get(count);
+			if (buffer!=NULL) 
+			{	
+				int available = count;
+				int done      = 0;
+				int written   = 0;
+				char data[100];
+				int  rcvCount;
+
+				rcvCount=m_StreamClient.Read(data, 10);
+				if (rcvCount > 0)
+				{
+					isyslog("[ffnetdev] Streamer: empfangen:%d Bytes\n", rcvCount);
+				}
+
+				if (oldPacketTime == 0)
+					oldPacketTime = get_time()- UDP_SEND_INTERVALL;
+
+				while ((available > 0) && (have_Streamclient == true) &&
+					   (!close_Streamclient_request))
+				{
+					while ((tsData.packsCount * TS_PACKET_SIZE < UDP_PACKET_SIZE) && (available > 0))
+					{
+						int moveCount = (available >= TS_PACKET_SIZE) ?  TS_PACKET_SIZE : available;
+						moveCount = (moveCount + restData >= TS_PACKET_SIZE) ?  TS_PACKET_SIZE - restData : moveCount;
+						memcpy(&tsData.data[(tsData.packsCount * TS_PACKET_SIZE) + restData], &buffer[done], moveCount);
+						available -= moveCount;
+						done += moveCount;
+						if (restData + moveCount == TS_PACKET_SIZE)
+						{
+							char *data = &tsData.data[tsData.packsCount * TS_PACKET_SIZE];
+							for (int i = 0; i < 4; i++)
+							{	
+								tsData.tsHeaderCRC += (char)*(data + i);
+							}
+							restData = 0;
+							tsData.packsCount ++;
+						}
+						else
+						{
+							restData = moveCount;
+							continue;
+						}
+					}
+
+					if (restData > 0)
+						continue;
+
+					while (get_time() < oldPacketTime + UDP_SEND_INTERVALL)
+						cCondWait::SleepMs(1);
+
+					if (toSend == 0)
+						toSend = (long)(UDP_MAX_BITRATE * (((double)get_time() - oldPacketTime) / 1000000) / 8);
+					
+					int sendcount = tsData.packsCount * TS_PACKET_SIZE + 3;
+					if (toSend < sendcount)
+					{
+						toSend = 0;
+						oldPacketTime = get_time();
+						continue;
+					}
+
+					char* pTsData = (char*)&tsData;
+					while ((sendcount > 0) && (have_Streamclient == true) &&
+					   (!close_Streamclient_request))
+					{
+						if (((written=m_StreamClient.Write(pTsData, sendcount)) < 0) && 
+							(errno != EAGAIN)) 
+						{
+							isyslog("[ffnetdev] Streamer: Couldn't send data: %d %s Len:%d\n", errno, strerror(errno), sendcount);
+							CloseStreamClient();
+						}
+						
+						if (written > 0)
+						{
+							sendcount -= written;
+							pTsData   += written;
+							toSend	  -= written;
+						}
+						else
+						{
+							cCondWait::SleepMs(5);
+						}
+					}
+
+					if (sendcount == 0)
+					{
+						tsData.packsCount = 0;
+						tsData.tsHeaderCRC = 0;
+						tsData.packNr ++;
+					}
+				}
+				m_StreamDevice->Del(count);
+
+				gettimeofday(&curtime, 0);
+				if (oldtime.tv_sec == 0)
+				{
+					oldtime = curtime;
+					bytessend    = 0;
+					oldbytessend = 0;
+				}
+					
+				bytessend += count;
+				if (curtime.tv_sec > oldtime.tv_sec + 10)
+				{
+					double secs = (curtime.tv_sec * 1000 + (curtime.tv_usec / 1000)) / 1000 
+						- (oldtime.tv_sec * 1000 + (oldtime.tv_usec / 1000)) / 1000;
+					double rate = (double)((bytessend - oldbytessend) / secs) * 8 / 1024 / 1024;
+#ifdef DEBUG
+					fprintf(stderr, "[ffnetdev] Streamer: current TransferRate %2.3f MBit/Sec, %d Bytes send\n",
+						rate, bytessend - oldbytessend);
+#endif						
+					dsyslog("[ffnetdev] Streamer: current TransferRate %2.3f MBit/Sec, %d Bytes send\n",
+						rate, bytessend - oldbytessend);
+					
+					oldbytessend = bytessend;
+					oldtime = curtime;
+				}
+			}
+			m_StreamDevice->UnlockOutput();
+
+		} 
+		else 
+		{  
+			/* simply discard all data in ringbuffer */
+			int count=0;
+			if ( (m_StreamDevice->Get(count)) !=NULL ) 
+			{
+				m_StreamDevice->Del(count);
+#ifdef DEBUG
+				fprintf (stderr, "[ffnetdev] Streamer: Bytes not sent, but deleted from ringbuffer: %d\n",count);
+#endif
+				dsyslog("[ffnetdev] Streamer: Bytes not sent, but deleted from ringbuffer: %d\n",count);
+			}
+		}
+		cCondWait::SleepMs(3);
+
+	} // while(m_Active)
+
+}
+
+/* Returns time since 1970 in microseconds */
+u64 cTSWorker::get_time(void)
+{
+	struct timeval tv;
+	struct timezone tz={0,0};
+
+	gettimeofday(&tv,&tz);
+	return ((u64)tv.tv_sec)*1000000+((u64)tv.tv_usec);
+}
diff --git a/tsworker.h b/tsworker.h
index 45e13ab..12b3452 100644
--- a/tsworker.h
+++ b/tsworker.h
@@ -16,6 +16,8 @@
 #include "netosd.h"
 #include "ffnetdev.h"
 
+typedef unsigned long long u64;
+
 
 // --- cTSWorker -------------------------------------------------------------
 
@@ -30,9 +32,12 @@ private:
 	int TSPort;
 	int origPrimaryDevice;
 	cPluginFFNetDev *m_pPlugin;	
+	u64 get_time(void);
 
 protected:
 	virtual void Action(void);
+	virtual void ActionTCP(void);
+	virtual void ActionUDP(void);
 	void Stop(void);
 public:
 	cTSWorker(void);
diff --git a/vncEncodeCoRRE.c b/vncEncodeCoRRE.c
index af398f3..ed6f183 100644
--- a/vncEncodeCoRRE.c
+++ b/vncEncodeCoRRE.c
@@ -203,8 +203,8 @@ vncEncodeCoRRE::EncodeRect(BYTE *source, BYTE *dest, const RECT &rect)
 	// Do the encoding
 	UINT size = InternalEncodeRect(source, dest, rect);
 
-	const int rectW = rect.right - rect.left;
-	const int rectH = rect.bottom - rect.top;
+	const int rectW = rect.right - rect.left + 1;
+	const int rectH = rect.bottom - rect.top + 1;
 
 	// Will this rectangle have been split for encoding?
 	if ((rectW>m_maxwidth) || (rectH>m_maxheight))
@@ -286,8 +286,8 @@ vncEncodeCoRRE::EncodeSmallRect(BYTE *source, BYTE *dest, const RECT &rect)
 {
 	int subrects = -1;
 
-	const UINT rectW = rect.right - rect.left;
-	const UINT rectH = rect.bottom - rect.top;
+	const UINT rectW = rect.right - rect.left + 1;
+	const UINT rectH = rect.bottom - rect.top + 1;
 
 	// Create the rectangle header
 	rfbFramebufferUpdateRectHeader *surh=(rfbFramebufferUpdateRectHeader *)dest;
diff --git a/vncEncodeHexT.c b/vncEncodeHexT.c
index ce7c3f0..a478c98 100644
--- a/vncEncodeHexT.c
+++ b/vncEncodeHexT.c
@@ -81,8 +81,8 @@ vncEncodeHexT::NumCodedRects(RECT &rect)
 UINT
 vncEncodeHexT::EncodeRect(BYTE *source, BYTE *dest, const RECT &rect)
 {
-	const UINT rectW = rect.right - rect.left;
-	const UINT rectH = rect.bottom - rect.top;
+	const UINT rectW = rect.right - rect.left + 1;
+	const UINT rectH = rect.bottom - rect.top + 1;
 	int encodedResult;
 
 	// Create the rectangle header
@@ -176,8 +176,8 @@ vncEncodeHexT::EncodeHextiles##bpp(BYTE *source, BYTE *dest,				\
 			RECT hexrect;													\
 			hexrect.left = x;												\
 			hexrect.top = y;												\
-			hexrect.right = x+w;											\
-			hexrect.bottom = y+h;											\
+			hexrect.right = x+w-1;											\
+			hexrect.bottom = y+h-1;											\
 			Translate(source, (BYTE *) &clientPixelData, hexrect);			\
 																			\
 			rectoffset = destoffset;										\
diff --git a/vncEncodeRRE.c b/vncEncodeRRE.c
index 3a9d44e..e479939 100644
--- a/vncEncodeRRE.c
+++ b/vncEncodeRRE.c
@@ -258,8 +258,8 @@ vncEncodeRRE::EncodeRect(BYTE *source, BYTE *dest, const RECT &rect)
 {
 	int subrects = -1;
 
-	const UINT rectW = rect.right - rect.left;
-	const UINT rectH = rect.bottom - rect.top;
+	const UINT rectW = rect.right - rect.left + 1;
+	const UINT rectH = rect.bottom - rect.top + 1;
 
 	// Create the rectangle header
 	rfbFramebufferUpdateRectHeader *surh=(rfbFramebufferUpdateRectHeader *)dest;
diff --git a/vncEncoder.c b/vncEncoder.c
index 1ade07d..3f6b73c 100644
--- a/vncEncoder.c
+++ b/vncEncoder.c
@@ -121,8 +121,8 @@ vncEncoder::Translate(BYTE *source, BYTE *dest, const RECT &rect)
 					(char *)sourcepos,
 					(char *)dest,
 					m_bytesPerRow,
-					rect.right-rect.left,
-					rect.bottom-rect.top
+					rect.right-rect.left+1,
+					rect.bottom-rect.top+1
 					);
 }
 
@@ -142,8 +142,8 @@ inline UINT
 vncEncoder::EncodeRect(BYTE *source, BYTE *dest, const RECT &rect)
 {
 
-	const int rectW = rect.right - rect.left;
-	const int rectH = rect.bottom - rect.top;
+	const int rectW = rect.right - rect.left + 1;
+	const int rectH = rect.bottom - rect.top + 1;
 
 
 	// Create the header for the update in the destination area