comparison core/ReceiveAudioThread.cpp @ 119:c692827083e1 scope-refactoring

Enabled multi channel audio receive

 #include "ReceiveAudioThread.h"
+
+#ifdef JUCE
+#else
 //initialise static members
+bool ReceiveAudioThread::staticConstructed=false;
 AuxiliaryTask ReceiveAudioThread::receiveDataTask=NULL;
+std::vector<ReceiveAudioThread *> ReceiveAudioThread::objAddrs(0);
+bool ReceiveAudioThread::threadRunning;
+bool ReceiveAudioThread::threadIsExiting;
+int ReceiveAudioThread::sleepTime;
+
+void receiveData(){
+	ReceiveAudioThread::run();
+}
+void ReceiveAudioThread::staticConstructor(){
+	if(staticConstructed==true)
+		return;
+	staticConstructed=true;
+    threadIsExiting=false;
+	receiveDataTask=BeagleRT_createAuxiliaryTask(receiveData, 90, "receiveDataTask"); //TODO: allow different priorities
+}
+#endif
 
 void ReceiveAudioThread::dealloc(){
     free(buffer);
     buffer=NULL;
     free(stackBuffer);

                                                         // but it would have made it more difficult for the thread to be killed
         pushPayload(writePointer); //backup headerLength samples. This could be skipped if writePointer==0
         //read header+payload
 //JUCE        int numBytes=socket.read(buffer+writePointer, bytesToRead,1);
         int numBytes=socket.read(buffer+writePointer, bytesToRead);
-            //TODO: do something with the data in the header (e.g.: check that timestamp is sequential,
-            // check the channel number)
             //TODO: (if using variable-length payload) validate the actual numBytes read against the size declared in the header
         if(numBytes<0){
             printf("error numBytes1\n");
             return -3; //TODO: something went wrong, you have to discard the rest of the packet!
         }
         if(numBytes==0){//TODO: when inplementing waitUntilReady, this should not happen unless you actually receive a packate of size zero (is it at all possible?)
 //        	printf("received 0 bytes\n");
         	return 0;
         }
-        if(numBytes != bytesToRead){ //this is equivalent to (numBytes<numBytes)
+        if(numBytes != bytesToRead){ //this is equivalent to (numBytes<bytesToRead)
             printf("error numBytes2: %d\n", numBytes);
             return -4; //TODO: something went wrong, less bytes than expected in the payload.
         }
-        if(buffer[writePointer]!=0)
-        	return 0; //wrong channel
-        //  printf("Received a message of length %d, it was on channel %d and timestamp %d\n", numBytes, (int)buffer[writePointer], (int)buffer[writePointer+1]);
+        if(channel!=(int)buffer[writePointer]){
+//        	printf("I am channel %d, but I received data for channel %d\n", channel, (int)buffer[writePointer]);
+        	return -5;
+        }
+		//TODO: do something else with the data in the header (e.g.: check that timestamp is sequential)
+//        rt_printf("Received a message of length %d, it was on channel %d and timestamp %d\n", numBytes, (int)buffer[writePointer], (int)buffer[writePointer+1]);
         popPayload(writePointer); //restore headerLength payload samples. This could be skipped if writePointer==0
 
         //even though we just wrote (payloadLength+headerLength) samples in the buffer,
         //we only increment by payloadLength. This way, next time a socket.read is performed, we will
         //backup the last headerLength samples that we just wrote and we will overwrite them with
         //the header from the new read. After parsing the header we will then restore the backed up samples.
         //This way we guarantee that, apart from the first headerLength samples, buffer is a circular buffer!
-        printf("writepointer:%d\n", writePointer);
+//        printf("writepointer:%d\n", writePointer);
         writePointer+=payloadLength;
 
         if(writePointer>lastValidPointer){
             //  lastValidPointer=writePointer+headerLength;
         }

 ReceiveAudioThread::ReceiveAudioThread() :
 //JUCE    Thread(threadName),
     socket(0),
     listening(false),
     bufferReady(false),
-    threadIsExiting(false),
     buffer(NULL),
     stackBuffer(NULL),
     bufferLength(0),
     lastValidPointer(0),
-    sleepTime(2000),
     waitForSocketTime(100),
     threadPriority(95)
 {};
 ReceiveAudioThread::~ReceiveAudioThread(){
 //JUCE    stopThread(1000);
 	while(threadRunning){
-		sleep(sleepTime*2);	//wait for thread to stop
+		usleep(sleepTime*2);	//wait for thread to stop
 		std::cout<< "Waiting for receiveAudioTask to stop" << std::endl;
 	}
 	//TODO: check if thread stopped, otherwise kill it before dealloc
     dealloc();
 }
-ReceiveAudioThread *gRAT;
-void receiveData(){
-	gRAT->run();
-}
-void ReceiveAudioThread::init(int aSamplesPerBlock){
+void ReceiveAudioThread::init(int aPort, int aSamplesPerBlock, int aChannel){
   dealloc();
+#ifdef JUCE
+#else
+  staticConstructor();
+  objAddrs.push_back(this);//TODO: this line should be in the constructor
+#endif
+  bindToPort(aPort);
+  channel=aChannel;
   //  fd=fopen("output.m","w"); //DEBUG
   //  fprintf(fd,"var=["); //DEBUG
-  gRAT=this;
   headerLength=2;
   payloadLength=300; //TODO: make sure that payloadLength and headerLength are the same as the client is sending.
   bufferLength=std::max(headerLength+(payloadLength*4), headerLength+(aSamplesPerBlock*4)); //there are many considerations that can be done here ...
                       //We keep a headerLength padding at the beginning of the array to allow full reads from the socket
   buffer=(float*)malloc(sizeof(float)*bufferLength);

   stackBuffer=(float*)malloc(sizeof(float)*headerLength);
   bytesToRead=sizeof(float)*(payloadLength + headerLength);
   writePointer=-1;
   readPointer=0; //TODO: this *4 is sortof a security margin
   sleepTime=payloadLength/(float)44100 /4.0; //set sleepTime so that you do not check too often or too infrequently
-	receiveDataTask=BeagleRT_createAuxiliaryTask( receiveData, threadPriority, "receiveDataTask");
 //JUCE    startThread(threadPriority);
 }
 
 void ReceiveAudioThread::bindToPort(int aPort){
     listening=socket.bindToPort(aPort);
+#ifdef JUCE
+#else
+    if(listening==false)
+    	printf("Could not bind to port %d\n",aPort);
+#endif
 }
 bool ReceiveAudioThread::isListening(){
     return listening;
 }
 float* ReceiveAudioThread::getCurrentBuffer(int length){ // NOTE: this cannot work all the time unless samplesPerBuffer and payloadLength are multiples

 void ReceiveAudioThread::run(){
     //  fd2=fopen("buffer.m","w"); //DEBUG
     //  fprintf(fd2, "buf=["); //DEBUG
 	threadRunning=true;
     while(!threadShouldExit()){ //TODO: check that the socket buffer is empty before starting
+#ifdef JUCE
         readUdpToBuffer(); // read into the oldBuffer
         usleep(sleepTime);
+#else
+		for(unsigned int n=0; n<ReceiveAudioThread::objAddrs.size(); n++){
+			ReceiveAudioThread::objAddrs[n]->readUdpToBuffer();
+		}
+		usleep(sleepTime); //TODO: use rt_task_sleep instead
+#endif
     }
     threadRunning=false;
     //  fprintf(fd,"];readPointer,writePointer,lastValidPointer,destination]=deal(var(:,1), var(:,2), var(:,3), var(:,4));"); //DEBUG
     //  fclose(fd);//DEBUG
     //  fprintf(fd2,"];");//DEBUG