Add preliminary streaming code for testing

 - Add some temporary streaming code using FFmpeg.  FFmpeg itself is not
   very ideal for streaming; lack of direct control of the sockets and
   no framedrop handling means that FFmpeg is definitely not something
   you want to use without wrapper code.  I'd prefer writing my own
   network framework in this particular case just because you give away
   so much control of the network interface.  Wasted an entire day
   trying to go through FFmpeg issues.

   There's just no way FFmpeg should be used for real streaming (at
   least without being patched or submitting some sort of patch, but I'm
   sort of feeling "meh" on that idea)

   I had to end up writing multiple threads just to handle both
   connecting and writing, because av_interleaved_write_frame blocks
   every call, stalling the main encoder thread, and thus also stalling
   draw signals.

 - Add some temporary user interface for streaming settings.  This is
   just temporary for the time being.  It's in the outputs section of
   the basic-mode settings

 - Make it so that dynamic arrays do not free all their data when the
   size just happens to be reduced to 0.  This prevents constant
   reallocation when an array keeps going from 1 item to 0 items.  Also,
   it was bad to become dependent upon that functionality.  You must now
   always explicitly call "free" on it to ensure the data is free, and
   that's how it should be.  Implicit functionality can lead to
   confusion and maintainability issues.

libobs/obs-internal.h

@@ -262,6 +262,11 @@ struct obs_output {
 	void                            *data;
 	struct obs_output_info          info;
 	obs_data_t                      settings;
+
+	signal_handler_t                signals;
+	proc_handler_t                  procs;
+
+	bool                            valid;
 };
 
 

libobs/obs-output.c

@@ -39,38 +39,57 @@ obs_output_t obs_output_create(const char *id, const char *name,
 		return NULL;
 	}
 
-	output = bmalloc(sizeof(struct obs_output));
+	output = bzalloc(sizeof(struct obs_output));
+
+	output->signals = signal_handler_create();
+	if (!output->signals)
+		goto fail;
+
+	output->procs = proc_handler_create();
+	if (!output->procs)
+		goto fail;
+
 	output->info = *info;
 	output->settings  = obs_data_newref(settings);
 	output->data      = info->create(output->settings, output);
-
-	if (!output->data) {
-		obs_data_release(output->settings);
-		bfree(output);
-		return NULL;
-	}
+	if (!output->data)
+		goto fail;
 
 	output->name = bstrdup(name);
 
 	pthread_mutex_lock(&obs->data.outputs_mutex);
 	da_push_back(obs->data.outputs, &output);
 	pthread_mutex_unlock(&obs->data.outputs_mutex);
+
+	output->valid = true;
+
 	return output;
+
+fail:
+	obs_output_destroy(output);
+	return NULL;
 }
 
 void obs_output_destroy(obs_output_t output)
 {
 	if (output) {
-		if (output->info.active) {
-			if (output->info.active(output->data))
-				output->info.stop(output->data);
+		if (output->valid) {
+			if (output->info.active) {
+				if (output->info.active(output->data))
+					output->info.stop(output->data);
+			}
+
+			pthread_mutex_lock(&obs->data.outputs_mutex);
+			da_erase_item(obs->data.outputs, &output);
+			pthread_mutex_unlock(&obs->data.outputs_mutex);
 		}
 
-		pthread_mutex_lock(&obs->data.outputs_mutex);
-		da_erase_item(obs->data.outputs, &output);
-		pthread_mutex_unlock(&obs->data.outputs_mutex);
+		if (output->data)
+			output->info.destroy(output->data);
+
+		signal_handler_destroy(output->signals);
+		proc_handler_destroy(output->procs);
 
-		output->info.destroy(output->data);
 		obs_data_release(output->settings);
 		bfree(output->name);
 		bfree(output);
@@ -143,3 +162,13 @@ void obs_output_pause(obs_output_t output)
 	if (output && output->info.pause)
 		output->info.pause(output->data);
 }
+
+signal_handler_t obs_output_signalhandler(obs_output_t output)
+{
+	return output->signals;
+}
+
+proc_handler_t obs_output_prochandler(obs_output_t output)
+{
+	return output->procs;
+}

libobs/obs.c

@@ -487,6 +487,11 @@ void obs_shutdown(void)
 		free_module(obs->modules.array+i);
 	da_free(obs->modules);
 
+	da_free(obs->data.sources);
+	da_free(obs->data.outputs);
+	da_free(obs->data.encoders);
+	da_free(obs->data.displays);
+
 	bfree(obs);
 	obs = NULL;
 }

libobs/obs.h

@@ -677,6 +677,12 @@ EXPORT void obs_output_pause(obs_output_t output);
 /* Gets the current output settings string */
 EXPORT obs_data_t obs_output_get_settings(obs_output_t output);
 
+/** Returns the signal handler for an output  */
+EXPORT signal_handler_t obs_output_signalhandler(obs_output_t output);
+
+/** Returns the procedure handler for an output */
+EXPORT proc_handler_t obs_output_prochandler(obs_output_t output);
+
 
 /* ------------------------------------------------------------------------- */
 /* Encoders */

libobs/util/darray.h

@@ -125,7 +125,7 @@ static inline void darray_resize(const size_t element_size,
 	if (size == dst->num) {
 		return;
 	} else if (size == 0) {
-		darray_free(dst);
+		dst->num = 0;
 		return;
 	}
 
@@ -305,7 +305,7 @@ static inline void darray_erase(const size_t element_size, struct darray *dst,
 		return;
 
 	if (!--dst->num) {
-		darray_free(dst);
+		dst->num = 0;
 		return;
 	}
 
@@ -336,7 +336,7 @@ static inline void darray_erase_range(const size_t element_size,
 		darray_erase(element_size, dst, start);
 		return;
 	} else if (count == dst->num) {
-		darray_free(dst);
+		dst->num = 0;
 		return;
 	}
 

obs/forms/OBSBasic.ui

@@ -319,7 +319,7 @@
           <item>
            <widget class="QPushButton" name="streamButton">
             <property name="enabled">
-             <bool>false</bool>
+             <bool>true</bool>
             </property>
             <property name="text">
              <string>Start Streaming</string>
@@ -331,6 +331,9 @@
           </item>
           <item>
            <widget class="QPushButton" name="recordButton">
+            <property name="enabled">
+             <bool>false</bool>
+            </property>
             <property name="text">
              <string>Start Recording</string>
             </property>

obs/forms/OBSBasicSettings.ui

@@ -77,7 +77,7 @@
      <item>
       <widget class="QStackedWidget" name="settingsPages">
        <property name="currentIndex">
-        <number>3</number>
+        <number>1</number>
        </property>
        <widget class="QWidget" name="generalPage">
         <layout class="QFormLayout" name="formLayout_2">
@@ -118,12 +118,101 @@
          </item>
         </layout>
        </widget>
-       <widget class="QWidget" name="outputsPage"/>
-       <widget class="QWidget" name="audioPage">
-        <layout class="QFormLayout" name="formLayout">
+       <widget class="QWidget" name="outputsPage">
+        <layout class="QFormLayout" name="formLayout_5">
          <property name="fieldGrowthPolicy">
           <enum>QFormLayout::AllNonFixedFieldsGrow</enum>
          </property>
+         <property name="labelAlignment">
+          <set>Qt::AlignRight|Qt::AlignTrailing|Qt::AlignVCenter</set>
+         </property>
+         <item row="0" column="1">
+          <widget class="QLabel" name="label_16">
+           <property name="text">
+            <string>NOTE: This is a test, just some temporary user interface</string>
+           </property>
+           <property name="wordWrap">
+            <bool>true</bool>
+           </property>
+          </widget>
+         </item>
+         <item row="2" column="0">
+          <widget class="QLabel" name="label_17">
+           <property name="minimumSize">
+            <size>
+             <width>170</width>
+             <height>0</height>
+            </size>
+           </property>
+           <property name="text">
+            <string>Video Bitrate:</string>
+           </property>
+           <property name="alignment">
+            <set>Qt::AlignRight|Qt::AlignTrailing|Qt::AlignVCenter</set>
+           </property>
+          </widget>
+         </item>
+         <item row="3" column="0">
+          <widget class="QLabel" name="label_18">
+           <property name="text">
+            <string>URL:</string>
+           </property>
+          </widget>
+         </item>
+         <item row="3" column="1">
+          <widget class="QLineEdit" name="streamURL"/>
+         </item>
+         <item row="4" column="0">
+          <widget class="QLabel" name="label_19">
+           <property name="text">
+            <string>Stream Key:</string>
+           </property>
+          </widget>
+         </item>
+         <item row="4" column="1">
+          <widget class="QLineEdit" name="streamKey">
+           <property name="echoMode">
+            <enum>QLineEdit::Password</enum>
+           </property>
+          </widget>
+         </item>
+         <item row="2" column="1">
+          <widget class="QSpinBox" name="streamVBitrate">
+           <property name="minimum">
+            <number>100</number>
+           </property>
+           <property name="maximum">
+            <number>60000</number>
+           </property>
+           <property name="value">
+            <number>2500</number>
+           </property>
+          </widget>
+         </item>
+         <item row="1" column="1">
+          <widget class="QSpinBox" name="streamABitrate">
+           <property name="minimum">
+            <number>24</number>
+           </property>
+           <property name="maximum">
+            <number>320</number>
+           </property>
+           <property name="value">
+            <number>128</number>
+           </property>
+          </widget>
+         </item>
+         <item row="1" column="0">
+          <widget class="QLabel" name="label_20">
+           <property name="text">
+            <string>Audio Bitrate:</string>
+           </property>
+          </widget>
+         </item>
+        </layout>
+       </widget>
+       <widget class="QWidget" name="audioPage">
+        <layout class="QFormLayout" name="formLayout">
          <property name="labelAlignment">
           <set>Qt::AlignRight|Qt::AlignTrailing|Qt::AlignVCenter</set>
          </property>

obs/obs-app.cpp

@@ -43,7 +43,7 @@ static void do_log(int log_level, const char *msg, va_list args)
 	OutputDebugStringA(bla);
 	OutputDebugStringA("\n");
 
-	if (log_level <= LOG_WARNING && IsDebuggerPresent())
+	if (log_level <= LOG_ERROR && IsDebuggerPresent())
 		__debugbreak();
 #else
 	vprintf(msg, args);

obs/window-basic-main.cpp

@@ -42,6 +42,7 @@ OBSBasic::OBSBasic(QWidget *parent)
 	: OBSMainWindow (parent),
 	  outputTest    (NULL),
 	  sceneChanging (false),
+	  resizeTimer   (0),
 	  ui            (new Ui::OBSBasic)
 {
 	ui->setupUi(this);
@@ -66,6 +67,12 @@ bool OBSBasic::InitBasicConfigDefaults()
 	uint32_t cx = monitors[0].cx;
 	uint32_t cy = monitors[0].cy;
 
+	/* TODO: temporary */
+	config_set_default_string(basicConfig, "OutputTemp", "URL", "");
+	config_set_default_string(basicConfig, "OutputTemp", "Key", "");
+	config_set_default_uint  (basicConfig, "OutputTemp", "VBitrate", 2500);
+	config_set_default_uint  (basicConfig, "OutputTemp", "ABitrate", 128);
+
 	config_set_default_uint  (basicConfig, "Video", "BaseCX",   cx);
 	config_set_default_uint  (basicConfig, "Video", "BaseCY",   cy);
 
@@ -732,33 +739,62 @@ void OBSBasic::on_actionSourceDown_triggered()
 {
 }
 
-void OBSBasic::on_recordButton_clicked()
+void OBSBasic::OutputConnect(bool success)
 {
-	if (outputTest) {
+	if (!success) {
 		obs_output_destroy(outputTest);
 		outputTest = NULL;
-		ui->recordButton->setText("Start Recording");
 	} else {
-		QString path = QFileDialog::getSaveFileName(this,
-				"Please enter a file name", QString(),
-				"MP4 Files (*.mp4)");
+		ui->streamButton->setText("Stop Streaming");
+	}
 
-		if (path.isNull() || path.isEmpty())
-			return;
+	ui->streamButton->setEnabled(true);
+}
+
+static void OBSOutputConnect(void *data, calldata_t params)
+{
+	bool success = calldata_bool(params, "success");
+
+	QMetaObject::invokeMethod(static_cast<OBSBasic*>(data),
+			"OutputConnect", Q_ARG(bool, success));
+}
+
+/* TODO: lots of temporary code */
+void OBSBasic::on_streamButton_clicked()
+{
+	if (outputTest) {
+		obs_output_destroy(outputTest);
+		outputTest = NULL;
+		ui->streamButton->setText("Start Streaming");
+	} else {
+		const char *url = config_get_string(basicConfig, "OutputTemp",
+				"URL");
+		const char *key = config_get_string(basicConfig, "OutputTemp",
+				"Key");
+		int vBitrate = config_get_uint(basicConfig, "OutputTemp",
+				"VBitrate");
+		int aBitrate = config_get_uint(basicConfig, "OutputTemp",
+				"ABitrate");
+
+		string fullURL = url;
+		fullURL = fullURL + "/" + key;
 
 		obs_data_t data = obs_data_create();
-		obs_data_setstring(data, "filename", QT_TO_UTF8(path));
+		obs_data_setstring(data, "filename", fullURL.c_str());
+		obs_data_setint(data, "audio_bitrate", aBitrate);
+		obs_data_setint(data, "video_bitrate", vBitrate);
 
 		outputTest = obs_output_create("ffmpeg_output", "test", data);
 		obs_data_release(data);
 
-		if (!obs_output_start(outputTest)) {
-			obs_output_destroy(outputTest);
-			outputTest = NULL;
+		if (!outputTest)
 			return;
-		}
 
-		ui->recordButton->setText("Stop Recording");
+		signal_handler_connect(obs_output_signalhandler(outputTest),
+				"connect", OBSOutputConnect, this);
+
+		obs_output_start(outputTest);
+		ui->streamButton->setEnabled(false);
 	}
 }
 

obs/window-basic-main.hpp

@@ -57,6 +57,9 @@ private:
 	void UpdateSources(OBSScene scene);
 	void InsertSceneItem(obs_sceneitem_t item);
 
+public slots:
+	void         OutputConnect(bool success);
+
 private slots:
 	void AddSceneItem(OBSSceneItem item);
 	void RemoveSceneItem(OBSSceneItem item);
@@ -116,7 +119,7 @@ private slots:
 	void on_actionSourceProperties_triggered();
 	void on_actionSourceUp_triggered();
 	void on_actionSourceDown_triggered();
-	void on_recordButton_clicked();
+	void on_streamButton_clicked();
 	void on_settingsButton_clicked();
 
 public:

obs/window-basic-settings.cpp

@@ -86,10 +86,12 @@ void OBSBasicSettings::HookWidget(QWidget *widget, const char *signal,
 
 #define COMBO_CHANGED   SIGNAL(currentIndexChanged(int))
 #define COMBO_CHANGED   SIGNAL(currentIndexChanged(int))
+#define EDIT_CHANGED    SIGNAL(textChanged(const QString &))
 #define CBEDIT_CHANGED  SIGNAL(editTextChanged(const QString &))
 #define SCROLL_CHANGED  SIGNAL(valueChanged(int))
 
 #define GENERAL_CHANGED SLOT(GeneralChanged())
+#define OUTPUTS_CHANGED SLOT(OutputsChanged())
 #define AUDIO_RESTART   SLOT(AudioChangedRestart())
 #define AUDIO_CHANGED   SLOT(AudioChanged())
 #define VIDEO_RESTART   SLOT(VideoChangedRestart())
@@ -117,6 +119,10 @@ OBSBasicSettings::OBSBasicSettings(QWidget *parent)
 		throw "Could not open locale.ini";
 
 	HookWidget(ui->language,            COMBO_CHANGED,  GENERAL_CHANGED);
+	HookWidget(ui->streamVBitrate,      SCROLL_CHANGED, OUTPUTS_CHANGED);
+	HookWidget(ui->streamABitrate,      SCROLL_CHANGED, OUTPUTS_CHANGED);
+	HookWidget(ui->streamURL,           EDIT_CHANGED,   OUTPUTS_CHANGED);
+	HookWidget(ui->streamKey,           EDIT_CHANGED,   OUTPUTS_CHANGED);
 	HookWidget(ui->channelSetup,        COMBO_CHANGED,  AUDIO_RESTART);
 	HookWidget(ui->sampleRate,          COMBO_CHANGED,  AUDIO_RESTART);
 	HookWidget(ui->desktopAudioDevice1, COMBO_CHANGED,  AUDIO_CHANGED);
@@ -310,6 +316,27 @@ void OBSBasicSettings::LoadVideoSettings()
 	loading = false;
 }
 
+void OBSBasicSettings::LoadOutputSettings()
+{
+	loading = true;
+
+	const char *url = config_get_string(main->Config(), "OutputTemp",
+			"URL");
+	const char *key = config_get_string(main->Config(), "OutputTemp",
+			"Key");
+	int videoBitrate = config_get_uint(main->Config(), "OutputTemp",
+			"VBitrate");
+	int audioBitrate = config_get_uint(main->Config(), "OutputTemp",
+			"ABitrate");
+
+	ui->streamURL->setText(QT_UTF8(url));
+	ui->streamKey->setText(QT_UTF8(key));
+	ui->streamVBitrate->setValue(videoBitrate);
+	ui->streamABitrate->setValue(audioBitrate);
+
+	loading = false;
+}
+
 static inline void LoadListValue(QComboBox *widget, const char *text,
 		const char *val)
 {
@@ -409,8 +436,8 @@ void OBSBasicSettings::LoadSettings(bool changedOnly)
 {
 	if (!changedOnly || generalChanged)
 		LoadGeneralSettings();
-	//if (!changedOnly || outputChanged)
-	//	LoadOutputSettings();
+	if (!changedOnly || outputsChanged)
+		LoadOutputSettings();
 	if (!changedOnly || audioChanged)
 		LoadAudioSettings();
 	if (!changedOnly || videoChanged)
@@ -464,6 +491,20 @@ void OBSBasicSettings::SaveVideoSettings()
 	main->ResetVideo();
 }
 
+/* TODO: Temporary! */
+void OBSBasicSettings::SaveOutputSettings()
+{
+	int videoBitrate = ui->streamVBitrate->value();
+	int audioBitrate = ui->streamABitrate->value();
+	QString url = ui->streamURL->text();
+	QString key = ui->streamKey->text();
+
+	config_set_uint(main->Config(), "OutputTemp", "VBitrate", videoBitrate);
+	config_set_uint(main->Config(), "OutputTemp", "ABitrate", audioBitrate);
+	config_set_string(main->Config(), "OutputTemp", "URL", QT_TO_UTF8(url));
+	config_set_string(main->Config(), "OutputTemp", "Key", QT_TO_UTF8(key));
+}
+
 static inline QString GetComboData(QComboBox *combo)
 {
 	int idx = combo->currentIndex();
@@ -517,8 +558,8 @@ void OBSBasicSettings::SaveSettings()
 {
 	if (generalChanged)
 		SaveGeneralSettings();
-	//if (outputChanged)
-	//	SaveOutputSettings();
+	if (outputsChanged)
+		SaveOutputSettings();
 	if (audioChanged)
 		SaveAudioSettings();
 	if (videoChanged)
@@ -622,6 +663,12 @@ void OBSBasicSettings::GeneralChanged()
 		generalChanged = true;
 }
 
+void OBSBasicSettings::OutputsChanged()
+{
+	if (!loading)
+		outputsChanged = true;
+}
+
 void OBSBasicSettings::AudioChanged()
 {
 	if (!loading)

obs/window-basic-settings.hpp

@@ -63,7 +63,7 @@ private:
 	bool QueryChanges();
 
 	void LoadGeneralSettings();
-	//void LoadOutputSettings();
+	void LoadOutputSettings();
 	void LoadAudioSettings();
 	void LoadVideoSettings();
 	void LoadSettings(bool changedOnly);
@@ -83,7 +83,7 @@ private:
 	void LoadFPSData();
 
 	void SaveGeneralSettings();
-	//void SaveOutputSettings();
+	void SaveOutputSettings();
 	void SaveAudioSettings();
 	void SaveVideoSettings();
 	void SaveSettings();
@@ -97,6 +97,7 @@ private slots:
 	void GeneralChanged();
 	void AudioChanged();
 	void AudioChangedRestart();
+	void OutputsChanged();
 	void VideoChanged();
 	void VideoChangedResolution();
 	void VideoChangedRestart();

plugins/obs-ffmpeg/obs-ffmpeg-output.c

@@ -18,6 +18,9 @@
 #include <obs.h>
 #include <util/circlebuf.h>
 #include <util/threading.h>
+#include <util/dstr.h>
+#include <util/darray.h>
+#include <util/platform.h>
 
 #include <libavutil/opt.h>
 #include <libavformat/avformat.h>
@@ -34,6 +37,7 @@ struct ffmpeg_data {
 	AVFormatContext    *output;
 	struct SwsContext  *swscale;
 
+	int                video_bitrate;
 	AVPicture          dst_picture;
 	AVFrame            *vframe;
 	int                frame_size;
@@ -41,6 +45,7 @@ struct ffmpeg_data {
 
 	uint64_t           start_timestamp;
 
+	int                audio_bitrate;
 	uint32_t           audio_samplerate;
 	enum audio_format  audio_format;
 	size_t             audio_planes;
@@ -50,8 +55,6 @@ struct ffmpeg_data {
 	AVFrame            *aframe;
 	int                total_samples;
 
-	pthread_mutex_t    write_mutex;
-
 	const char         *filename_test;
 
 	bool               initialized;
@@ -61,6 +64,17 @@ struct ffmpeg_output {
 	obs_output_t       output;
 	volatile bool      active;
 	struct ffmpeg_data ff_data;
+
+	bool               connecting;
+	pthread_t          start_thread;
+
+	bool               write_thread_active;
+	pthread_mutex_t    write_mutex;
+	pthread_t          write_thread;
+	sem_t              write_sem;
+	event_t            stop_event;
+
+	DARRAY(AVPacket)   packets;
 };
 
 /* ------------------------------------------------------------------------- */
@@ -127,8 +141,10 @@ static bool open_video_codec(struct ffmpeg_data *data)
 	AVCodecContext *context = data->video->codec;
 	int ret;
 
-	if (data->vcodec->id == AV_CODEC_ID_H264)
+	if (data->vcodec->id == AV_CODEC_ID_H264) {
 		av_opt_set(context->priv_data, "preset", "veryfast", 0);
+		av_opt_set(context->priv_data, "x264-params", "nal-hrd=cbr", 0);
+	}
 
 	ret = avcodec_open2(context, data->vcodec, NULL);
 	if (ret < 0) {
@@ -188,15 +204,17 @@ static bool create_video_stream(struct ffmpeg_data *data)
 				data->output->oformat->video_codec))
 		return false;
 
-	context                = data->video->codec;
-	context->codec_id      = data->output->oformat->video_codec;
-	context->bit_rate      = 6000000;
-	context->width         = ovi.output_width;
-	context->height        = ovi.output_height;
-	context->time_base.num = ovi.fps_den;
-	context->time_base.den = ovi.fps_num;
-	context->gop_size      = 12;
-	context->pix_fmt       = AV_PIX_FMT_YUV420P;
+	context                 = data->video->codec;
+	context->codec_id       = data->output->oformat->video_codec;
+	context->bit_rate       = data->video_bitrate * 1000;
+	context->rc_buffer_size = data->video_bitrate * 1000;
+	context->rc_max_rate    = data->video_bitrate * 1000;
+	context->width          = ovi.output_width;
+	context->height         = ovi.output_height;
+	context->time_base.num  = ovi.fps_den;
+	context->time_base.den  = ovi.fps_num;
+	context->gop_size       = 120;
+	context->pix_fmt        = AV_PIX_FMT_YUV420P;
 
 	if (data->output->oformat->flags & AVFMT_GLOBALHEADER)
 		context->flags |= CODEC_FLAG_GLOBAL_HEADER;
@@ -259,7 +277,7 @@ static bool create_audio_stream(struct ffmpeg_data *data)
 		return false;
 
 	context              = data->audio->codec;
-	context->bit_rate    = 128000;
+	context->bit_rate    = data->audio_bitrate * 1000;
 	context->channels    = get_audio_channels(aoi.speakers);
 	context->sample_rate = aoi.samples_per_sec;
 	context->sample_fmt  = data->acodec->sample_fmts ?
@@ -335,7 +353,6 @@ static void close_audio(struct ffmpeg_data *data)
 
 static void ffmpeg_data_free(struct ffmpeg_data *data)
 {
-	pthread_mutex_lock(&data->write_mutex);
 	if (data->initialized)
 		av_write_trailer(data->output);
 
@@ -348,30 +365,30 @@ static void ffmpeg_data_free(struct ffmpeg_data *data)
 
 	avformat_free_context(data->output);
 
-	pthread_mutex_unlock(&data->write_mutex);
-	pthread_mutex_destroy(&data->write_mutex);
-
 	memset(data, 0, sizeof(struct ffmpeg_data));
 }
 
 static bool ffmpeg_data_init(struct ffmpeg_data *data, const char *filename)
 {
-	memset(data, 0, sizeof(struct ffmpeg_data));
-	pthread_mutex_init_value(&data->write_mutex);
+	bool is_rtmp = false;
 
+	memset(data, 0, sizeof(struct ffmpeg_data));
 	data->filename_test = filename;
 
 	if (!filename || !*filename)
 		return false;
 
-	if (pthread_mutex_init(&data->write_mutex, NULL) != 0)
-		return false;
-
 	av_register_all();
+	avformat_network_init();
+
+	is_rtmp = (astrcmp_n(filename, "rtmp://", 7) == 0);
 
 	/* TODO: settings */
-	avformat_alloc_output_context2(&data->output, NULL, NULL,
-			data->filename_test);
+	avformat_alloc_output_context2(&data->output, NULL,
+			is_rtmp ? "flv" : NULL, data->filename_test);
+	data->output->oformat->video_codec = AV_CODEC_ID_H264;
+	data->output->oformat->audio_codec = AV_CODEC_ID_AAC;
+
 	if (!data->output) {
 		blog(LOG_WARNING, "Couldn't create avformat context");
 		goto fail;
@@ -382,6 +399,8 @@ static bool ffmpeg_data_init(struct ffmpeg_data *data, const char *filename)
 	if (!open_output_file(data))
 		goto fail;
 
+	av_dump_format(data->output, 0, NULL, 1);
+
 	data->initialized = true;
 	return true;
 
@@ -411,21 +430,46 @@ static void ffmpeg_log_callback(void *param, int level, const char *format,
 static void *ffmpeg_output_create(obs_data_t settings, obs_output_t output)
 {
 	struct ffmpeg_output *data = bzalloc(sizeof(struct ffmpeg_output));
+	pthread_mutex_init_value(&data->write_mutex);
 	data->output = output;
 
+	if (pthread_mutex_init(&data->write_mutex, NULL) != 0)
+		goto fail;
+	if (event_init(&data->stop_event, EVENT_TYPE_AUTO) != 0)
+		goto fail;
+	if (sem_init(&data->write_sem, 0, 0) != 0)
+		goto fail;
+
+	signal_handler_add(obs_output_signalhandler(output),
+			"void connect(ptr output, bool success)");
+
 	av_log_set_callback(ffmpeg_log_callback);
 
 	UNUSED_PARAMETER(settings);
 	return data;
+
+fail:
+	pthread_mutex_destroy(&data->write_mutex);
+	event_destroy(data->stop_event);
+	bfree(data);
+	return NULL;
 }
 
+static void ffmpeg_output_stop(void *data);
+
 static void ffmpeg_output_destroy(void *data)
 {
 	struct ffmpeg_output *output = data;
 
 	if (output) {
-		if (output->active)
-			ffmpeg_data_free(&output->ff_data);
+		if (output->connecting)
+			pthread_join(output->start_thread, NULL);
+
+		ffmpeg_output_stop(output);
+
+		pthread_mutex_destroy(&output->write_mutex);
+		sem_destroy(&output->write_sem);
+		event_destroy(output->stop_event);
 		bfree(data);
 	}
 }
@@ -488,9 +532,10 @@ static void receive_video(void *param, const struct video_data *frame)
 		packet.data          = data->dst_picture.data[0];
 		packet.size          = sizeof(AVPicture);
 
-		pthread_mutex_lock(&data->write_mutex);
-		ret = av_interleaved_write_frame(data->output, &packet);
-		pthread_mutex_unlock(&data->write_mutex);
+		pthread_mutex_lock(&output->write_mutex);
+		da_push_back(output->packets, &packet);
+		pthread_mutex_unlock(&output->write_mutex);
+		sem_post(&output->write_sem);
 
 	} else {
 		data->vframe->pts = data->total_frames;
@@ -511,10 +556,10 @@ static void receive_video(void *param, const struct video_data *frame)
 					context->time_base,
 					data->video->time_base);
 
-			pthread_mutex_lock(&data->write_mutex);
-			ret = av_interleaved_write_frame(data->output,
-					&packet);
-			pthread_mutex_unlock(&data->write_mutex);
+			pthread_mutex_lock(&output->write_mutex);
+			da_push_back(output->packets, &packet);
+			pthread_mutex_unlock(&output->write_mutex);
+			sem_post(&output->write_sem);
 		} else {
 			ret = 0;
 		}
@@ -528,20 +573,22 @@ static void receive_video(void *param, const struct video_data *frame)
 	data->total_frames++;
 }
 
-static inline void encode_audio(struct ffmpeg_data *output,
+static inline void encode_audio(struct ffmpeg_output *output,
 		struct AVCodecContext *context, size_t block_size)
 {
+	struct ffmpeg_data *data = &output->ff_data;
+
 	AVPacket packet = {0};
 	int ret, got_packet;
-	size_t total_size = output->frame_size * block_size * context->channels;
+	size_t total_size = data->frame_size * block_size * context->channels;
 
-	output->aframe->nb_samples = output->frame_size;
-	output->aframe->pts = av_rescale_q(output->total_samples,
+	data->aframe->nb_samples = data->frame_size;
+	data->aframe->pts = av_rescale_q(data->total_samples,
 			(AVRational){1, context->sample_rate},
 			context->time_base);
 
-	ret = avcodec_fill_audio_frame(output->aframe, context->channels,
-			context->sample_fmt, output->samples[0],
+	ret = avcodec_fill_audio_frame(data->aframe, context->channels,
+			context->sample_fmt, data->samples[0],
 			(int)total_size, 1);
 	if (ret < 0) {
 		blog(LOG_WARNING, "receive_audio: avcodec_fill_audio_frame "
@@ -549,9 +596,9 @@ static inline void encode_audio(struct ffmpeg_data *output,
 		return;
 	}
 
-	output->total_samples += output->frame_size;
+	data->total_samples += data->frame_size;
 
-	ret = avcodec_encode_audio2(context, &packet, output->aframe,
+	ret = avcodec_encode_audio2(context, &packet, data->aframe,
 			&got_packet);
 	if (ret < 0) {
 		blog(LOG_WARNING, "receive_audio: Error encoding audio: %s",
@@ -562,18 +609,16 @@ static inline void encode_audio(struct ffmpeg_data *output,
 	if (!got_packet)
 		return;
 
-	packet.pts = rescale_ts(packet.pts, context, output->audio);
-	packet.dts = rescale_ts(packet.dts, context, output->audio);
+	packet.pts = rescale_ts(packet.pts, context, data->audio);
+	packet.dts = rescale_ts(packet.dts, context, data->audio);
 	packet.duration = (int)av_rescale_q(packet.duration, context->time_base,
-			output->audio->time_base);
-	packet.stream_index = output->audio->index;
+			data->audio->time_base);
+	packet.stream_index = data->audio->index;
 
 	pthread_mutex_lock(&output->write_mutex);
-	ret = av_interleaved_write_frame(output->output, &packet);
-	if (ret != 0)
-		blog(LOG_WARNING, "receive_audio: Error writing audio: %s",
-				av_err2str(ret));
+	da_push_back(output->packets, &packet);
 	pthread_mutex_unlock(&output->write_mutex);
+	sem_post(&output->write_sem);
 }
 
 static bool prepare_audio(struct ffmpeg_data *data,
@@ -627,16 +672,76 @@ static void receive_audio(void *param, const struct audio_data *frame)
 			circlebuf_pop_front(&data->excess_frames[i],
 					data->samples[i], frame_size_bytes);
 
-		encode_audio(data, context, data->audio_size);
+		encode_audio(output, context, data->audio_size);
 	}
 }
 
-static bool ffmpeg_output_start(void *data)
+static bool process_packet(struct ffmpeg_output *output)
+{
+	AVPacket packet;
+	bool new_packet = false;
+	uint64_t time1, time2;
+	int ret;
+
+	pthread_mutex_lock(&output->write_mutex);
+	if (output->packets.num) {
+		packet = output->packets.array[0];
+		da_erase(output->packets, 0);
+		new_packet = true;
+	}
+	pthread_mutex_unlock(&output->write_mutex);
+
+	if (!new_packet)
+		return true;
+
+	time1 = os_gettime_ns();
+
+	ret = av_interleaved_write_frame(output->ff_data.output, &packet);
+	if (ret < 0) {
+		blog(LOG_WARNING, "receive_audio: Error writing packet: %s",
+				av_err2str(ret));
+
+		pthread_detach(output->write_thread);
+		output->write_thread_active = false;
+		return false;
+	}
+
+	time2 = os_gettime_ns();
+	/*blog(LOG_DEBUG, "%llu, size = %d, flags = %lX, stream = %d, "
+			"packets queued: %lu: time1; %llu",
+			time2-time1, packet.size, packet.flags,
+			packet.stream_index, output->packets.num, time1);*/
+
+	return true;
+}
+
+static void *write_thread(void *data)
 {
 	struct ffmpeg_output *output = data;
 
+	while (sem_wait(&output->write_sem) == 0) {
+		/* check to see if shutting down */
+		if (event_try(output->stop_event) == 0)
+			break;
+
+		if (!process_packet(output)) {
+			ffmpeg_output_stop(output);
+			break;
+		}
+	}
+
+	output->active = false;
+	return NULL;
+}
+
+static bool try_connect(struct ffmpeg_output *output)
+{
 	video_t video = obs_video();
 	audio_t audio = obs_audio();
+	const char *filename_test;
+	obs_data_t settings;
+	int audio_bitrate, video_bitrate;
+	int ret;
 
 	if (!video || !audio) {
 		blog(LOG_WARNING, "ffmpeg_output_start: audio and video must "
@@ -644,14 +749,18 @@ static bool ffmpeg_output_start(void *data)
 		return false;
 	}
 
-	const char *filename_test;
-	obs_data_t settings = obs_output_get_settings(output->output);
+	settings = obs_output_get_settings(output->output);
 	filename_test = obs_data_getstring(settings, "filename");
+	video_bitrate = (int)obs_data_getint(settings, "video_bitrate");
+	audio_bitrate = (int)obs_data_getint(settings, "audio_bitrate");
 	obs_data_release(settings);
 
 	if (!filename_test || !*filename_test)
 		return false;
 
+	output->ff_data.video_bitrate = video_bitrate;
+	output->ff_data.audio_bitrate = audio_bitrate;
+
 	if (!ffmpeg_data_init(&output->ff_data, filename_test))
 		return false;
 
@@ -663,21 +772,71 @@ static bool ffmpeg_output_start(void *data)
 		.format = VIDEO_FORMAT_I420
 	};
 
-	video_output_connect(video, &vsi, receive_video, output);
-	audio_output_connect(audio, &aci, receive_audio, output);
 	output->active = true;
 
+	ret = pthread_create(&output->write_thread, NULL, write_thread, output);
+	if (ret != 0) {
+		blog(LOG_WARNING, "ffmpeg_output_start: failed to create write "
+		                  "thread.");
+		ffmpeg_output_stop(output);
+		return false;
+	}
+
+	video_output_connect(video, &vsi, receive_video, output);
+	audio_output_connect(audio, &aci, receive_audio, output);
+	output->write_thread_active = true;
 	return true;
 }
 
+static void *start_thread(void *data)
+{
+	struct ffmpeg_output *output = data;
+	struct calldata params       = {0};
+
+	bool success = try_connect(output);
+
+	output->connecting = false;
+
+	calldata_setbool(&params, "success", success);
+	calldata_setptr(&params, "output", output->output);
+	signal_handler_signal(obs_output_signalhandler(output->output),
+			"connect", &params);
+	calldata_free(&params);
+
+	return NULL;
+}
+
+static bool ffmpeg_output_start(void *data)
+{
+	struct ffmpeg_output *output = data;
+	int ret;
+
+	if (output->connecting)
+		return false;
+
+	ret = pthread_create(&output->start_thread, NULL, start_thread, output);
+	return (output->connecting = (ret == 0));
+}
+
 static void ffmpeg_output_stop(void *data)
 {
 	struct ffmpeg_output *output = data;
 
 	if (output->active) {
-		output->active = false;
 		video_output_disconnect(obs_video(), receive_video, data);
 		audio_output_disconnect(obs_audio(), receive_audio, data);
+
+		if (output->write_thread_active) {
+			event_signal(output->stop_event);
+			sem_post(&output->write_sem);
+			pthread_join(output->write_thread, NULL);
+			output->write_thread_active = false;
+		}
+
+		for (size_t i = 0; i < output->packets.num; i++)
+			av_free_packet(output->packets.array+i);
+
+		da_free(output->packets);
 		ffmpeg_data_free(&output->ff_data);
 	}
 }