【C++】gstreamerを使ってRTSP(Video+Audio)を再生するサンプル

はじめに

gstreamer をコマンドベースで利用する場合の記事はいくつかあるのですが、
C++ でライブラリとして利用する場合のサンプルが少なく、どのようなものなのかと使ってみた記録です。

今回は以下のようなことができる実現案を、サンプルコードと共にご紹介いたします。

Windows で動作するアプリケーション
MPEG2-TTS に対応できるようにしておきたい
デコードした映像は、OpenCV の Mat 形式でアプリ側で使用したい
デコードした音声は、gstreamer に任せて再生したい

SANACHAN

本サンプルでは、デコードした映像をブラウザで表示します。

開発環境は以下を使用しています。

OS	Windows10 Pro
CPU	Intel Core i5 11400
GPU	NVIDIA GTX 1650
IDE	Visual Studio 2022
CMake	v3.28.3
gstreamer	v1.22.9
OpenCV	v4.9.0
boost	v1.84.0
websocketpp	v0.8.2

環境構築

先ずは、gstreamer を利用できる環境を作ります。

インストール

gstreamer の公式ページから、インストーラーをダウンロードします。
以下、2つのファイルをダウンロードしてインストールします。

Runtime Installer
Development Installer

注意ポイント①

インストールする際、下図のように全てのプラグインをインストールしましょう。
多くのコーデックに対応するためのプラグイン gst-plugins-good などが含まれています。

注意ポイント②

インストールは、Runtime → Development の順で行いましょう。

gstreamer インストール画面

環境変数の設定

gstreamer をインストールしたパスに合わせて、システム環境変数を以下のように設定します。

SANACHAN

設定後、再起動しないと反映されませんでした。

変数	値
GST_PLUGIN_SYSTEM_PATH	[インストールしたパス]\1.0\msvc_x86_64\lib\gstreamer-1.0
GSTREAMER_1_0_ROOT_MSVC_X86_64	[インストールしたパス]\1.0\msvc_x86_64

システム環境変数の設定

Visual Studio プロジェクト

C++ で記述するため、コンパイラへ「追加のインクルードディレクトリ」、
リンカーへ「追加のライブラリディレクトリ」と「依存ファイル」の設定が必要です。

gstreamer をインストールしたパスに合わせて、以下のように設定します。

設定項目	設定内容
C/C++＞全般＞追加のインクルードディレクトリ	[インストールしたパス]\opencv-4.9.0\build\include; [インストールしたパス]\1.0\msvc_x86_64\lib\glib-2.0\include; [インストールしたパス]\1.0\msvc_x86_64\include\gstreamer-1.0; [インストールしたパス]\1.0\msvc_x86_64\include\glib-2.0; [インストールしたパス]\1.0\msvc_x86_64\include\glib-2.0\glib;
リンカー＞全般＞追加のライブラリディレクトリ	[インストールしたパス]\opencv-4.9.0\build\x64\vc16\lib; [インストールしたパス]\1.0\msvc_x86_64\lib
リンカー＞入力＞追加の依存ファイル	gobject-2.0.lib; glib-2.0.lib; gstreamer-1.0.lib; gstaudio-1.0.lib; gstpbutils-1.0.lib; gstvideo-1.0.lib; gstapp-1.0.lib; opencv_world490.lib; (Debug 構成の場合は opencv_world490d.lib;）

追加のインクルードディレクトリ

追加のライブラリディレクトリ

追加の依存ファイル

SANACHAN

これで環境構築、及び空のVisual Studioプロジェクト作成完了です。

gstreamer pipeline の検討

今回は gstreamer を使って、少し特殊なことを行います。
順に説明、解説していきます。

アプリで映像を取得

gstreamer でデコードした映像を、gstreamer が備えている DirectX で表示するのではなく、
アプリ側で独自に画像処理を行って表示することを条件としてプログラミングします。

この場合、gstreamer へ指定する pipeline を以下のように記述します。

 gstreamer pipeline
"rtspsrc location=rtsp://127.0.0.1:554/stream"
" ! rtph264depay"
" ! h264parse"
" ! avdec_h264"
" ! videoconvert"
" ! appsink name = sink"

RTSP を入力とするため、rtspsrc を使用します。
また、アプリ側でデコード結果を取得したいので、末端は appsink を利用し名前を付けます。

SANACHAN

これで、アプリ側でデコード結果の画素データを取得できるようになります。

音声の出力

映像データはアプリで使用し、音声データは gstreamer に任せて再生することにしています。
この場合は、gstreamer へ指定する pipeline を以下のように記述します。

 gstreamer pipeline
"rtspsrc location=rtsp://127.0.0.1:554/stream name=src"
" src. ! queue"
" ! rtph264depay"
" ! h264parse"
" ! avdec_h264"
" ! videoconvert"
" ! appsink name=sink"
" src. ! queue leaky=1"
" ! decodebin"
" ! audioconvert"
" ! audioresample"
" ! autoaudiosink sync=false"

分かりやすくするため細かく改行していますが、rtspsrc の入力に名前を付け、
入力 src. から映像と音声、それぞれの pipeline を queue で繋げるイメージで記述します。

音声の末端は autoaudiosink としているため、gstreamer が自動で再生してくれます。

SANACHAN

これで映像データはアプリ、音声データは自動で再生する条件を満たします。

MPEG2-TTS の対応

以下で紹介している方法で、rtph264depay を拡張し「rtpmp2tdepay tts=true」と指定、
更に「tsdemux」を追加すれば再生できるようになります。

こちらもCHECK

: MPEG2-TTSをOSSでデコードする方法

はじめに MPEG-2（Moving Picture Experts Group）は、映像や音声を多重化する ISO/IEC の標準規格です。 DVD や放送で主に使用されており、ごく一部の公共インフ ...

続きを見る

SANACHAN

pipeline の記述は、「gstreamerのpipeline集」を参照ください。

gstreamer 制御クラス

少し汎用的にプログラムを記述するため、gstreamer を制御するためのクラスを用意します。

#pragma once
#include <gst/gst.h>
#include <gst/app/gstappsink.h>
#include <glib.h>
#include <opencv2/opencv.hpp>

class GstHandler {
public:
    GstHandler(int argc, char** argv, GstDebugLevel level = GST_LEVEL_ERROR)
    : m_pipeline(nullptr)
    , m_sink(nullptr)
    , m_appsink(nullptr)　{
        // initialize
        gst_init(&argc, &argv);

        // set debug log level
        if (!gst_debug_is_active()) {
            gst_debug_set_active(TRUE);

            GstDebugLevel dbglevel = gst_debug_get_default_threshold();
            if (dbglevel < level)
                gst_debug_set_default_threshold(level);
        }
    }

    virtual ~GstHandler() {
        if (m_pipeline) {
            gst_element_set_state(m_pipeline, GST_STATE_NULL);
            if (m_sink)
                gst_object_unref(m_sink);
            gst_object_unref(m_pipeline);
        }
    }

    bool CreatePipeline(std::string& command, const char* sink_name = nullptr) {
        if (command.empty())
            return false;

        GError* err = nullptr;
        m_pipeline = gst_parse_launch((gchar*)command.c_str(), &err);
        if (!m_pipeline) {
            if (err) {
                g_printerr("Failed to create pipeline: msg=%s\n", err->message);
                g_error_free(err);
            }
            return false;
        }

        if (sink_name) {
            m_sink = gst_bin_get_by_name(GST_BIN(m_pipeline), sink_name);
            m_appsink = GST_APP_SINK(m_sink);
        }

        m_command = command;
        return true;
    }

    bool StartPipeline() {
        if (!m_pipeline)
            return false;

        if (gst_element_set_state(m_pipeline, GST_STATE_PLAYING) == GST_STATE_CHANGE_FAILURE) {
            g_printerr("Failed to start pipeline\n");
            return false;
        }
        g_print("Pipeline started with: %s\n", m_command.c_str());
        return true;
    }

    void ShowPipelineState() {
        if (!m_pipeline)
            return;

        GstState state, pending;
        GstStateChangeReturn ret = gst_element_get_state(m_pipeline, &state, &pending, GST_CLOCK_TIME_NONE);
        if (ret == GST_STATE_CHANGE_FAILURE) {
            g_printerr("Failed to get pipeline state\n");
            return;
        }
        g_print("Pipeline state: %s\n", gst_element_state_get_name(state));
    }

    bool ExecPipeline() {
        if (!m_pipeline)
            return false;

        GstBus* bus = gst_element_get_bus(m_pipeline);
        GstMessage* msg = gst_bus_timed_pop_filtered(bus, GST_CLOCK_TIME_NONE,
            (GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS | GST_MESSAGE_STATE_CHANGED));
        if (msg) {
            gst_message_unref(msg);
        }
        return true;
    }

    cv::Mat ReadFrame() {
        cv::Mat empty;

        if (!m_pipeline || !m_appsink)
            return empty;

        GstSample* sample = gst_app_sink_pull_sample(m_appsink);
        if (!sample) {
            g_printerr("sample is NULL.\n");
            return empty;
        }

        // read frame from GStreamer
        GstBuffer* buffer = gst_sample_get_buffer(sample);
        GstCaps* caps = gst_sample_get_caps(sample);
        GstStructure* structure = gst_caps_get_structure(caps, 0);

        gint width, height;
        gst_structure_get_int(structure, "width", &width);
        gst_structure_get_int(structure, "height", &height);

        GstMapInfo map;
        if (gst_buffer_map(buffer, &map, GST_MAP_READ) != TRUE) {
            g_printerr("Failed to map buffer\n");
            gst_sample_unref(sample);
            return empty;
        }

        // check YUV or RGB, and convert to RGB if YUV
        cv::Mat frame;
        if (map.size / width / height != 3) {
            cv::Mat tmp = cv::Mat(height + height / 2, width, CV_8UC1, map.data);
            cv::cvtColor(tmp, frame, cv::COLOR_YUV2BGR_I420);
        }
        else {
            frame = cv::Mat(height, width, CV_8UC3, map.data).clone();
        }

        gst_buffer_unmap(buffer, &map);
        gst_sample_unref(sample);

        return frame;
    }


private:
    std::string m_command;

    GstElement* m_pipeline;
    GstElement* m_sink;
    GstAppSink* m_appsink;
};

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

#pragma once

#include <gst/gst.h>

#include <gst/app/gstappsink.h>

#include <glib.h>

#include <opencv2/opencv.hpp>

class GstHandler {

public:

GstHandler(int argc, char** argv, GstDebugLevel level = GST_LEVEL_ERROR)

: m_pipeline(nullptr)

, m_sink(nullptr)

, m_appsink(nullptr)　{

// initialize

gst_init(&argc, &argv);

// set debug log level

if (!gst_debug_is_active()) {

gst_debug_set_active(TRUE);

GstDebugLevel dbglevel = gst_debug_get_default_threshold();

if (dbglevel < level)

gst_debug_set_default_threshold(level);

}

virtual ~GstHandler() {

if (m_pipeline) {

gst_element_set_state(m_pipeline, GST_STATE_NULL);

if (m_sink)

gst_object_unref(m_sink);

gst_object_unref(m_pipeline);

}

bool CreatePipeline(std::string& command, const char* sink_name = nullptr) {

if (command.empty())

return false;

GError* err = nullptr;

m_pipeline = gst_parse_launch((gchar*)command.c_str(), &err);

if (!m_pipeline) {

if (err) {

g_printerr("Failed to create pipeline: msg=%s\n", err->message);

g_error_free(err);

}

return false;

}

if (sink_name) {

m_sink = gst_bin_get_by_name(GST_BIN(m_pipeline), sink_name);

m_appsink = GST_APP_SINK(m_sink);

}

m_command = command;

return true;

}

bool StartPipeline() {

if (!m_pipeline)

return false;

if (gst_element_set_state(m_pipeline, GST_STATE_PLAYING) == GST_STATE_CHANGE_FAILURE) {

g_printerr("Failed to start pipeline\n");

return false;

}

g_print("Pipeline started with: %s\n", m_command.c_str());

return true;

}

void ShowPipelineState() {

if (!m_pipeline)

return;

GstState state, pending;

GstStateChangeReturn ret = gst_element_get_state(m_pipeline, &state, &pending, GST_CLOCK_TIME_NONE);

if (ret == GST_STATE_CHANGE_FAILURE) {

g_printerr("Failed to get pipeline state\n");

return;

}

g_print("Pipeline state: %s\n", gst_element_state_get_name(state));

}

bool ExecPipeline() {

if (!m_pipeline)

return false;

GstBus* bus = gst_element_get_bus(m_pipeline);

GstMessage* msg = gst_bus_timed_pop_filtered(bus, GST_CLOCK_TIME_NONE,

(GstMessageType)(GST_MESSAGE_ERROR | GST_MESSAGE_EOS | GST_MESSAGE_STATE_CHANGED));

if (msg) {

gst_message_unref(msg);

}

return true;

}

cv::Mat ReadFrame() {

cv::Mat empty;

if (!m_pipeline || !m_appsink)

return empty;

GstSample* sample = gst_app_sink_pull_sample(m_appsink);

if (!sample) {

g_printerr("sample is NULL.\n");

return empty;

}

// read frame from GStreamer

GstBuffer* buffer = gst_sample_get_buffer(sample);

GstCaps* caps = gst_sample_get_caps(sample);

GstStructure* structure = gst_caps_get_structure(caps, 0);

gint width, height;

gst_structure_get_int(structure, "width", &width);

gst_structure_get_int(structure, "height", &height);

GstMapInfo map;

if (gst_buffer_map(buffer, &map, GST_MAP_READ) != TRUE) {

g_printerr("Failed to map buffer\n");

gst_sample_unref(sample);

return empty;

}

// check YUV or RGB, and convert to RGB if YUV

cv::Mat frame;

if (map.size / width / height != 3) {

cv::Mat tmp = cv::Mat(height + height / 2, width, CV_8UC1, map.data);

cv::cvtColor(tmp, frame, cv::COLOR_YUV2BGR_I420);

}

else {

frame = cv::Mat(height, width, CV_8UC3, map.data).clone();

}

gst_buffer_unmap(buffer, &map);

gst_sample_unref(sample);

return frame;

}

private:

std::string m_command;

GstElement* m_pipeline;

GstElement* m_sink;

GstAppSink* m_appsink;

};

前半は、gstreamer の初期化、pipeline の初期化や起動を記述しています。

また、デコードした画像データを取得できる ReadFrame() というメソッドを用意し、
OpenCV の VideoCapture.read() と同様に、フレームが到着するたびに読み出せるようにしています。

注意ポイント

gstreamer が確保するメモリ map.data を引数に cv::Mat を生成すると、
同じメモリアドレスを指す cv::Mat になるため、clone() してから解放します。

特に指定しなければ、取得できるのは YUV のカラーフォーマットなため、RGB へ変換して返します。
pipeline で、appsink の前に " ! video/x-raw, format=BGR " とすることで、
OpenCV に合わせた BGR フォーマットをデコード結果の色空間として指定することもできます。

SANACHAN

動きを見ていると、フレームレートに応じて gst_app_sink_pull_sample から戻ってくるため、特に意識せずに表示してもフレームレートを制御できるようです。

ストリーミングした映像をブラウザで表示

デコードした映像のフレームデータを cv::Mat 形式で受け取れるようになったため、
「ブラウザでRTSPストリーミング再生」で紹介している方法で、ブラウザに表示してみます。

今回は以下のような Bse64 変換関数と、

#pragma once
#include <string>
#include <vector>

namespace algorithm {

    bool encode_base64(const std::vector<unsigned char>& src, std::string& dst) {
        const std::string table("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/");
        std::string       cdst;

        for (std::size_t i = 0; i < src.size(); ++i) {
            switch (i % 3) {
            case 0:
                cdst.push_back(table[(src[i] & 0xFC) >> 2]);
                if (i + 1 == src.size()) {
                    cdst.push_back(table[(src[i] & 0x03) << 4]);
                    cdst.push_back('=');
                    cdst.push_back('=');
                }
                break;
            case 1:
                cdst.push_back(table[((src[i - 1] & 0x03) << 4) | ((src[i + 0] & 0xF0) >> 4)]);
                if (i + 1 == src.size()) {
                    cdst.push_back(table[(src[i] & 0x0F) << 2]);
                    cdst.push_back('=');
                }
                break;
            case 2:
                cdst.push_back(table[((src[i - 1] & 0x0F) << 2) | ((src[i + 0] & 0xC0) >> 6)]);
                cdst.push_back(table[src[i] & 0x3F]);
                break;
            }
        }
        dst.swap(cdst);
        return true;
    }

    bool decode_base64(const std::string& src, std::vector<unsigned char>& dst) {
        if (src.size() & 0x00000003)
            return false;

        const std::string          table("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/");
        std::vector<unsigned char> cdst;

        for (std::size_t i = 0; i < src.size(); i += 4) {
            if (src[i + 0] == '=')
                return false;
            else if (src[i + 1] == '=')
                return false;
            else if (src[i + 2] == '=') {
                const std::string::size_type s1 = table.find(src[i + 0]);
                const std::string::size_type s2 = table.find(src[i + 1]);

                if (s1 == std::string::npos || s2 == std::string::npos)
                    return false;

                cdst.push_back(static_cast<unsigned char>(((s1 & 0x3F) << 2) | ((s2 & 0x30) >> 4)));
                break;
            }
            else if (src[i + 3] == '=') {
                const std::string::size_type s1 = table.find(src[i + 0]);
                const std::string::size_type s2 = table.find(src[i + 1]);
                const std::string::size_type s3 = table.find(src[i + 2]);

                if (s1 == std::string::npos || s2 == std::string::npos || s3 == std::string::npos)
                    return false;

                cdst.push_back(static_cast<unsigned char>(((s1 & 0x3F) << 2) | ((s2 & 0x30) >> 4)));
                cdst.push_back(static_cast<unsigned char>(((s2 & 0x0F) << 4) | ((s3 & 0x3C) >> 2)));
                break;
            }
            else {
                const std::string::size_type s1 = table.find(src[i + 0]);
                const std::string::size_type s2 = table.find(src[i + 1]);
                const std::string::size_type s3 = table.find(src[i + 2]);
                const std::string::size_type s4 = table.find(src[i + 3]);

                if (s1 == std::string::npos || s2 == std::string::npos || s3 == std::string::npos || s4 == std::string::npos)
                    return false;

                cdst.push_back(static_cast<unsigned char>(((s1 & 0x3F) << 2) | ((s2 & 0x30) >> 4)));
                cdst.push_back(static_cast<unsigned char>(((s2 & 0x0F) << 4) | ((s3 & 0x3C) >> 2)));
                cdst.push_back(static_cast<unsigned char>(((s3 & 0x03) << 6) | ((s4 & 0x3F) >> 0)));
            }
        }

        dst.swap(cdst);
        return true;
    }

}; // namespace algorithm

#pragma once

#include <string>

#include <vector>

namespace algorithm {

bool encode_base64(const std::vector<unsigned char>& src, std::string& dst) {

const std::string table("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/");

std::string cdst;

for (std::size_t i = 0; i < src.size(); ++i) {

switch (i % 3) {

case 0:

cdst.push_back(table[(src[i] & 0xFC) >> 2]);

if (i + 1 == src.size()) {

cdst.push_back(table[(src[i] & 0x03) << 4]);

cdst.push_back('=');

}

break;

case 1:

cdst.push_back(table[((src[i - 1] & 0x03) << 4) | ((src[i + 0] & 0xF0) >> 4)]);

if (i + 1 == src.size()) {

cdst.push_back(table[(src[i] & 0x0F) << 2]);

cdst.push_back('=');

}

break;

case 2:

cdst.push_back(table[((src[i - 1] & 0x0F) << 2) | ((src[i + 0] & 0xC0) >> 6)]);

cdst.push_back(table[src[i] & 0x3F]);

break;

}

dst.swap(cdst);

return true;

}

bool decode_base64(const std::string& src, std::vector<unsigned char>& dst) {

if (src.size() & 0x00000003)

return false;

const std::string table("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/");

std::vector<unsigned char> cdst;

for (std::size_t i = 0; i < src.size(); i += 4) {

if (src[i + 0] == '=')

return false;

else if (src[i + 1] == '=')

return false;

else if (src[i + 2] == '=') {

const std::string::size_type s1 = table.find(src[i + 0]);

const std::string::size_type s2 = table.find(src[i + 1]);

if (s1 == std::string::npos || s2 == std::string::npos)

return false;

cdst.push_back(static_cast<unsigned char>(((s1 & 0x3F) << 2) | ((s2 & 0x30) >> 4)));

break;

}

else if (src[i + 3] == '=') {

const std::string::size_type s1 = table.find(src[i + 0]);

const std::string::size_type s2 = table.find(src[i + 1]);

const std::string::size_type s3 = table.find(src[i + 2]);

if (s1 == std::string::npos || s2 == std::string::npos || s3 == std::string::npos)

return false;

cdst.push_back(static_cast<unsigned char>(((s1 & 0x3F) << 2) | ((s2 & 0x30) >> 4)));

cdst.push_back(static_cast<unsigned char>(((s2 & 0x0F) << 4) | ((s3 & 0x3C) >> 2)));

break;

}

else {

const std::string::size_type s1 = table.find(src[i + 0]);

const std::string::size_type s2 = table.find(src[i + 1]);

const std::string::size_type s3 = table.find(src[i + 2]);

const std::string::size_type s4 = table.find(src[i + 3]);

if (s1 == std::string::npos || s2 == std::string::npos || s3 == std::string::npos || s4 == std::string::npos)

return false;

cdst.push_back(static_cast<unsigned char>(((s1 & 0x3F) << 2) | ((s2 & 0x30) >> 4)));

cdst.push_back(static_cast<unsigned char>(((s2 & 0x0F) << 4) | ((s3 & 0x3C) >> 2)));

cdst.push_back(static_cast<unsigned char>(((s3 & 0x03) << 6) | ((s4 & 0x3F) >> 0)));

}

dst.swap(cdst);

return true;

}

}; // namespace algorithm

libwebsocketpp を使って WebSocket で通信するクラスを用意します。

#pragma once
#include <opencv2/core/utils/logger.hpp>
#include <opencv2/opencv.hpp>

#include <websocketpp/config/asio_no_tls.hpp>
#include <websocketpp/server.hpp>

#include "Base64.h"

typedef websocketpp::server<websocketpp::config::asio> WsServer;
typedef websocketpp::connection_hdl WsClientHandle;

class OutputWebSocket {
public:
    static OutputWebSocket* Get() {
        static OutputWebSocket* inst = nullptr;
        if (!inst)
            inst = new OutputWebSocket();
        return inst;
    }

    void PutFrame(cv::Mat frame) {
        std::lock_guard<std::mutex> lock(m_mtx_img);
        if (!m_frames.empty())
            m_frames.clear();
        m_frames.push_back(frame);
        m_cond.notify_one();
    }

private:
    OutputWebSocket()
        : m_td_wss(new std::thread(&OutputWebSocket::ThreadWss, this))
        , m_td_send(new std::thread(&OutputWebSocket::ThreadSend, this)) {
    }

    void ThreadWss() {
        using websocketpp::lib::bind;
        using websocketpp::lib::placeholders::_1;
        using websocketpp::lib::placeholders::_2;

        m_wss.set_open_handler(bind(&OutputWebSocket::OnOpen, this, _1));
        m_wss.set_message_handler(bind(&OutputWebSocket::OnMessage, this, _1, _2));
        m_wss.set_close_handler(bind(&OutputWebSocket::OnClose, this, _1));

        m_wss.set_access_channels(websocketpp::log::alevel::none);
        m_wss.set_error_channels(websocketpp::log::elevel::all);

        m_wss.set_max_message_size(16 * 1024 * 1024 /* 16 MiB */);

        m_wss.init_asio();
        m_wss.listen(60000);
        m_wss.start_accept();

        m_wss.run();
    }

    void ThreadSend() {
        while (1) {
            cv::Mat frame;

            // wait frame to be available
            {
                std::unique_lock<std::mutex> lock(m_mtx_img);
                if (m_frames.empty())
                    m_cond.wait(lock);
                frame = m_frames.front();
                m_frames.pop_front();
            }

            try {
                SendFrame(frame);
            } catch (websocketpp::exception e) {
                ; // nop
            }
        }
    }

    void SendFrame(cv::Mat frame) {
        cv::Mat resized_frame;
        cv::resize(frame, resized_frame, cv::Size(1920, 1080), cv::INTER_LINEAR);

        std::vector<unsigned char> buf;
        std::vector<int> params(2);
        params[0] = cv::IMWRITE_JPEG_QUALITY;
        params[1] = 80;
        cv::imencode(".jpg", resized_frame, buf, params);

        std::string b64;
        algorithm::encode_base64(buf, b64);

        std::lock_guard<std::mutex> lock(m_mtx_conn);
        for (auto it : m_conn_list) {
            m_wss.send(it, b64, websocketpp::frame::opcode::text);
        }
    }

    void OnOpen(WsClientHandle hd) {
        std::lock_guard<std::mutex> lock(m_mtx_conn);
        m_conn_list.insert(hd);
    }

    void OnMessage(WsClientHandle hd, WsServer::message_ptr msg) {
        std::cout << msg->get_payload() << std::endl;
        std::cout << "DONE." << std::endl;
    }

    void OnClose(WsClientHandle hd) {
        std::lock_guard<std::mutex> lock(m_mtx_conn);
        m_conn_list.erase(hd);
    }

    WsServer m_wss;

    std::thread* m_td_wss;
    std::thread* m_td_send;

    std::mutex m_mtx_conn;
    std::set<WsClientHandle, std::owner_less<WsClientHandle>> m_conn_list;

    std::mutex m_mtx_img;
    std::condition_variable m_cond;
    std::list<cv::Mat> m_frames;
};

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

#pragma once

#include <opencv2/core/utils/logger.hpp>

#include <opencv2/opencv.hpp>

#include <websocketpp/config/asio_no_tls.hpp>

#include <websocketpp/server.hpp>

#include "Base64.h"

typedef websocketpp::server<websocketpp::config::asio> WsServer;

typedef websocketpp::connection_hdl WsClientHandle;

class OutputWebSocket {

public:

static OutputWebSocket* Get() {

static OutputWebSocket* inst = nullptr;

if (!inst)

inst = new OutputWebSocket();

return inst;

}

void PutFrame(cv::Mat frame) {

std::lock_guard<std::mutex> lock(m_mtx_img);

if (!m_frames.empty())

m_frames.clear();

m_frames.push_back(frame);

m_cond.notify_one();

}

private:

OutputWebSocket()

: m_td_wss(new std::thread(&OutputWebSocket::ThreadWss, this))

, m_td_send(new std::thread(&OutputWebSocket::ThreadSend, this)) {

}

void ThreadWss() {

using websocketpp::lib::bind;

using websocketpp::lib::placeholders::_1;

using websocketpp::lib::placeholders::_2;

m_wss.set_open_handler(bind(&OutputWebSocket::OnOpen, this, _1));

m_wss.set_message_handler(bind(&OutputWebSocket::OnMessage, this, _1, _2));

m_wss.set_close_handler(bind(&OutputWebSocket::OnClose, this, _1));

m_wss.set_access_channels(websocketpp::log::alevel::none);

m_wss.set_error_channels(websocketpp::log::elevel::all);

m_wss.set_max_message_size(16 * 1024 * 1024 /* 16 MiB */);

m_wss.init_asio();

m_wss.listen(60000);

m_wss.start_accept();

m_wss.run();

}

void ThreadSend() {

while (1) {

cv::Mat frame;

// wait frame to be available

{

std::unique_lock<std::mutex> lock(m_mtx_img);

if (m_frames.empty())

m_cond.wait(lock);

frame = m_frames.front();

m_frames.pop_front();

}

try {

SendFrame(frame);

} catch (websocketpp::exception e) {

; // nop

}

void SendFrame(cv::Mat frame) {

cv::Mat resized_frame;

cv::resize(frame, resized_frame, cv::Size(1920, 1080), cv::INTER_LINEAR);

std::vector<unsigned char> buf;

std::vector<int> params(2);

params[0] = cv::IMWRITE_JPEG_QUALITY;

params[1] = 80;

cv::imencode(".jpg", resized_frame, buf, params);

std::string b64;

algorithm::encode_base64(buf, b64);

std::lock_guard<std::mutex> lock(m_mtx_conn);

for (auto it : m_conn_list) {

m_wss.send(it, b64, websocketpp::frame::opcode::text);

}

void OnOpen(WsClientHandle hd) {

std::lock_guard<std::mutex> lock(m_mtx_conn);

m_conn_list.insert(hd);

}

void OnMessage(WsClientHandle hd, WsServer::message_ptr msg) {

std::cout << msg->get_payload() << std::endl;

std::cout << "DONE." << std::endl;

}

void OnClose(WsClientHandle hd) {

std::lock_guard<std::mutex> lock(m_mtx_conn);

m_conn_list.erase(hd);

}

WsServer m_wss;

std::thread* m_td_wss;

std::thread* m_td_send;

std::mutex m_mtx_conn;

std::set<WsClientHandle, std::owner_less<WsClientHandle>> m_conn_list;

std::mutex m_mtx_img;

std::condition_variable m_cond;

std::list<cv::Mat> m_frames;

};

SANACHAN

websocketpp と boost のビルドが必要になります。また、Visual Studio のプロジェクトにインクルードディレクトリの追加も必要です。双方、ヘッダーだけで完結するため、インクルードディレクトリだけでOKです。

設定項目	設定内容
C/C++＞全般＞追加のインクルードディレクトリ	[ビルドしたパス]\include\boost-1_84; [ビルドしたパス]\websocketpp-0.8.2\install\include;

サンプルの Main.cpp

これまで準備してきたクラス達を使用し、以下のように記述することで、
所望の動作をするアプリケーションを作成できます。

#include <iostream>
#include <fstream>

#include <opencv2/core/utils/logger.hpp>
#include <opencv2/opencv.hpp>

#include "GstHandler.h"
#include "OutputWebSocket.h"

int main(int argc, char* argv[]) {

    GstHandler* handl = new GstHandler(argc, argv, GST_LEVEL_WARNING);
    std::string pipeline_comm_video_only(
        "rtspsrc location=rtsp://127.0.0.1:554/stream"
        " ! rtph264depay"
        " ! h264parse"
        " ! avdec_h264"
        " ! videoconvert"
        " ! appsink name = sink"
    );
    std::string pipeline_comm_video_and_audio(
        "rtspsrc location=rtsp://127.0.0.1:554/stream name=src"
        " src. ! queue"
        " ! rtph264depay"
        " ! h264parse"
        " ! avdec_h264"
        " ! videoconvert"
        " ! appsink name=sink"
        " src. ! queue leaky=1"
        " ! decodebin"
        " ! audioconvert"
        " ! audioresample"
        " ! autoaudiosink sync=false"
    );
    if (!handl->CreatePipeline(pipeline_comm_video_and_audio, "sink"))
        return -1;

    if (!handl->StartPipeline())
        return -1;

    if (!handl->ExecPipeline())
        return -1;

    // start websocket thread
    OutputWebSocket::Get();

    while (1) {
        cv::Mat frame = handl->ReadFrame();
        if (frame.empty()) {
            Sleep(500);
            continue;
        }
#if 1
        OutputWebSocket::Get()->PutFrame(frame);
#else
        cv::imshow("Video", frame);
        if (cv::waitKey(1) == 27) // ESC key
            break;
#endif
    }

    cv::destroyAllWindows();
    delete handl;

    return 0;
}

#include <iostream>

#include <fstream>

#include <opencv2/core/utils/logger.hpp>

#include <opencv2/opencv.hpp>

#include "GstHandler.h"

#include "OutputWebSocket.h"

int main(int argc, char* argv[]) {

GstHandler* handl = new GstHandler(argc, argv, GST_LEVEL_WARNING);

std::string pipeline_comm_video_only(

"rtspsrc location=rtsp://127.0.0.1:554/stream"

" ! rtph264depay"

" ! h264parse"

" ! avdec_h264"

" ! videoconvert"

" ! appsink name = sink"

);

std::string pipeline_comm_video_and_audio(

"rtspsrc location=rtsp://127.0.0.1:554/stream name=src"

" src. ! queue"

" ! rtph264depay"

" ! h264parse"

" ! avdec_h264"

" ! videoconvert"

" ! appsink name=sink"

" src. ! queue leaky=1"

" ! decodebin"

" ! audioconvert"

" ! audioresample"

" ! autoaudiosink sync=false"

);

if (!handl->CreatePipeline(pipeline_comm_video_and_audio, "sink"))

return -1;

if (!handl->StartPipeline())

return -1;

if (!handl->ExecPipeline())

return -1;

// start websocket thread

OutputWebSocket::Get();

while (1) {

cv::Mat frame = handl->ReadFrame();

if (frame.empty()) {

Sleep(500);

continue;

}

#if 1

OutputWebSocket::Get()->PutFrame(frame);

#else

cv::imshow("Video", frame);

if (cv::waitKey(1) == 27) // ESC key

break;

#endif

}

cv::destroyAllWindows();

delete handl;

return 0;

}

検討した pipeline を使用すると、デコードした映像データはアプリ側で表示処理でき、
音声データは gstreamer で自動で再生されます。

SANACHAN

映像と音声を別経路で再生しているため、処理系によってはリップシンクの制御が別途必要かもしれません。

おわりに

いかがでしたでしょうか。

gstreamer は、pipeline を構成するエレメントを組み替えることで、
きめ細やかな制御が簡単にできるようになっています。

pipeline の構成検討に障壁があるのか、あまり解説している記事が無いと感じており、
本記事が参考になれば幸いです。

【C++】gstreamerを使ってRTSP(Video+Audio)を再生するサンプル

はじめに

環境構築

インストール

環境変数の設定

Visual Studio プロジェクト

gstreamer pipeline の検討

アプリで映像を取得

音声の出力

MPEG2-TTS の対応

MPEG2-TTSをOSSでデコードする方法

gstreamer 制御クラス

ストリーミングした映像をブラウザで表示

サンプルの Main.cpp

おわりに