Add application media type and APIs

Fixes #16315

9 files changed, 849 insertions, 6 deletions

diff --git a/configure.ac b/configure.ac
index 727c2cf859..8596bb63dc 100644
--- a/configure.ac
+++ b/configure.ac
@@ -916,6 +916,20 @@ fi
 AM_CONDITIONAL(USE_VV, test "x$enable_vv" != "xno")
 
 dnl #######################################################################
+dnl # Check for Raw data streams support in Farstream 
+dnl #######################################################################
+if test "x$enable_vv" != "xno" -a "x$with_gstreamer" == "x1.0"; then
+	AC_MSG_CHECKING(for raw data support in Farstream)
+	PKG_CHECK_MODULES(GSTAPP, [gstreamer-app-1.0], [
+		AC_DEFINE(USE_GSTAPP, 1, [Use GStreamer Video Overlay support])
+		AC_SUBST(GSTAPP_CFLAGS)
+		AC_SUBST(GSTAPP_LIBS)
+		AC_DEFINE(HAVE_MEDIA_APPLICATION, 1, [Define if we have support for application media type.])
+		AC_MSG_RESULT(yes)
+		], [AC_MSG_RESULT(no)])
+fi
+
+dnl #######################################################################
 dnl # Check for Internationalized Domain Name support
 dnl #######################################################################
 
diff --git a/libpurple/Makefile.am b/libpurple/Makefile.am
index 2e99c866dc..01149003d3 100644
--- a/libpurple/Makefile.am
+++ b/libpurple/Makefile.am
@@ -309,6 +309,7 @@ libpurple_la_LIBADD = \
 	$(FARSTREAM_LIBS) \
 	$(GSTREAMER_LIBS) \
 	$(GSTVIDEO_LIBS) \
+	$(GSTAPP_LIBS) \
 	$(GSTINTERFACES_LIBS) \
 	$(IDN_LIBS) \
 	ciphers/libpurple-ciphers.la \
@@ -326,6 +327,7 @@ AM_CPPFLAGS = \
 	$(FARSTREAM_CFLAGS) \
 	$(GSTREAMER_CFLAGS) \
 	$(GSTVIDEO_CFLAGS) \
+	$(GSTAPP_CFLAGS) \
 	$(GSTINTERFACES_CFLAGS) \
 	$(IDN_CFLAGS) \
 	$(NETWORKMANAGER_CFLAGS)
diff --git a/libpurple/media-gst.h b/libpurple/media-gst.h
index 91d9986d8a..31fbf3a871 100644
--- a/libpurple/media-gst.h
+++ b/libpurple/media-gst.h
@@ -71,6 +71,7 @@ typedef enum {
 
 	PURPLE_MEDIA_ELEMENT_SRC = 1 << 9,		/** can be set as an active src */
 	PURPLE_MEDIA_ELEMENT_SINK = 1 << 10,		/** can be set as an active sink */
+	PURPLE_MEDIA_ELEMENT_APPLICATION = 1 << 11,	/** supports application data */
 } PurpleMediaElementType;
 
 #ifdef __cplusplus
diff --git a/libpurple/media/backend-fs2.c b/libpurple/media/backend-fs2.c
index c57078b806..e3949a6268 100644
--- a/libpurple/media/backend-fs2.c
+++ b/libpurple/media/backend-fs2.c
@@ -577,6 +577,10 @@ session_type_to_fs_media_type(PurpleMediaSessionType type)
 		return FS_MEDIA_TYPE_AUDIO;
 	else if (type & PURPLE_MEDIA_VIDEO)
 		return FS_MEDIA_TYPE_VIDEO;
+#ifdef HAVE_MEDIA_APPLICATION
+	else if (type & PURPLE_MEDIA_APPLICATION)
+		return FS_MEDIA_TYPE_APPLICATION;
+#endif
 	else
 		return 0;
 }
@@ -585,7 +589,7 @@ static FsStreamDirection
 session_type_to_fs_stream_direction(PurpleMediaSessionType type)
 {
 	if ((type & PURPLE_MEDIA_AUDIO) == PURPLE_MEDIA_AUDIO ||
-			(type & PURPLE_MEDIA_VIDEO) == PURPLE_MEDIA_VIDEO)
+		(type & PURPLE_MEDIA_VIDEO) == PURPLE_MEDIA_VIDEO)
 		return FS_DIRECTION_BOTH;
 	else if ((type & PURPLE_MEDIA_SEND_AUDIO) ||
 			(type & PURPLE_MEDIA_SEND_VIDEO))
@@ -593,6 +597,14 @@ session_type_to_fs_stream_direction(PurpleMediaSessionType type)
 	else if ((type & PURPLE_MEDIA_RECV_AUDIO) ||
 			(type & PURPLE_MEDIA_RECV_VIDEO))
 		return FS_DIRECTION_RECV;
+#ifdef HAVE_MEDIA_APPLICATION
+	else if ((type & PURPLE_MEDIA_APPLICATION) == PURPLE_MEDIA_APPLICATION)
+		return FS_DIRECTION_BOTH;
+	else if (type & PURPLE_MEDIA_SEND_APPLICATION)
+		return FS_DIRECTION_SEND;
+	else if (type & PURPLE_MEDIA_RECV_APPLICATION)
+		return FS_DIRECTION_RECV;
+#endif
 	else
 		return FS_DIRECTION_NONE;
 }
@@ -611,6 +623,13 @@ session_type_from_fs(FsMediaType type, FsStreamDirection direction)
 			result |= PURPLE_MEDIA_SEND_VIDEO;
 		if (direction & FS_DIRECTION_RECV)
 			result |= PURPLE_MEDIA_RECV_VIDEO;
+#ifdef HAVE_MEDIA_APPLICATION
+	} else if (type == FS_MEDIA_TYPE_APPLICATION) {
+		if (direction & FS_DIRECTION_SEND)
+			result |= PURPLE_MEDIA_SEND_APPLICATION;
+		if (direction & FS_DIRECTION_RECV)
+			result |= PURPLE_MEDIA_RECV_APPLICATION;
+#endif
 	}
 	return result;
 }
@@ -1367,7 +1386,8 @@ gst_handle_message_error(GstBus *bus, GstMessage *msg,
 				& PURPLE_MEDIA_AUDIO)
 			purple_media_error(priv->media,
 					_("Error with your microphone"));
-		else
+		else if (purple_media_get_session_type(priv->media,
+                        sessions->data) & PURPLE_MEDIA_VIDEO)
 			purple_media_error(priv->media,
 					_("Error with your webcam"));
 
@@ -1790,6 +1810,21 @@ create_session(PurpleMediaBackendFs2 *self, const gchar *sess_id,
 	session->session = fs_conference_new_session(priv->conference,
 			session_type_to_fs_media_type(type), &err);
 
+#ifdef HAVE_MEDIA_APPLICATION
+	if (type == PURPLE_MEDIA_APPLICATION) {
+		GstCaps *caps;
+		GObject *rtpsession = NULL;
+
+		caps = gst_caps_new_empty_simple ("application/octet-stream");
+		fs_session_set_allowed_caps (session->session, caps, caps, NULL);
+		gst_caps_unref (caps);
+		g_object_get (session->session, "internal-session", &rtpsession, NULL);
+		if (rtpsession) {
+			g_object_set (rtpsession, "probation", 0, NULL);
+			g_object_unref (rtpsession);
+		}
+	}
+#endif
 	if (err != NULL) {
 		purple_media_error(priv->media,
 				_("Error creating session: %s"),
@@ -2004,6 +2039,21 @@ src_pad_added_cb(FsStream *fsstream, GstPad *srcpad,
 			gst_bin_add(GST_BIN(priv->confbin), sink);
 			gst_element_set_state(sink, GST_STATE_PLAYING);
 			stream->fakesink = sink;
+#ifdef HAVE_MEDIA_APPLICATION
+		} else if (codec->media_type == FS_MEDIA_TYPE_APPLICATION) {
+#if GST_CHECK_VERSION(1,0,0)
+			stream->src = gst_element_factory_make("funnel", NULL);
+#else
+			stream->src = gst_element_factory_make("fsfunnel", NULL);
+#endif
+			sink = purple_media_manager_get_element(
+					purple_media_get_manager(priv->media),
+					PURPLE_MEDIA_RECV_APPLICATION, priv->media,
+					stream->session->id,
+					stream->participant);
+			gst_bin_add(GST_BIN(priv->confbin), sink);
+			gst_element_set_state(sink, GST_STATE_PLAYING);
+#endif
 		}
 		stream->tee = gst_element_factory_make("tee", NULL);
 		gst_bin_add_many(GST_BIN(priv->confbin),
@@ -2366,6 +2416,9 @@ purple_media_backend_fs2_codecs_ready(PurpleMediaBackend *self,
 			return FALSE;
 
 		if (session->type & (PURPLE_MEDIA_SEND_AUDIO |
+#ifdef HAVE_MEDIA_APPLICATION
+				PURPLE_MEDIA_SEND_APPLICATION |
+#endif
 				PURPLE_MEDIA_SEND_VIDEO)) {
 #ifdef HAVE_FARSIGHT
 			g_object_get(session->session,
@@ -2389,6 +2442,9 @@ purple_media_backend_fs2_codecs_ready(PurpleMediaBackend *self,
 			PurpleMediaBackendFs2Session *session = values->data;
 
 			if (session->type & (PURPLE_MEDIA_SEND_AUDIO |
+#ifdef HAVE_MEDIA_APPLICATION
+					PURPLE_MEDIA_SEND_APPLICATION |
+#endif
 					PURPLE_MEDIA_SEND_VIDEO)) {
 #ifdef HAVE_FARSIGHT
 				g_object_get(session->session,
diff --git a/libpurple/media/codec.c b/libpurple/media/codec.c
index 24fd2ad461..7f99848488 100644
--- a/libpurple/media/codec.c
+++ b/libpurple/media/codec.c
@@ -188,7 +188,7 @@ purple_media_codec_class_init(PurpleMediaCodecClass *klass)
 	g_object_class_install_property(gobject_class, PROP_MEDIA_TYPE,
 			g_param_spec_flags("media-type",
 			"Media Type",
-			"Whether this is an audio of video codec.",
+			"Whether this is an audio, video or application codec.",
 			PURPLE_TYPE_MEDIA_SESSION_TYPE,
 			PURPLE_MEDIA_NONE,
 			G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE));
@@ -402,6 +402,8 @@ purple_media_codec_to_string(const PurpleMediaCodec *codec)
 		media_type_str = "audio";
 	else if (priv->media_type & PURPLE_MEDIA_VIDEO)
 		media_type_str = "video";
+	else if (priv->media_type & PURPLE_MEDIA_APPLICATION)
+		media_type_str = "application";
 
 	g_string_printf(string, "%d: %s %s clock:%d channels:%d", priv->id,
 			media_type_str, priv->encoding_name,
diff --git a/libpurple/media/enum-types.c b/libpurple/media/enum-types.c
index 02b9ab74d0..7e8b0a35d7 100644
--- a/libpurple/media/enum-types.c
+++ b/libpurple/media/enum-types.c
@@ -182,10 +182,16 @@ purple_media_session_type_get_type()
 				"PURPLE_MEDIA_RECV_VIDEO", "recv-video" },
 			{ PURPLE_MEDIA_SEND_VIDEO,
 				"PURPLE_MEDIA_SEND_VIDEO", "send-video" },
+			{ PURPLE_MEDIA_RECV_APPLICATION,
+				"PURPLE_MEDIA_RECV_APPLICATION", "recv-application" },
+			{ PURPLE_MEDIA_SEND_APPLICATION,
+				"PURPLE_MEDIA_SEND_APPLICATION", "send-application" },
 			{ PURPLE_MEDIA_AUDIO,
 				"PURPLE_MEDIA_AUDIO", "audio" },
 			{ PURPLE_MEDIA_VIDEO,
 				"PURPLE_MEDIA_VIDEO", "video" },
+			{ PURPLE_MEDIA_APPLICATION,
+				"PURPLE_MEDIA_APPLICATION", "application" },
 			{ 0, NULL, NULL }
 		};
 		type = g_flags_register_static(
diff --git a/libpurple/media/enum-types.h b/libpurple/media/enum-types.h
index a53cf2a9e6..8294604b37 100644
--- a/libpurple/media/enum-types.h
+++ b/libpurple/media/enum-types.h
@@ -94,8 +94,12 @@ typedef enum {
 	PURPLE_MEDIA_SEND_AUDIO = 1 << 1,
 	PURPLE_MEDIA_RECV_VIDEO = 1 << 2,
 	PURPLE_MEDIA_SEND_VIDEO = 1 << 3,
+	PURPLE_MEDIA_RECV_APPLICATION = 1 << 4,
+	PURPLE_MEDIA_SEND_APPLICATION = 1 << 5,
 	PURPLE_MEDIA_AUDIO = PURPLE_MEDIA_RECV_AUDIO | PURPLE_MEDIA_SEND_AUDIO,
-	PURPLE_MEDIA_VIDEO = PURPLE_MEDIA_RECV_VIDEO | PURPLE_MEDIA_SEND_VIDEO
+	PURPLE_MEDIA_VIDEO = PURPLE_MEDIA_RECV_VIDEO | PURPLE_MEDIA_SEND_VIDEO,
+	PURPLE_MEDIA_APPLICATION = PURPLE_MEDIA_RECV_APPLICATION |
+                                   PURPLE_MEDIA_SEND_APPLICATION
 } PurpleMediaSessionType;
 
 /** Media state-changed types */
diff --git a/libpurple/mediamanager.c b/libpurple/mediamanager.c
index 1127ebf3aa..47faed92dd 100644
--- a/libpurple/mediamanager.c
+++ b/libpurple/mediamanager.c
@@ -44,6 +44,9 @@
 #else
 #include <farstream/fs-element-added-notifier.h>
 #endif
+#ifdef HAVE_MEDIA_APPLICATION
+#include <gst/app/app.h>
+#endif
 
 #if GST_CHECK_VERSION(1,0,0)
 #include <gst/video/videooverlay.h>
@@ -97,14 +100,45 @@ struct _PurpleMediaManagerPrivate
 	PurpleMediaElementInfo *video_sink;
 	PurpleMediaElementInfo *audio_src;
 	PurpleMediaElementInfo *audio_sink;
+
+#ifdef HAVE_MEDIA_APPLICATION
+	/* Application data streams */
+	GList *appdata_info; /* holds PurpleMediaAppDataInfo */
+	GMutex appdata_mutex;
+#endif
 };
 
+#ifdef HAVE_MEDIA_APPLICATION
+typedef struct {
+	PurpleMedia *media;
+	GWeakRef media_ref;
+	gchar *session_id;
+	gchar *participant;
+	PurpleMediaAppDataCallbacks callbacks;
+	gpointer user_data;
+	GDestroyNotify notify;
+	GstAppSrc *appsrc;
+	GstAppSink *appsink;
+	gint num_samples;
+	GstSample *current_sample;
+	guint sample_offset;
+	gboolean writable;
+	gboolean connected;
+	guint writable_timer_id;
+	guint readable_timer_id;
+	GCond readable_cond;
+} PurpleMediaAppDataInfo;
+#endif
+
 #define PURPLE_MEDIA_MANAGER_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE((obj), PURPLE_TYPE_MEDIA_MANAGER, PurpleMediaManagerPrivate))
 #define PURPLE_MEDIA_ELEMENT_INFO_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE((obj), PURPLE_TYPE_MEDIA_ELEMENT_INFO, PurpleMediaElementInfoPrivate))
 
 static void purple_media_manager_class_init (PurpleMediaManagerClass *klass);
 static void purple_media_manager_init (PurpleMediaManager *media);
 static void purple_media_manager_finalize (GObject *object);
+#ifdef HAVE_MEDIA_APPLICATION
+static void free_appdata_info_locked (PurpleMediaAppDataInfo *info);
+#endif
 
 static GObjectClass *parent_class = NULL;
 
@@ -190,8 +224,10 @@ purple_media_manager_init (PurpleMediaManager *media)
 	media->priv->medias = NULL;
 	media->priv->private_medias = NULL;
 	media->priv->next_output_window_id = 1;
-#ifdef USE_VV
 	media->priv->backend_type = PURPLE_TYPE_MEDIA_BACKEND_FS2;
+#ifdef HAVE_MEDIA_APPLICATION
+	media->priv->appdata_info = NULL;
+	g_mutex_init (&media->priv->appdata_mutex);
 #endif
 
 	purple_prefs_add_none("/purple/media");
@@ -220,6 +256,13 @@ purple_media_manager_finalize (GObject *media)
 	}
 	if (priv->video_caps)
 		gst_caps_unref(priv->video_caps);
+#ifdef HAVE_MEDIA_APPLICATION
+	if (priv->appdata_info)
+		g_list_free_full (priv->appdata_info,
+			(GDestroyNotify) free_appdata_info_locked);
+	g_mutex_clear (&priv->appdata_mutex);
+#endif
+
 	parent_class->finalize(media);
 }
 #endif
@@ -440,8 +483,23 @@ purple_media_manager_remove_media(PurpleMediaManager *manager, PurpleMedia *medi
 		medias = &manager->priv->private_medias;
 	}
 
-	if (list)
+	if (list) {
 		*medias = g_list_delete_link(*medias, list);
+
+#ifdef HAVE_MEDIA_APPLICATION
+		g_mutex_lock (&manager->priv->appdata_mutex);
+		for (list = manager->priv->appdata_info; list; list = list->next) {
+			PurpleMediaAppDataInfo *info = list->data;
+
+			if (info->media == media) {
+				manager->priv->appdata_info = g_list_delete_link (
+					manager->priv->appdata_info, list);
+				free_appdata_info_locked (info);
+			}
+		}
+		g_mutex_unlock (&manager->priv->appdata_mutex);
+#endif
+	}
 #endif
 }
 
@@ -493,6 +551,92 @@ purple_media_manager_get_private_media_by_account(PurpleMediaManager *manager,
 	return get_media_by_account (manager, account, TRUE);
 }
 
+#ifdef HAVE_MEDIA_APPLICATION
+static void
+free_appdata_info_locked (PurpleMediaAppDataInfo *info)
+{
+	if (info->notify)
+		info->notify (info->user_data);
+
+	/* Make sure no other thread is using the structure */
+	g_free (info->session_id);
+	g_free (info->participant);
+
+	if (info->readable_timer_id) {
+		purple_timeout_remove (info->readable_timer_id);
+		info->readable_timer_id = 0;
+	}
+
+	if (info->writable_timer_id) {
+		purple_timeout_remove (info->writable_timer_id);
+		info->writable_timer_id = 0;
+	}
+
+	if (info->current_sample)
+		gst_sample_unref (info->current_sample);
+	info->current_sample = NULL;
+
+	/* Unblock any reading thread before destroying the GCond */
+	g_cond_broadcast (&info->readable_cond);
+
+	g_cond_clear (&info->readable_cond);
+
+	g_slice_free (PurpleMediaAppDataInfo, info);
+}
+
+/*
+ * Get an app data info struct associated with a session and lock the mutex
+ * We don't want to return an info struct and unlock then it gets destroyed
+ * so we need to return it with the lock still taken
+ */
+static PurpleMediaAppDataInfo *
+get_app_data_info_and_lock (PurpleMediaManager *manager,
+	PurpleMedia *media, const gchar *session_id, const gchar *participant)
+{
+	GList *i;
+
+	g_mutex_lock (&manager->priv->appdata_mutex);
+	for (i = manager->priv->appdata_info; i; i = i->next) {
+		PurpleMediaAppDataInfo *info = i->data;
+
+		if (info->media == media &&
+			g_strcmp0 (info->session_id, session_id) == 0 &&
+			(participant == NULL ||
+				g_strcmp0 (info->participant, participant) == 0)) {
+			return info;
+		}
+	}
+
+	return NULL;
+}
+
+/*
+ * Get an app data info struct associated with a session and lock the mutex
+ * if it doesn't exist, we create it.
+ */
+static PurpleMediaAppDataInfo *
+ensure_app_data_info_and_lock (PurpleMediaManager *manager, PurpleMedia *media,
+	const gchar *session_id, const gchar *participant)
+{
+	PurpleMediaAppDataInfo * info = get_app_data_info_and_lock (manager, media,
+		session_id, participant);
+
+	if (info == NULL) {
+		info = g_slice_new0 (PurpleMediaAppDataInfo);
+		info->media = media;
+		g_weak_ref_init (&info->media_ref, media);
+		info->session_id = g_strdup (session_id);
+		info->participant = g_strdup (participant);
+		g_cond_init (&info->readable_cond);
+		manager->priv->appdata_info = g_list_prepend (
+			manager->priv->appdata_info, info);
+	}
+
+	return info;
+}
+#endif
+
+
 #ifdef USE_VV
 static void
 request_pad_unlinked_cb(GstPad *pad, GstPad *peer, gpointer user_data)
@@ -587,6 +731,351 @@ purple_media_manager_get_video_caps(PurpleMediaManager *manager)
 #endif
 }
 
+#ifdef HAVE_MEDIA_APPLICATION
+/*
+ * Calls the appdata writable callback from the main thread.
+ * This needs to grab the appdata lock and make sure it didn't get destroyed
+ * before calling the callback.
+ */
+static gboolean
+appsrc_writable (gpointer user_data)
+{
+	PurpleMediaManager *manager = purple_media_manager_get ();
+	PurpleMediaAppDataInfo *info = user_data;
+	void (*writable_cb) (PurpleMediaManager *manager, PurpleMedia *media,
+		const gchar *session_id, const gchar *participant, gboolean writable,
+		gpointer user_data);
+	PurpleMedia *media;
+	gchar *session_id;
+	gchar *participant;
+	gboolean writable;
+	gpointer cb_data;
+	guint *timer_id_ptr = &info->writable_timer_id;
+	guint timer_id = *timer_id_ptr;
+
+
+	g_mutex_lock (&manager->priv->appdata_mutex);
+	if (timer_id == 0 || timer_id != *timer_id_ptr) {
+		/* In case info was freed while we were waiting for the mutex to unlock
+		 * we still have a pointer to the timer_id which should still be
+		 * accessible since it's in the Glib slice allocator. It gets set to 0
+		 * just after the timeout is canceled which happens also before the
+		 * AppDataInfo is freed, so even if that memory slice gets reused, the
+		 * timer_id would be different from its previous value (unless
+		 * extremely unlucky). So checking if the value for the timer_id changed
+		 * should be enough to prevent any kind of race condition in which the
+		 * media/AppDataInfo gets destroyed in one thread while the timeout was
+		 * triggered and is waiting on the mutex to get unlocked in this thread
+		 */
+		g_mutex_unlock (&manager->priv->appdata_mutex);
+		return FALSE;
+	}
+	writable_cb = info->callbacks.writable;
+	media = g_weak_ref_get (&info->media_ref);
+	session_id = g_strdup (info->session_id);
+	participant = g_strdup (info->participant);
+	writable = info->writable && info->connected;
+	cb_data = info->user_data;
+
+    info->writable_timer_id = 0;
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+
+
+	if (writable_cb && media)
+		writable_cb (manager, media, session_id, participant, writable,
+			cb_data);
+
+	g_object_unref (media);
+	g_free (session_id);
+	g_free (participant);
+
+	return FALSE;
+}
+
+/*
+ * Schedule a writable callback to be called from the main thread.
+ * We need to do this because need-data/enough-data signals from appsrc
+ * will come from the streaming thread and we need to create
+ * a source that we attach to the main context but we can't use
+ * g_main_context_invoke since we need to be able to cancel the source if the
+ * media gets destroyed.
+ * We use a timeout source instead of idle source, so the callback gets a higher
+ * priority
+ */
+static void
+call_appsrc_writable_locked (PurpleMediaAppDataInfo *info)
+{
+	/* We already have a writable callback scheduled, don't create another one */
+	if (info->writable_timer_id || info->callbacks.writable == NULL)
+		return;
+
+	info->writable_timer_id = purple_timeout_add (0, appsrc_writable, info);
+}
+
+static void
+appsrc_need_data (GstAppSrc *appsrc, guint length, gpointer user_data)
+{
+	PurpleMediaAppDataInfo *info = user_data;
+	PurpleMediaManager *manager = purple_media_manager_get ();
+
+	g_mutex_lock (&manager->priv->appdata_mutex);
+	if (!info->writable) {
+		info->writable = TRUE;
+		/* Only signal writable if we also established a connection */
+		if (info->connected)
+			call_appsrc_writable_locked (info);
+	}
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+}
+
+static void
+appsrc_enough_data (GstAppSrc *appsrc, gpointer user_data)
+{
+	PurpleMediaAppDataInfo *info = user_data;
+	PurpleMediaManager *manager = purple_media_manager_get ();
+
+	g_mutex_lock (&manager->priv->appdata_mutex);
+	if (info->writable) {
+		info->writable = FALSE;
+		call_appsrc_writable_locked (info);
+	}
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+}
+
+static gboolean
+appsrc_seek_data (GstAppSrc *appsrc, guint64 offset, gpointer user_data)
+{
+	return FALSE;
+}
+
+static void
+appsrc_destroyed (PurpleMediaAppDataInfo *info)
+{
+	PurpleMediaManager *manager = purple_media_manager_get ();
+
+	g_mutex_lock (&manager->priv->appdata_mutex);
+	info->appsrc = NULL;
+	if (info->writable) {
+		info->writable = FALSE;
+		call_appsrc_writable_locked (info);
+	}
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+}
+
+static void
+media_established_cb (PurpleMedia *media,const gchar *session_id,
+	const gchar *participant, PurpleMediaCandidate *local_candidate,
+	PurpleMediaCandidate *remote_candidate, PurpleMediaAppDataInfo *info)
+{
+	PurpleMediaManager *manager = purple_media_manager_get ();
+
+	g_mutex_lock (&manager->priv->appdata_mutex);
+	info->connected = TRUE;
+	/* We established the connection, if we were writable, then we need to
+	 * signal it now */
+	if (info->writable)
+		call_appsrc_writable_locked (info);
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+}
+
+static GstElement *
+create_send_appsrc(PurpleMedia *media,
+		const gchar *session_id, const gchar *participant)
+{
+	PurpleMediaManager *manager = purple_media_manager_get ();
+	PurpleMediaAppDataInfo * info = ensure_app_data_info_and_lock (manager,
+		media, session_id, participant);
+	GstElement *appsrc = (GstElement *)info->appsrc;
+
+	if (appsrc == NULL) {
+		GstAppSrcCallbacks callbacks = {appsrc_need_data, appsrc_enough_data,
+										appsrc_seek_data, {NULL}};
+		GstCaps *caps = gst_caps_new_empty_simple ("application/octet-stream");
+
+		appsrc = gst_element_factory_make("appsrc", NULL);
+
+		info->appsrc = (GstAppSrc *)appsrc;
+
+		gst_app_src_set_caps (info->appsrc, caps);
+		gst_app_src_set_callbacks (info->appsrc,
+			&callbacks, info, (GDestroyNotify) appsrc_destroyed);
+		g_signal_connect (media, "candidate-pair-established",
+			(GCallback) media_established_cb, info);
+		gst_caps_unref (caps);
+	}
+
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+	return appsrc;
+}
+
+static void
+appsink_eos (GstAppSink *appsink, gpointer user_data)
+{
+}
+
+static GstFlowReturn
+appsink_new_preroll (GstAppSink *appsink, gpointer user_data)
+{
+	return GST_FLOW_OK;
+}
+
+static gboolean
+appsink_readable (gpointer user_data)
+{
+	PurpleMediaManager *manager = purple_media_manager_get ();
+	PurpleMediaAppDataInfo *info = user_data;
+	void (*readable_cb) (PurpleMediaManager *manager, PurpleMedia *media,
+		const gchar *session_id, const gchar *participant, gpointer user_data);
+	PurpleMedia *media;
+	gchar *session_id;
+	gchar *participant;
+	gpointer cb_data;
+	guint *timer_id_ptr = &info->readable_timer_id;
+	guint timer_id = *timer_id_ptr;
+
+	g_mutex_lock (&manager->priv->appdata_mutex);
+	if (timer_id == 0 || timer_id != *timer_id_ptr) {
+		/* Avoided a race condition (see writable callback) */
+		g_mutex_unlock (&manager->priv->appdata_mutex);
+		return FALSE;
+	}
+	/* We need to signal readable until there are no more samples */
+	while (info->callbacks.readable &&
+		(info->num_samples > 0 || info->current_sample != NULL)) {
+		readable_cb = info->callbacks.readable;
+		media = g_weak_ref_get (&info->media_ref);
+		session_id = g_strdup (info->session_id);
+		participant = g_strdup (info->participant);
+		cb_data = info->user_data;
+		g_mutex_unlock (&manager->priv->appdata_mutex);
+
+		if (readable_cb)
+			readable_cb (manager, media, session_id, participant, cb_data);
+
+		g_mutex_lock (&manager->priv->appdata_mutex);
+		g_object_unref (media);
+		g_free (session_id);
+		g_free (participant);
+		if (timer_id == 0 || timer_id != *timer_id_ptr) {
+			/* We got cancelled */
+			g_mutex_unlock (&manager->priv->appdata_mutex);
+			return FALSE;
+		}
+	}
+    info->readable_timer_id = 0;
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+	return FALSE;
+}
+
+static void
+call_appsink_readable_locked (PurpleMediaAppDataInfo *info)
+{
+	/* We must signal that a new sample has arrived to release blocking reads */
+	g_cond_broadcast (&info->readable_cond);
+
+	/* We already have a writable callback scheduled, don't create another one */
+	if (info->readable_timer_id || info->callbacks.readable == NULL)
+		return;
+
+	info->readable_timer_id = purple_timeout_add (0, appsink_readable, info);
+}
+
+static GstFlowReturn
+appsink_new_sample (GstAppSink *appsink, gpointer user_data)
+{
+	PurpleMediaManager *manager = purple_media_manager_get ();
+	PurpleMediaAppDataInfo *info = user_data;
+
+	g_mutex_lock (&manager->priv->appdata_mutex);
+	info->num_samples++;
+	call_appsink_readable_locked (info);
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+
+	return GST_FLOW_OK;
+}
+
+static void
+appsink_destroyed (PurpleMediaAppDataInfo *info)
+{
+	PurpleMediaManager *manager = purple_media_manager_get ();
+
+	g_mutex_lock (&manager->priv->appdata_mutex);
+	info->appsink = NULL;
+	info->num_samples = 0;
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+}
+
+static GstElement *
+create_recv_appsink(PurpleMedia *media,
+		const gchar *session_id, const gchar *participant)
+{
+	PurpleMediaManager *manager = purple_media_manager_get ();
+	PurpleMediaAppDataInfo * info = ensure_app_data_info_and_lock (manager,
+		media, session_id, participant);
+	GstElement *appsink = (GstElement *)info->appsink;
+
+	if (appsink == NULL) {
+		GstAppSinkCallbacks callbacks = {appsink_eos, appsink_new_preroll,
+										 appsink_new_sample, {NULL}};
+		GstCaps *caps = gst_caps_new_empty_simple ("application/octet-stream");
+
+		appsink = gst_element_factory_make("appsink", NULL);
+
+		info->appsink = (GstAppSink *)appsink;
+
+		gst_app_sink_set_caps (info->appsink, caps);
+		gst_app_sink_set_callbacks (info->appsink,
+			&callbacks, info, (GDestroyNotify) appsink_destroyed);
+		gst_caps_unref (caps);
+
+	}
+
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+	return appsink;
+}
+#endif
+
+static PurpleMediaElementInfo *
+get_send_application_element_info ()
+{
+	static PurpleMediaElementInfo *info = NULL;
+
+#ifdef HAVE_MEDIA_APPLICATION
+	if (info == NULL) {
+		info = g_object_new(PURPLE_TYPE_MEDIA_ELEMENT_INFO,
+			"id", "pidginappsrc",
+			"name", "Pidgin Application Source",
+			"type", PURPLE_MEDIA_ELEMENT_APPLICATION
+					| PURPLE_MEDIA_ELEMENT_SRC
+					| PURPLE_MEDIA_ELEMENT_ONE_SRC,
+			"create-cb", create_send_appsrc, NULL);
+	}
+#endif
+
+	return info;
+}
+
+
+static PurpleMediaElementInfo *
+get_recv_application_element_info ()
+{
+	static PurpleMediaElementInfo *info = NULL;
+
+#ifdef HAVE_MEDIA_APPLICATION
+	if (info == NULL) {
+		info = g_object_new(PURPLE_TYPE_MEDIA_ELEMENT_INFO,
+			"id", "pidginappsink",
+			"name", "Pidgin Application Sink",
+			"type", PURPLE_MEDIA_ELEMENT_APPLICATION
+					| PURPLE_MEDIA_ELEMENT_SINK
+					| PURPLE_MEDIA_ELEMENT_ONE_SINK,
+			"create-cb", create_recv_appsink, NULL);
+	}
+#endif
+
+	return info;
+}
+
 GstElement *
 purple_media_manager_get_element(PurpleMediaManager *manager,
 		PurpleMediaSessionType type, PurpleMedia *media,
@@ -605,6 +1094,10 @@ purple_media_manager_get_element(PurpleMediaManager *manager,
 		info = manager->priv->video_src;
 	else if (type & PURPLE_MEDIA_RECV_VIDEO)
 		info = manager->priv->video_sink;
+	else if (type & PURPLE_MEDIA_SEND_APPLICATION)
+		info = get_send_application_element_info ();
+	else if (type & PURPLE_MEDIA_RECV_APPLICATION)
+		info = get_recv_application_element_info ();
 
 	if (info == NULL)
 		return NULL;
@@ -848,11 +1341,16 @@ purple_media_manager_get_active_element(PurpleMediaManager *manager,
 			return manager->priv->audio_src;
 		else if (type & PURPLE_MEDIA_ELEMENT_VIDEO)
 			return manager->priv->video_src;
+		else if (type & PURPLE_MEDIA_ELEMENT_APPLICATION)
+			return get_send_application_element_info ();
 	} else if (type & PURPLE_MEDIA_ELEMENT_SINK) {
 		if (type & PURPLE_MEDIA_ELEMENT_AUDIO)
 			return manager->priv->audio_sink;
 		else if (type & PURPLE_MEDIA_ELEMENT_VIDEO)
 			return manager->priv->video_sink;
+		else if (type & PURPLE_MEDIA_ELEMENT_APPLICATION)
+			return get_recv_application_element_info ();
+
 	}
 #endif
 
@@ -1152,6 +1650,174 @@ purple_media_manager_get_backend_type(PurpleMediaManager *manager)
 #endif
 }
 
+void
+purple_media_manager_set_application_data_callbacks(PurpleMediaManager *manager,
+		PurpleMedia *media, const gchar *session_id,
+		const gchar *participant, PurpleMediaAppDataCallbacks *callbacks,
+		gpointer user_data, GDestroyNotify notify)
+{
+#ifdef HAVE_MEDIA_APPLICATION
+	PurpleMediaAppDataInfo * info = ensure_app_data_info_and_lock (manager,
+		media, session_id, participant);
+
+	if (info->notify)
+		info->notify (info->user_data);
+
+	if (info->readable_timer_id) {
+		purple_timeout_remove (info->readable_timer_id);
+		info->readable_timer_id = 0;
+	}
+
+	if (info->writable_timer_id) {
+		purple_timeout_remove (info->writable_timer_id);
+		info->writable_timer_id = 0;
+	}
+
+	if (callbacks) {
+		info->callbacks = *callbacks;
+	} else {
+		info->callbacks.writable = NULL;
+		info->callbacks.readable = NULL;
+	}
+	info->user_data = user_data;
+	info->notify = notify;
+
+	call_appsrc_writable_locked (info);
+	if (info->num_samples > 0 || info->current_sample != NULL)
+		call_appsink_readable_locked (info);
+
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+#endif
+}
+
+gint
+purple_media_manager_send_application_data (
+	PurpleMediaManager *manager, PurpleMedia *media, const gchar *session_id,
+	const gchar *participant, gpointer buffer, guint size, gboolean blocking)
+{
+#ifdef HAVE_MEDIA_APPLICATION
+	PurpleMediaAppDataInfo * info = get_app_data_info_and_lock (manager,
+		media, session_id, participant);
+
+	if (info && info->appsrc && info->connected) {
+		GstBuffer *gstbuffer = gst_buffer_new_wrapped (g_memdup (buffer, size),
+			size);
+		GstAppSrc *appsrc = gst_object_ref (info->appsrc);
+
+		g_mutex_unlock (&manager->priv->appdata_mutex);
+		if (gst_app_src_push_buffer (appsrc, gstbuffer) == GST_FLOW_OK) {
+			if (blocking) {
+				GstPad *srcpad;
+
+				srcpad = gst_element_get_static_pad (GST_ELEMENT (appsrc),
+					"src");
+				if (srcpad) {
+					gst_pad_peer_query (srcpad, gst_query_new_drain ());
+					gst_object_unref (srcpad);
+				}
+			}
+			gst_object_unref (appsrc);
+			return size;
+		} else {
+			gst_object_unref (appsrc);
+			return -1;
+		}
+	}
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+	return -1;
+#else
+	return -1;
+#endif
+}
+
+gint
+purple_media_manager_receive_application_data (
+	PurpleMediaManager *manager, PurpleMedia *media, const gchar *session_id,
+	const gchar *participant, gpointer buffer, guint max_size,
+	gboolean blocking)
+{
+#ifdef HAVE_MEDIA_APPLICATION
+	PurpleMediaAppDataInfo * info = get_app_data_info_and_lock (manager,
+		media, session_id, participant);
+	guint bytes_read = 0;
+
+	if (info) {
+		/* If we are in a blocking read, we need to loop until max_size data
+		 * is read into the buffer, if we're not, then we need to read as much
+		 * data as possible
+		 */
+		do {
+			if (!info->current_sample && info->appsink && info->num_samples > 0) {
+				info->current_sample = gst_app_sink_pull_sample (info->appsink);
+				info->sample_offset = 0;
+				if (info->current_sample)
+					info->num_samples--;
+			}
+
+			if (info->current_sample) {
+				GstBuffer *gstbuffer = gst_sample_get_buffer (
+					info->current_sample);
+
+				if (gstbuffer) {
+					GstMapInfo mapinfo;
+					guint bytes_to_copy;
+
+					gst_buffer_map (gstbuffer, &mapinfo, GST_MAP_READ);
+					/* We must copy only the data remaining in the buffer without
+					 * overflowing the buffer */
+					bytes_to_copy = max_size - bytes_read;
+					if (bytes_to_copy > mapinfo.size - info->sample_offset)
+						bytes_to_copy = mapinfo.size - info->sample_offset;
+					memcpy ((guint8 *)buffer + bytes_read,
+						mapinfo.data + info->sample_offset,	bytes_to_copy);
+
+					gst_buffer_unmap (gstbuffer, &mapinfo);
+					info->sample_offset += bytes_to_copy;
+					bytes_read += bytes_to_copy;
+					if (info->sample_offset == mapinfo.size) {
+						gst_sample_unref (info->current_sample);
+						info->current_sample = NULL;
+						info->sample_offset = 0;
+					}
+				} else {
+					/* In case there's no buffer in the sample (should never
+					 * happen), we need to at least unref it */
+					gst_sample_unref (info->current_sample);
+					info->current_sample = NULL;
+					info->sample_offset = 0;
+				}
+			}
+
+			/* If blocking, wait until there's an available sample */
+			while (bytes_read < max_size && blocking &&
+				info->current_sample == NULL && info->num_samples == 0) {
+				g_cond_wait (&info->readable_cond, &manager->priv->appdata_mutex);
+
+				/* We've been signaled, we need to unlock and regrab the info
+				 * struct to make sure nothing changed */
+				g_mutex_unlock (&manager->priv->appdata_mutex);
+				info = get_app_data_info_and_lock (manager,
+					media, session_id, participant);
+				if (info == NULL || info->appsink == NULL) {
+					/* The session was destroyed while we were waiting, we
+					 * should return here */
+					g_mutex_unlock (&manager->priv->appdata_mutex);
+					return bytes_read;
+				}
+			}
+		} while (bytes_read < max_size &&
+			(blocking || info->num_samples > 0));
+
+		g_mutex_unlock (&manager->priv->appdata_mutex);
+		return bytes_read;
+	}
+	g_mutex_unlock (&manager->priv->appdata_mutex);
+	return -1;
+#else
+	return -1;
+#endif
+}
+
 #ifdef USE_GSTREAMER
 
 /*
@@ -1199,6 +1865,8 @@ purple_media_element_type_get_type()
 				"PURPLE_MEDIA_ELEMENT_SRC", "src" },
 			{ PURPLE_MEDIA_ELEMENT_SINK,
 				"PURPLE_MEDIA_ELEMENT_SINK", "sink" },
+			{ PURPLE_MEDIA_ELEMENT_APPLICATION,
+				"PURPLE_MEDIA_ELEMENT_APPLICATION", "application" },
 			{ 0, NULL, NULL }
 		};
 		type = g_flags_register_static(
@@ -1424,5 +2092,6 @@ purple_media_element_info_call_create(PurpleMediaElementInfo *info,
 	return NULL;
 }
 
+
 #endif /* USE_GSTREAMER */
 
diff --git a/libpurple/mediamanager.h b/libpurple/mediamanager.h
index 3a82cd83f5..ff457f9e06 100644
--- a/libpurple/mediamanager.h
+++ b/libpurple/mediamanager.h
@@ -38,6 +38,30 @@ typedef struct _PurpleMediaManagerClass PurpleMediaManagerClass;
 #include "account.h"
 #include "media.h"
 
+/**
+ * PurpleMediaAppDataCallbacks:
+ * @readable: Called when the stream has received data and is readable.
+ * @writable: Called when the stream has become writable or has stopped being
+ * writable.
+ *
+ * A set of callbacks that can be installed on an Application data session with
+ * purple_media_manager_set_application_data_callbacks()
+ *
+ * Once installed the @readable callback will get called as long as data is
+ * available to read, so the data must be read completely.
+ * The @writable callback will only be called when the writable state of the
+ * stream changes. The @writable argument defines whether the stream has
+ * become writable or stopped being writable.
+ *
+ */
+typedef struct {
+	void      (*readable)    (PurpleMediaManager *manager, PurpleMedia *media,
+		const gchar *session_id, const gchar *participant, gpointer user_data);
+	void      (*writable)    (PurpleMediaManager *manager, PurpleMedia *media,
+		const gchar *session_id, const gchar *participant, gboolean writable,
+		gpointer user_data);
+} PurpleMediaAppDataCallbacks;
+
 G_BEGIN_DECLS
 
 #define PURPLE_TYPE_MEDIA_MANAGER            (purple_media_manager_get_type())
@@ -285,6 +309,71 @@ void purple_media_manager_set_backend_type(PurpleMediaManager *manager,
  */
 GType purple_media_manager_get_backend_type(PurpleMediaManager *manager);
 
+/**
+ * purple_media_manager_set_application_data_callbacks:
+ * @manager: The manager to register the callbacks with.
+ * @media: The media instance to register the callbacks with.
+ * @session_id: The session to register the callbacks with.
+ * @participant: The participant to register the callbacks with.
+ * @callbacks: The callbacks to be set on the session.
+ * @user_data: a user_data argument for the callbacks.
+ * @notify: a destroy notify function.
+ *
+ * Set callbacks on a session to be called when the stream becomes writable
+ * or readable for media sessions of type #PURPLE_MEDIA_APPLICATION
+ */
+void purple_media_manager_set_application_data_callbacks(
+	PurpleMediaManager *manager, PurpleMedia *media, const gchar *session_id,
+	const gchar *participant, PurpleMediaAppDataCallbacks *callbacks,
+	gpointer user_data, GDestroyNotify notify);
+
+/**
+ * purple_media_manager_send_application_data:
+ * @manager: The manager to send data with.
+ * @media: The media instance to which the session belongs.
+ * @session_id: The session to send data to.
+ * @participant: The participant to send data to.
+ * @buffer: The buffer of data to send.
+ * @size: The size of @buffer
+ * @blocking: Whether to block until the data was send or not.
+ *
+ * Sends a buffer of data to a #PURPLE_MEDIA_APPLICATION session.
+ * If @blocking is set, unless an error occured, the function will not return
+ * until the data has been flushed into the network.
+ * If the stream is not writable, the data will be queued. It is the
+ * responsability of the user to stop sending data when the stream isn't
+ * writable anymore. It is also the responsability of the user to only start
+ * sending data after the stream has been configured correctly (encryption
+ * parameters for example).
+ *
+ * Returns: Number of bytes sent or -1 in case of error.
+ */
+gint purple_media_manager_send_application_data (
+	PurpleMediaManager *manager, PurpleMedia *media, const gchar *session_id,
+	const gchar *participant, gpointer buffer, guint size, gboolean blocking);
+
+/**
+ * purple_media_manager_receive_application_data:
+ * @manager: The manager to receive data with.
+ * @media: The media instance to which the session belongs.
+ * @session_id: The session to receive data from.
+ * @participant: The participant to receive data from.
+ * @buffer: The buffer to receive data into.
+ * @max_size: The max_size of @buffer
+ * @blocking: Whether to block until the buffer is entirely filled or return
+ * with currently available data.
+ *
+ * Receive a buffer of data from a #PURPLE_MEDIA_APPLICATION session.
+ * If @blocking is set, unless an error occured, the function will not return
+ * until @max_size bytes are read.
+ *
+ * Returns: Number of bytes received or -1 in case of error.
+ */
+gint purple_media_manager_receive_application_data (
+	PurpleMediaManager *manager, PurpleMedia *media, const gchar *session_id,
+	const gchar *participant, gpointer buffer, guint max_size,
+	gboolean blocking);
+
 /*}@*/
 
 #ifdef __cplusplus