From ac2c08927f31f4ec15b15fed30956182053ab37f Mon Sep 17 00:00:00 2001
From: jp9000 <obs.jim@gmail.com>
Date: Thu, 24 Oct 2013 00:57:55 -0700
Subject: [PATCH] added intial async audio/video code, fixed a few bugs,
 improved thread safety, and made a few other minor adjustments

---
 README                               |   2 +-
 build/data/effects/yuv_to_rgb.effect |  37 ++++
 libobs/media-io/audio-io.c           | 125 ++++++++++--
 libobs/media-io/audio-io.h           |  55 +++++-
 libobs/media-io/video-io.h           |  23 ++-
 libobs/obs-defs.h                    |   6 +-
 libobs/obs-scene.c                   |   6 +-
 libobs/obs-source.c                  | 284 +++++++++++++++++++++++++--
 libobs/obs-source.h                  |  58 ++++--
 libobs/obs.h                         |  67 ++++++-
 libobs/util/circlebuf.h              | 202 +++++++++++++++++++
 11 files changed, 786 insertions(+), 79 deletions(-)
 create mode 100644 build/data/effects/yuv_to_rgb.effect
 create mode 100644 libobs/util/circlebuf.h

diff --git a/README b/README
index 4cb6c90b5..a61947001 100644
--- a/README
+++ b/README
@@ -14,7 +14,7 @@ What's the goal of rewriting OBS?
    necessary for user interface.  It also means allowing the use of OpenGL as
    well as Direct3D.
 
- - Separate the application from the core, allowing custom appliction of
+ - Separate the application from the core, allowing custom application of
    the core if desired, and easier extending of the user interface.
 
  - Simplify complex systems to not only make it easier to use, but easier to
diff --git a/build/data/effects/yuv_to_rgb.effect b/build/data/effects/yuv_to_rgb.effect
new file mode 100644
index 000000000..5cb3de13b
--- /dev/null
+++ b/build/data/effects/yuv_to_rgb.effect
@@ -0,0 +1,37 @@
+uniform float4x4 ViewProj;
+uniform float4x4 yuv_matrix;
+texture2d tex;
+
+sampler_state def_sampler {
+	Filter   = Linear;
+	AddressU = Clamp;
+	AddressV = Clamp;
+};
+
+struct VertInOut {
+	float4 pos : POSITION;
+	float2 uv  : TEXCOORD0;
+};
+
+VertInOut VSConvert(VertInOut vert_in)
+{
+	VertInOut vert_out;
+	vert_out.pos = mul(vert_in.pos, ViewProj);
+	vert_out.uv  = vert_in.uv;
+	return vert_out;
+}
+
+float4 PSConvert(VertInOut vert_in)
+{
+	float4 yuv = tex.Sample(def_sampler, vert_in.uv);
+	return saturate(mul(float4(yuv.xyz, 1.0), yuv_matrix));
+}
+
+technique ConvertYUV
+{
+	pass
+	{
+		vertex_shader = VSConvert(vert_in);
+		pixel_shader  = PSConvert(vert_in);
+	}
+}
diff --git a/libobs/media-io/audio-io.c b/libobs/media-io/audio-io.c
index 952340898..5aac18bf5 100644
--- a/libobs/media-io/audio-io.c
+++ b/libobs/media-io/audio-io.c
@@ -17,26 +17,56 @@
 
 #include "../util/threading.h"
 #include "../util/darray.h"
+#include "../util/circlebuf.h"
 #include "../util/platform.h"
 
 #include "audio-io.h"
 
 /* TODO: Incomplete */
 
-struct audio_output {
-	struct audio_info info;
-	media_t           media;
-	media_output_t    output;
+struct audio_line {
+	struct audio_output        *audio;
+	struct circlebuf           buffer;
+	uint64_t                   base_timestamp;
+	uint64_t                   last_timestamp;
 
-	pthread_t         thread;
-	pthread_mutex_t   data_mutex;
-	event_t           stop_event;
-
-	struct darray     pending_frames;
-
-	bool              initialized;
+	/* states whether this line is still being used.  if not, then when the
+	 * buffer is depleted, it's destroyed */
+	bool                       alive;
 };
 
+static inline void audio_line_destroy_data(struct audio_line *line)
+{
+	circlebuf_free(&line->buffer);
+	bfree(line);
+}
+
+struct audio_output {
+	struct audio_info          info;
+	size_t                     block_size;
+	media_t                    media;
+	media_output_t             output;
+
+	pthread_t                  thread;
+	event_t                    stop_event;
+
+	DARRAY(uint8_t)            pending_bytes;
+
+	bool                       initialized;
+
+	pthread_mutex_t            line_mutex;
+	DARRAY(struct audio_line*) lines;
+};
+
+static inline void audio_output_removeline(struct audio_output *audio,
+		struct audio_line *line)
+{
+	pthread_mutex_lock(&audio->line_mutex);
+	da_erase_item(audio->lines, &line);
+	pthread_mutex_unlock(&audio->line_mutex);
+	audio_line_destroy_data(line);
+}
+
 /* ------------------------------------------------------------------------- */
 
 static void *audio_thread(void *param)
@@ -44,7 +74,6 @@ static void *audio_thread(void *param)
 	struct audio_output *audio = param;
 
 	while (event_try(&audio->stop_event) == EAGAIN) {
-		os_sleep_ms(5);
 		/* TODO */
 	}
 
@@ -55,8 +84,8 @@ static void *audio_thread(void *param)
 
 static inline bool valid_audio_params(struct audio_info *info)
 {
-	return info->channels > 0 && info->format && info->name &&
-		info->samples_per_sec > 0 && info->speakers > 0;
+	return info->format && info->name && info->samples_per_sec > 0 &&
+	       info->speakers > 0;
 }
 
 static inline bool ao_add_to_media(audio_t audio)
@@ -85,9 +114,12 @@ int audio_output_open(audio_t *audio, media_t media, struct audio_info *info)
 	memset(out, 0, sizeof(struct audio_output));
 
 	memcpy(&out->info, info, sizeof(struct audio_info));
+	pthread_mutex_init_value(&out->line_mutex);
 	out->media = media;
+	out->block_size = get_audio_channels(info->speakers) *
+	                  get_audio_bytes_per_channel(info->type);
 
-	if (pthread_mutex_init(&out->data_mutex, NULL) != 0)
+	if (pthread_mutex_init(&out->line_mutex, NULL) != 0)
 		goto fail;
 	if (event_init(&out->stop_event, true) != 0)
 		goto fail;
@@ -105,16 +137,27 @@ fail:
 	return AUDIO_OUTPUT_FAIL;
 }
 
-void audio_output_data(audio_t audio, struct audio_data *data)
+audio_line_t audio_output_createline(audio_t audio)
 {
-	pthread_mutex_lock(&audio->data_mutex);
-	/* TODO */
-	pthread_mutex_unlock(&audio->data_mutex);
+	struct audio_line *line = bmalloc(sizeof(struct audio_line));
+	memset(line, 0, sizeof(struct audio_line));
+	line->alive = true;
+
+	pthread_mutex_lock(&audio->line_mutex);
+	da_push_back(audio->lines, &line);
+	pthread_mutex_unlock(&audio->line_mutex);
+	return line;
+}
+
+const struct audio_info *audio_output_getinfo(audio_t audio)
+{
+	return &audio->info;
 }
 
 void audio_output_close(audio_t audio)
 {
 	void *thread_ret;
+	size_t i;
 
 	if (!audio)
 		return;
@@ -124,8 +167,50 @@ void audio_output_close(audio_t audio)
 		pthread_join(audio->thread, &thread_ret);
 	}
 
+	for (i = 0; i < audio->lines.num; i++)
+		audio_line_destroy_data(audio->lines.array[i]);
+
+	da_free(audio->lines);
 	media_remove_output(audio->media, audio->output);
 	event_destroy(&audio->stop_event);
-	pthread_mutex_destroy(&audio->data_mutex);
+	pthread_mutex_destroy(&audio->line_mutex);
 	bfree(audio);
 }
+
+void audio_line_destroy(struct audio_line *line)
+{
+	if (line) {
+		if (!line->buffer.size)
+			audio_output_removeline(line->audio, line);
+		else
+			line->alive = false;
+	}
+}
+
+size_t audio_output_blocksize(audio_t audio)
+{
+	return audio->block_size;
+}
+
+static inline uint64_t convert_to_sample_offset(audio_t audio, uint64_t offset)
+{
+	return (uint64_t)((double)offset *
+	                  (1000000000.0 / (double)audio->info.samples_per_sec));
+}
+
+void audio_line_output(audio_line_t line, const struct audio_data *data)
+{
+	if (!line->buffer.size) {
+		line->base_timestamp = data->timestamp;
+
+		circlebuf_push_back(&line->buffer, data->data,
+				data->frames * line->audio->block_size);
+	} else {
+		uint64_t position = data->timestamp - line->base_timestamp;
+		position = convert_to_sample_offset(line->audio, position);
+		position *= line->audio->block_size;
+
+		circlebuf_place(&line->buffer, position, data->data,
+				data->frames * line->audio->block_size);
+	}
+}
diff --git a/libobs/media-io/audio-io.h b/libobs/media-io/audio-io.h
index 0a0f6495a..d59811de7 100644
--- a/libobs/media-io/audio-io.h
+++ b/libobs/media-io/audio-io.h
@@ -30,7 +30,9 @@ extern "C" {
  */
 
 struct audio_output;
+struct audio_line;
 typedef struct audio_output *audio_t;
+typedef struct audio_line   *audio_line_t;
 
 enum audio_type {
 	AUDIO_FORMAT_UNKNOWN,
@@ -56,11 +58,8 @@ enum speaker_setup {
 };
 
 struct audio_data {
-	void            *data;
+	const void      *data;
 	uint32_t        frames;
-	uint32_t        speakers;
-	uint32_t        samples_per_sec;
-	enum audio_type type;
 	uint64_t        timestamp;
 };
 
@@ -68,21 +67,65 @@ struct audio_info {
 	const char         *name;
 	const char         *format;
 
-	uint32_t           channels;
 	uint32_t           samples_per_sec;
 	enum audio_type    type;
 	enum speaker_setup speakers;
 };
 
+
+static inline uint32_t get_audio_channels(enum speaker_setup speakers)
+{
+	switch (speakers) {
+	case SPEAKERS_MONO:             return 1;
+	case SPEAKERS_STEREO:           return 2;
+	case SPEAKERS_2POINT1:          return 3;
+	case SPEAKERS_SURROUND:
+	case SPEAKERS_QUAD:             return 4;
+	case SPEAKERS_4POINT1:          return 5;
+	case SPEAKERS_5POINT1:
+	case SPEAKERS_5POINT1_SURROUND: return 6;
+	case SPEAKERS_7POINT1:          return 8;
+	case SPEAKERS_7POINT1_SURROUND: return 8;
+	default:
+	case SPEAKERS_UNKNOWN:          return 0;
+	}
+}
+
+static inline size_t get_audio_bytes_per_channel(enum audio_type type)
+{
+	switch (type) {
+	case AUDIO_FORMAT_8BIT:    return 1;
+	case AUDIO_FORMAT_16BIT:   return 2;
+	case AUDIO_FORMAT_24BIT:   return 3;
+	case AUDIO_FORMAT_FLOAT:
+	case AUDIO_FORMAT_32BIT:   return 4;
+	default:
+	case AUDIO_FORMAT_UNKNOWN: return 0;
+	}
+}
+
+static inline size_t get_audio_size(enum audio_type type,
+		enum speaker_setup speakers, uint32_t frames)
+{
+	return get_audio_channels(speakers) *
+	       get_audio_bytes_per_channel(type) *
+	       frames;
+}
+
 #define AUDIO_OUTPUT_SUCCESS       0
 #define AUDIO_OUTPUT_INVALIDPARAM -1
 #define AUDIO_OUTPUT_FAIL         -2
 
 EXPORT int audio_output_open(audio_t *audio, media_t media,
 		struct audio_info *info);
-EXPORT void audio_output_data(audio_t audio, struct audio_data *data);
+EXPORT audio_line_t audio_output_createline(audio_t audio);
+EXPORT size_t audio_output_blocksize(audio_t audio);
+EXPORT const struct audio_info *audio_output_getinfo(audio_t audio);
 EXPORT void audio_output_close(audio_t audio);
 
+EXPORT void audio_line_destroy(audio_line_t line);
+EXPORT void audio_line_output(audio_line_t line, const struct audio_data *data);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/libobs/media-io/video-io.h b/libobs/media-io/video-io.h
index 7881c9e38..a0694cb5a 100644
--- a/libobs/media-io/video-io.h
+++ b/libobs/media-io/video-io.h
@@ -32,6 +32,16 @@ extern "C" {
 struct video_output;
 typedef struct video_output *video_t;
 
+enum video_type {
+	VIDEO_FORMAT_UNKNOWN,
+	VIDEO_FORMAT_YUV444,
+	VIDEO_FORMAT_YUV422,
+	VIDEO_FORMAT_YUV420,
+	VIDEO_FORMAT_RGBA,
+	VIDEO_FORMAT_BGRA,
+	VIDEO_FORMAT_BGRX,
+};
+
 struct video_frame {
 	const void *data;
 	uint32_t   row_size;
@@ -39,13 +49,14 @@ struct video_frame {
 };
 
 struct video_info {
-	const char *name;
-	const char *format;
+	const char      *name;
+	const char      *format;
 
-	uint32_t   fps_num; /* numerator */
-	uint32_t   fps_den; /* denominator */
-	uint32_t   width;
-	uint32_t   height;
+	enum video_type type;
+	uint32_t        fps_num; /* numerator */
+	uint32_t        fps_den; /* denominator */
+	uint32_t        width;
+	uint32_t        height;
 };
 
 #define VIDEO_OUTPUT_SUCCESS       0
diff --git a/libobs/obs-defs.h b/libobs/obs-defs.h
index 39087656f..fde45bef6 100644
--- a/libobs/obs-defs.h
+++ b/libobs/obs-defs.h
@@ -22,6 +22,6 @@
 #define MODULE_FILENOTFOUND     -2
 #define MODULE_FUNCTIONNOTFOUND -3
 
-#define SOURCE_VIDEO (1<<0)
-#define SOURCE_AUDIO (1<<1)
-#define SOURCE_ASYNC (1<<2)
+#define SOURCE_VIDEO          (1<<0)
+#define SOURCE_AUDIO          (1<<1)
+#define SOURCE_ASYNC_VIDEO    (1<<2)
diff --git a/libobs/obs-scene.c b/libobs/obs-scene.c
index 4f490177f..e3b5b6995 100644
--- a/libobs/obs-scene.c
+++ b/libobs/obs-scene.c
@@ -41,7 +41,7 @@ static void scene_destroy(void *data)
 
 static uint32_t scene_get_output_flags(void *data)
 {
-	return SOURCE_VIDEO | SOURCE_AUDIO;
+	return SOURCE_VIDEO;
 }
 
 static void scene_video_render(void *data)
@@ -111,6 +111,8 @@ obs_scene_t obs_scene_create(void)
 	struct obs_source *source = bmalloc(sizeof(struct obs_source));
 	struct obs_scene  *scene  = scene_create(NULL, source);
 
+	memset(source, 0, sizeof(struct obs_source));
+
 	source->data = scene;
 	if (!source->data) {
 		bfree(source);
@@ -118,7 +120,7 @@ obs_scene_t obs_scene_create(void)
 	}
 
 	scene->source = source;
-	obs_source_init(source);
+	obs_source_init(source, NULL, &scene_info);
 	memcpy(&source->callbacks, &scene_info, sizeof(struct source_info));
 	return scene;
 }
diff --git a/libobs/obs-source.c b/libobs/obs-source.c
index 05bb5febe..9fd70f15e 100644
--- a/libobs/obs-source.c
+++ b/libobs/obs-source.c
@@ -15,6 +15,8 @@
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 ******************************************************************************/
 
+#include "util/platform.h"
+
 #include "obs.h"
 #include "obs-data.h"
 
@@ -77,17 +79,39 @@ static inline const struct source_info *find_source(struct darray *list,
 	return NULL;
 }
 
-void obs_source_init(struct obs_source *source)
+bool obs_source_init(struct obs_source *source, const char *settings,
+		const struct source_info *info)
 {
-	source->filter_target    = NULL;
-	source->rendering_filter = false;
+	uint32_t flags = info->get_output_flags(source->data);
 
-	dstr_init(&source->settings);
-	da_init(source->filters);
+	pthread_mutex_init_value(&source->filter_mutex);
+	pthread_mutex_init_value(&source->video_mutex);
+	pthread_mutex_init_value(&source->audio_mutex);
+	dstr_copy(&source->settings, settings);
+	memcpy(&source->callbacks, info, sizeof(struct source_info));
 
+	if (pthread_mutex_init(&source->filter_mutex, NULL) != 0)
+		return false;
+	if (pthread_mutex_init(&source->audio_mutex, NULL) != 0)
+		return false;
+	if (pthread_mutex_init(&source->video_mutex, NULL) != 0)
+		return false;
+
+	if (flags & SOURCE_AUDIO) {
+		source->audio_line = audio_output_createline(obs->audio);
+		if (!source->audio_line) {
+			blog(LOG_ERROR, "Failed to create audio line for "
+			                "source");
+			return false;
+		}
+	}
+
+	source->valid = true;
 	pthread_mutex_lock(&obs->source_list_mutex);
 	da_push_back(obs->sources, &source);
 	pthread_mutex_unlock(&obs->source_list_mutex);
+
+	return true;
 }
 
 obs_source_t obs_source_create(enum obs_source_type type, const char *name,
@@ -113,27 +137,57 @@ obs_source_t obs_source_create(enum obs_source_type type, const char *name,
 	}
 
 	source = bmalloc(sizeof(struct obs_source));
-	source->data = info->create(settings, source);
-	if (!source->data) {
-		bfree(source);
-		return NULL;
-	}
+	memset(source, 0, sizeof(struct obs_source));
+
+	source->data = info->create(settings, source);
+	if (!source->data)
+		goto fail;
+
+	if (!obs_source_init(source, settings, info))
+		goto fail;
 
-	obs_source_init(source);
-	dstr_copy(&source->settings, settings);
-	memcpy(&source->callbacks, info, sizeof(struct source_info));
 	return source;
+
+fail:
+	blog(LOG_ERROR, "obs_source_create failed");
+	obs_source_destroy(source);
+	return NULL;
 }
 
 void obs_source_destroy(obs_source_t source)
 {
 	if (source) {
-		pthread_mutex_lock(&obs->source_list_mutex);
-		da_erase_item(obs->sources, &source);
-		pthread_mutex_unlock(&obs->source_list_mutex);
+		size_t i;
+		if (source->filter_parent)
+			obs_source_filter_remove(source->filter_parent, source);
+
+		for (i = 0; i < source->filters.num; i++)
+			obs_source_destroy(source->filters.array[i]);
+
+		if (source->valid) {
+			pthread_mutex_lock(&obs->source_list_mutex);
+			da_erase_item(obs->sources, &source);
+			pthread_mutex_unlock(&obs->source_list_mutex);
+		}
+
+		for (i = 0; i < source->audio_buffer.num; i++)
+			audiobuf_free(source->audio_buffer.array+i);
+		for (i = 0; i < source->video_frames.num; i++)
+			source_frame_destroy(source->video_frames.array[i]);
+
+		gs_entercontext(obs->graphics);
+		texture_destroy(source->output_texture);
+		gs_leavecontext();
 
 		da_free(source->filters);
-		source->callbacks.destroy(source->data);
+		if (source->data)
+			source->callbacks.destroy(source->data);
+
+		audio_line_destroy(source->audio_line);
+
+		pthread_mutex_destroy(&source->filter_mutex);
+		pthread_mutex_destroy(&source->audio_mutex);
+		pthread_mutex_destroy(&source->video_mutex);
 		dstr_free(&source->settings);
 		bfree(source);
 	}
@@ -173,6 +227,74 @@ void obs_source_video_tick(obs_source_t source, float seconds)
 		source->callbacks.video_tick(source->data, seconds);
 }
 
+#define MAX_VARIANCE 2000000000ULL
+
+static void source_output_audio_line(obs_source_t source,
+		const struct audio_data *data)
+{
+	struct audio_data in = *data;
+
+	if (!in.timestamp) {
+		in.timestamp = os_gettime_ns();
+		if (!source->timing_set) {
+			source->timing_set    = true;
+			source->timing_adjust = 0;
+		}
+	}
+
+	if (!source->timing_set) {
+		source->timing_set    = true;
+		source->timing_adjust = in.timestamp - os_gettime_ns();
+
+		/* detects 'directly' set timestamps as long as they're within
+		 * a certain threashold */
+		if ((source->timing_adjust+MAX_VARIANCE) < MAX_VARIANCE*2)
+			source->timing_adjust = 0;
+	}
+
+	in.timestamp += source->timing_adjust;
+	audio_line_output(source->audio_line, &in);
+}
+
+static void obs_source_flush_audio_buffer(obs_source_t source)
+{
+	size_t i;
+
+	pthread_mutex_lock(&source->audio_mutex);
+	source->timing_set = true;
+
+	for (i = 0; i < source->audio_buffer.num; i++) {
+		struct audiobuf *buf = source->audio_buffer.array+i;
+		struct audio_data data;
+
+		data.data      = buf->data;
+		data.frames    = buf->frames;
+		data.timestamp = buf->timestamp + source->timing_adjust;
+		audio_line_output(source->audio_line, &data);
+		audiobuf_free(buf);
+	}
+
+	da_free(source->audio_buffer);
+	pthread_mutex_unlock(&source->audio_mutex);
+}
+
+static void obs_source_render_async_video(obs_source_t source)
+{
+	struct source_frame *frame = obs_source_getframe(source);
+	if (!frame)
+		return;
+
+	source->timing_adjust = frame->timestamp - os_gettime_ns();
+	if (!source->timing_set && source->audio_buffer.num)
+		obs_source_flush_audio_buffer(source);
+
+	if (!source->output_texture) {
+		/* TODO */
+	}
+
+	obs_source_releaseframe(source, frame);
+}
+
 void obs_source_video_render(obs_source_t source)
 {
 	if (source->callbacks.video_render) {
@@ -183,6 +305,12 @@ void obs_source_video_render(obs_source_t source)
 		} else {
 			source->callbacks.video_render(source->data);
 		}
+
+	} else if (source->filter_target) {
+		obs_source_video_render(source->filter_target);
+
+	} else {
+		obs_source_render_async_video(source);
 	}
 }
 
@@ -231,6 +359,8 @@ obs_source_t obs_filter_gettarget(obs_source_t filter)
 
 void obs_source_filter_add(obs_source_t source, obs_source_t filter)
 {
+	pthread_mutex_lock(&source->filter_mutex);
+
 	if (da_find(source->filters, &filter, 0) != DARRAY_INVALID) {
 		blog(LOG_WARNING, "Tried to add a filter that was already "
 		                  "present on the source");
@@ -243,12 +373,20 @@ void obs_source_filter_add(obs_source_t source, obs_source_t filter)
 	}
 
 	da_push_back(source->filters, &filter);
+
+	pthread_mutex_unlock(&source->filter_mutex);
+
+	filter->filter_parent = source;
 	filter->filter_target = source;
 }
 
 void obs_source_filter_remove(obs_source_t source, obs_source_t filter)
 {
-	size_t idx = da_find(source->filters, &filter, 0);
+	size_t idx;
+
+	pthread_mutex_lock(&source->filter_mutex);
+
+	idx = da_find(source->filters, &filter, 0);
 	if (idx == DARRAY_INVALID)
 		return;
 
@@ -258,6 +396,10 @@ void obs_source_filter_remove(obs_source_t source, obs_source_t filter)
 	}
 
 	da_erase(source->filters, idx);
+
+	pthread_mutex_unlock(&source->filter_mutex);
+
+	filter->filter_parent = NULL;
 	filter->filter_target = NULL;
 }
 
@@ -308,12 +450,112 @@ void obs_source_save_settings(obs_source_t source, const char *settings)
 	dstr_copy(&source->settings, settings);
 }
 
-void obs_source_output_video(obs_source_t source, struct video_frame *frame)
+static inline struct filter_frame *filter_async_video(obs_source_t source,
+		struct filter_frame *in)
 {
-	/* TODO */
+	size_t i;
+	for (i = source->filters.num; i > 0; i--) {
+		struct obs_source *filter = source->filters.array[i-1];
+		if (filter->callbacks.filter_video) {
+			in = filter->callbacks.filter_video(filter->data, in);
+			if (!in)
+				return NULL;
+		}
+	}
+
+	return in;
 }
 
-void obs_source_output_audio(obs_source_t source, struct audio_data *audio)
+static struct filter_frame *process_video(obs_source_t source,
+		const struct source_video *frame)
+{
+	/* TODO: convert to YUV444 or RGB */
+	return NULL;
+}
+
+void obs_source_output_video(obs_source_t source,
+		const struct source_video *frame)
+{
+	struct filter_frame *output;
+
+	output = process_video(source, frame);
+
+	pthread_mutex_lock(&source->filter_mutex);
+	output = filter_async_video(source, output);
+	pthread_mutex_unlock(&source->filter_mutex);
+
+	pthread_mutex_lock(&source->video_mutex);
+	da_push_back(source->video_frames, &output);
+	pthread_mutex_unlock(&source->video_mutex);
+}
+
+static inline const struct audio_data *filter_async_audio(obs_source_t source,
+		const struct audio_data *in)
+{
+	size_t i;
+	for (i = source->filters.num; i > 0; i--) {
+		struct obs_source *filter = source->filters.array[i-1];
+		if (filter->callbacks.filter_audio) {
+			in = filter->callbacks.filter_audio(filter->data, in);
+			if (!in)
+				return NULL;
+		}
+	}
+
+	return in;
+}
+
+static struct audio_data *process_audio(obs_source_t source,
+		const struct source_audio *audio)
+{
+	/* TODO: upmix/downmix/resample */
+	return NULL;
+}
+
+void obs_source_output_audio(obs_source_t source,
+		const struct source_audio *audio)
+{
+	uint32_t flags = obs_source_get_output_flags(source);
+	struct audio_data *data;
+	const struct audio_data *output;
+
+	data = process_audio(source, audio);
+
+	pthread_mutex_lock(&source->filter_mutex);
+	output = filter_async_audio(source, data);
+
+	if (output) {
+		pthread_mutex_lock(&source->audio_mutex);
+
+		if (!source->timing_set && flags & SOURCE_ASYNC_VIDEO) {
+			struct audiobuf newbuf;
+			size_t audio_size = audio_output_blocksize(obs->audio) *
+				output->frames;
+
+			newbuf.data      = bmalloc(audio_size);
+			newbuf.frames    = output->frames;
+			newbuf.timestamp = output->timestamp;
+			memcpy(newbuf.data, output->data, audio_size);
+
+			da_push_back(source->audio_buffer, &newbuf);
+
+		} else {
+			source_output_audio_line(source, output);
+		}
+
+		pthread_mutex_unlock(&source->audio_mutex);
+	}
+
+	pthread_mutex_unlock(&source->filter_mutex);
+}
+
+struct source_frame *obs_source_getframe(obs_source_t source)
+{
+	/* TODO */
+	return NULL;
+}
+
+void obs_source_releaseframe(obs_source_t source, struct source_frame *frame)
 {
 	/* TODO */
 }
diff --git a/libobs/obs-source.h b/libobs/obs-source.h
index a96cf9534..09c1213ec 100644
--- a/libobs/obs-source.h
+++ b/libobs/obs-source.h
@@ -20,6 +20,7 @@
 #include "util/c99defs.h"
 #include "util/darray.h"
 #include "util/dstr.h"
+#include "util/threading.h"
 #include "media-io/media-io.h"
 
 /*
@@ -135,16 +136,20 @@
  *       Return value: true if sources remaining, otherwise false.
  *
  * ---------------------------------------------------------
- *   void [name]_filter_video(void *data, struct video_frame *frame);
+ *   struct filter_frame *[name]_filter_video(void *data,
+ *                                     struct filter_frame *frame);
  *       Filters audio data.  Used with audio filters.
  *
  *       frame: Video frame data.
+ *       returns: New video frame data (or NULL if pending)
  *
  * ---------------------------------------------------------
- *   void [name]_filter_audio(void *data, struct audio_data *audio);
+ *   const struct audio_data *[name]_filter_audio(void *data,
+ *                                     const struct audio_data *audio);
  *       Filters video data.  Used with async video data.
  *
  *       audio: Audio data.
+ *       reutrns New audio data (or NULL if pending)
  */
 
 struct obs_source;
@@ -180,24 +185,53 @@ struct source_info {
 
 	bool (*enum_children)(void *data, size_t idx, obs_source_t *child);
 
-	void (*filter_video)(void *data, struct video_frame *frame);
-	void (*filter_audio)(void *data, struct audio_data *audio);
+	struct filter_frame *(*filter_video)(void *data,
+			struct filter_frame *frame);
+	const struct audio_data *(*filter_audio)(void *data,
+			const struct audio_data *audio);
 };
 
-struct obs_source {
-	void                       *data;
-	struct source_info         callbacks;
-	struct dstr                settings;
-	bool                       rendering_filter;
+struct audiobuf {
+	void     *data;
+	uint32_t frames;
+	uint64_t timestamp;
+};
 
-	struct obs_source          *filter_target;
-	DARRAY(struct obs_source*) filters;
+static inline void audiobuf_free(struct audiobuf *buf)
+{
+	bfree(buf->data);
+}
+
+struct obs_source {
+	void                         *data; /* source-specific data */
+	struct source_info           callbacks;
+	struct dstr                  settings;
+	bool                         valid;
+
+	/* async video and audio */
+	bool                         timing_set;
+	uint64_t                     timing_adjust;
+	texture_t                    output_texture;
+
+	audio_line_t                 audio_line;
+	DARRAY(struct audiobuf)      audio_buffer;
+	DARRAY(struct source_frame*) video_frames;
+	pthread_mutex_t              audio_mutex;
+	pthread_mutex_t              video_mutex;
+
+	/* filters */
+	struct obs_source            *filter_parent;
+	struct obs_source            *filter_target;
+	DARRAY(struct obs_source*)   filters;
+	pthread_mutex_t              filter_mutex;
+	bool                         rendering_filter;
 };
 
 extern bool get_source_info(void *module, const char *module_name,
 		const char *source_name, struct source_info *info);
 
-extern void obs_source_init(struct obs_source *source);
+extern bool obs_source_init(struct obs_source *source, const char *settings,
+		const struct source_info *info);
 
 extern void obs_source_activate(obs_source_t source);
 extern void obs_source_deactivate(obs_source_t source);
diff --git a/libobs/obs.h b/libobs/obs.h
index d4625459f..0a1163121 100644
--- a/libobs/obs.h
+++ b/libobs/obs.h
@@ -39,6 +39,8 @@ enum obs_source_type {
 	SOURCE_SCENE
 };
 
+/* used for changing the order of items (for example, filters in a source,
+ * or items in a scene */
 enum order_movement {
 	ORDER_MOVE_UP,
 	ORDER_MOVE_DOWN,
@@ -46,6 +48,49 @@ enum order_movement {
 	ORDER_MOVE_BOTTOM
 };
 
+struct source_audio {
+	const void         *data;
+	uint32_t           frames;
+
+	/* audio will be automatically resampled/upmixed/downmixed */
+	enum speaker_setup speakers;
+	enum audio_type    type;
+	uint32_t           samples_per_sec;
+
+	/* can be 0 if 'immediate' */
+	uint64_t           timestamp;
+};
+
+struct source_video {
+	const void         *data;
+	uint32_t           width;
+	uint32_t           height;
+	uint32_t           row_bytes;
+	uint64_t           timestamp;
+
+	enum video_type    type;
+	float              yuv_matrix[16];
+};
+
+struct source_frame {
+	void               *data;
+	uint32_t           width;
+	uint32_t           height;
+	uint32_t           row_bytes;
+	uint64_t           timestamp;
+
+	bool               yuv;
+	float              yuv_matrix[16];
+};
+
+static inline void source_frame_destroy(struct source_frame *frame)
+{
+	if (frame) {
+		bfree(frame->data);
+		bfree(frame);
+	}
+}
+
 /* opaque types */
 struct obs_display;
 struct obs_source;
@@ -63,7 +108,7 @@ typedef struct obs_output     *obs_output_t;
 /* OBS context */
 
 /**
- * Creates the OBS context.
+ * Starts up and shuts down OBS
  *
  *   Using the graphics module specified, creates an OBS context and sets the
  * primary video/audio output information.
@@ -214,9 +259,8 @@ EXPORT size_t obs_source_getparam(obs_source_t source, const char *param,
  */
 EXPORT void obs_source_setparam(obs_source_t source, const char *param,
 		const void *data, size_t size);
-/**
- * Enumerates child sources this source has
- */
+
+/** Enumerates child sources this source has */
 EXPORT bool obs_source_enum_children(obs_source_t source, size_t idx,
 		obs_source_t *child);
 
@@ -243,12 +287,19 @@ EXPORT const char *obs_source_get_settings(obs_source_t source);
 EXPORT void obs_source_save_settings(obs_source_t source, const char *settings);
 
 /** Outputs asynchronous video data */
-EXPORT void obs_source_obs_output_video(obs_source_t source,
-		struct video_frame *frame);
+EXPORT void obs_source_obs_async_video(obs_source_t source,
+		const struct video_frame *frame);
 
 /** Outputs audio data (always asynchronous) */
-EXPORT void obs_source_obs_output_audio(obs_source_t source,
-		struct audio_data *audio);
+EXPORT void obs_source_obs_async_audio(obs_source_t source,
+		const struct source_audio *audio);
+
+/** Gets the current async video frame */
+EXPORT struct source_frame *obs_source_getframe(obs_source_t source);
+
+/** Relases the current async frame */
+EXPORT void obs_source_releaseframe(obs_source_t source,
+		struct source_frame *frame);
 
 
 /* ------------------------------------------------------------------------- */
diff --git a/libobs/util/circlebuf.h b/libobs/util/circlebuf.h
new file mode 100644
index 000000000..5ed864102
--- /dev/null
+++ b/libobs/util/circlebuf.h
@@ -0,0 +1,202 @@
+/******************************************************************************
+  Copyright (c) 2013 by Hugh Bailey <obs.jim@gmail.com>
+
+  This software is provided 'as-is', without any express or implied
+  warranty. In no event will the authors be held liable for any damages
+  arising from the use of this software.
+
+  Permission is granted to anyone to use this software for any purpose,
+  including commercial applications, and to alter it and redistribute it
+  freely, subject to the following restrictions:
+
+     1. The origin of this software must not be misrepresented; you must not
+     claim that you wrote the original software. If you use this software
+     in a product, an acknowledgment in the product documentation would be
+     appreciated but is not required.
+
+     2. Altered source versions must be plainly marked as such, and must not be
+     misrepresented as being the original software.
+
+     3. This notice may not be removed or altered from any source
+     distribution.
+******************************************************************************/
+
+#pragma once
+
+#include "c99defs.h"
+#include <string.h>
+#include <stdlib.h>
+#include <assert.h>
+
+#include "bmem.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/* Dynamic circular buffer */
+
+struct circlebuf {
+	void   *data;
+	size_t size;
+
+	size_t start_pos;
+	size_t end_pos;
+	size_t capacity;
+};
+
+static inline void circlebuf_init(struct circlebuf *cb)
+{
+	memset(cb, 0, sizeof(struct circlebuf));
+}
+
+static inline void circlebuf_free(struct circlebuf *cb)
+{
+	bfree(cb->data);
+	memset(cb, 0, sizeof(struct circlebuf));
+}
+
+static inline void circlebuf_reorder_data(struct circlebuf *cb,
+		size_t new_capacity)
+{
+	size_t difference;
+	uint8_t *data;
+
+	if (!cb->size || !cb->start_pos || cb->end_pos > cb->start_pos)
+		return;
+
+	difference = new_capacity - cb->capacity;
+	data = (uint8_t*)cb->data + cb->start_pos;
+	memmove(data+difference, data, cb->capacity - cb->start_pos);
+	cb->start_pos += difference;
+}
+
+static inline void circlebuf_ensure_capacity(struct circlebuf *cb)
+{
+	size_t new_capacity;
+	if (cb->size <= cb->capacity)
+		return;
+
+	new_capacity = cb->capacity*2;
+	if (cb->size > new_capacity)
+		new_capacity = cb->size;
+
+	cb->data = brealloc(cb->data, new_capacity);
+	circlebuf_reorder_data(cb, new_capacity);
+	cb->capacity = new_capacity;
+}
+
+static inline void circlebuf_reserve(struct circlebuf *cb, size_t capacity)
+{
+	if (capacity <= cb->capacity)
+		return;
+
+	cb->data = brealloc(cb->data, capacity);
+	circlebuf_reorder_data(cb, capacity);
+	cb->capacity = capacity;
+}
+
+static inline void circlebuf_upsize(struct circlebuf *cb, size_t size)
+{
+	size_t add_size = size - cb->size;
+	size_t new_end_pos = cb->end_pos + add_size;
+
+	if (size <= cb->size)
+		return;
+
+	cb->size = size;
+	circlebuf_ensure_capacity(cb);
+
+	if (new_end_pos > cb->capacity) {
+		size_t back_size = cb->capacity - cb->end_pos;
+		size_t loop_size = add_size - back_size;
+
+		if (back_size)
+			memset((uint8_t*)cb->data + cb->end_pos, 0, back_size);
+
+		memset(cb->data, 0, loop_size);
+		new_end_pos -= cb->capacity;
+	} else {
+		memset((uint8_t*)cb->data + cb->end_pos, 0, add_size);
+	}
+
+	cb->end_pos = new_end_pos;
+}
+
+/** Overwrites data at a specific point in the buffer (relative).  */
+static inline void circlebuf_place(struct circlebuf *cb, size_t position,
+		const void *data, size_t size)
+{
+	size_t end_point = position + size;
+	size_t data_end_pos;
+
+	if (end_point > cb->size)
+		circlebuf_upsize(cb, end_point);
+
+	position += cb->start_pos;
+	if (position >= cb->capacity)
+		position -= cb->capacity;
+
+	data_end_pos = position + size;
+	if (data_end_pos > cb->capacity) {
+		size_t back_size = cb->capacity - data_end_pos;
+		size_t loop_size = size - back_size;
+
+		if (back_size)
+			memcpy((uint8_t*)cb->data + position, data, back_size);
+		memcpy(cb->data, (uint8_t*)data + back_size, loop_size);
+	} else {
+		memcpy((uint8_t*)cb->data + position, data, size);
+	}
+}
+
+static inline void circlebuf_push_back(struct circlebuf *cb, const void *data,
+		size_t size)
+{
+	size_t new_end_pos = cb->end_pos + size;
+
+	cb->size += size;
+	circlebuf_ensure_capacity(cb);
+
+	if (new_end_pos > cb->capacity) {
+		size_t back_size = cb->capacity - cb->end_pos;
+		size_t loop_size = size - back_size;
+
+		if (back_size)
+			memcpy((uint8_t*)cb->data + cb->end_pos, data,
+					back_size);
+		memcpy(cb->data, (uint8_t*)data + back_size, loop_size);
+
+		new_end_pos -= cb->capacity;
+	} else {
+		memcpy((uint8_t*)cb->data + cb->end_pos, data, size);
+	}
+
+	cb->end_pos = new_end_pos;
+}
+
+static inline void circlebuf_pop_front(struct circlebuf *cb, void *data,
+		size_t size)
+{
+	size_t start_size;
+	assert(size <= cb->size);
+
+	start_size = cb->capacity - cb->start_pos;
+
+	if (start_size < size) {
+		memcpy(data, (uint8_t*)cb->data + cb->start_pos, start_size);
+		memcpy((uint8_t*)data + start_size, cb->data,
+				size - start_size);
+	} else {
+		memcpy(data, (uint8_t*)cb->data + cb->start_pos, size);
+	}
+
+	cb->size -= size;
+	cb->start_pos += size;
+	if (cb->start_pos >= cb->capacity)
+		cb->start_pos -= cb->capacity;
+}
+
+#ifdef __cplusplus
+}
+#endif