vodfile.cpp

组播服务端程序

    pref_length = 0;
    pref_header = pref_buffer;

    // judge file type
    
    //begin added by xj
    read(connfd,&filetype,1);
    
    if(lseek(connfd,0,SEEK_SET)==-1L)
    {
       dbg_print0(COMMON_LOG,"seek to file beginning failed!\n ");
		
       goto broadcast_quit;
    }
        
    if(filetype==0x47)
    {
    	StreamFileType=NATIVE_MPEG2_TS;
    	
    	goto AAA;
    }
    //end added by xj
     
//    if ( ssread(pFileHeader, sizeof(char), sizeof(STREAMFILEHEADER), connfd)!=sizeof(STREAMFILEHEADER) ) {
    if ( ssread(pFileHeader, sizeof(STREAMFILEHEADER), connfd)!=sizeof(STREAMFILEHEADER) ) {
      dbg_print0(COMMON_LOG, "read file header error\n");
      main_broadcast_err = ERR_FILE;
	  
      goto broadcast_quit;
    }
	
    if ( pFileHeader->type==MPEG1 ) 
		StreamFileType=MPEG1;
    else if ( pFileHeader->type==MPEG1_NEW ) 
		StreamFileType=MPEG1_NEW;
    else if ( pFileHeader->type==TYPE_TSFILEHEADER_OLD ) 
		StreamFileType=MPEG2_TS_OLD;
    else if ( pFileHeader->type==TYPE_TSFILEHEADER_NEW ) 
		StreamFileType=MPEG2_TS_NEW;
    else if ( pFileHeader->type==TYPE_WSFILEHEADER ) 
		StreamFileType=MPEG2_WS;
    else if ( pFileHeader->type==MPEG2_PS ) 
		StreamFileType=MPEG2_PS;
    else
    {
	StreamFileType=UNKNOWN;
    }

    if ( pFileHeader->type==MPEG2_TS_OLD ) {
        signal(SIGALRM, token_proc);
        if ( setitimer(ITIMER_REAL, &timer_value, NULL)!=0 ) { 
          dbg_print0(COMMON_LOG, "main_broadcast setitimer Quit\n");
          dbg_print1(FATAL_ERROR, "setitimer return error %d\n", errno);
          main_broadcast_err = ERR_SYSTEM;
          goto broadcast_quit;
        } 
    }

 
AAA:

//  if using peak limit    global_type = type;
    switch(StreamFileType) { 
    case MPEG2_WS:
    case MPEG1:
      {
	  	
        timer_value.it_interval.tv_sec = timer_value.it_value.tv_sec =0;
        timer_value.it_interval.tv_usec= timer_value.it_value.tv_usec=0;
		
        setitimer(ITIMER_REAL, &timer_value, NULL);
        signal(SIGALRM, SIG_DFL);
		
        sem_init(&(pshared->nempty),0,0);
        sem_init(&(pshared->nstored),0,MAXBUFFLEN);
		
        bufTail = bufHead = 0;
		
        if( (ret=pthread_create(&sendpid,NULL,sendThread,NULL)) ){
           dbg_print1(FATAL_ERROR,"sendthread create error, return %d\n", ret);
           main_broadcast_err = ERR_SYSTEM;
           goto broadcast_quit;
        }
		
        dbg_print1(COMMON_LOG, "sendpid=%d\n", sendpid);
        dbg_print0(COMMON_LOG, "create send thread success\n");

        index = getEmptySpace();
        pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);

	if ( notimeofday==0 ) {// normal operation
	        
//	        if ( ssread(pPackHeader, sizeof(char), STREAMPACKETHEADER_PREFIX, connfd)!=STREAMPACKETHEADER_PREFIX ) {
	        if ( ssread(pPackHeader, STREAMPACKETHEADER_PREFIX, connfd)!=STREAMPACKETHEADER_PREFIX ) {
	          dbg_print0(COMMON_LOG, "read first STREAM HEADER error\n");
	          main_broadcast_err = ERR_FILE;
			  
	          goto broadcast_quit;
	        }
			
	        while ( sendpid ) {
				
//		  ret = ssread((char*)&(pPackHeader->packet), sizeof(char), pPackHeader->length+STREAMPACKETHEADER_PREFIX, connfd);
		  ret = ssread((char*)&(pPackHeader->packet), pPackHeader->length+STREAMPACKETHEADER_PREFIX, connfd);
		  
		  if ( ret<pPackHeader->length) {
		  	
	            pPackHeader->length = ERR_READ_LESS_DATA; // -1
	            
	            putDataSpace();
	            dbg_print1(COMMON_LOG, "one data packet read error, length=%d\n", pPackHeader->length);
				
	            break;
	          }
		  
	          if ( ret==pPackHeader->length ) { // the last packet
	            putDataSpace();
	            index = getEmptySpace();
				
	            pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	            *pPackHeader = PackHeader;        
	            pPackHeader->length = ERR_READ_ZERO_DATA; // 0
	            
	            putDataSpace();
	            dbg_print0(COMMON_LOG, "the last packet\n");
				
	            break;
	          }
	          else if ( ret!=pPackHeader->length+STREAMPACKETHEADER_PREFIX ) {
	            pPackHeader->length = ERR_READ_MORE_DATA; //-2

	            putDataSpace();
	            dbg_print0(COMMON_LOG, "next packet header is not there\n");
				
	            break;
	          }
			  
	          PackHeader = *(STREAMPACKETHEADER *)((char*)&(pPackHeader->packet)+pPackHeader->length);
			  
	          putDataSpace();
	          index = getEmptySpace();
			  
	          pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	          *pPackHeader = PackHeader;          
			  
	        }
	}
	else { // test only

	        int num=0;
	        u_int32_t size = 0;
	        double timestamp = 0.0;

	        pPackHeader->length = 15000;
	        size = 15000;
			
	        while ( size<totalSize && sendpid ) {
				
	          pPackHeader->start_timestamp=timestamp;
	          pPackHeader->length = 15000;
	          pPackHeader->packet.num = num++;
	          PackHeader = *(STREAMPACKETHEADER *)((char*)&(pPackHeader->packet)+pPackHeader->length);
			  
	          putDataSpace();
			  
	          size += 15000;
			  
	          index = getEmptySpace();
			  
	          pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	          timestamp += 15000.0*8*1000.0/1441200;
	        }
	        
	} //end if (notimeofday==0)
	
        // said to sendThread that this is the last packet and wait its quit
        pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
        pPackHeader->length = 0;        
        putDataSpace();
		
        if ( sendpid ) {
          pthread_join(sendpid,NULL);
          sendpid=0;
        }
		
        sem_destroy(&(pshared->nempty));
        sem_destroy(&(pshared->nstored));
      }
	
      break;

    case MPEG2_TS_NEW:
      {

        timer_value.it_interval.tv_sec = timer_value.it_value.tv_sec =0;
        timer_value.it_interval.tv_usec= timer_value.it_value.tv_usec=0;
		
        setitimer(ITIMER_REAL, &timer_value, NULL);
        signal(SIGALRM, SIG_DFL);
		
        sem_init(&(pshared->nempty),0,0);
        sem_init(&(pshared->nstored),0,MAXBUFFLEN);
		
        bufTail = bufHead = 0;
		
        if( (ret=pthread_create(&sendpid,NULL,sendThread,NULL)) ){
           dbg_print1(FATAL_ERROR,"sendthread create error, return %d\n", ret);
           main_broadcast_err = ERR_SYSTEM;
           goto broadcast_quit;
        }
		
        dbg_print1(COMMON_LOG, "sendpid=%d\n", sendpid);
        dbg_print0(COMMON_LOG, "create send thread success\n");

        index = getEmptySpace();
        pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);

	        
//	if ( ssread(pPackHeader, sizeof(char), STREAMPACKETHEADER_PREFIX, connfd)!=STREAMPACKETHEADER_PREFIX ) 
	if ( ssread(pPackHeader, STREAMPACKETHEADER_PREFIX, connfd)!=STREAMPACKETHEADER_PREFIX ) 
	{
		dbg_print0(COMMON_LOG, "read first STREAM HEADER error\n");
		main_broadcast_err = ERR_FILE;
			  
	        goto broadcast_quit;
	}
			
	while ( sendpid ) {

				
//		ret = ssread((char*)&(pPackHeader->packet), sizeof(char), pPackHeader->length+STREAMPACKETHEADER_PREFIX, connfd);
		ret = ssread((char*)&(pPackHeader->packet), pPackHeader->length+STREAMPACKETHEADER_PREFIX, connfd);
		  
		if ( ret<pPackHeader->length) 
		{
		  	
			pPackHeader->length = ERR_READ_LESS_DATA; // -1
	            
	            	putDataSpace();
	            	dbg_print1(COMMON_LOG, "one data packet read error, length=%d\n", pPackHeader->length);
				
	            	break;
	     	}
		  
	      	if ( ret==pPackHeader->length ) { // the last packet
	            	putDataSpace();
	            	index = getEmptySpace();
				
	            	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	            	*pPackHeader = PackHeader;        
	            	pPackHeader->length = ERR_READ_ZERO_DATA; // 0
	            
	            	putDataSpace();
	            	dbg_print0(COMMON_LOG, "the last packet\n");
				
	            	break;
	   	}
	   	else if ( ret!=pPackHeader->length+STREAMPACKETHEADER_PREFIX ) {
	            	pPackHeader->length = ERR_READ_MORE_DATA; //-2

	            	putDataSpace();
	            	dbg_print0(COMMON_LOG, "next packet header is not there\n");
				
	            	break;
	  	}
			  
	   	PackHeader = *(STREAMPACKETHEADER *)((char*)&(pPackHeader->packet)+pPackHeader->length);
			  
	     	putDataSpace();
	      	index = getEmptySpace();
			  
	     	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	      	*pPackHeader = PackHeader;          
			  
	}
	
        // said to sendThread that this is the last packet and wait its quit
        pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
        pPackHeader->length = 0;        
        putDataSpace();
		
        if ( sendpid ) {
          pthread_join(sendpid,NULL);
          sendpid=0;
        }
		
        sem_destroy(&(pshared->nempty));
        sem_destroy(&(pshared->nstored));

      }
	
      break;

    case MPEG1_NEW:
      {

        timer_value.it_interval.tv_sec = timer_value.it_value.tv_sec =0;
        timer_value.it_interval.tv_usec= timer_value.it_value.tv_usec=0;
		
        setitimer(ITIMER_REAL, &timer_value, NULL);
        signal(SIGALRM, SIG_DFL);
		
        sem_init(&(pshared->nempty),0,0);
        sem_init(&(pshared->nstored),0,MAXBUFFLEN);
		
        bufTail = bufHead = 0;
		
        if( (ret=pthread_create(&sendpid,NULL,sendThread,NULL)) ){
           dbg_print1(FATAL_ERROR,"sendthread create error, return %d\n", ret);
           main_broadcast_err = ERR_SYSTEM;
           goto broadcast_quit;
        }
		
        dbg_print1(COMMON_LOG, "sendpid=%d\n", sendpid);
        dbg_print0(COMMON_LOG, "create send thread success\n");

        index = getEmptySpace();
        pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);

	        
//	if ( ssread(pPackHeader, sizeof(char), STREAMPACKETHEADER_PREFIX, connfd)!=STREAMPACKETHEADER_PREFIX ) 
	if ( ssread(pPackHeader, STREAMPACKETHEADER_PREFIX, connfd)!=STREAMPACKETHEADER_PREFIX ) 
	{
		dbg_print0(COMMON_LOG, "read first STREAM HEADER error\n");
		main_broadcast_err = ERR_FILE;
			  
	        goto broadcast_quit;
	}
			
	while ( sendpid ) {

				
//		ret = ssread((char*)&(pPackHeader->packet), sizeof(char), pPackHeader->length+STREAMPACKETHEADER_PREFIX, connfd);
		ret = ssread((char*)&(pPackHeader->packet), pPackHeader->length+STREAMPACKETHEADER_PREFIX, connfd);
		  
		if ( ret<pPackHeader->length) 
		{
		  	
			pPackHeader->length = ERR_READ_LESS_DATA; // -1
	            
	            	putDataSpace();
	            	dbg_print1(COMMON_LOG, "one data packet read error, length=%d\n", pPackHeader->length);
				
	            	break;
	     	}
		  
	      	if ( ret==pPackHeader->length ) { // the last packet
	            	putDataSpace();
	            	index = getEmptySpace();
				
	            	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	            	*pPackHeader = PackHeader;        
	            	pPackHeader->length = ERR_READ_ZERO_DATA; // 0
	            
	            	putDataSpace();
	            	dbg_print0(COMMON_LOG, "the last packet\n");
				
	            	break;
	   	}
	   	else if ( ret!=pPackHeader->length+STREAMPACKETHEADER_PREFIX ) {
	            	pPackHeader->length = ERR_READ_MORE_DATA; //-2

	            	putDataSpace();
	            	dbg_print0(COMMON_LOG, "next packet header is not there\n");
				
	            	break;
	  	}
			  
	   	PackHeader = *(STREAMPACKETHEADER *)((char*)&(pPackHeader->packet)+pPackHeader->length);
			  
	     	putDataSpace();
	      	index = getEmptySpace();
			  
	     	pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
	      	*pPackHeader = PackHeader;          
			  
	}
	
        // said to sendThread that this is the last packet and wait its quit
        pPackHeader = (STREAMPACKETHEADER *)(pshared->buffer[index]);
        pPackHeader->length = 0;        
        putDataSpace();
		
        if ( sendpid ) {
          pthread_join(sendpid,NULL);
          sendpid=0;
        }
		
        sem_destroy(&(pshared->nempty));
        sem_destroy(&(pshared->nstored));

      }
	
      break;


    case MPEG2_TS_OLD:
      {
        double band;
        int max_length=MAXPACKETLEN-sizeof(LIVE_PACKET);
        int one_shot = 0;
        LIVE_PACKET *pPacket;
 
     	dbg_print0(COMMON_LOG ,"Enter MPEG2_TS_OLD first\n");
        
        pTSFileHeader=(TSFILEHEADER*)pFileHeader;
        
  //	dbg_print1(COMMON_LOG,"The bandwidth is %lld!\n ", pTSFileHeader->band);
 	dbg_print1(COMMON_LOG,"sizeof(LIVE_PACKET) is %d\n ", sizeof(LIVE_PACKET));
  		       
        band = pTSFileHeader->band;
        
        
        if (band<=2000000 || band>=7000000) {
          band = 5000000;
        }
        
 	dbg_print1(COMMON_LOG,"band is %f\n ", band);
 
	       
        pPacket = (LIVE_PACKET*) buffer;
        pPacket->num = 0;

        timer_value.it_interval.tv_sec = timer_value.it_value.tv_sec =0;
        timer_value.it_interval.tv_usec= timer_value.it_value.tv_usec=0;
        
        setitimer(ITIMER_REAL, &timer_value, NULL);
        signal(SIGALRM, SIG_DFL);
        
        one_token   = 0;    
        nextPTS = 0;
        
        GETTIMEOFDAY(&firstTime, NULL);

        token   = 0;
        bChannelStart=TRUE;
        
 //       one_shot = max_length;
        one_shot = read(connfd,pPacket->str, max_length);
  
   	dbg_print1(COMMON_LOG,"one_shot is %d!\n ", one_shot);
       
        while ( one_shot!=0 ) {
          pPacket->num++;
          if ( !t_sock.SendMsg((char*)pPacket, one_shot+sizeof(int)) ) {
            dbg_print0(COMMON_LOG, "send error\n");
            goto broadcast_quit;
          } 
          
//      	dbg_print0(COMMON_LOG ,"Enter MPEG2_TS_OLD middle\n");
      	         
          nextPTS +=one_shot*8000000.0/band;
          one_shot = read(connfd,pPacket->str, max_length);

   		
//          dbg_print1(COMMON_LOG,"one_shot is %d!\n ", one_shot);
          
          GETTIMEOFDAY(&secondTime, NULL);
          
          token = one_token+(secondTime.tv_sec-firstTime.tv_sec)*1000000.0+secondTime.tv_usec-firstTime.tv_usec;
          if ( token<nextPTS ) {
//            printf("token %f, nextPTS %f, usleep %ld\n", token, nextPTS, (long)(nextPTS-token));
            usleep((long)(nextPTS-token)); 
          }
        }
        
     	dbg_print0(COMMON_LOG ,"Enter MPEG2_TS_OLD last\n");
        
      }
      break;

    case NATIVE_MPEG2_TS:
      {