Implement mjr postproessing and upload

Cargo.toml

@@ -8,6 +8,7 @@ edition = "2018"
 crate-type = ["cdylib"]
 
 [dependencies]
+async-trait = "0.1"
 anyhow = "1.0"
 chrono = "0.4"
 janus-plugin-sys = "0.7"
@@ -20,26 +21,17 @@ lazy_static = "1.4"
 libc = "0.2"
 atom = "0.3"
 multimap = "0.8"
-glib-sys = "0.9"
-glib = "0.8"
-gstreamer = "0.14"
-gstreamer-app = "0.14" 
-gstreamer-base = "0.14"
-gstreamer-pbutils="0.14"
-rusoto_core = "0.42"
-rusoto_credential = "0.42"
-rusoto_s3 = "0.42"
-rusoto_signature = "0.42"
 http = "0.1"
 futures = { version = "0.3", features = ["thread-pool"] }
 futures-channel = "0.3"
 svc-error = { version = "0.1", features = ["sentry-extension"] }
 sentry = "0.18"
+tokio = { version = "0.2", features = ["process"] }
 uuid = { version = "0.8", features = ["serde", "v4"] }
 
 [dev-dependencies]
 rand = "0.6"
 rumqtt = "0.30"
 svc-agent = "0.7"
 tempfile = "3.0"
-janus-plugin-sys = "0.6"
+janus-plugin-sys = "0.7"

docker/Dockerfile

@@ -9,22 +9,17 @@ RUN set -xe \
     && apt-get -y --no-install-recommends install \
         autoconf \
         automake \
+        awscli \
         ca-certificates \
         curl \
         ffmpeg \
         gengetopt \
         git \
-        gstreamer1.0-libav \
-        gstreamer1.0-plugins-bad \
-        gstreamer1.0-plugins-base \
-        gstreamer1.0-plugins-good \
-        gstreamer1.0-plugins-ugly \
+        libavformat-dev \
+        libavcodec-dev \
         libconfig-dev \
         libcurl4-openssl-dev \
         libglib2.0-dev \
-        libgstreamer1.0-dev \
-        libgstreamer-plugins-base1.0-dev \
-        libgstrtspserver-1.0-dev \
         libjansson-dev \
         libmicrohttpd-dev \
         libogg-dev \
@@ -90,7 +85,7 @@ RUN set -xe \
     && git clone 'https://github.com/netology-group/janus-gateway' . \
     && git checkout "${JANUS_GATEWAY_COMMIT}" \
     && ./autogen.sh \
-    && ./configure --prefix=/opt/janus \
+    && ./configure --prefix=/opt/janus --enable-post-processing  \
     && make -j $(nproc) \
     && make install \
     && make configs \
@@ -134,11 +129,7 @@ RUN set -eux; \
 
 RUN set -xe \
     && apt-get update \
-    && apt-get -y --no-install-recommends install \
-        libjansson-dev libgstreamer1.0-dev libgstreamer-plugins-base1.0-dev \
-        gstreamer1.0-plugins-base gstreamer1.0-plugins-good \
-        gstreamer1.0-plugins-bad gstreamer1.0-plugins-ugly \
-        gstreamer1.0-libav libgstrtspserver-1.0-dev
+    && apt-get -y --no-install-recommends install libjansson-dev 
 
 WORKDIR /build
 COPY . .
@@ -151,3 +142,4 @@ FROM build-janus
 ARG PLUGIN=libjanus_conference.so
 WORKDIR /opt/janus
 COPY --from=build-plugin-deploy /build/target/release/${PLUGIN} ./lib/janus/plugins/${PLUGIN}
+COPY ./scripts/upload_record.sh /opt/janus/bin/upload_record.sh

docker/develop.dockerfile

@@ -9,22 +9,17 @@ RUN set -xe \
     && apt-get -y --no-install-recommends install \
         autoconf \
         automake \
+        awscli \
         ca-certificates \
         curl \
         ffmpeg \
         gengetopt \
         git \
-        gstreamer1.0-libav \
-        gstreamer1.0-plugins-bad \
-        gstreamer1.0-plugins-base \
-        gstreamer1.0-plugins-good \
-        gstreamer1.0-plugins-ugly \
+        libavformat-dev \
+        libavcodec-dev \
         libconfig-dev \
         libcurl4-openssl-dev \
         libglib2.0-dev \
-        libgstreamer1.0-dev \
-        libgstreamer-plugins-base1.0-dev \
-        libgstrtspserver-1.0-dev \
         libjansson-dev \
         libmicrohttpd-dev \
         libogg-dev \
@@ -110,7 +105,7 @@ RUN set -xe \
     && git clone 'https://github.com/netology-group/janus-gateway' . \
     && git checkout "${JANUS_GATEWAY_COMMIT}" \
     && ./autogen.sh \
-    && ./configure --prefix=/opt/janus \
+    && ./configure --prefix=/opt/janus --enable-post-processing \
     && make -j $(nproc) \
     && make install \
     && make configs \

docker/docker-compose.yml

@@ -13,6 +13,7 @@ services:
       - ..:/build
       - ../recordings:/recordings
       - janus-conference-cargo:/usr/local/cargo
+      - ../scripts/upload_record.sh:/opt/janus/bin/upload_record.sh
     env_file: janus.plugin.conference.environment
     links:
       - vernemq

examples/src/sdp.js

@@ -37,7 +37,7 @@ export function transformOfferSDP (sdp, opts) {
       modifiedPayload: 109,
     },
     video: {
-      codecName: 'H264',
+      codecName: 'VP8',
       modifiedPayload: 126,
     }
   }

scripts/upload_record.sh

@@ -0,0 +1,60 @@
+#!/bin/bash -e
+
+function REPORT_ERROR() { >&2 echo ${@}; }
+
+# Arguments.
+RTC_ID=$1
+BUCKET=$2
+OBJECT=$3
+
+if [[ ! ${RTC_ID} ]]; then $(REPORT_ERROR "RTC_ID isn't specified"); exit 1; fi
+if [[ ! ${BUCKET} ]]; then $(REPORT_ERROR "BUCKET isn't specified"); exit 1; fi
+if [[ ! ${OBJECT} ]]; then $(REPORT_ERROR "OBJECT isn't specified"); exit 1; fi
+
+# Environment.
+if [[ ! ${APP_UPLOADING__ACCESS_KEY_ID} ]]; then $(REPORT_ERROR "APP_UPLOADING__ACCESS_KEY_ID isn't specified"); exit 1; fi
+if [[ ! ${APP_UPLOADING__SECRET_ACCESS_KEY} ]]; then $(REPORT_ERROR "APP_UPLOADING__SECRET_ACCESS_KEY isn't specified"); exit 1; fi
+if [[ ! ${APP_UPLOADING__ENDPOINT} ]]; then $(REPORT_ERROR "APP_UPLOADING__ENDPOINT isn't specified"); exit 1; fi
+if [[ ! ${APP_UPLOADING__REGION} ]]; then $(REPORT_ERROR "APP_UPLOADING__REGION isn't specified"); exit 1; fi
+
+export AWS_ACCESS_KEY_ID=${APP_UPLOADING__ACCESS_KEY_ID}
+export AWS_SECRET_ACCESS_KEY=${APP_UPLOADING__SECRET_ACCESS_KEY}
+export AWS_ENDPOINT=${APP_UPLOADING__ENDPOINT}
+export AWS_REGION=${APP_UPLOADING__REGION}
+
+# Working directory.
+cd /recordings/${RTC_ID}
+
+# Convert video .mjr dumps into .webm files.
+for FILE in *.video.mjr; do
+  /opt/janus/bin/janus-pp-rec $FILE ${FILE%.*}.webm
+  echo "file '${FILE%.*}.webm'" >> video_sources.txt
+done
+
+# Get video segments durations and write to segments.csv file.
+for FILE in *.video.webm; do
+  DURATION=$(ffprobe -i ${FILE} -show_entries format=duration -v quiet -of csv="p=0")
+  echo "${FILE%%.*},${DURATION}" >> segments.csv
+done
+
+# Concat video segments into a single .webm file.
+ffmpeg -f concat -i video_sources.txt -c copy -y concat.webm
+
+# Convert audio .mjr dumps into .opus files.
+for FILE in *.audio.mjr; do
+  /opt/janus/bin/janus-pp-rec $FILE ${FILE%.*}.opus
+  echo "file '${FILE%.*}.opus'" >> audio_sources.txt
+done
+
+# Concat audio segments into a single .opus file.
+ffmpeg -f concat -i audio_sources.txt -c copy -y concat.opus
+
+# Mux video & audio into a single .webm file.
+ffmpeg -i concat.webm -i concat.opus -c copy full.webm
+
+# Upload record.
+aws --endpoint-url=${AWS_ENDPOINT} --region=${AWS_REGION} s3 cp full.webm s3://${BUCKET}/${OBJECT}
+
+# Clean up.
+cd ..
+rm -rf ./${RTC_ID}

src/app.rs

@@ -7,7 +7,6 @@ use chrono::Duration;
 use crate::conf::Config;
 use crate::message_handler::{JanusSender, MessageHandlingLoop};
 use crate::switchboard::LockedSwitchboard as Switchboard;
-use crate::uploader::Uploader;
 
 lazy_static! {
     pub static ref APP: AtomSetOnce<Box<App>> = AtomSetOnce::empty();
@@ -25,7 +24,6 @@ pub struct App {
     pub config: Config,
     pub switchboard: Switchboard,
     pub message_handling_loop: MessageHandlingLoop,
-    pub uploader: Uploader,
 }
 
 impl App {
@@ -58,13 +56,10 @@ impl App {
     }
 
     pub fn new(config: Config) -> Result<Self> {
-        let uploader = Uploader::build(config.uploading.clone())?;
-
         Ok(Self {
             config,
             switchboard: Switchboard::new(),
             message_handling_loop: MessageHandlingLoop::new(JanusSender::new()),
-            uploader,
         })
     }
 }

src/conf.rs

@@ -4,15 +4,13 @@ use anyhow::Result;
 use config;
 
 use crate::recorder;
-use crate::uploader;
 
 const CONFIG_FILE_NAME: &str = "janus.plugin.conference.toml";
 
 #[derive(Clone, Deserialize, Debug)]
 pub struct Config {
     pub general: General,
     pub recordings: recorder::Config,
-    pub uploading: uploader::Config,
     pub constraint: Constraint,
     pub sentry: Option<svc_error::extension::sentry::Config>,
 }

src/janus_recorder.rs

@@ -4,7 +4,6 @@ use std::ffi::CString;
 use std::os::raw::{c_char, c_int, c_long, c_uint};
 
 use anyhow::{bail, format_err, Context, Result};
-use glib_sys::gboolean;
 use janus_plugin_sys::janus_refcount;
 use libc::{pthread_mutex_t, FILE};
 
@@ -77,6 +76,7 @@ impl Drop for JanusRecorder<'_> {
 
 ///////////////////////////////////////////////////////////////////////////////
 
+type gboolean = c_int;
 type gint = c_int;
 type gint64 = c_long;
 type janus_mutex = pthread_mutex_t;

src/lib.rs

@@ -35,7 +35,6 @@ mod switchboard;
 mod utils;
 #[cfg(test)]
 mod test_stubs;
-mod uploader;
 
 use app::App;
 use conf::Config;