livethread.h

Go to the documentation of this file.

#ifndef LIVETHREAD_HEADER_GUARD 
#define LIVETHREAD_HEADER_GUARD
/*
 * livethread.h : A live555 thread
 * 
 * Copyright 2017-2020 Valkka Security Ltd. and Sampsa Riikonen
 * 
 * Authors: Sampsa Riikonen <sampsa.riikonen@iki.fi>
 * 
 * This file is part of the Valkka library.
 * 
 * Valkka is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation, either version 3 of the
 * License, or (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program.  If not, see <https://www.gnu.org/licenses/>
 *
 */
 
#include "live.h"
#include "liveserver.h"
#include "thread.h"
#include "framefilter.h"
#include "framefifo.h"
#include "event.h"
 
 
void setLiveOutPacketBuffermaxSize(unsigned i); // <pyapi>
 
 
class LiveFifo : public FrameFifo {                       
  
public:                                                    
  LiveFifo(const char* name, FrameFifoContext ctx);       
  ~LiveFifo();                                            
  
protected:
  void* live_thread;
  
public:
  void setLiveThread(void* live_thread);
  bool writeCopy(Frame* f, bool wait=false);
};                                                        
 
 
enum class LiveConnectionType { // <pyapi>
  none,                         // <pyapi>
  rtsp,                         // <pyapi>
  sdp                           // <pyapi>
};                              // <pyapi>
 
 
struct LiveConnectionContext {                                                                        // <pyapi>
  LiveConnectionContext(LiveConnectionType ct, std::string address, SlotNumber slot,                  // <pyapi>
                        FrameFilter* framefilter) :                                                   // <pyapi>
  connection_type(ct), address(address), slot(slot), framefilter(framefilter), msreconnect(10000),       // <pyapi>
  request_multicast(false), request_tcp(false), recv_buffer_size(0), reordering_time(0),              // <pyapi>
  time_correction(TimeCorrectionType::smart)                                                          // <pyapi>
  {}                                                                                                  // <pyapi>
  LiveConnectionContext() :                                                                           // <pyapi>
  connection_type(LiveConnectionType::none), address(""), slot(0), framefilter(NULL), msreconnect(10000), // <pyapi>
  request_multicast(false), request_tcp(false),time_correction(TimeCorrectionType::smart)             // <pyapi>
  {}                                                                                                  // <pyapi>
  LiveConnectionType connection_type;   
  std::string        address;           
  SlotNumber         slot;              
  FrameFilter*       framefilter;       
  long unsigned int  msreconnect;       
  bool               request_multicast; 
  bool               request_tcp;       
  unsigned           recv_buffer_size;  
  unsigned           reordering_time;   
  TimeCorrectionType time_correction;   
};                                                                                                    // <pyapi>
 
 
struct LiveOutboundContext {                                                                     // <pyapi>
  LiveOutboundContext(LiveConnectionType ct, std::string address, SlotNumber slot,               // <pyapi>
                      unsigned short int portnum) :                                              // <pyapi>
  connection_type(ct), address(address), slot(slot), portnum(portnum), ttl(225)                  // <pyapi>
  {}                                                                                             // <pyapi>
  LiveOutboundContext() :                                                                        // <pyapi>
  connection_type(LiveConnectionType::none), address(""), slot(0), portnum(0), ttl(255)          // <pyapi>
  {}                                                                                             // <pyapi>
  LiveConnectionType  connection_type; 
  std::string         address;         
  SlotNumber          slot;            
  unsigned short int  portnum;         
  unsigned char       ttl;             
};                                                                                               // <pyapi>
 
 
 
enum class LiveSignal {
  none,
  exit,
  // inbound streams
  register_stream,
  deregister_stream,
  play_stream,
  stop_stream,
  // outbound streams
  register_outbound,
  deregister_outbound
};
 
 
struct LiveSignalContext {
  LiveSignal              signal;
  LiveConnectionContext   *connection_context;
  LiveOutboundContext     *outbound_context;
};
 
 
 
class Connection {
  
public:
  Connection(UsageEnvironment& env, LiveConnectionContext& ctx);
  virtual ~Connection(); 
  
protected:
  LiveConnectionContext   &ctx;           
  
  // internal framefilter chain.. if we'd like to modify the frames before they are passed to the API user
  // more framefilters could be generated here and initialized in the constructor init list
  // the starting filter should always be named as "inputfilter" .. this is where Live555 writes the frames
  // TimestampFrameFilter2   timestampfilter; ///< Internal framefilter: correct timestamp
  // SlotFrameFilter         inputfilter;     ///< Internal framefilter: set slot number
  
  FrameFilter*            timestampfilter;
  FrameFilter*            inputfilter;
  FrameFilter*            repeat_sps_filter;        
  
  long int                frametimer;      
  long int                pendingtimer;    
  
public:
  UsageEnvironment &env;                   
  bool is_playing;
  
public:
  virtual void playStream() =0;   
  virtual void stopStream(bool cut = true) =0;   
  virtual void reStartStream();   
  virtual void reStartStreamIf(); 
  virtual bool isClosed();        
  virtual void forceClose();      
  SlotNumber getSlot();           
};
 
 
class Outbound { // will leave this quite generic .. don't know at this point how the rtsp server is going to be // analogy: AVThread
  
public:
  Outbound(UsageEnvironment& env, FrameFifo& fifo, LiveOutboundContext& ctx);  
  virtual ~Outbound(); 
  
public: // init'd at constructor time
  LiveOutboundContext  &ctx;     
  UsageEnvironment     &env;     
  FrameFifo            &fifo;    
  
protected:
  bool setup_ok, at_setup; 
  
public:
  virtual void reinit();              
  virtual void handleFrame(Frame *f); 
};
 
 
class RTSPConnection : public Connection {
 
public:
  RTSPConnection(UsageEnvironment& env, LiveConnectionContext& ctx);
  ~RTSPConnection();
  // RTSPConnection(const RTSPConnection& cp); ///< Copy constructor .. nopes, default copy constructor good enough
  
  
private:
  ValkkaRTSPClient* client; 
  LiveStatus livestatus;    
  bool termplease;          
  
public:
  void playStream();      
  void stopStream(bool cut = true);      
  void reStartStreamIf(); 
  bool isClosed();        
  void forceClose();
};
 
 
class SDPConnection : public Connection {
 
public:
  SDPConnection(UsageEnvironment& env, LiveConnectionContext& ctx);
  ~SDPConnection();
 
private:
  StreamClientState *scs;
  bool passthrough;
  
public:
  void playStream(); 
  void stopStream(bool cut = true); 
 
};
 
 
class SDPOutbound : public Outbound {
  
public: 
  SDPOutbound(UsageEnvironment &env, FrameFifo &fifo, LiveOutboundContext& ctx);
  ~SDPOutbound();  
  
public: // virtual redefined
  void reinit();
  void handleFrame(Frame *f);
  
public:
  std::vector<Stream*> streams;  
};
 
 
class RTSPOutbound : public Outbound {
  
public: 
  RTSPOutbound(UsageEnvironment &env, RTSPServer &server, FrameFifo &fifo, LiveOutboundContext& ctx);
  ~RTSPOutbound();  
  
protected: // init'd at constructor time
  RTSPServer &server;   
  
public: // virtual redefined
  void reinit();
  void handleFrame(Frame *f);
  
public:
  ServerMediaSession *media_session; 
  std::vector<ValkkaServerMediaSubsession*> media_subsessions;
  
};
 
 
 
class LiveThread : public Thread { // <pyapi>
  
public:                           
  static void periodicTask(void* cdata); 
 
public:                                                // <pyapi>
  LiveThread(const char* name, FrameFifoContext fifo_ctx=FrameFifoContext());  // <pyapi>
  ~LiveThread();                                                               // <pyapi>
  
protected: // frame input
  LiveFifo          infifo;     
  FifoFrameFilter   infilter;   
  
protected: // redefinitions
  std::deque<LiveSignalContext> signal_fifo;    
  
protected:
  TaskScheduler*    scheduler;               
  UsageEnvironment* env;                     
  char              eventLoopWatchVariable;  
  std::vector<Connection*>   slots_;         
  std::vector<Outbound*>     out_slots_;     
  std::list<Connection*>     pending;        
  bool                       exit_requested; 
  EventTriggerId    event_trigger_id_hello_world;
  EventTriggerId    event_trigger_id_frame_arrived;
  EventTriggerId    event_trigger_id_got_frames;
  int fc;                                    
  
protected: // rtsp server for live and/or recorded stream
  UserAuthenticationDatabase  *authDB;
  RTSPServer                  *server;
  
public: // redefined virtual functions
  void run();
  void preRun();
  void postRun();
  void sendSignal(LiveSignalContext signal_ctx);
  
protected:
  void handlePending();       
  void checkAlive();          
  void closePending();        
  void handleSignals();       
  void handleFrame(Frame* f); 
  
private: // internal
  int  safeGetSlot         (SlotNumber slot, Connection*& con);
  int  safeGetOutboundSlot (SlotNumber slot, Outbound*& outbound);
  // inbound streams
  void registerStream   (LiveConnectionContext &connection_ctx);
  void deregisterStream (LiveConnectionContext &connection_ctx);
  void playStream       (LiveConnectionContext &connection_ctx);
  // outbound streams
  void registerOutbound    (LiveOutboundContext &outbound_ctx); 
  void deregisterOutbound  (LiveOutboundContext &outbound_ctx);
  // thread control
  void stopStream       (LiveConnectionContext &connection_ctx);
  
public: // *** C & Python API *** .. these routines go through the condvar/mutex locking                                                // <pyapi>
  // inbound streams
  void registerStreamCall   (LiveConnectionContext &connection_ctx); 
  void deregisterStreamCall (LiveConnectionContext &connection_ctx); 
  void playStreamCall       (LiveConnectionContext &connection_ctx); 
  void stopStreamCall       (LiveConnectionContext &connection_ctx); 
  // outbound streams
  void registerOutboundCall   (LiveOutboundContext &outbound_ctx);   
  void deregisterOutboundCall (LiveOutboundContext &outbound_ctx);   
  // thread control
  void requestStopCall();                                            
  // LiveFifo &getFifo();                                            ///< API method: get fifo for sending frames with live555          // <pyapi>
  FifoFrameFilter &getFrameFilter();                                 
  void setRTSPServer(int portnum=8554);                              
  virtual void waitReady();                                          
 
public: // live555 events and tasks
  static void helloWorldEvent(void* clientData);   
  static void frameArrivedEvent(void* clientData); 
  static void gotFramesEvent(void* clientData);    
  static void readFrameFifoTask(void* clientData); 
 
public:  
  void testTrigger();         
  void triggerGotFrames();    
}; // <pyapi>
 
#endif