wait4sessionprocess.c

RTSP PROXY Server and client

				else {
					s->amtInServerOutBuffer -= len;
					if (s->amtInServerOutBuffer == 0)
                                        {
						s->state = stServerTransactionRecv;
                                        }
				}
			}

			break;

		case stServerTransactionRecv:
			// check to see if we've got a response from the server
	  	       KRTSPROXYD_OUT(KERN_INFO "in stServerTransactionRecv\n");
			count++;

                 	if (s->server_skt == NULL) {
				s->state = stServerShutdown;
				break;
			}
			pBuf = s->sinbuf + s->amtInServerInBuffer;

       	       canRead = RTSP_SESSION_BUF_SIZE - s->amtInServerInBuffer - 1;
                     if (canRead > 0 && !skb_queue_empty(&(s->server_skt->sk->receive_queue))) {
				if ((len = ReceiveBuffer(s->server_skt, pBuf, canRead)) < 0) {
				 switch (errno) {
						case EAGAIN:
							
							break;
						case EPIPE:			// connection broke
						case ENOTCONN:		// shut down
						case ECONNRESET:
							s->state = stServerShutdown;
							break;
						default:
							KRTSPROXYD_OUT(KERN_ERR "problems reading from server\n");
							break;
					}
				}
				else {
					pBuf[len] = '\0';
					KRTSPROXYD_OUT(KERN_INFO "read %d bytes from server:%s\n", len, pBuf);
					s->amtInServerInBuffer += len;
				}
			}
			else{ 
                         KRTSPROXYD_OUT(KERN_INFO "Cannot read now\n");
                         break;
                        }
			// DMS - if there is a content-length, make sure we've gotten all that data too.
			if ((s->totalContentLength > 0) && (s->amtInServerInBuffer < s->totalContentLength))
			{	
				break;
			}
                     if ( s->totalContentLength == 0 )
			{	//-rt if totalContentLength != 0,then we must have seen "has_two_crlfs"
				// and already parsed out the content-length header.
				// now we won't be able to find it again becuase str_sep
				// in the parsing code below has already replaced the CRLFs with \0's.
			
				// did we get complete response headers yet?
				responseHeaderLen = has_two_crlfs(s->sinbuf);
				
				if (responseHeaderLen == 0)
				{	
				       KRTSPROXYD_OUT(KERN_INFO "we have not get complete response headers!!\n");
					break;
				}
			}
			
		if ( !s->haveParsedServerReplyHeaders )
		{				
				// we can only do this one time!
				
            			// munge the data for the client, while snarfing what we need
			
			for (pBuf = s->sinbuf; (p = str_sep(&pBuf, "\r\n")) != NULL; ) {
				// Count the empty fields; three in a row is end of the header
				if (*p == '\0') {
					if (++numDelims == 3)
						break;
					continue;
				}
				else
					numDelims = 0;

                            if (0 == strn_casecmp(p, "x-Accept-Dynamic-Rate", 21))
		                 p = "x-Accept-Dynamic-Rate: 0";
					 
				// see if we can snarf any data out of the headers
				if (has_content_length(p, &s->contentLength)) {
					s->totalContentLength = s->contentLength + responseHeaderLen;
				}
				else if (has_sessionID(p, temp)) {
					if (!s->sessionID)
					{
						s->sessionID = kmalloc(strlen(temp) + 1, GFP_KERNEL);
						if(!s->sessionID)
                                             {
                                               KRTSPROXYD_OUT(KERN_ERR "cannot alloate session ID's Mem!\n");
               				     s->state = stError;
               				     return count;
                                             }
						strcpy(s->sessionID, temp);
					}
					else if (str_casecmp(s->sessionID, temp) != 0)
							KRTSPROXYD_OUT(KERN_ERR "Bad session ID in response from server");
				}
				else if (s->transaction_type == ttSetup && has_ports(p, &i, &len)) {
					t = s->trk + s->cur_trk;
					t->ServerRTPPort = len;
					
					KRTSPROXYD_OUT(KERN_INFO "Server ports for track %d are %d-%d \n",
								s->cur_trk, t->ServerRTPPort, t->ServerRTPPort + 1);

					// make rtp/rtcp port pair here proxy=>client
					if (make_udp_port_pair(gProxyIP, s->client_ip, &t->RTP_P2C, 
						&t->RTCP_P2C, CPUNR) == -1) 
					{		
						s->server_skt = NULL;
						KRTSPROXYD_OUT(KERN_INFO "Couldn't create udp port pair for proxy=>client \n");
						s->state = stError;
						break;
					}

					// set up transfer param blocks
					t->RTP_S2C_tpb.status = &s->die;
					t->RTP_S2C_tpb.send_from = t->RTP_P2C;
					t->RTP_S2C_tpb.send_to_ip = s->client_ip;
					t->RTP_S2C_tpb.send_to_port = t->ClientRTPPort;
					strcpy(t->RTP_S2C_tpb.socketName, "RTP Server to Client");
					upon_receipt_from(t->RTP_S2P, s->server_ip, 
						(do_routine)transfer_data, (void *)&(t->RTP_S2C_tpb));

					t->RTCP_S2C_tpb.status = &s->die;
					t->RTCP_S2C_tpb.send_from = t->RTCP_P2C;
					t->RTCP_S2C_tpb.send_to_ip = s->client_ip;
					t->RTCP_S2C_tpb.send_to_port = t->ClientRTPPort + 1;
					strcpy(t->RTCP_S2C_tpb.socketName, "RTCP Server to Client");

					upon_receipt_from(t->RTCP_S2P, s->server_ip, 
						(do_routine)transfer_data, (void *)&(t->RTCP_S2C_tpb));

					t->RTCP_C2S_tpb.status = &s->die;
					t->RTCP_C2S_tpb.send_from = t->RTCP_S2P;
					t->RTCP_C2S_tpb.send_to_ip = s->server_ip;
					t->RTCP_C2S_tpb.send_to_port = t->ServerRTPPort + 1;
					strcpy(t->RTCP_C2S_tpb.socketName,"RTCP Client to Server");

					upon_receipt_from(t->RTCP_P2C, s->client_ip, 
						(do_routine)transfer_data, (void *)&(t->RTCP_C2S_tpb));

					KRTSPROXYD_OUT(KERN_INFO "Created ports for proxy to client on track %d are %d-%d\n",
								s->cur_trk,
								t->RTP_P2C->port, t->RTCP_P2C->port);

					// reconstruct the client;server string
						w = find_transport_header(p);
						if (w != NULL)
							*w = '\0';
						sprintf(temp, "%sclient_port=%d-%d;server_port=%d-%d;source=%s", p,
							t->ClientRTPPort, t->ClientRTPPort+1,
							t->RTP_P2C->port, t->RTCP_P2C->port,
							ntoa(gProxyIP));
						p = temp;
				}

				// put the line in the outgoing buffer
				len = strlen(p);
				if(len + s->amtInClientOutBuffer + 2 > RTSP_SESSION_BUF_SIZE)
				{
				    s->state = stError;
				    return count;
				}
				memcpy(s->coutbuf + s->amtInClientOutBuffer, p, len);
				s->amtInClientOutBuffer += len;
				s->coutbuf[s->amtInClientOutBuffer++] = '\r';
				s->coutbuf[s->amtInClientOutBuffer++] = '\n';
				
				s->responseBodyP = pBuf + 3;
			}
                     if(s->amtInClientOutBuffer +2 > RTSP_SESSION_BUF_SIZE)
                     {
                            KRTSPROXYD_OUT(KERN_ERR "s->amtInClientOutBuffer + 2>RTSP_SESSION_BUF_SIZE\n");
                     	s->state = stError;
                            return count;
                     }
			s->coutbuf[s->amtInClientOutBuffer++] = '\r';
			s->coutbuf[s->amtInClientOutBuffer++] = '\n';
		}

              // the headers are done now.
			s->haveParsedServerReplyHeaders = 1;
			
			// DMS - if there is a content-length, make sure we've gotten all that data too.
			if ((s->totalContentLength > 0) && (s->amtInServerInBuffer < s->totalContentLength))
			{	
				break;
			}
			
	
			pBuf = s->responseBodyP;
			
			// munge and add the content if there is any
			if (s->contentLength) 
			{
				if (s->transaction_type == ttDescribe) 
				{
				       char *nextBuffPos = pBuf;
				       nextBuffPos = get_line_str( lineBuff, nextBuffPos, MAX_LINE_BUFF );
                                //for dbg
                                if(nextBuffPos == NULL){
                                  KRTSPROXYD_OUT(KERN_ERR "nextBuffPos == NULL\n");
                                  s->state = stError;
                                  return count;
                                 }
				while ( *lineBuff )
				{
						//  c=IN IP0 ?
						if (has_IN_IP(lineBuff, temp)) 
						{
						}
				
						// put the line in the outgoing buffer
						len = strlen(lineBuff);
						
					if( len + s->amtInClientOutBuffer > RTSP_SESSION_BUF_SIZE )
					{
					        KRTSPROXYD_OUT(KERN_ERR "len + s->amtInClientOutBuffer > RTSP_SESSION_BUF_SIZE \n");
					        s->state = stError;
					        return count;
					}
	
						memcpy(s->coutbuf + s->amtInClientOutBuffer, lineBuff, len);
						s->amtInClientOutBuffer += len;
						
						nextBuffPos = get_line_str( lineBuff, nextBuffPos, MAX_LINE_BUFF );
                                         if(nextBuffPos == NULL)
                                         	  {//for dbg
                                                s->state = stError;
                                                 return count;
                                             }
				  }
				}
				else {
					if( s->contentLength + s->amtInClientOutBuffer > RTSP_SESSION_BUF_SIZE )
					{
					  KRTSPROXYD_OUT(KERN_ERR "s->contentLength + s->amtInClientOutBuffer > RTSP_SESSION_BUF_SIZE\n"); 
					  s->state=stError;
					  return count;
					}
					memcpy(s->coutbuf + s->amtInClientOutBuffer, pBuf, s->contentLength);
					s->amtInClientOutBuffer += s->contentLength;
				}
			}

			s->state = stSendClientResponse;
			s->amtInServerInBuffer = 0;
			s->totalContentLength = 0;
			s->haveParsedServerReplyHeaders = 0;
			s->contentLength = 0;
			break;

		case stSendClientResponse:
			// check to see that we're still connected (writable) and send the response
                     KRTSPROXYD_OUT(KERN_INFO "in stSendClientResponse\n");
			count++;

			if (s->amtInClientOutBuffer){// && isWritable(s->client_skt)) {
	                        
                        KRTSPROXYD_OUT(KERN_INFO "the Length of the message is %d\n",s->amtInClientOutBuffer);
                        KRTSPROXYD_OUT(KERN_INFO "and the message is %s\n", s->coutbuf); 	
                           if ((len = SendBuffer(s->client_skt, s->coutbuf, s->amtInClientOutBuffer)) < 0) {
          				switch (errno) {
						case EPIPE:			// connection broke
						case ENOTCONN:		// shut down
						case ECONNRESET:
							s->state = stClientShutdown;
							break;
						case EAGAIN:	// was busy - try again
						case EINTR:		// got interrupted - try again
							break;
						default:
							KRTSPROXYD_OUT(KERN_ERR "writing to client error\n");
							s->state = stError;
							return count;
					}
				}
				else {
					s->amtInClientOutBuffer -= len;
					if (s->amtInClientOutBuffer == 0){
						if (s->transaction_type == ttTeardown)
							s->state = stClientShutdown;
						else
							s->state = stRecvClientCommand;
                                                
                                        }
				}
			}

			break;
			

		case stClientShutdown:
			s->die = 1;
                     KRTSPROXYD_OUT(KERN_INFO "stClientShutdown\n");
                     count = 0;
			break;

		case stBadServerName:
                     KRTSPROXYD_OUT(KERN_INFO "stBadServerName\n");
			send_rtsp_error(s->client_skt, kServerNotFound);
			s->state = stServerShutdown;
		       count = 0;
			break;

		case stCantConnectToServer:
                     KRTSPROXYD_OUT(KERN_INFO "stCantConnectToServer\n");
			send_rtsp_error(s->client_skt, kServerNotFound);
			s->state = stServerShutdown;
			count = 0;
			break;
			
		case stServerShutdown:
			KRTSPROXYD_OUT(KERN_ERR "Server shutdown (ip %s)", ntoa(s->server_ip));
			s->die = 1;
			count = 0;
			break;

		case stError:
			send_rtsp_error(s->client_skt, kUnknownError);
			KRTSPROXYD_OUT(KERN_ERR "error condition\n");
			s->die = 1;
			count = 0;
			break;
          }

    LeaveFunction("process_session");

    return count; 
}

int Wait4SessionProcess(const int CPUNR)
{
	struct rtsp_session *CurrentSession,**Prev;
	int count = 0;
	EnterFunction("Wait4SessionProcess");
      
	CurrentSession = threadinfo[CPUNR].RtspSessionQueue;
	
	Prev = &(threadinfo[CPUNR].RtspSessionQueue);
	
	while (CurrentSession!=NULL)
	{
              if(CurrentSession->new_session == 1)
              {
                CurrentSession->new_session = 0;
		if (gUserLimit && (atomic_read(&gNumUsers) > gUserLimit)){
                CurrentSession->die = 1;
                send_rtsp_error(CurrentSession->client_skt, kTooManyUsers);
               }
              }
		/* If the connection is lost, remove from queue */
		
		if ((CurrentSession->client_skt->sk->state != TCP_ESTABLISHED
		    && CurrentSession->client_skt->sk->state != TCP_CLOSE_WAIT)
		    ||CurrentSession->die)
		{
			struct rtsp_session*Next;
			
			Next = CurrentSession->next;
			
			*Prev = CurrentSession->next;
			CurrentSession->next = NULL;
			
                        KRTSPROXYD_OUT(KERN_INFO "cleanupsession!\n");		
			CleanUpSession(CurrentSession, CPUNR);
			CurrentSession = Next;
			continue;
		}
	//service session	
                count += process_session(CurrentSession, CPUNR);

		Prev = &(CurrentSession->next);
		CurrentSession = CurrentSession->next;
	}

	LeaveFunction("Wait4SessionProcess");
	return count;
}

void StopSessionProcess(const int CPUNR)
{
	struct rtsp_session *CurrentSession,*Next;
	
	EnterFunction("StopSessionProcess");
	CurrentSession = threadinfo[CPUNR].RtspSessionQueue;

	while (CurrentSession!=NULL)
	{
		Next = CurrentSession->next;
		CleanUpSession(CurrentSession, CPUNR);
		CurrentSession = Next;		
	}
	
	threadinfo[CPUNR].RtspSessionQueue = NULL;
	
	free_page((unsigned long)Buffer[CPUNR]);
	Buffer[CPUNR]=NULL;
	
	EnterFunction("StopWaitingForHeaders");
}