rstpserver.h

/* libradar_stp
 * <http://www.k9nl.net/radar/> <radar@k9nl.net>
 * 
 * rstpserver.h - RSTP server class
 *
 * THIS FILE IS NOT MEANT FOR LIBRARY CLIENT USE
 * It is included in case it's helpful
 *
 * Copyright (C) 2007 Nicholas J. Luther
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License,
 * version 2, as published by the Free Software Foundation.
 * 
 * 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 General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 * You may contact the author by writing to Nick Luther, 1655 South
 * Westhaven Drive, Oshkosh, Wisconsin 54904, USA.
 * 
 */
 
#ifndef _LIBRSTP_INC_RSTPSERVER_H
#define _LIBRSTP_INC_RSTPSERVER_H
 
#include <string>
#include <set>
#include <queue>
#include <list>
#include <map>
#include <vector>

#include "lplproj.h"
#include "thread/thread.h"
#include "thread/mutex.h"
#include "socket/tcpsocket.h"
#include "filter.h"
#include "loghandler.h"



namespace libradar_stp {

// CAUTION: be careful copying RSTPServers
class 
RSTPServer : public LPORTLIB_NAMESPACE::thread::Thread {
public:
  
  RSTPServer();
  ~RSTPServer();
  
  bool listen( const std::string& local_addr, unsigned short port, int backlog = 10 );
  bool listen( unsigned short port );
  
  bool notifyData( const char* data, unsigned int size );
  
  void quit();
  
  void setAuthRequired();
  void setAuthCredential( const std::string& login, const std::string& pass );
  void setAuthPermission( const std::string& login, const std::string& stream );
  void clearAuthSettings();
  void setMaxClients( unsigned int max_clients = 0 );
  
  void setFilterStream( const std::string& stream, const std::string& file );
  void setPassStream( const std::string& stream );
  void setPipeThrough( const std::string& stream, const std::string& cmd );
  
  void setStreamCache( const std::string& stream, int cache_size, const std::string& file );
  
  typedef void (*STRCB)( void*, const std::string& );
  void setStreamAddCallback( STRCB func, void* arg = 0 );
  void setStreamDropCallback( STRCB func, void* arg = 0 );
  
  static std::string pipeThroughCmd( const std::string& data,
      const std::string& cmd );
  static std::string getTempPath( unsigned int seed = 0 );
  
  void addBroadcastData( const std::string& data );
  void addBroadcastFile( const std::string& file );
  void setBroadcastInterval( unsigned int interval );
  
  void setLogging( const std::string& file, int level = 1, std::string progid = "" );
  void setStatusFile( const std::string& file );
  
protected:

  void beginThread();
  
private:

  void cleanup();

  bool m_quit_request;
  std::queue< std::string > m_data_queue;
  LPORTLIB_NAMESPACE::thread::Mutex m_data_queue_mutex;
  LPORTLIB_NAMESPACE::socket::TCPSocket m_sock;
  
  bool m_auth_required;
  std::map< std::string, std::string > m_auth_db;
  LPORTLIB_NAMESPACE::thread::Mutex m_auth_db_mutex;
  std::map< std::string, std::set<std::string> > m_auth_streams;
  LPORTLIB_NAMESPACE::thread::Mutex m_auth_streams_mutex;
  
  std::set< std::string > m_pass_streams;
  std::map< std::string, Filter > m_filter_streams;
  std::map< std::string, std::string > m_pipe_streams;
  
  unsigned int m_max_clients;
  
  class ClientHelper : public LPORTLIB_NAMESPACE::thread::Thread {
  public:
    ClientHelper( RSTPServer* parent, const LPORTLIB_NAMESPACE::socket::TCPSocket& sock );
    ~ClientHelper();
    void notifyData( const std::set<std::string>& streams, const char* data, unsigned int size );
    void quit();
    void join();
    LogHandler m_log;
  protected:
    void beginThread();
  private:
    struct queue_datum { std::string m_stream; std::string m_data; };
    //std::queue<std::string> m_data_queue;
    std::queue<queue_datum> m_data_queue;
    LPORTLIB_NAMESPACE::thread::Mutex m_data_queue_mutex;
    std::set<std::string> m_streams;
    LPORTLIB_NAMESPACE::thread::Mutex m_streams_mutex;
    LPORTLIB_NAMESPACE::socket::TCPSocket m_sock;
    std::string m_login;
    bool m_quit_request;
    unsigned int m_lh;
    bool m_nop;
    RSTPServer* m_parent;
    bool m_authenticated;
    bool m_supp_bzip2;
    bool m_supp_zlib;
  }; // class ClientHelper
  friend class ClientHelper;
  
  std::set<ClientHelper*> m_clients;
  LPORTLIB_NAMESPACE::thread::Mutex m_clients_mutex;
  std::queue<ClientHelper*> m_close_queue;
  LPORTLIB_NAMESPACE::thread::Mutex m_close_queue_mutex;
  
  void notifyClose( ClientHelper* helper );
  
  std::map<std::string,unsigned int> m_stream_count;
  LPORTLIB_NAMESPACE::thread::Mutex m_stream_count_mutex;
  
  void notifyStreamAdd( const std::string& stream, ClientHelper* ch = 0 );
  void notifyStreamDrop( const std::string& stream );
  
  STRCB m_stream_add_cb;
  STRCB m_stream_drop_cb;
  void* m_stream_add_arg;
  void* m_stream_drop_arg;
  
  std::vector<std::string> m_broadcast_data;
  std::vector<std::string> m_broadcast_files;
  unsigned int m_broadcast_time;
  unsigned int m_last_broadcast;
  void processBroadcastFile( const std::string& file );
  
  LogHandler m_log;
  
  struct ConnStatus {
    string m_addr;
    string m_conn_time;
    string m_login;
    string m_streams;
  }; // struct
  std::map<int,ConnStatus> m_status;
  std::string              m_status_file;
  void reportConnect( int id, string addr );
  void reportLogin( int id, string login );
  void reportStreams( int id, const std::set<std::string>& streams );
  void reportDisconnect( int id );
  void writeReport();
  void reportMinLen( std::string& s, int len ) const;
  LPORTLIB_NAMESPACE::thread::Mutex m_status_mutex;
  
  
  // a thread-safe class!
  class CacheCtrl {
    public:
      CacheCtrl();
      ~CacheCtrl();
      
      bool ok() const;
      
      void setFilter( const Filter& filter );
      Filter getFilter();
      
      void setMaxSize( int max_size );
      int getMaxSize() const;
      
      void setStream( std::string stream );
      std::string getStream();
      
      void notifyData( const char* data, unsigned int size );
      void pushCacheData( ClientHelper* ch );
      
    private:
      int m_max_size;
      std::string m_stream;
      Filter m_filter;
      std::map< std::string, std::list< std::string > > m_cache;
      LPORTLIB_NAMESPACE::thread::Mutex m_mutex;
  }; // class CacheCtrl
  friend class CacheCtrl;
  std::map< std::string, CacheCtrl > m_cache_ctrl;

}; // class


};// namespace

#endif

---