<commit_before>#include "chunkstream.cpp" //chunkstream decoding
#include "amf.cpp" //simple AMF0 parsing
std::string streamname = "/tmp/shared_socket";

//gets and parses one chunk
void parseChunk(){
  static chunkpack next;
  static AMFType amfdata("empty", (unsigned char)0xFF);
  static AMFType amfelem("empty", (unsigned char)0xFF);
  next = getWholeChunk();
  switch (next.msg_type_id){
    case 0://does not exist
      break;//happens when connection breaks unexpectedly
    case 1://set chunk size
      chunk_rec_max = ntohl(*(int*)next.data);
      #if DEBUG >= 4
      fprintf(stderr, "CTRL: Set chunk size: %i\n", chunk_rec_max);
      #endif
      break;
    case 2://abort message - we ignore this one
      #if DEBUG >= 4
      fprintf(stderr, "CTRL: Abort message\n");
      #endif
      //4 bytes of stream id to drop
      break;
    case 3://ack
      #if DEBUG >= 4
      fprintf(stderr, "CTRL: Acknowledgement\n");
      #endif
      snd_window_at = ntohl(*(int*)next.data);
      snd_window_at = snd_cnt;
      break;
    case 4:{
      #if DEBUG >= 4
      short int ucmtype = ntohs(*(short int*)next.data);
      fprintf(stderr, "CTRL: User control message %hi\n", ucmtype);
      #endif
      //2 bytes event type, rest = event data
      //types:
      //0 = stream begin, 4 bytes ID
      //1 = stream EOF, 4 bytes ID
      //2 = stream dry, 4 bytes ID
      //3 = setbufferlen, 4 bytes ID, 4 bytes length
      //4 = streamisrecorded, 4 bytes ID
      //6 = pingrequest, 4 bytes data
      //7 = pingresponse, 4 bytes data
      //we don't need to process this
      SendCTL(3, rec_cnt);//send ack (msg 3)
    } break;
    case 5://window size of other end
      #if DEBUG >= 4
      fprintf(stderr, "CTRL: Window size\n");
      #endif
      rec_window_size = ntohl(*(int*)next.data);
      rec_window_at = rec_cnt;
      SendCTL(3, rec_cnt);//send ack (msg 3)
      break;
    case 6:
      #if DEBUG >= 4
      fprintf(stderr, "CTRL: Set peer bandwidth\n");
      #endif
      //4 bytes window size, 1 byte limit type (ignored)
      snd_window_size = ntohl(*(int*)next.data);
      SendCTL(5, snd_window_size);//send window acknowledgement size (msg 5)
      break;
    case 8:
      #if DEBUG >= 4
      fprintf(stderr, "Received audio data\n");
      #endif
      break;
    case 9:
      #if DEBUG >= 4
      fprintf(stderr, "Received video data\n");
      #endif
      break;
    case 15:
      #if DEBUG >= 4
      fprintf(stderr, "Received AFM3 data message\n");
      #endif
      break;
    case 16:
      #if DEBUG >= 4
      fprintf(stderr, "Received AFM3 shared object\n");
      #endif
      break;
    case 17:
      #if DEBUG >= 4
      fprintf(stderr, "Received AFM3 command message\n");
      #endif
      break;
    case 18:
      #if DEBUG >= 4
      fprintf(stderr, "Received AFM0 data message\n");
      #endif
      break;
    case 19:
      #if DEBUG >= 4
      fprintf(stderr, "Received AFM0 shared object\n");
      #endif
      break;
    case 20:{//AMF0 command message
      bool parsed = false;
      amfdata = parseAMF(next.data, next.real_len);
      #if DEBUG >= 4
      amfdata.Print();
      #endif
      if (amfdata.getContentP(0)->StrValue() == "connect"){
        #if DEBUG >= 4
        int tmpint;
        tmpint = amfdata.getContentP(2)->getContentP("videoCodecs")->NumValue();
        if (tmpint & 0x04){fprintf(stderr, "Sorensen video support detected\n");}
        if (tmpint & 0x80){fprintf(stderr, "H264 video support detected\n");}
        tmpint = amfdata.getContentP(2)->getContentP("audioCodecs")->NumValue();
        if (tmpint & 0x04){fprintf(stderr, "MP3 audio support detected\n");}
        if (tmpint & 0x400){fprintf(stderr, "AAC video support detected\n");}
        #endif
        SendCTL(6, rec_window_size, 0);//send peer bandwidth (msg 6)
        SendCTL(5, snd_window_size);//send window acknowledgement size (msg 5)
        SendUSR(0, 0);//send UCM StreamBegin (0), stream 0
        //send a _result reply
        AMFType amfreply("container", (unsigned char)0xFF);
        amfreply.addContent(AMFType("", "_result"));//result success
        amfreply.addContent(amfdata.getContent(1));//same transaction ID
//        amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
        amfreply.addContent(AMFType(""));//server properties
        amfreply.getContentP(2)->addContent(AMFType("fmsVer", "FMS/3,0,1,123"));//stolen from examples
        amfreply.getContentP(2)->addContent(AMFType("capabilities", (double)31));//stolen from examples
        amfreply.addContent(AMFType(""));//info
        amfreply.getContentP(3)->addContent(AMFType("level", "status"));
        amfreply.getContentP(3)->addContent(AMFType("code", "NetConnection.Connect.Success"));
        amfreply.getContentP(3)->addContent(AMFType("description", "Connection succeeded."));
        amfreply.getContentP(3)->addContent(AMFType("capabilities", (double)33));//from red5 server
        amfreply.getContentP(3)->addContent(AMFType("fmsVer", "PLS/1,0,0,0"));//from red5 server
        #if DEBUG >= 4
        amfreply.Print();
        #endif
        SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
        //send onBWDone packet
        //amfreply = AMFType("container", (unsigned char)0xFF);
        //amfreply.addContent(AMFType("", "onBWDone"));//result success
        //amfreply.addContent(AMFType("", (double)0));//zero
        //amfreply.addContent(AMFType("", (double)0, 0x05));//null
        //SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
        parsed = true;
      }//connect
      if (amfdata.getContentP(0)->StrValue() == "createStream"){
        //send a _result reply
        AMFType amfreply("container", (unsigned char)0xFF);
        amfreply.addContent(AMFType("", "_result"));//result success
        amfreply.addContent(amfdata.getContent(1));//same transaction ID
        amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
        amfreply.addContent(AMFType("", (double)1));//stream ID - we use 1
        #if DEBUG >= 4
        amfreply.Print();
        #endif
        SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
        SendUSR(0, 0);//send UCM StreamBegin (0), stream 0
        parsed = true;
      }//createStream<commit_msg>Contd<commit_after>#include "chunkstream.cpp" //chunkstream decoding
#include "amf.cpp" //simple AMF0 parsing
std::string streamname = "/tmp/shared_socket";

//gets and parses one chunk
void parseChunk(){
  static chunkpack next;
  static AMFType amfdata("empty", (unsigned char)0xFF);
  static AMFType amfelem("empty", (unsigned char)0xFF);
  next = getWholeChunk();
  switch (next.msg_type_id){
    case 0://does not exist
      break;//happens when connection breaks unexpectedly
    case 1://set chunk size
      chunk_rec_max = ntohl(*(int*)next.data);
      #if DEBUG >= 4
      fprintf(stderr, "CTRL: Set chunk size: %i\n", chunk_rec_max);
      #endif
      break;
    case 2://abort message - we ignore this one
      #if DEBUG >= 4
      fprintf(stderr, "CTRL: Abort message\n");
      #endif
      //4 bytes of stream id to drop
      break;
    case 3://ack
      #if DEBUG >= 4
      fprintf(stderr, "CTRL: Acknowledgement\n");
      #endif
      snd_window_at = ntohl(*(int*)next.data);
      snd_window_at = snd_cnt;
      break;
    case 4:{
      #if DEBUG >= 4
      short int ucmtype = ntohs(*(short int*)next.data);
      fprintf(stderr, "CTRL: User control message %hi\n", ucmtype);
      #endif
      //2 bytes event type, rest = event data
      //types:
      //0 = stream begin, 4 bytes ID
      //1 = stream EOF, 4 bytes ID
      //2 = stream dry, 4 bytes ID
      //3 = setbufferlen, 4 bytes ID, 4 bytes length
      //4 = streamisrecorded, 4 bytes ID
      //6 = pingrequest, 4 bytes data
      //7 = pingresponse, 4 bytes data
      //we don't need to process this
      SendCTL(3, rec_cnt);//send ack (msg 3)
    } break;
    case 5://window size of other end
      #if DEBUG >= 4
      fprintf(stderr, "CTRL: Window size\n");
      #endif
      rec_window_size = ntohl(*(int*)next.data);
      rec_window_at = rec_cnt;
      SendCTL(3, rec_cnt);//send ack (msg 3)
      break;
    case 6:
      #if DEBUG >= 4
      fprintf(stderr, "CTRL: Set peer bandwidth\n");
      #endif
      //4 bytes window size, 1 byte limit type (ignored)
      snd_window_size = ntohl(*(int*)next.data);
      SendCTL(5, snd_window_size);//send window acknowledgement size (msg 5)
      break;
    case 8:
      #if DEBUG >= 4
      fprintf(stderr, "Received audio data\n");
      #endif
      break;
    case 9:
      #if DEBUG >= 4
      fprintf(stderr, "Received video data\n");
      #endif
      break;
    case 15:
      #if DEBUG >= 4
      fprintf(stderr, "Received AFM3 data message\n");
      #endif
      break;
    case 16:
      #if DEBUG >= 4
      fprintf(stderr, "Received AFM3 shared object\n");
      #endif
      break;
    case 17:
      #if DEBUG >= 4
      fprintf(stderr, "Received AFM3 command message\n");
      #endif
      break;
    case 18:
      #if DEBUG >= 4
      fprintf(stderr, "Received AFM0 data message\n");
      #endif
      break;
    case 19:
      #if DEBUG >= 4
      fprintf(stderr, "Received AFM0 shared object\n");
      #endif
      break;
    case 20:{//AMF0 command message
      bool parsed = false;
      amfdata = parseAMF(next.data, next.real_len);
      #if DEBUG >= 4
      amfdata.Print();
      #endif
      if (amfdata.getContentP(0)->StrValue() == "connect"){
        #if DEBUG >= 4
        int tmpint;
        tmpint = amfdata.getContentP(2)->getContentP("videoCodecs")->NumValue();
        if (tmpint & 0x04){fprintf(stderr, "Sorensen video support detected\n");}
        if (tmpint & 0x80){fprintf(stderr, "H264 video support detected\n");}
        tmpint = amfdata.getContentP(2)->getContentP("audioCodecs")->NumValue();
        if (tmpint & 0x04){fprintf(stderr, "MP3 audio support detected\n");}
        if (tmpint & 0x400){fprintf(stderr, "AAC video support detected\n");}
        #endif
        SendCTL(6, rec_window_size, 0);//send peer bandwidth (msg 6)
        SendCTL(5, snd_window_size);//send window acknowledgement size (msg 5)
        SendUSR(0, 1);//send UCM StreamBegin (0), stream 1
        //send a _result reply
        AMFType amfreply("container", (unsigned char)0xFF);
        amfreply.addContent(AMFType("", "_result"));//result success
        amfreply.addContent(amfdata.getContent(1));//same transaction ID
//        amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
        amfreply.addContent(AMFType(""));//server properties
        amfreply.getContentP(2)->addContent(AMFType("fmsVer", "FMS/3,0,1,123"));//stolen from examples
        amfreply.getContentP(2)->addContent(AMFType("capabilities", (double)31));//stolen from examples
        amfreply.addContent(AMFType(""));//info
        amfreply.getContentP(3)->addContent(AMFType("level", "status"));
        amfreply.getContentP(3)->addContent(AMFType("code", "NetConnection.Connect.Success"));
        amfreply.getContentP(3)->addContent(AMFType("description", "Connection succeeded."));
        amfreply.getContentP(3)->addContent(AMFType("capabilities", (double)33));//from red5 server
        amfreply.getContentP(3)->addContent(AMFType("fmsVer", "PLS/1,0,0,0"));//from red5 server
        #if DEBUG >= 4
        amfreply.Print();
        #endif
        SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
        //send onBWDone packet
        //amfreply = AMFType("container", (unsigned char)0xFF);
        //amfreply.addContent(AMFType("", "onBWDone"));//result success
        //amfreply.addContent(AMFType("", (double)0));//zero
        //amfreply.addContent(AMFType("", (double)0, 0x05));//null
        //SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
        parsed = true;
      }//connect
      if (amfdata.getContentP(0)->StrValue() == "createStream"){
        //send a _result reply
        AMFType amfreply("container", (unsigned char)0xFF);
        amfreply.addContent(AMFType("", "_result"));//result success
        amfreply.addContent(amfdata.getContent(1));//same transaction ID
        amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
        amfreply.addContent(AMFType("", (double)1));//stream ID - we use 1
        #if DEBUG >= 4
        amfreply.Print();
        #endif
        SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
        SendUSR(0, 1);//send UCM StreamBegin (0), stream 1
        parsed = true;
      }//createStream
{'fraction_non_alphanumeric': 0.14454183266932272, 'fraction_numerical': 0.025179282868525898, 'mean_word_length': 2.630604570436795, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5806081', 'n_tokens_mistral': 4246, 'n_tokens_neox': 4071, 'n_words': 1029}