streamer.h

#ifndef INCLUDE_MUXER_H_
#define INCLUDE_MUXER_H_
 
#include <log.h>
#include <net.h>
#include <ffmpeg.h>
#include <live_input_session.h>
#include <telemetry.h>
#include <output_streamer.h>
#include <transcoder.h>
#include <video_streaming_protocol.h>
#include <urtp/handler.h>
 
 
/** Top-level component for producing MISP-compliant video stream. */
class Streamer {
public:
    /** Should match MispStreamer.Params in managed code. */
    struct Params {
        /** Vehicle ID (typically serial number). */
        std::string vehicleId;
        /** User-defined tail number, may be empty. */
        Optional<std::string> tailNumber;
        /** Target URI for streaming. Examples:
         * * `udp://1.2.3.4:3338` - Plain UDP streaming, suitable for external MISP-compliant players
         *      such as GIS software or just a regular media player (metadata will not be available
         *      in such case).
         * * `urtp+connect://1.2.3.4://3341` - URTP streaming with outgoing connection. URTP is UgCS
         *      proprietary protocol supported by its Command Center components. It ensures better
         *      quality on unreliable links.
         */
        std::string targetUri;
        /** Zero value disables transcoding, -1 for auto (based on input bitrate and scale),
         * otherwise value in bits/s.
         */
        int64_t transcodeBitrate = 0;
        /** Frame size scale factor for transcoding output, 1.0 to preserve input size. */
        float transcodeScale = 1.0;
        /** Frame rate limit for transcoding output, fps. Zero to preserve all input frames. */
        float transcodeFramerateLimit = 0;
    };
 
    /** Should match MispStreamer.Status in managed code. */
    enum class Status {
        /** Stream is being initialized (format is probed, codec parameters determined etc.). */
        INITIALIZING,
        /** Stream is being sent normally. */
        OPERATIONAL,
        /** Connection has been closed unexpectedly. */
        CONNECTION_FAILURE,
        /** Streaming protocol version mismatch. */
        STREAMING_PROTOCOL_BAD_VERSION,
        /** Other streaming failure (e.g. network interface lost). */
        OTHER_FAILURE,
        /** Streamer was stopped normally. */
        FINISHED
    };
 
    /** Handler methods are called in a dedicated thread. */
    class EventsListener {
    public:
        virtual
        ~EventsListener()
        {}
 
        /** Called on streamer status change. */
        virtual void
        OnStatus(Status status) = 0;
 
        /** Called when queue is overflowed or when it is consumed again.
         * @param status True when overflowed, false when back in normal state.
         */
        virtual void
        OnQueueOverflow(bool status) = 0;
    };
 
    /**
     * @param params Streaming parameters.
     * @param eventsListener Optional listener for several event types.
     */
    Streamer(const Params &params, std::unique_ptr<EventsListener> &&eventsListener = nullptr);
 
    /** Start the component. It cannot be used until started. */
    void
    Start();
 
    /** Stop the component releasing all used resources. The component cannot be used after that. */
    void
    Stop();
 
    /** Set new tail number. */
    void
    SetTailNumber(const std::string &tailNumber);
 
    /** Feed next chunk of raw H.264 video stream. There are no requirements for chunks alignment
     * and sizes respectively to H.264 protocol data units.
     */
    void
    FeedData(const net::SingleBuffer::Ptr &pkt);
 
    /** Set most recent telemetry values to send in next metadata packet. Fields may be updated
     * selectively, only non-empty values are accounted in each invocation.
     */
    void
    FeedTelemetry(const Telemetry &telemetry);
 
    /** Get various streaming statistics. The statistics does not have any fixed structure (actual
     *  content is a subject for change at any time) and is intended for display/logging only. No
     *  any logic should be bound to.
     */
    StructuredData
    GetStatistics() const;
 
    /** @return Peer protocol version if known, zero if not known. The call is legal only for URTP
     * target.
     */
    int
    GetUrtpPeerVersion() const;
 
private:
    constexpr static auto TELEMETRY_SEND_INTERVAL = std::chrono::milliseconds(500);
    constexpr static int MIN_SCALED_FRAME_SIZE = 32;
 
    Logger _log{Log::GetLogger("Streamer")};
    Params params;
    const std::unique_ptr<EventsListener> eventsListener;
    misp::StreamingTarget target;
    asio::io_context ioCtx;
    std::thread thread;
    misp::LiveInputSession inputSession;
    std::unique_ptr<misp::OutputStreamer> outputStreamer;
    Optional<misp::Transcoder> transcoder;
    ev::ConnectionHandle inputStateConn, inputQueueStatusConn, outputQueueStatusConn, urtpStateConn;
    /** Values pending for next transmission. */
    Telemetry pendingTelemetry;
    asio::steady_timer telemetryTimer;
    ProcessingContext statusEventsCtx;
    std::thread statusEventsThread;
    /** Tail number value to send in KLV. */
    std::string klvTailNumber;
    Status curStatus = Status::INITIALIZING;
    bool inputQueueOverflow = false, outputQueueOveflow = false;
 
    void
    Run();
 
    /** @return true on success, false on failure (status set). */
    bool
    CreateOutput();
 
    void
    HandleInputStateChange(misp::InputSession::State state);
 
    void
    SetTelemetryTimer();
 
    void
    OnTelemetryTimer(const asio::error_code &error);
 
    void
    SendTelemetry();
 
    void
    StatusEventsLoop();
 
    /** Should be called in muxer context only. */
    void
    SetStatus(Status status);
 
    void
    OnQueueOverflowStatus(bool isInput, bool status);
 
    void
    OnUrtpState(misp::urtp::UrtpHandler::State state, misp::urtp::UrtpHandler::ErrorCode errorCode,
                const std::string &errorMsg);
 
    bool
    CanAcceptData() const
    {
        return curStatus < Status::CONNECTION_FAILURE;
    }
};
 
#endif /* INCLUDE_MUXER_H_ */