Ringbuffer werkend. Denk ik. Testing needed. Also - commentaar toegevoegd en zo

Server/main.cpp

@@ -6,93 +6,104 @@
 #include <cstdio>
 #include "buffer.h"
 #include "user.h"
+#include "string.h"
 
 #define BUFLEN 1000000
 
 int get_empty( user ** list, int amount ) {
   for (int i = 0; i < amount; i++ ){
-    if (!list[i]->is_connected()) { return i; }
+    if (!list[i]->is_connected){return i;}
   }
   return -1;
 }
 
 int main( int argc, char * argv[] ) {
-  if (argc < 3) { std::cout << "Not the right amount of arguments!\n"; exit(1);}
+  if (argc < 4) {
+    std::cout << "usage: " << argv[0] << " buffers_count total_buffersize max_clients" << std::endl;
+    return 1;
+  }
   int buffers = atoi(argv[1]);
   int total_buffersize = atoi(argv[2]);
+  int connections = atoi(argv[3]);
   int size_per_buffer = total_buffersize/buffers;
-  std::cout << "Size per buffer: " << size_per_buffer << "\n";
+  std::cout << "Size per buffer: " << size_per_buffer << std::endl;
   buffer ** ringbuf = (buffer**) calloc (buffers,sizeof(buffer*));
   for (int i = 0; i < buffers; i ++ ) {
     ringbuf[i] = new buffer;
     ringbuf[i]->data = (char*) malloc(size_per_buffer);
   }
-  for (int i = 0; i < buffers; i ++ ) { std::cout << "Buffer[" << i << "][0]: " << ringbuf[i]->data[0] << "\n"; }
+  std::cout << "Successfully allocated " << total_buffersize << " bytes total buffer." << std::endl;
-  int connections = 2;
   user ** connectionList = (user**) calloc (connections,sizeof(user*));
   for (int i = 0; i < connections; i++) { connectionList[i] = new user; }
-  char input[BUFLEN];
+  char header[13];//FLV header is always 13 bytes
-  char header[BUFLEN];
+  int ret = 0;
-  int inp_amount;
-  int cur_header_pos;
-  int position_current = 0;
-  int position_startframe = 0;
   int frame_bodylength = 0;
   int current_buffer = 0;
   int open_connection = -1;
+  int lastproper = 0;//last properly finished buffer number
   unsigned int loopcount = 0;
   SWUnixSocket listener;
-  SWUnixSocket *mySocket = NULL;
+  SWBaseSocket * incoming = 0;
   SWBaseSocket::SWBaseError BError;
-  cur_header_pos = fread(&header,1,13,stdin);
+  //read FLV header - 13 bytes
+  ret = fread(&header,1,13,stdin);
+  //TODO: check ret?
 
   listener.bind("/tmp/socketfile");
   listener.listen();
   listener.set_timeout(0,50000);
   
+  //TODO: not while true, but while running - set running to false when kill signal is received!
   while(true) {
     loopcount ++;
-    inp_amount = fread(&input,1,11,stdin);
+    //invalidate the current buffer
-    if (input[0] == 9) {
+    ringbuf[current_buffer]->size = 0;
+    ringbuf[current_buffer]->number = -1;
+    //read FLV frame header - 11 bytes
+    ret = fread(ringbuf[current_buffer]->data,1,11,stdin);
+    //TODO: Check ret?
+    //if video frame? (id 9) check for incoming connections
+    if (ringbuf[current_buffer]->data[0] == 9) {
+      incoming = listener.accept(&BError);
+      if (incoming){
         open_connection = get_empty(connectionList,connections);
         if (open_connection != -1) {
-        connectionList[open_connection]->connect( (SWUnixSocket *)listener.accept(&BError) );
+          connectionList[open_connection]->connect(incoming);
-        if (connectionList[open_connection]->is_connected()) {
+          //send the FLV header
-          std::cout << "#" << loopcount << ": ";
+          std::cout << "Client " << open_connection << " connected." << std::endl;
-          std::cout << "!!Client " << open_connection << " connected.!!\n";
+          connectionList[open_connection]->MyBuffer = lastproper;
-          connectionList[open_connection]->send_msg(&header[0],13,NULL);
+          connectionList[open_connection]->MyBuffer_num = ringbuf[lastproper]->number;
-        }
+          //TODO: Do this more nicely?
-      }
+          if (connectionList[open_connection]->Conn->send(&header[0],13,NULL) != 13){
-    }
+            connectionList[open_connection]->disconnect();
-    position_current = 0;
+            std::cout << "Client " << open_connection << " failed to receive the header!" << std::endl;
-    position_startframe = position_current;
+          }
-    for(int i = 0; i < 11; i++) { ringbuf[current_buffer]->data[position_current] = input[i]; position_current ++; }
+          std::cout << "Client " << open_connection << " received header!" << std::endl;
-    frame_bodylength = 0;
+        }else{
-    frame_bodylength += input[3];
+          std::cout << "New client not connected: no more connections!" << std::endl;
-    frame_bodylength += (input[2] << 8);
-    frame_bodylength += (input[1] << 16);
-    for (int i = 0; i < frame_bodylength + 4; i++) {
-      inp_amount = fread(&input,1,1,stdin);
-      ringbuf[current_buffer]->data[position_current] = input[0];
-      position_current ++;
-    }
-    ringbuf[current_buffer]->size = position_current;
-//    std::cout << "Total message read!\n";
-    for (int i = 0; i < connections; i++) {
-      if (connectionList[i]->is_connected()) {
-//        std::cout << "Client " << i << " connected, sending..\n";
-        if ( connectionList[i]->myConnection->send(&ringbuf[current_buffer]->data[0], ringbuf[current_buffer]->size, &BError) == -1) {
-          connectionList[i]->disconnect();
-          std::cout << "#" << loopcount << ": ";
-          std::cout << "!!Client " << i << " disconnected.!!\n";
         }
       }
     }
+    //calculate body length of frame
+    frame_bodylength = 4;
+    frame_bodylength += ringbuf[current_buffer]->data[3];
+    frame_bodylength += (ringbuf[current_buffer]->data[2] << 8);
+    frame_bodylength += (ringbuf[current_buffer]->data[1] << 16);
+    //read the rest of the frame
+    ret = fread(&ringbuf[current_buffer]->data[11],1,frame_bodylength,stdin);
+    //TODO: Check ret?
+    ringbuf[current_buffer]->size = frame_bodylength + 11;
+    ringbuf[current_buffer]->number = loopcount;
+    //send all connections what they need, if and when they need it
+    for (int i = 0; i < connections; i++) {connectionList[i]->Send(ringbuf, buffers);}
+    //keep track of buffers
+    lastproper = current_buffer;
     current_buffer++;
-    current_buffer = current_buffer % buffers;
+    current_buffer %= buffers;
   }
 
-  // disconnect and clean up
+  // disconnect listener
   listener.disconnect();
+  //TODO: cleanup
   return 0;
 }

Server/user.cpp

@@ -1,36 +1,51 @@
 #include "user.h"
 
 user::user() {
-  myBuffer = NULL;
+  Conn = NULL;
-  myConnection = NULL;
+  is_connected = false;
 }
 
 user::~user() {
-  myConnection->disconnect();
+  Conn->disconnect();
-  myConnection = NULL;
+  Conn = NULL;
-}
+  is_connected = false;
-
-void user::set_buffer(buffer * newBuffer) {
-  myBuffer = newBuffer;
-  sent = 0;
-}
-
-int user::get_number() { return myBuffer->number; }
-
-bool user::complete_send() {
-  if (sent == myBuffer->size) { return true; }
-  return false;
 }
 
 void user::disconnect() {
-  if (myConnection) {
+  if (Conn) {
-    myConnection->disconnect();
+    Conn->disconnect();
-    myConnection = NULL;
+    Conn = NULL;
   }
+  std::cout << "Disconnected user" << std::endl;
+  is_connected = false;
+}
+void user::connect(SWBaseSocket * newConn) {
+  Conn = (SWUnixSocket*)newConn;
+  is_connected = (Conn != 0);
 }
 
-void user::connect(SWUnixSocket * newConnection) { myConnection = newConnection; }
+void user::Send(buffer ** ringbuf, int buffers){
-
+  //not connected? cancel
-bool user::is_connected( ) { return myConnection; }
+  if (!is_connected){return;}
-
+  //still waiting for next buffer? check it
-int user::send_msg(char * message, int length, SWBaseSocket::SWBaseError * BError) { return myConnection->send(message,length,BError); }
+  if (MyBuffer_num < 0){MyBuffer_num = ringbuf[MyBuffer]->number;}
+  //still waiting? don't crash - wait longer.
+  if (MyBuffer_num < 0){return;}
+  //buffer number changed? disconnect
+  if ((ringbuf[MyBuffer]->number != MyBuffer_num)){
+    disconnect();
+    return;
+  }
+  SWBaseSocket::SWBaseError err;
+  int ret = Conn->fsend(ringbuf[MyBuffer]->data, ringbuf[MyBuffer]->size, &err);
+  if ((err != SWBaseSocket::ok) && (err != SWBaseSocket::notReady)){
+    disconnect();
+    return;
+  }
+  if (ret == ringbuf[MyBuffer]->size){
+    //completed a send - switch to next buffer
+    MyBuffer++;
+    MyBuffer %= buffers;
+    MyBuffer_num = -1;
+  }
+}

Server/user.h

@@ -7,16 +7,12 @@ class user{
   public:
     user();
     ~user();
-    void set_buffer(buffer * newBuffer);
-    int get_number();
-    bool complete_send();
     void disconnect();
-    void connect(SWUnixSocket * newConnection);
+    void connect(SWBaseSocket * newConnection);
-    bool is_connected();
+    void Send(buffer ** ringbuf, int buffers);
-    int send_msg(char * message, int length, SWBaseSocket::SWBaseError * BError);
+    bool is_connected;
-    int sent;
+    SWUnixSocket * Conn;
-    buffer * myBuffer;
+    int MyBuffer;
-    SWUnixSocket * myConnection;
+    int MyBuffer_num;
-  private:
 };//user