mpeg2t_thread.cpp

完整的RTP RTSP代码库

 */
static mpeg2t_client_t *mpeg2t_create_client (CPlayerSession *psptr,
					      const char *address,
					      in_port_t rx_port,
					      in_port_t tx_port,
					      int use_rtp,
					      double rtcp_bw, 
					      int ttl,
					      rtsp_client_t *client = NULL,
					      char *rtsp_url = NULL)
{
  mpeg2t_client_t *info;
  int ret;
  mpeg2t_msg_type_t msg;
  mpeg2t_msg_resp_t resp;
  rtsp_command_t cmd;
  rtsp_decode_t *decode;
  int err;

  info = MALLOC_STRUCTURE(mpeg2t_client_t);
  if (info == NULL) return (NULL);
  memset(info, 0, sizeof(mpeg2t_client_t));
  info->m_rtsp_client = client;
  if (client != NULL) {
    info->m_have_rtsp = 1;
  } else {
    info->m_have_rtsp = 0;
  }

  info->m_rtsp_url = rtsp_url;
  info->address = strdup(address);
  info->rx_port = rx_port;
  info->tx_port = tx_port;
  info->useRTP = use_rtp;
  info->rtcp_bw = rtcp_bw;
  info->ttl = ttl;
  info->recv_timeout = 100;
  info->decoder = create_mpeg2_transport();
  if (info->decoder == NULL) {
    mpeg2t_delete_client(info);
    return (NULL);
  }
  info->decoder->save_frames_at_start = 1;
  info->pam_recvd_sem = SDL_CreateSemaphore(0);
  info->msg_mutex = SDL_CreateMutex();
  // start the thread, which starts the data being received, so we
  // can look at the program maps
  if (mpeg2t_create_thread(info) != 0) {
    mpeg2t_delete_client(info);
    return (NULL);
  }
  SDL_Delay(100);
  msg = MPEG2T_MSG_START;
  ret = mpeg2t_thread_ipc_send_wait(info,
				  (unsigned char *)&msg,
				  sizeof(msg),
				  &resp);
  if (ret < 0 || resp < 0) {
    mpeg2t_delete_client(info);
    psptr->set_message("Couldn't create client - error %d", resp);
    return NULL;
  }
  if (info->m_rtsp_client) {
    // issue play, so we've got the thread ready to receive data, 
    // and we can look for the program maps
    memset(&cmd, 0, sizeof(rtsp_command_t));
    decode = NULL;
    cmd.scale=1.0;
    cmd.range="0-";
    err = rtsp_send_aggregate_play(info->m_rtsp_client, 
				   info->m_rtsp_url, 
				   &cmd, 
				   &decode);
    if (err != RTSP_RESPONSE_GOOD) {
      psptr->set_message("RTSP play error %s %s", 
	       decode->retcode, decode->retresp);
      free_decode_response(decode);
      memset(&cmd, 0, sizeof(rtsp_command_t));
      decode = NULL;
      rtsp_send_aggregate_teardown(info->m_rtsp_client, 
				   info->m_rtsp_url, 
				   &cmd, 
				   &decode);
      free_rtsp_client(info->m_rtsp_client);
      mpeg2t_delete_client(info);
      return NULL;
    }
    free_decode_response(decode);
  }
  
  // Wait until we get the program maps (should probably be PMAP_WAIT), 
  // so we can determine audio/video
  int max = config.get_config_value(CONFIG_MPEG2T_PAM_WAIT_SECS);
  do {
    ret = SDL_SemWaitTimeout(info->pam_recvd_sem, 1000);
    if (ret == SDL_MUTEX_TIMEDOUT) {
      max--;
      if (psptr->ShouldStopProcessing()) {
	max = -1;
      }
      mpeg2t_message(LOG_DEBUG, "timeout - still left %d", max);
    }
  } while (ret == SDL_MUTEX_TIMEDOUT && max >= 0);

  if (ret == SDL_MUTEX_TIMEDOUT) {
    if (info->decoder->program_maps_recvd != 0) {
      mpeg2t_message(LOG_INFO, "Program count received %d doesn't match program count %d", 
		     info->decoder->program_maps_recvd, 
		     info->decoder->program_count);
    } else {
      if (info->m_rtsp_client) {
	memset(&cmd, 0, sizeof(rtsp_command_t));
	decode = NULL;
	rtsp_send_aggregate_teardown(info->m_rtsp_client, 
				     info->m_rtsp_url, 
				     &cmd, 
				     &decode);
      free_rtsp_client(info->m_rtsp_client);
      }
      psptr->set_message("Did not receive Transport Stream Program Map in %d seconds", config.get_config_value(CONFIG_MPEG2T_PAM_WAIT_SECS));
      mpeg2t_delete_client(info);
      return NULL;
    }
  }
  return (info);
}

void mpeg2t_delete_client (mpeg2t_client_t *info)
{
  mpeg2t_stream_t *p;

  mpeg2t_thread_close(info);

  CHECK_AND_FREE(info->address);
  delete_mpeg2t_transport(info->decoder);
  info->decoder = NULL;
  SDL_DestroyMutex(info->msg_mutex);
  SDL_DestroySemaphore(info->pam_recvd_sem);
  p = info->stream;
  while (p != NULL) {
    info->stream = p->next_stream;
    free(p);
    p = info->stream;
  }
  free(info);
}

static void close_mpeg2t_client (void *data)
{
  mpeg2t_delete_client((mpeg2t_client_t *)data);
}

static int mpeg2t_create_video(mpeg2t_client_t *info,
			       CPlayerSession *psptr, 
			       video_query_t *vq, 
			       int video_offset,
			       int &sdesc)
{
  int ix;
  CPlayerMedia *mptr;
  codec_plugin_t *plugin;
  int created = 0;

  for (ix = 0; ix < video_offset; ix++) {
    mpeg2t_pid_t *pidptr;
    mpeg2t_es_t *es_pid;
    pidptr = mpeg2t_lookup_pid(info->decoder,vq[ix].track_id);
    if (pidptr->pak_type != MPEG2T_ES_PAK) {
      mpeg2t_message(LOG_CRIT, "mpeg2t video type is not es pak - pid %x",
		     vq[ix].track_id);
      exit(1);
    }
    es_pid = (mpeg2t_es_t *)pidptr;
    if (vq[ix].enabled != 0 && created == 0) {
      created = 1;
      mptr = new CPlayerMedia(psptr, VIDEO_SYNC);
      if (mptr == NULL) {
	return (-1);
      }
      video_info_t *vinfo;
      vinfo = MALLOC_STRUCTURE(video_info_t);
      vinfo->height = vq[ix].h;
      vinfo->width = vq[ix].w;
      plugin = check_for_video_codec(STREAM_TYPE_MPEG2_TRANSPORT_STREAM,
				     NULL,
				     NULL,
				     vq[ix].type,
				     vq[ix].profile,
				     vq[ix].config, 
				     vq[ix].config_len,
				     &config);

      int ret = mptr->create_video_plugin(plugin, 
					  STREAM_TYPE_MPEG2_TRANSPORT_STREAM,
					  NULL,
					  vq[ix].type,
					  vq[ix].profile,
					  NULL, // sdp info
					  vinfo, // video info
					  vq[ix].config,
					  vq[ix].config_len);

      if (ret < 0) {
	mpeg2t_message(LOG_ERR, "Failed to create plugin data");
	psptr->set_message("Failed to start plugin");
	delete mptr;
	return -1;
      }
      mpeg2t_stream_t *stream;
      stream = MALLOC_STRUCTURE(mpeg2t_stream_t);
      stream->m_parent = info;
      stream->m_mptr = mptr;
      stream->m_is_video = 1;
      stream->m_buffering = 1;
      stream->m_have_info = 0;
      stream->m_have_eof = 0;
      stream->next_stream = info->stream;
      info->stream = stream;
      mpeg2t_set_userdata(&es_pid->pid, stream);

      CMpeg2tVideoByteStream *vbyte;
      vbyte = new CMpeg2tVideoByteStream(es_pid, info->m_have_rtsp);
      if (vbyte == NULL) {
	mpeg2t_message(LOG_CRIT, "failed to create byte stream");
	delete mptr;
	free(stream);
	return (-1);
      }
      ret = mptr->create_media("video", vbyte, true);
      if (ret != 0) {
	mpeg2t_message(LOG_CRIT, "failed to create from file");
	free(stream);
	return (-1);
      }
      if (es_pid->info_loaded) {
	char buffer[80];
	if (mpeg2t_write_stream_info(es_pid, buffer, 80) >= 0) {
	  psptr->set_session_desc(sdesc, buffer);
	  sdesc++;
	}
      }
      mpeg2t_set_frame_status(es_pid, MPEG2T_PID_SAVE_FRAME);
    }  else {
      mpeg2t_set_frame_status(es_pid, MPEG2T_PID_NOTHING);
    }
  }
  return created;
}

static int mpeg2t_create_audio (mpeg2t_client_t *info,
				CPlayerSession *psptr, 
				audio_query_t *aq, 
				int audio_offset,
				int &sdesc)
{
  int ix;
  CPlayerMedia *mptr;
  codec_plugin_t *plugin;
  int created = 0;

  for (ix = 0; ix < audio_offset; ix++) {
    mpeg2t_pid_t *pidptr;
    mpeg2t_es_t *es_pid;
    pidptr = mpeg2t_lookup_pid(info->decoder,aq[ix].track_id);
    if (pidptr->pak_type != MPEG2T_ES_PAK) {
      mpeg2t_message(LOG_CRIT, "mpeg2t audio type is not es pak -pid %x", 
		     aq[ix].track_id);
      exit(1);
    }
    es_pid = (mpeg2t_es_t *)pidptr;
    if (aq[ix].enabled != 0 && created == 0) {
      mptr = new CPlayerMedia(psptr, AUDIO_SYNC);
      if (mptr == NULL) {
	return (-1);
      }
      audio_info_t *ainfo;
      ainfo = MALLOC_STRUCTURE(audio_info_t);
      ainfo->freq = aq[ix].sampling_freq;
      ainfo->chans = aq[ix].chans;
      ainfo->bitspersample = 0;
      plugin = check_for_audio_codec(STREAM_TYPE_MPEG2_TRANSPORT_STREAM,
				     NULL,
				     NULL,
				     aq[ix].type,
				     aq[ix].profile,
				     aq[ix].config, 
				     aq[ix].config_len,
				     &config);

      int ret = mptr->create_audio_plugin(plugin, 
					  STREAM_TYPE_MPEG2_TRANSPORT_STREAM,
					  NULL,
					  aq[ix].type,
					  aq[ix].profile,
					  NULL, // sdp info
					  ainfo, // video info
					  aq[ix].config,
					  aq[ix].config_len);

      if (ret < 0) {
	mpeg2t_message(LOG_ERR, "Failed to create plugin data");
	psptr->set_message("Failed to start plugin");
	delete mptr;
	return -1;
      }
      mpeg2t_stream_t *stream;
      stream = MALLOC_STRUCTURE(mpeg2t_stream_t);
      stream->m_parent = info;
      stream->m_mptr = mptr;
      stream->m_is_video = 0;
      stream->m_buffering = 1;
      stream->m_have_info = 0;
      stream->next_stream = info->stream;
      info->stream = stream;

      mpeg2t_set_userdata(&es_pid->pid, stream);

      created = 1;
      CMpeg2tAudioByteStream *abyte;
      abyte = new CMpeg2tAudioByteStream(es_pid, info->m_have_rtsp);
      if (abyte == NULL) {
	mpeg2t_message(LOG_CRIT, "failed to create byte stream");
	delete mptr;
	free(stream);
	return (-1);
      }
      ret = mptr->create_media("audio", abyte, true);
      if (ret != 0) {
	mpeg2t_message(LOG_CRIT, "failed to create from file");
	free(stream);
	return (-1);
      }
      if (es_pid->info_loaded) {
	char buffer[80];
	if (mpeg2t_write_stream_info(es_pid, buffer, 80) >= 0) {
	  psptr->set_session_desc(sdesc, buffer);
	  sdesc++;
	}
      }
      mpeg2t_set_frame_status(es_pid, MPEG2T_PID_SAVE_FRAME);
    }  else {
      mpeg2t_set_frame_status(es_pid, MPEG2T_PID_NOTHING);
    }
  }
  return created;
}

mpeg2t_client_t *mpeg2t_start_rtsp (CPlayerSession *psptr, 
				    const char *orig_name)
{
  char *rtsp_url;
  rtsp_client_t *rptr;
  int err;
  rtsp_command_t cmd;
  rtsp_decode_t *decode;
  int rtsp_resp;
  int have_duration = 0;
  uint64_t duration = 0;
  mpeg2t_client_t *mp2t;

  rtsp_url = strdup(orig_name + 2);
  rtsp_url[0] = 'r';
  rtsp_url[1] = 't';
  rtsp_url[2] = 's';
  rtsp_url[3] = 'p';
  // That's to change mpeg2t:// to rtsp://
  rptr = rtsp_create_client(rtsp_url, &err,
			    config.GetStringValue(CONFIG_RTSP_PROXY_ADDR),
			    config.GetIntegerValue(CONFIG_RTSP_PROXY_PORT));
  if (rptr == NULL) {
    psptr->set_message("Couldn't create rtsp client %d", err);
    free(rtsp_url);
    return NULL;
  }
  memset(&cmd, 0, sizeof(rtsp_command_t));
  // don't set accept
  rtsp_resp = rtsp_send_describe(rptr, &cmd, &decode);
  if (rtsp_resp == RTSP_RESPONSE_GOOD) {
    // good response - see what they returned.
    if (decode->content_type != NULL) {
      if (strncasecmp(decode->content_type, 
		      "application/sdp", 
		      strlen("application/sdp")) == 0) {
	free_decode_response(decode);
	free(rtsp_url);
	psptr->set_message("Returned content type is SDP - please use rtsp://");
	return NULL;
      } else if (strncasecmp(decode->content_type, 
			     "application/x-rtsp-mh",
			     strlen("application/x-rtsp-mh")) == 0) {
	// 
	if (decode->body) {
	  const char *dptr = strcasestr(decode->body, "Duration");
	  if (dptr != NULL) {
	    dptr += strlen("duration");
	    ADV_SPACE(dptr);
	    if (*dptr == '=') {
	      dptr++;
	      ADV_SPACE(dptr);
	      char *endptr;
	      duration = strtoull(dptr, &endptr, 10);
	      duration = (duration + 999) / TO_U64(1000); // convert usec to msec