<commit_before>#include <kms-core.h>
#include <rtmp/kms-rtmp-receiver.h>
#include <rtmp/kms-rtmp-session.h>
#include "internal/kms-utils.h"

#define KMS_RTMP_RECEIVER_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE ((obj), KMS_TYPE_RTMP_RECEIVER, KmsRtmpReceiverPriv))

#define LOCK(obj) (g_mutex_lock(KMS_RTMP_RECEIVER(obj)->priv->mutex))
#define UNLOCK(obj) (g_mutex_unlock(KMS_RTMP_RECEIVER(obj)->priv->mutex))

#define MEDIA_TYPE_DATA "type"

struct _KmsRtmpReceiverPriv {
	GMutex *mutex;

	KmsRtmpSession *neg_spec;
	gboolean offerer;
	gboolean finished;
};

enum {
	PROP_0,

	PROP_NEG_SPEC,
	PROP_OFFERER,
};

G_DEFINE_TYPE(KmsRtmpReceiver, kms_rtmp_receiver, KMS_TYPE_MEDIA_HANDLER_SRC)

static void
dispose_neg_spec(KmsRtmpReceiver *self) {
	if (self->priv->neg_spec != NULL) {
		g_object_unref(self->priv->neg_spec);
		self->priv->neg_spec = NULL;
	}
}

static void
set_property (GObject *object, guint property_id, const GValue *value,
							GParamSpec *pspec) {
	KmsRtmpReceiver *self = KMS_RTMP_RECEIVER(object);

	switch (property_id) {
		case PROP_NEG_SPEC:
			LOCK(self);
			dispose_neg_spec(self);
			self->priv->neg_spec = g_value_dup_object(value);
			UNLOCK(self);
			break;
		case PROP_OFFERER:
			LOCK(self);
			self->priv->offerer = g_value_get_boolean(value);
			UNLOCK(self);
			break;
		default:
			/* We don't have any other property... */
			G_OBJECT_WARN_INVALID_PROPERTY_ID(object, property_id, pspec);
			break;
	}
}

static void
get_property(GObject *object, guint property_id, GValue *value,
							GParamSpec *pspec) {
	KmsRtmpReceiver *self = KMS_RTMP_RECEIVER(object);

	switch (property_id) {
		case PROP_NEG_SPEC:
			LOCK(self);
			g_value_set_object(value, self->priv->neg_spec);
			UNLOCK(self);
			break;
		case PROP_OFFERER:
			LOCK(self);
			g_value_set_boolean(value, self->priv->offerer);
			UNLOCK(self);
			break;
		default:
			/* We don't have any other property... */
			G_OBJECT_WARN_INVALID_PROPERTY_ID(object, property_id, pspec);
			break;
	}
}

static GstElement *generate_rtmpsrc(KmsRtmpReceiver *self);

static gboolean
remove_rtmp(KmsRtmpReceiver *self) {
	GstElement *rtmpsrc, *flvdemux;

	g_print("Remove rtmp for element: %s\n", GST_OBJECT_NAME(self));
	LOCK(self);
	if (self->priv->finished)
		goto end;

	rtmpsrc = gst_bin_get_by_name(GST_BIN(self), "rtmp");
	if (rtmpsrc == NULL)
		goto end;

	gst_bin_remove(GST_BIN(self), rtmpsrc);
	gst_element_set_state(rtmpsrc, GST_STATE_NULL);
	g_object_unref(rtmpsrc);

	rtmpsrc = generate_rtmpsrc(self);
	gst_bin_add(GST_BIN(self), rtmpsrc);
	gst_element_set_state(rtmpsrc, GST_STATE_PLAYING);

	flvdemux = gst_bin_get_by_name(GST_BIN(self), "demux");

	if (flvdemux == NULL)
		goto end;

	gst_element_link(rtmpsrc, flvdemux);

end:
	UNLOCK(self);
	return FALSE;
}

static gboolean
event_handler(GstPad *pad, GstEvent *event, KmsRtmpReceiver *self) {
	if (!GST_IS_EVENT(event))
		return TRUE;

	if (GST_EVENT_TYPE(event) == GST_EVENT_EOS) {
		g_timeout_add_full(G_PRIORITY_DEFAULT, 500,<commit_msg>Retry rtmp connection each 3 seconds instead of each 500ms<commit_after>#include <kms-core.h>
#include <rtmp/kms-rtmp-receiver.h>
#include <rtmp/kms-rtmp-session.h>
#include "internal/kms-utils.h"

#define KMS_RTMP_RECEIVER_GET_PRIVATE(obj) (G_TYPE_INSTANCE_GET_PRIVATE ((obj), KMS_TYPE_RTMP_RECEIVER, KmsRtmpReceiverPriv))

#define LOCK(obj) (g_mutex_lock(KMS_RTMP_RECEIVER(obj)->priv->mutex))
#define UNLOCK(obj) (g_mutex_unlock(KMS_RTMP_RECEIVER(obj)->priv->mutex))

#define MEDIA_TYPE_DATA "type"

struct _KmsRtmpReceiverPriv {
	GMutex *mutex;

	KmsRtmpSession *neg_spec;
	gboolean offerer;
	gboolean finished;
};

enum {
	PROP_0,

	PROP_NEG_SPEC,
	PROP_OFFERER,
};

G_DEFINE_TYPE(KmsRtmpReceiver, kms_rtmp_receiver, KMS_TYPE_MEDIA_HANDLER_SRC)

static void
dispose_neg_spec(KmsRtmpReceiver *self) {
	if (self->priv->neg_spec != NULL) {
		g_object_unref(self->priv->neg_spec);
		self->priv->neg_spec = NULL;
	}
}

static void
set_property (GObject *object, guint property_id, const GValue *value,
							GParamSpec *pspec) {
	KmsRtmpReceiver *self = KMS_RTMP_RECEIVER(object);

	switch (property_id) {
		case PROP_NEG_SPEC:
			LOCK(self);
			dispose_neg_spec(self);
			self->priv->neg_spec = g_value_dup_object(value);
			UNLOCK(self);
			break;
		case PROP_OFFERER:
			LOCK(self);
			self->priv->offerer = g_value_get_boolean(value);
			UNLOCK(self);
			break;
		default:
			/* We don't have any other property... */
			G_OBJECT_WARN_INVALID_PROPERTY_ID(object, property_id, pspec);
			break;
	}
}

static void
get_property(GObject *object, guint property_id, GValue *value,
							GParamSpec *pspec) {
	KmsRtmpReceiver *self = KMS_RTMP_RECEIVER(object);

	switch (property_id) {
		case PROP_NEG_SPEC:
			LOCK(self);
			g_value_set_object(value, self->priv->neg_spec);
			UNLOCK(self);
			break;
		case PROP_OFFERER:
			LOCK(self);
			g_value_set_boolean(value, self->priv->offerer);
			UNLOCK(self);
			break;
		default:
			/* We don't have any other property... */
			G_OBJECT_WARN_INVALID_PROPERTY_ID(object, property_id, pspec);
			break;
	}
}

static GstElement *generate_rtmpsrc(KmsRtmpReceiver *self);

static gboolean
remove_rtmp(KmsRtmpReceiver *self) {
	GstElement *rtmpsrc, *flvdemux;

	g_print("Remove rtmp for element: %s\n", GST_OBJECT_NAME(self));
	LOCK(self);
	if (self->priv->finished)
		goto end;

	rtmpsrc = gst_bin_get_by_name(GST_BIN(self), "rtmp");
	if (rtmpsrc == NULL)
		goto end;

	gst_bin_remove(GST_BIN(self), rtmpsrc);
	gst_element_set_state(rtmpsrc, GST_STATE_NULL);
	g_object_unref(rtmpsrc);

	rtmpsrc = generate_rtmpsrc(self);
	gst_bin_add(GST_BIN(self), rtmpsrc);
	gst_element_set_state(rtmpsrc, GST_STATE_PLAYING);

	flvdemux = gst_bin_get_by_name(GST_BIN(self), "demux");

	if (flvdemux == NULL)
		goto end;

	gst_element_link(rtmpsrc, flvdemux);

end:
	UNLOCK(self);
	return FALSE;
}

static gboolean
event_handler(GstPad *pad, GstEvent *event, KmsRtmpReceiver *self) {
	if (!GST_IS_EVENT(event))
		return TRUE;

	if (GST_EVENT_TYPE(event) == GST_EVENT_EOS) {
		g_timeout_add_full(G_PRIORITY_DEFAULT, 3000,
{'fraction_non_alphanumeric': 0.12338736354614621, 'fraction_numerical': 0.0021501819384717167, 'mean_word_length': 5.615973741794311, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6013891', 'n_tokens_mistral': 2705, 'n_tokens_neox': 2504, 'n_words': 418}