srs/trunk/src/app/srs_app_recv_thread.hpp

/**
 * The MIT License (MIT)
 *
 * Copyright (c) 2013-2020 Winlin
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of
 * this software and associated documentation files (the "Software"), to deal in
 * the Software without restriction, including without limitation the rights to
 * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
 * the Software, and to permit persons to whom the Software is furnished to do so,
 * subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
 * FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
 * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
 * IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
 * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

#ifndef SRS_APP_RECV_THREAD_HPP
#define SRS_APP_RECV_THREAD_HPP

#include <srs_core.hpp>

#include <vector>
#include <string>

#include <srs_app_st.hpp>
#include <srs_protocol_stream.hpp>
#include <srs_core_performance.hpp>
#include <srs_app_reload.hpp>

class SrsRtmpServer;
class SrsCommonMessage;
class SrsRtmpConn;
class SrsSource;
class SrsRequest;
class SrsConsumer;
class SrsHttpConn;
class SrsResponseOnlyHttpConn;

// The message consumer which consume a message.
class ISrsMessageConsumer
{
public:
    ISrsMessageConsumer();
    virtual ~ISrsMessageConsumer();
public:
    // Consume the received message.
    // @remark user must free this message.
    virtual srs_error_t consume(SrsCommonMessage* msg) = 0;
};

// The message pumper to pump messages to processer.
class ISrsMessagePumper : public ISrsMessageConsumer
{
public:
    ISrsMessagePumper();
    virtual ~ISrsMessagePumper();
public:
    // Whether the pumper is interrupted.
    // For example, when pumpter is busy, it's interrupted,
    // please wait for a while then try to feed the pumper.
    virtual bool interrupted() = 0;
    // Interrupt the pumper for a error.
    virtual void interrupt(srs_error_t error) = 0;
    // When start the pumper.
    virtual void on_start() = 0;
    // When stop the pumper.
    virtual void on_stop() = 0;
};

// The recv thread, use message handler to handle each received message.
class SrsRecvThread : public ISrsCoroutineHandler
{
protected:
    SrsCoroutine* trd;
    ISrsMessagePumper* pumper;
    SrsRtmpServer* rtmp;
    SrsContextId _parent_cid;
    // The recv timeout in srs_utime_t.
    srs_utime_t timeout;
public:
    // Constructor.
    // @param tm The receive timeout in srs_utime_t.
    SrsRecvThread(ISrsMessagePumper* p, SrsRtmpServer* r, srs_utime_t tm, SrsContextId parent_cid);
    virtual ~SrsRecvThread();
public:
    virtual SrsContextId cid();
public:
    virtual srs_error_t start();
    virtual void stop();
    virtual void stop_loop();
// Interface ISrsReusableThread2Handler
public:
    virtual srs_error_t cycle();
private:
    virtual srs_error_t do_cycle();
};

// The recv thread used to replace the timeout recv,
// which hurt performance for the epoll_ctrl is frequently used.
// @see: SrsRtmpConn::playing
// @see: https://github.com/ossrs/srs/issues/217
class SrsQueueRecvThread : public ISrsMessagePumper
{
private:
    std::vector<SrsCommonMessage*> queue;
    SrsRecvThread trd;
    SrsRtmpServer* rtmp;
    // The recv thread error code.
    srs_error_t recv_error;
    SrsConsumer* _consumer;
public:
	// TODO: FIXME: Refine timeout in time unit.
    SrsQueueRecvThread(SrsConsumer* consumer, SrsRtmpServer* rtmp_sdk, srs_utime_t tm, SrsContextId parent_cid);
    virtual ~SrsQueueRecvThread();
public:
    virtual srs_error_t start();
    virtual void stop();
public:
    virtual bool empty();
    virtual int size();
    virtual SrsCommonMessage* pump();
    virtual srs_error_t error_code();
// Interface ISrsMessagePumper
public:
    virtual srs_error_t consume(SrsCommonMessage* msg);
    virtual bool interrupted();
    virtual void interrupt(srs_error_t err);
    virtual void on_start();
    virtual void on_stop();
};

// The publish recv thread got message and callback the source method to process message.
// @see: https://github.com/ossrs/srs/issues/237
class SrsPublishRecvThread : virtual public ISrsMessagePumper, virtual public ISrsReloadHandler
#ifdef SRS_PERF_MERGED_READ
    , virtual public IMergeReadHandler
#endif
{
private:
    SrsRecvThread trd;
    SrsRtmpServer* rtmp;
    SrsRequest* req;
    // The msgs already got.
    int64_t _nb_msgs;
    // The video frames we got.
    uint64_t video_frames;
    // For mr(merged read),
    // @see https://github.com/ossrs/srs/issues/241
    bool mr;
    int mr_fd;
    srs_utime_t mr_sleep;
    // For realtime
    // @see https://github.com/ossrs/srs/issues/257
    bool realtime;
    // The recv thread error code.
    srs_error_t recv_error;
    SrsRtmpConn* _conn;
    // The params for conn callback.
    SrsSource* _source;
    // The error timeout cond
    // @see https://github.com/ossrs/srs/issues/244
    srs_cond_t error;
    // The merged context id.
    SrsContextId cid;
    SrsContextId ncid;
public:
    SrsPublishRecvThread(SrsRtmpServer* rtmp_sdk, SrsRequest* _req,
        int mr_sock_fd, srs_utime_t tm, SrsRtmpConn* conn, SrsSource* source, SrsContextId parent_cid);
    virtual ~SrsPublishRecvThread();
public:
    // Wait for error for some timeout.
    virtual srs_error_t wait(srs_utime_t tm);
    virtual int64_t nb_msgs();
    virtual uint64_t nb_video_frames();
    virtual srs_error_t error_code();
    virtual void set_cid(SrsContextId v);
    virtual SrsContextId get_cid();
public:
    virtual srs_error_t start();
    virtual void stop();
// Interface ISrsMessagePumper
public:
    virtual srs_error_t consume(SrsCommonMessage* msg);
    virtual bool interrupted();
    virtual void interrupt(srs_error_t err);
    virtual void on_start();
    virtual void on_stop();
// Interface IMergeReadHandler
public:
#ifdef SRS_PERF_MERGED_READ
    virtual void on_read(ssize_t nread);
#endif
// Interface ISrsReloadHandler
public:
    virtual srs_error_t on_reload_vhost_publish(std::string vhost);
    virtual srs_error_t on_reload_vhost_realtime(std::string vhost);
private:
    virtual void set_socket_buffer(srs_utime_t sleep_v);
};

// The HTTP receive thread, try to read messages util EOF.
// For example, the HTTP FLV serving thread will use the receive thread to break
// when client closed the request, to avoid FD leak.
// @see https://github.com/ossrs/srs/issues/636#issuecomment-298208427
class SrsHttpRecvThread : public ISrsCoroutineHandler
{
private:
    SrsResponseOnlyHttpConn* conn;
    SrsCoroutine* trd;
public:
    SrsHttpRecvThread(SrsResponseOnlyHttpConn* c);
    virtual ~SrsHttpRecvThread();
public:
    virtual srs_error_t start();
public:
    virtual srs_error_t pull();
// Interface ISrsOneCycleThreadHandler
public:
    virtual srs_error_t cycle();
};

#endif