OBS源码分析三RTMP推流输出_obs源代码

1.第一步先看RTMP推流输出流程图

2.第二步RTMP输出代码详细步骤说明 

首先是RtmpOutput输出对象的创建和编码器绑定

RtmpOutput::RtmpOutput() {
	CreateOutputWithUniqueName();
}
 
RtmpOutput::~RtmpOutput()
{
	obs_encoder_release(videoEncoder_);
	obs_encoder_release(delayVideoEncoder_);
}
 
void RtmpOutput::CreateOutputWithUniqueName() {
	uid_++;
	auto uid = uid_.load();
	std::string rtmpNameFormat = "rtmp_stream_%d";
	std::string rtmpOutputName = string_format(rtmpNameFormat, uid);
	obs_data_t* rtmpSettings = obs_data_create();
	SetDefaultRtmpSettings(rtmpSettings);
	auto rtmpSettingsGuard = gsl::finally([&]() { obs_data_release(rtmpSettings); });
	output_ = obs_output_create("rtmp_output", rtmpOutputName.c_str(),
		nullptr, nullptr);
	UpdateOutputSettings(rtmpSettings);
	auto rtmpOutputGuard = gsl::finally([&]() {	obs_output_release(output_); });
 
	std::string rtmpServiceNameFormat = "rtmp_service_%d";
	std::string rtmpServiceName = string_format(rtmpServiceNameFormat, uid);
 
	service_ = obs_service_create("rtmp_custom", rtmpServiceName.c_str(), nullptr,
		nullptr);
	auto serviceGuard = gsl::finally([&]() { obs_service_release(service_); });
 
	obs_output_set_service(output_, service_);
}

调用obs_output_create()函数,根据输出id创建推流对象,与创建编码对象类似,推流对象在加载模块时已添加到obs->output_types中,获取到的推流输出对象赋值给streamOutput指针调用obs_output_set_video_encoder()函数,将推流输出video_encoder设置为编码器；

bool obs_output_start(obs_output_t *output)
{
	bool encoded;
	if (!obs_output_valid(output, "obs_output_start"))
		return false;
	if (!output->context.data)
		return false;
	output->reconnect_total_retries = 0;
	encoded = (output->info.flags & OBS_OUTPUT_ENCODED) != 0;
 
	if (encoded && output->delay_sec) {
		return obs_output_delay_start(output);
	} else {
		if (obs_output_actual_start(output)) {
			do_output_signal(output, "starting");
			return true;
		}
 
		return false;
	}
}

1. 调用obs_output_start() -> obs_output_actual_start() 回调推流对象output->info.start()回调函数开启推流,其中start绑定至rtmp_stream_start()

bool obs_output_actual_start(obs_output_t *output)
{
	bool success = false;
 
	os_event_wait(output->connecting_event);
	if (delay_active(output) && delay_capturing(output)) {
		os_event_signal(output->connecting_event);
		return true;
	}
		
	os_event_wait(output->stopping_event);
	output->stop_code = 0;
	if (output->last_error_message) {
		bfree(output->last_error_message);
		output->last_error_message = NULL;
	}
	blog(LOG_INFO, "output %s actual start", obs_output_get_name(output));
	if (output->context.data) {
		success = output->info.start(output->context.data);
		if (!success) {
			if (output->mixer) {
				obs_stop_core_audio_mix(output->mixer);
				output->mixer = NULL;
			}
			if (output->video_output) {
				obs_stop_video_output(output->video_output);
				output->video_output = NULL;
			}
 
			os_event_signal(output->connecting_event);
			blog(LOG_INFO, "output %s actual start return false", obs_output_get_name(output));
		}
	}
 
	if (success && output->video) {
		output->starting_frame_count =
			video_output_get_total_frames(output->video);
		output->starting_drawn_count = obs->video.total_frames;
		output->starting_lagged_count = obs->video.lagged_frames;
	}
 
	if (os_atomic_load_long(&output->delay_restart_refs))
		os_atomic_dec_long(&output->delay_restart_refs);
 
	output->caption_timestamp = 0;
	return success;
}

 static bool rtmp_stream_start(void *data),创建线程,执行connect_thread()函数,static void *connect_thread(void *data)

init_connect()初始化推流,调用free_packets清空stream->packets,获取推流设置,赋值到stream,try_connect()连接rtmp服务器​​​​​​​;

static void *connect_thread(void *data)
{
	struct rtmp_stream *stream = data;
	int ret;
 
	os_set_thread_name("rtmp-stream: connect_thread");
 
	if (!init_connect(stream)) {
		obs_output_signal_stop(stream->output, OBS_OUTPUT_BAD_PATH);
		return NULL;
	}
 
	ret = try_connect(stream);
 
	if (ret != OBS_OUTPUT_SUCCESS) {
		obs_output_signal_stop(stream->output, ret);
		info("Connection to %s failed: %d", stream->path.array, ret);
	}
 
	if (!stopping(stream))
		pthread_detach(stream->connect_thread);
 
	os_atomic_set_bool(&stream->connecting, false);
	return NULL;
}

 RTMP_Init(&stream->rtmp)初始化rtmp客户端,设置推流服务器地址、用户名、密码、流地址、音频编码名称(为何没有添加视频编码名称);

RTMP_Connect()连接rtmp服务器;

RTMP_ConnectStream()连接rtmp流地址;
init_send()启动发送函数,reset_semaphore()重置发送信号量,创建推流执行线程send_thread,发送视频关键数send_meta_data();

然后开启推流数据捕获obs_output_begin_data_capture();

send_thread()线程函数:循环等待信号量stream->send_sem 被唤醒,唤醒后 get_next_packet()取出队列中的第一个已编码数据包,执行 send_packet 函数,调用flv_packet_mux进行flv数据封包,再调用RTMP_Write()发送数据包,完成视频数据推流.

static void *send_thread(void *data)
{
	struct rtmp_stream *stream = data;
 
	os_set_thread_name("rtmp-stream: send_thread");
 
	while (os_sem_wait(stream->send_sem) == 0) {
		struct encoder_packet packet;
 
		if (stopping(stream) && stream->stop_ts == 0) {
			break;
		}
 
		if (!get_next_packet(stream, &packet))
			continue;
 
		if (stopping(stream)) {
			if (can_shutdown_stream(stream, &packet)) {
				obs_encoder_packet_release(&packet);
				break;
			}
		}
 
		if (!stream->sent_headers) {
			if (!send_headers(stream)) {
				os_atomic_set_bool(&stream->disconnected, true);
				break;
			}
		}
 
		if (send_packet(stream, &packet, false, packet.track_idx) < 0) {
			os_atomic_set_bool(&stream->disconnected, true);
			break;
		}
	}
 
	if (disconnected(stream)) {
		info("Disconnected from %s", stream->path.array);
	} else {
		info("User stopped the stream");
	}
 
	if (stream->new_socket_loop) {
		os_event_signal(stream->send_thread_signaled_exit);
		os_event_signal(stream->buffer_has_data_event);
		pthread_join(stream->socket_thread, NULL);
		stream->socket_thread_active = false;
		stream->rtmp.m_bCustomSend = false;
	}
 
	set_output_error(stream);
	RTMP_Close(&stream->rtmp);
 
	if (!stopping(stream)) {
		pthread_detach(stream->send_thread);
		obs_output_signal_stop(stream->output, stream->sent_headers? OBS_OUTPUT_DISCONNECTED: OBS_OUTPUT_CONNECT_FAILED);
	} else {
		obs_output_end_data_capture(stream->output);
	}
 
	free_packets(stream);
	os_event_reset(stream->stop_event);
	os_atomic_set_bool(&stream->active, false);
	stream->sent_headers = false;
	return NULL;
}