Add pause/resume recording functionality to Record&Play and SIP plugi…

…ns (#2724)

html/recordplaytest.html

@@ -89,7 +89,11 @@ <h3 class="panel-title">Recorder/Playout</h3>
 					<div class="col-md-6 hide" id="video">
 						<div class="panel panel-default">
 							<div class="panel-heading">
-								<h3 class="panel-title"><span id="videotitle">Remote Video</span> <button class="btn-xs btn-danger pull-right" autocomplete="off" id="stop">Stop</button></h3>
+								<h3 class="panel-title">
+									<span id="videotitle">Remote Video</span>
+									<button class="btn-xs btn-danger pull-right" autocomplete="off" id="stop">Stop</button>
+									<button class="btn-xs btn-primary pull-right" autocomplete="off" id="pause-resume">Pause</button>
+								</h3>
 							</div>
 							<div class="panel-body" id="videobox"></div>
 						</div>

html/recordplaytest.js

@@ -453,6 +453,7 @@ function startRecording() {
 		$('#list').unbind('click').attr('disabled', true);
 		$('#recset').attr('disabled', true);
 		$('#recslist').attr('disabled', true);
+		$('#pause-resume').removeClass('hide');
 
 		// bitrate and keyframe interval can be set at any time:
 		// before, after, during recording
@@ -497,6 +498,15 @@ function startRecording() {
 					recordplay.hangup();
 				}
 			});
+		$('#pause-resume').unbind('click').on('click', function() {
+			if($(this).text() === 'Pause') {
+				recordplay.send({message: {request: 'pause'}});
+				$(this).text('Resume');
+			} else {
+				recordplay.send({message: {request: 'resume'}});
+				$(this).text('Pause');
+			}
+		});
 	});
 }
 
@@ -515,6 +525,7 @@ function startPlayout() {
 	$('#list').unbind('click').attr('disabled', true);
 	$('#recset').attr('disabled', true);
 	$('#recslist').attr('disabled', true);
+	$('#pause-resume').addClass('hide');
 	var play = { request: "play", id: parseInt(selectedRecording) };
 	recordplay.send({ message: play });
 }

plugins/janus_recordplay.c

@@ -418,6 +418,7 @@ typedef struct janus_recordplay_recording {
 	char *offer;				/* The SDP offer that will be sent to watchers */
 	gboolean e2ee;				/* Whether media in the recording is encrypted, e.g., using Insertable Streams */
 	GList *viewers;				/* List of users watching this recording */
+	volatile gint paused;		/* Whether this recording is paused */
 	volatile gint completed;	/* Whether this recording was completed or still going on */
 	volatile gint destroyed;	/* Whether this recording has been marked as destroyed */
 	janus_refcount ref;			/* Reference counter */
@@ -1104,7 +1105,8 @@ struct janus_plugin_result *janus_recordplay_handle_message(janus_plugin_session
 		json_object_set_new(response, "settings", settings);
 		goto plugin_response;
 	} else if(!strcasecmp(request_text, "record") || !strcasecmp(request_text, "play")
-			|| !strcasecmp(request_text, "start") || !strcasecmp(request_text, "stop")) {
+			|| !strcasecmp(request_text, "start") || !strcasecmp(request_text, "stop")
+			|| !strcasecmp(request_text, "pause") || !strcasecmp(request_text, "resume")) {
 		/* These messages are handled asynchronously */
 		janus_recordplay_message *msg = g_malloc(sizeof(janus_recordplay_message));
 		msg->handle = handle;
@@ -1634,6 +1636,7 @@ static void *janus_recordplay_handler(void *data) {
 			rec->acodec = JANUS_AUDIOCODEC_NONE;
 			rec->vcodec = JANUS_VIDEOCODEC_NONE;
 			rec->e2ee = e2ee;
+			g_atomic_int_set(&rec->paused, 0);
 			g_atomic_int_set(&rec->destroyed, 0);
 			g_atomic_int_set(&rec->completed, 0);
 			janus_refcount_init(&rec->ref, janus_recordplay_recording_free);
@@ -2007,6 +2010,35 @@ static void *janus_recordplay_handler(void *data) {
 			}
 			/* Tell the core to tear down the PeerConnection, hangup_media will do the rest */
 			gateway->close_pc(session->handle);
+		} else if (!strcasecmp(request_text, "pause") || !strcasecmp(request_text, "resume")) {
+			JANUS_LOG(LOG_VERB, "Record&Play: Got pause/resume request\n");
+			if(session->recording) {
+				gboolean pause = !strcasecmp(request_text, "pause");
+				result = json_object();
+				json_object_set_new(result, "status", json_string(pause ? "paused" : "resumed"));
+				json_object_set_new(result, "id", json_integer(session->recording->id));
+				/* Also notify event handlers */
+				if(notify_events && gateway->events_is_enabled()) {
+					json_t *info = json_object();
+					json_object_set_new(info, "event", json_string(pause ? "paused" : "resumed"));
+					json_object_set_new(info, "id", json_integer(session->recording->id));
+					gateway->notify_event(&janus_recordplay_plugin, session->handle, info);
+				}
+				if (g_atomic_int_compare_and_exchange(&session->recording->paused, !pause, pause)) {
+					if(pause) {
+						janus_recorder_pause(session->arc);
+						janus_recorder_pause(session->vrc);
+						janus_recorder_pause(session->drc);
+					} else {
+						janus_recorder_resume(session->arc);
+						janus_recorder_resume(session->vrc);
+						janus_recorder_resume(session->drc);
+						gateway->send_pli(session->handle);
+					}
+				}
+			} else {
+				JANUS_LOG(LOG_VERB, "Record&Play: Not recording, ignoring pause/resume request\n");
+			}
 		} else {
 			JANUS_LOG(LOG_ERR, "Unknown request '%s'\n", request_text);
 			error_code = JANUS_RECORDPLAY_ERROR_INVALID_REQUEST;
@@ -2203,6 +2235,7 @@ void janus_recordplay_update_recordings_list(void) {
 		if(janus_recordplay_generate_offer(rec) < 0) {
 			JANUS_LOG(LOG_WARN, "Could not generate offer for recording %"SCNu64"...\n", rec->id);
 		}
+		g_atomic_int_set(&rec->paused, 0);
 		g_atomic_int_set(&rec->destroyed, 0);
 		g_atomic_int_set(&rec->completed, 1);
 		janus_refcount_init(&rec->ref, janus_recordplay_recording_free);

plugins/janus_sip.c

@@ -4377,10 +4377,11 @@ static void *janus_sip_handler(void *data) {
 				goto error;
 			json_t *action = json_object_get(root, "action");
 			const char *action_text = json_string_value(action);
-			if(strcasecmp(action_text, "start") && strcasecmp(action_text, "stop")) {
-				JANUS_LOG(LOG_ERR, "Invalid action (should be start|stop)\n");
+			if(strcasecmp(action_text, "start") && strcasecmp(action_text, "stop") &&
+					strcasecmp(action_text, "pause") && strcasecmp(action_text, "resume")) {
+				JANUS_LOG(LOG_ERR, "Invalid action (should be start|stop|pause|resume)\n");
 				error_code = JANUS_SIP_ERROR_INVALID_ELEMENT;
-				g_snprintf(error_cause, 512, "Invalid action (should be start|stop)");
+				g_snprintf(error_cause, 512, "Invalid action (should be start|stop|pause|resume)");
 				goto error;
 			}
 			gboolean record_audio = FALSE, record_video = FALSE,	/* No media is recorded by default */
@@ -4522,6 +4523,24 @@ static void *janus_sip_handler(void *data) {
 						gateway->send_pli(session->handle);
 					}
 				}
+			} else if(!strcasecmp(action_text, "pause")) {
+				if(record_audio)
+					janus_recorder_pause(session->arc);
+				if(record_video)
+					janus_recorder_pause(session->vrc);
+				if(record_peer_audio)
+					janus_recorder_pause(session->arc_peer);
+				if(record_peer_video)
+					janus_recorder_pause(session->vrc_peer);
+			} else if(!strcasecmp(action_text, "resume")) {
+				if(record_audio)
+					janus_recorder_resume(session->arc);
+				if(record_video && !janus_recorder_resume(session->vrc))
+					gateway->send_pli(session->handle);
+				if(record_peer_audio)
+					janus_recorder_resume(session->arc_peer);
+				if(record_peer_video)
+					janus_recorder_resume(session->vrc_peer);
 			} else {
 				/* Stop recording something: notice that this never returns an error, even when we were not recording anything */
 				janus_sip_recorder_close(session, record_audio, record_peer_audio, record_video, record_peer_video);

record.c

@@ -110,6 +110,7 @@ janus_recorder *janus_recorder_create_full(const char *dir, const char *codec, c
 	/* Create the recorder */
 	janus_recorder *rc = g_malloc0(sizeof(janus_recorder));
 	janus_refcount_init(&rc->ref, janus_recorder_free);
+	janus_rtp_switching_context_reset(&rc->context);
 	rc->dir = NULL;
 	rc->filename = NULL;
 	rc->file = NULL;
@@ -255,6 +256,35 @@ janus_recorder *janus_recorder_create_full(const char *dir, const char *codec, c
 	return rc;
 }
 
+int janus_recorder_pause(janus_recorder *recorder) {
+	if(!recorder)
+		return -1;
+	if(g_atomic_int_compare_and_exchange(&recorder->paused, 0, 1))
+		return 0;
+	return -2;
+}
+
+int janus_recorder_resume(janus_recorder *recorder) {
+	if(!recorder)
+		return -1;
+	janus_mutex_lock_nodebug(&recorder->mutex);
+	if(g_atomic_int_compare_and_exchange(&recorder->paused, 1, 0)) {
+		if(recorder->type == JANUS_RECORDER_AUDIO) {
+			recorder->context.a_ts_reset = TRUE;
+			recorder->context.a_seq_reset = TRUE;
+			recorder->context.a_last_time = janus_get_monotonic_time();
+		} else if(recorder->type == JANUS_RECORDER_VIDEO) {
+			recorder->context.v_ts_reset = TRUE;
+			recorder->context.v_seq_reset = TRUE;
+			recorder->context.v_last_time = janus_get_monotonic_time();
+		}
+		janus_mutex_unlock_nodebug(&recorder->mutex);
+		return 0;
+	}
+	janus_mutex_unlock_nodebug(&recorder->mutex);
+	return -2;
+}
+
 int janus_recorder_add_extmap(janus_recorder *recorder, int id, const char *extmap) {
 	if(!recorder || g_atomic_int_get(&recorder->header) || id < 1 || id > 15 || extmap == NULL)
 		return -1;
@@ -292,6 +322,10 @@ int janus_recorder_save_frame(janus_recorder *recorder, char *buffer, uint lengt
 		janus_mutex_unlock_nodebug(&recorder->mutex);
 		return -4;
 	}
+	if(g_atomic_int_get(&recorder->paused)) {
+		janus_mutex_unlock_nodebug(&recorder->mutex);
+		return -5;
+	}
 	gint64 now = janus_get_monotonic_time();
 	if(!g_atomic_int_get(&recorder->header)) {
 		/* Write info header as a JSON formatted info */
@@ -384,17 +418,39 @@ int janus_recorder_save_frame(janus_recorder *recorder, char *buffer, uint lengt
 				res, sizeof(gint64), g_strerror(errno));
 		}
 	}
+	/* Edit packet header if needed */
+	janus_rtp_header *header = (janus_rtp_header *)buffer;
+	uint32_t ssrc = 0;
+	uint16_t seq = 0;
+	if(recorder->type != JANUS_RECORDER_DATA) {
+		ssrc = ntohl(header->ssrc);
+		seq = ntohs(header->seq_number);
+		timestamp = ntohl(header->timestamp);
+		janus_rtp_header_update(header, &recorder->context, recorder->type == JANUS_RECORDER_VIDEO, 0);
+	}
 	/* Save packet on file */
 	int temp = 0, tot = length;
 	while(tot > 0) {
 		temp = fwrite(buffer+length-tot, sizeof(char), tot, recorder->file);
 		if(temp <= 0) {
 			JANUS_LOG(LOG_ERR, "Error saving frame...\n");
+			if(recorder->type != JANUS_RECORDER_DATA) {
+				/* Restore packet header data */
+				header->ssrc = htonl(ssrc);
+				header->seq_number = htons(seq);
+				header->timestamp = htonl(timestamp);
+			}
 			janus_mutex_unlock_nodebug(&recorder->mutex);
 			return -6;
 		}
 		tot -= temp;
 	}
+	if(recorder->type != JANUS_RECORDER_DATA) {
+		/* Restore packet header data */
+		header->ssrc = htonl(ssrc);
+		header->seq_number = htons(seq);
+		header->timestamp = htonl(timestamp);
+	}
 	/* Done */
 	janus_mutex_unlock_nodebug(&recorder->mutex);
 	return 0;

record.h

@@ -27,6 +27,7 @@
 
 #include "mutex.h"
 #include "refcount.h"
+#include "rtp.h"
 
 
 /*! \brief Media types we can record */
@@ -60,6 +61,10 @@ typedef struct janus_recorder {
 	volatile int header;
 	/*! \brief Whether this recorder instance can be used for writing or not */
 	volatile int writable;
+	/*! \brief Whether writing s/RTP packets/data is paused */
+	volatile int paused;
+	/*! \brief RTP switching context for rewriting RTP headers */
+	janus_rtp_switching_context context;
 	/*! \brief Mutex to lock/unlock this recorder instance */
 	janus_mutex mutex;
 	/*! \brief Atomic flag to check if this instance has been destroyed */
@@ -92,6 +97,16 @@ janus_recorder *janus_recorder_create(const char *dir, const char *codec, const
  * @param[in] filename Filename to use for the recording
  * @returns A valid janus_recorder instance in case of success, NULL otherwise */
 janus_recorder *janus_recorder_create_full(const char *dir, const char *codec, const char *fmtp, const char *filename);
+/*! \brief Pause recording packets
+ * \note This is to allow pause and resume recorder functionality.
+ * @param[in] recorder The janus_recorder to pause
+ * @returns 0 in case of success, a negative integer otherwise */
+int janus_recorder_pause(janus_recorder *recorder);
+/*! \brief Resume recording packets
+ * \note This is to allow pause and resume recorder functionality.
+ * @param[in] recorder The janus_recorder to resume
+ * @returns 0 in case of success, a negative integer otherwise */
+int janus_recorder_resume(janus_recorder *recorder);
 /*! \brief Add an RTP extension to this recording
  * \note This will only be possible BEFORE the first frame is written, as it needs to
  * be reflected in the .mjr header: doing this after that will return an error.

rtp.c

@@ -632,10 +632,17 @@ void janus_rtp_header_update(janus_rtp_header *header, janus_rtp_switching_conte
 			JANUS_LOG(LOG_VERB, "Video SSRC changed, %"SCNu32" --> %"SCNu32"\n",
 				context->v_last_ssrc, ssrc);
 			context->v_last_ssrc = ssrc;
+			context->v_ts_reset = TRUE;
+			context->v_seq_reset = TRUE;
+			/* Reset skew compensation data */
+			context->v_new_ssrc = TRUE;
+		}
+		if(context->v_ts_reset) {
+			/* Video timestamp was paused for a while */
+			JANUS_LOG(LOG_HUGE, "Video RTP timestamp reset requested");
+			context->v_ts_reset = FALSE;
 			context->v_base_ts_prev = context->v_last_ts;
 			context->v_base_ts = timestamp;
-			context->v_base_seq_prev = context->v_last_seq;
-			context->v_base_seq = seq;
 			/* How much time since the last video RTP packet? We compute an offset accordingly */
 			if(context->v_last_time > 0) {
 				gint64 time_diff = janus_get_monotonic_time() - context->v_last_time;
@@ -644,13 +651,12 @@ void janus_rtp_header_update(janus_rtp_header *header, janus_rtp_switching_conte
 					time_diff = 1;
 				context->v_base_ts_prev += (guint32)time_diff;
 				context->v_last_ts += (guint32)time_diff;
-				JANUS_LOG(LOG_VERB, "Computed offset for video RTP timestamp: %"SCNu32"\n", (guint32)time_diff);
+				JANUS_LOG(LOG_HUGE, "Computed offset for video RTP timestamp: %"SCNu32"\n", (guint32)time_diff);
 			}
-			/* Reset skew compensation data */
-			context->v_new_ssrc = TRUE;
 		}
 		if(context->v_seq_reset) {
-			/* Video sequence number was paused for a while: just update that */
+			/* Video sequence number was paused for a while */
+			JANUS_LOG(LOG_HUGE, "Video RTP sequence number reset requested");
 			context->v_seq_reset = FALSE;
 			context->v_base_seq_prev = context->v_last_seq;
 			context->v_base_seq = seq;
@@ -671,10 +677,17 @@ void janus_rtp_header_update(janus_rtp_header *header, janus_rtp_switching_conte
 			JANUS_LOG(LOG_VERB, "Audio SSRC changed, %"SCNu32" --> %"SCNu32"\n",
 				context->a_last_ssrc, ssrc);
 			context->a_last_ssrc = ssrc;
+			context->a_ts_reset = TRUE;
+			context->a_seq_reset = TRUE;
+			/* Reset skew compensation data */
+			context->a_new_ssrc = TRUE;
+		}
+		if(context->a_ts_reset) {
+			/* Audio timestamp was paused for a while */
+			JANUS_LOG(LOG_HUGE, "Audio RTP timestamp reset requested");
+			context->a_ts_reset = FALSE;
 			context->a_base_ts_prev = context->a_last_ts;
 			context->a_base_ts = timestamp;
-			context->a_base_seq_prev = context->a_last_seq;
-			context->a_base_seq = seq;
 			/* How much time since the last audio RTP packet? We compute an offset accordingly */
 			if(context->a_last_time > 0) {
 				gint64 time_diff = janus_get_monotonic_time() - context->a_last_time;
@@ -687,13 +700,12 @@ void janus_rtp_header_update(janus_rtp_header *header, janus_rtp_switching_conte
 				context->a_base_ts_prev += (guint32)time_diff;
 				context->a_prev_ts += (guint32)time_diff;
 				context->a_last_ts += (guint32)time_diff;
-				JANUS_LOG(LOG_VERB, "Computed offset for audio RTP timestamp: %"SCNu32"\n", (guint32)time_diff);
+				JANUS_LOG(LOG_HUGE, "Computed offset for audio RTP timestamp: %"SCNu32"\n", (guint32)time_diff);
 			}
-			/* Reset skew compensation data */
-			context->a_new_ssrc = TRUE;
 		}
 		if(context->a_seq_reset) {
-			/* Audio sequence number was paused for a while: just update that */
+			/* Audio sequence number was paused for a while */
+			JANUS_LOG(LOG_HUGE, "Audio RTP sequence number reset requested");
 			context->a_seq_reset = FALSE;
 			context->a_base_seq_prev = context->a_last_seq;
 			context->a_base_seq = seq;

rtp.h

@@ -242,8 +242,8 @@ typedef struct janus_rtp_switching_context {
 			v_last_ssrc, v_last_ts, v_base_ts, v_base_ts_prev, v_prev_ts, v_target_ts, v_start_ts;
 	uint16_t a_last_seq, a_prev_seq, a_base_seq, a_base_seq_prev,
 			v_last_seq, v_prev_seq, v_base_seq, v_base_seq_prev;
-	gboolean a_seq_reset, a_new_ssrc,
-			v_seq_reset, v_new_ssrc;
+	gboolean a_ts_reset, a_seq_reset, a_new_ssrc,
+			v_ts_reset, v_seq_reset, v_new_ssrc;
 	gint16 a_seq_offset,
 			v_seq_offset;
 	gint32 a_prev_delay, a_active_delay, a_ts_offset,