hydrar/mistserver
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Namespace refactor and documentation in the separate buffer files.

Browse source
This commit is contained in:
Erik Zandvliet 2013-03-28 15:42:46 +01:00
parent 9b41a07c2f
commit 1e2f1602f8
3 changed files with 376 additions and 332 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

170
src/buffer/buffer_stream.cpp
View file

@ -4,11 +4,13 @@
#include "buffer_stream.h"
#include <mist/timing.h>
/// Stores the globally equal reference.
Buffer::Stream * Buffer::Stream::ref = 0;
namespace Buffer {
///\brief Stores the singleton reference.
Stream * Stream::ref = 0;
/// Returns a globally equal reference to this class.
Buffer::Stream * Buffer::Stream::get(){
///\brief Returns a reference to the singleton instance of this class.
///\return A reference to the class.
Stream * Stream::get(){
static tthread::mutex creator;
if (ref == 0){
//prevent creating two at the same time
@ -19,17 +21,17 @@ Buffer::Stream * Buffer::Stream::get(){
creator.unlock();
}
return ref;
}
}
/// Creates a new DTSC::Stream object, private function so only one instance can exist.
Buffer::Stream::Stream(){
///\brief Creates a new DTSC::Stream object, private function so only one instance can exist.
Stream::Stream(){
Strm = new DTSC::Stream(5);
readers = 0;
writers = 0;
}
}
/// Do cleanup on delete.
Buffer::Stream::~Stream(){
///\brief Do cleanup on delete.
Stream::~Stream(){
tthread::lock_guard<tthread::mutex> guard(stats_mutex);
for (usersIt = users.begin(); usersIt != users.end(); usersIt++){
if (( * *usersIt).S.connected()){
@ -38,10 +40,11 @@ Buffer::Stream::~Stream(){
}
moreData.notify_all();
delete Strm;
}
}
/// Calculate and return the current statistics in JSON format.
std::string & Buffer::Stream::getStats(){
///\brief Calculate and return the current statistics.
///\return The current statistics in JSON format.
std::string & Stream::getStats(){
static std::string ret;
long long int now = Util::epoch();
unsigned int tot_up = 0, tot_down = 0, tot_count = 0;
@ -68,55 +71,66 @@ std::string & Buffer::Stream::getStats(){
ret = Storage.toString();
Storage["log"].null();
return ret;
}
}
/// Get a new DTSC::Ring object for a user.
DTSC::Ring * Buffer::Stream::getRing(){
///\brief Get a new DTSC::Ring object for a user.
///\return A new DTSC::Ring object.
DTSC::Ring * Stream::getRing(){
return Strm->getRing();
}
}
/// Drop a DTSC::Ring object.
void Buffer::Stream::dropRing(DTSC::Ring * ring){
///\brief Drop a DTSC::Ring object.
///\param ring The DTSC::Ring to be invalidated.
void Stream::dropRing(DTSC::Ring * ring){
Strm->dropRing(ring);
}
}
/// Get the (constant) header data of this stream.
std::string & Buffer::Stream::getHeader(){
///\brief Get the (constant) header data of this stream.
///\return A reference to the header data of the stream.
std::string & Stream::getHeader(){
return Strm->outHeader();
}
}
/// Set the IP address to accept push data from.
void Buffer::Stream::setWaitingIP(std::string ip){
///\brief Set the IP address to accept push data from.
///\param ip The new IP to accept push data from.
void Stream::setWaitingIP(std::string ip){
waiting_ip = ip;
}
}
/// Check if this is the IP address to accept push data from.
bool Buffer::Stream::checkWaitingIP(std::string ip){
///\brief Check if this is the IP address to accept push data from.
///\param ip The IP address to check.
///\return True if it is the correct address, false otherwise.
bool Stream::checkWaitingIP(std::string ip){
if (ip == waiting_ip || ip == "::ffff:" + waiting_ip){
return true;
}else{
std::cout << ip << " != (::ffff:)" << waiting_ip << std::endl;
return false;
}
}
}
/// Sets the current socket for push data.
bool Buffer::Stream::setInput(Socket::Connection S){
///\brief Sets the current socket for push data.
///\param S The new socket for accepting push data.
///\return True if succesful, false otherwise.
bool Stream::setInput(Socket::Connection S){
if (ip_input.connected()){
return false;
}else{
ip_input = S;
return true;
}
}
}
/// Gets the current socket for push data.
Socket::Connection & Buffer::Stream::getIPInput(){
///\brief Gets the current socket for push data.
///\return A reference to the push socket.
Socket::Connection & Stream::getIPInput(){
return ip_input;
}
}
/// Stores intermediate statistics.
void Buffer::Stream::saveStats(std::string username, Stats & stats){
///\brief Stores intermediate statistics.
///\param username The name of the user.
///\param stats The final statistics to store.
void Stream::saveStats(std::string username, Stats & stats){
tthread::lock_guard<tthread::mutex> guard(stats_mutex);
Storage["curr"][username]["connector"] = stats.connector;
Storage["curr"][username]["up"] = stats.up;
@ -124,17 +138,20 @@ void Buffer::Stream::saveStats(std::string username, Stats & stats){
Storage["curr"][username]["conntime"] = stats.conntime;
Storage["curr"][username]["host"] = stats.host;
Storage["curr"][username]["start"] = Util::epoch() - stats.conntime;
}
}
/// Stores final statistics.
void Buffer::Stream::clearStats(std::string username, Stats & stats, std::string reason){
///\brief Stores final statistics.
///\param username The name of the user.
///\param stats The final statistics to store.
///\param reason The reason for disconnecting.
void Stream::clearStats(std::string username, Stats & stats, std::string reason){
tthread::lock_guard<tthread::mutex> guard(stats_mutex);
if (Storage["curr"].isMember(username)){
Storage["curr"].removeMember(username);
#if DEBUG >= 4
#if DEBUG >= 4
std::cout << "Disconnected user " << username << ": " << reason << ". " << stats.connector << " transferred " << stats.up << " up and "
<< stats.down << " down in " << stats.conntime << " seconds to " << stats.host << std::endl;
#endif
#endif
}
Storage["log"][username]["connector"] = stats.connector;
Storage["log"][username]["up"] = stats.up;
@ -142,10 +159,10 @@ void Buffer::Stream::clearStats(std::string username, Stats & stats, std::string
Storage["log"][username]["conntime"] = stats.conntime;
Storage["log"][username]["host"] = stats.host;
Storage["log"][username]["start"] = Util::epoch() - stats.conntime;
}
}
/// Cleans up broken connections
void Buffer::Stream::cleanUsers(){
///\brief Clean up broken connections
void Stream::cleanUsers(){
bool repeat = false;
tthread::lock_guard<tthread::mutex> guard(stats_mutex);
do{
@ -169,21 +186,24 @@ void Buffer::Stream::cleanUsers(){
}
}
}while (repeat);
}
}
/// Blocks until writing is safe.
void Buffer::Stream::getWriteLock(){
///\brief Ask to obtain a write lock.
///
/// Blocks until writing is safe.
void Stream::getWriteLock(){
rw_mutex.lock();
writers++;
while (writers != 1 && readers != 0){
rw_change.wait(rw_mutex);
}
rw_mutex.unlock();
}
}
/// Drops a previously gotten write lock.
void Buffer::Stream::dropWriteLock(bool newpackets_available){
if (newpackets_available){
///\brief Drops a previously obtained write lock.
///\param newPacketsAvailable Whether new packets are available to update the index.
void Stream::dropWriteLock(bool newPacketsAvailable){
if (newPacketsAvailable){
if (Strm->getPacket(0).isMember("keyframe")){
stats_mutex.lock();
Strm->updateHeaders();
@ -194,47 +214,53 @@ void Buffer::Stream::dropWriteLock(bool newpackets_available){
writers--;
rw_mutex.unlock();
rw_change.notify_all();
if (newpackets_available){
if (newPacketsAvailable){
moreData.notify_all();
}
}
}
/// Blocks until reading is safe.
void Buffer::Stream::getReadLock(){
///\brief Ask to obtain a read lock.
///
///Blocks until reading is safe.
void Stream::getReadLock(){
rw_mutex.lock();
while (writers > 0){
rw_change.wait(rw_mutex);
}
readers++;
rw_mutex.unlock();
}
}
/// Drops a previously gotten read lock.
void Buffer::Stream::dropReadLock(){
///\brief Drops a previously obtained read lock.
void Stream::dropReadLock(){
rw_mutex.lock();
readers--;
rw_mutex.unlock();
rw_change.notify_all();
}
}
/// Retrieves a reference to the DTSC::Stream
DTSC::Stream * Buffer::Stream::getStream(){
///\brief Retrieves a reference to the DTSC::Stream
///\return A reference to the used DTSC::Stream
DTSC::Stream * Stream::getStream(){
return Strm;
}
}
/// Sets the buffer name.
void Buffer::Stream::setName(std::string n){
///\brief Sets the buffer name.
///\param n The new name of the buffer.
void Stream::setName(std::string n){
name = n;
}
}
/// Add a user to the userlist.
void Buffer::Stream::addUser(user * new_user){
///\brief Add a user to the userlist.
///\param newUser The user to be added.
void Stream::addUser(user * newUser){
tthread::lock_guard<tthread::mutex> guard(stats_mutex);
users.push_back(new_user);
}
users.push_back(newUser);
}
/// Blocks the thread until new data is available.
void Buffer::Stream::waitForData(){
///\brief Blocks the thread until new data is available.
void Stream::waitForData(){
tthread::lock_guard<tthread::mutex> guard(stats_mutex);
moreData.wait(stats_mutex);
}
}

2
src/buffer/buffer_stream.h
View file

@ -50,7 +50,7 @@ namespace Buffer {
/// Sets the buffer name.
void setName(std::string n);
/// Add a user to the userlist.
void addUser(user * new_user);
void addUser(user * newUser);
/// Blocks the thread until new data is available.
void waitForData();
/// Cleanup function

72
src/buffer/buffer_user.cpp
View file

@ -1,15 +1,19 @@
/// \file buffer_user.cpp
/// Contains code for buffer users.
#include "buffer_user.h"
#include "buffer_stream.h"
#include <sstream>
#include <stdlib.h> //for atoi and friends
int Buffer::user::UserCount = 0;
/// Creates a new user from a newly connected socket.
/// Also prints "User connected" text to stdout.
Buffer::user::user(Socket::Connection fd){
#include <sstream>
#include <stdlib.h>
namespace Buffer {
int user::UserCount = 0;
///\brief Creates a new user from a newly connected socket.
///
///Also prints "User connected" text to stdout.
///\param fd A connection to the user.
user::user(Socket::Connection fd){
S = fd;
MyNum = UserCount++;
std::stringstream st;
@ -22,25 +26,31 @@ Buffer::user::user(Socket::Connection fd){
Thread = 0;
gotproperaudio = false;
lastpointer = 0;
} //constructor
} //constructor
/// Drops held DTSC::Ring class, if one is held.
Buffer::user::~user(){
///\brief Drops held DTSC::Ring class, if one is held.
user::~user(){
Stream::get()->dropRing(myRing);
} //destructor
} //destructor
/// Disconnects the current user. Doesn't do anything if already disconnected.
/// Prints "Disconnected user" to stdout if disconnect took place.
void Buffer::user::Disconnect(std::string reason){
///\brief Disconnects the current user. Doesn't do anything if already disconnected.
///
///Prints "Disconnected user" to stdout if disconnect took place.
///\param reason The reason for disconnecting the user.
void user::Disconnect(std::string reason){
if (S.connected()){
S.close();
}
Stream::get()->clearStats(MyStr, lastStats, reason);
} //Disconnect
} //Disconnect
/// Tries to send the current buffer, returns true if success, false otherwise.
/// Has a side effect of dropping the connection if send will never complete.
bool Buffer::user::doSend(const char * ptr, int len){
///\brief Tries to send data to the user.
///
///Has a side effect of dropping the connection if send will never complete.
///\param ptr A pointer to the data that is to be sent.
///\param len The amount of bytes to be sent from this pointer.
///\return True if len bytes are sent, false otherwise.
bool user::doSend(const char * ptr, int len){
if ( !len){
return true;
} //do not do empty sends
@ -54,10 +64,12 @@ bool Buffer::user::doSend(const char * ptr, int len){
}
currsend += r;
return (currsend == len);
} //doSend
} //doSend
/// Try to send data to this user. Disconnects if any problems occur.
bool Buffer::user::Send(){
///\brief Try to send the current buffer.
///
///\return True if the send was succesful, false otherwise.
bool user::Send(){
if ( !myRing){
return false;
} //no ring!
@ -114,17 +126,22 @@ bool Buffer::user::Send(){
} //completed a send
Stream::get()->dropReadLock();
return true;
} //send
} //send
/// Default constructor - should not be in use.
Buffer::Stats::Stats(){
///\brief Default stats constructor.
///
///Should not be used.
Stats::Stats(){
up = 0;
down = 0;
conntime = 0;
}
}
/// Reads a stats string and parses it to the internal representation.
Buffer::Stats::Stats(std::string s){
///\brief Stats constructor reading a string.
///
///Reads a stats string and parses it to the internal representation.
///\param s The string of stats.
Stats::Stats(std::string s){
size_t f = s.find(' ');
if (f != std::string::npos){
host = s.substr(0, f);
@ -146,4 +163,5 @@ Buffer::Stats::Stats(std::string s){
s.erase(0, f + 1);
down = atoi(s.c_str());
}
}
}