Mix transferring data model for playback and capture

src/audio/audiortp.cpp

@@ -267,11 +267,11 @@ AudioRtpRTX::sendSessionFromMic(int timestamp)
     int maxBytesToGet = _layerSampleRate * _layerFrameSize * sizeof(SFLDataFormat) / 1000;
     // available bytes inside ringbuffer
     int availBytesFromMic = audiolayer->canGetMic();
-    //printf("%i \n", availBytesFromMic);
+    //printf("availBytesFromMic = %i \n", availBytesFromMic);
 
     // take the lowest
     int bytesAvail = (availBytesFromMic < maxBytesToGet) ? availBytesFromMic : maxBytesToGet;
-    //printf("%i\n", bytesAvail);
+    //printf("max bytes to get: %i\n", bytesAvail);
     // Get bytes from micRingBuffer to data_from_mic
     int nbSample = audiolayer->getMic(_dataAudioLayer, bytesAvail) / sizeof(SFLDataFormat);
     int nb_sample_up = nbSample;

src/audio/audiostream.cpp

@@ -81,7 +81,7 @@ AudioStream::createStream( pa_context* c )
 				pa_cvolume_set(&cv, sample_spec.channels , volume) , NULL );
   }
   else if( _streamType == CAPTURE_STREAM ){
-    pa_stream_connect_record( s , NULL , NULL , flag );
+    pa_stream_connect_record( s , NULL , NULL , PA_STREAM_START_CORKED );
   }
   else if( _streamType == UPLOAD_STREAM ){
     pa_stream_connect_upload( s , 1024  );

src/audio/pulselayer.cpp

@@ -19,6 +19,10 @@
 
 #include "pulselayer.h"
 
+static pa_channel_map channel_map ;
+static pa_stream_flags_t flag;
+static pa_sample_spec sample_spec ;
+static pa_volume_t volume;
 
 int framesPerBuffer = 4096;
 const pa_buffer_attr *a;
@@ -35,8 +39,6 @@ const pa_buffer_attr *a;
 // Destructor
 PulseLayer::~PulseLayer (void) 
 { 
-  delete playback;
-  delete record;
   ////delete cache;
   pa_context_disconnect(context);
 }
@@ -66,8 +68,7 @@ PulseLayer::connectPulseServer( void )
   _debug("Context creation done\n");
 }
 
-  void 
-PulseLayer::context_state_callback( pa_context* c, void* user_data )
+void PulseLayer::context_state_callback( pa_context* c, void* user_data )
 {
   _debug("The state of the context changed\n");
   PulseLayer* pulse = (PulseLayer*)user_data;
@@ -93,16 +94,64 @@ PulseLayer::context_state_callback( pa_context* c, void* user_data )
   }
 }
 
+  void 
+PulseLayer::stream_state_callback( pa_stream* s, void* user_data )
+{
+  _debug("The state of the stream changed\n");
+  assert(s);
+  switch(pa_stream_get_state(s)){
+    case PA_STREAM_CREATING:
+    case PA_STREAM_TERMINATED:
+      _debug("Stream is creating...\n");
+      break;
+    case PA_STREAM_READY:
+      _debug("Stream successfully created\n");
+      break;
+    case PA_STREAM_FAILED:
+    default:
+      _debug("Stream error: %s\n" , pa_strerror(pa_context_errno(pa_stream_get_context(s))));
+      break;
+  }
+}
+
   void
 PulseLayer::createStreams( pa_context* c )
 {
-  playback = new AudioStream(c, PLAYBACK_STREAM, "SFLphone out");
-  pa_stream_set_write_callback( playback->pulseStream() , audioCallback, this);
-  pa_stream_set_overflow_callback( playback->pulseStream() , overflow , this);
-  record = new AudioStream(c, CAPTURE_STREAM, "SFLphone in");
-  pa_stream_set_read_callback( record->pulseStream() , audioCallback, this);
-  pa_stream_set_underflow_callback( playback->pulseStream() , underflow , this);
-  cache = new AudioStream(c, UPLOAD_STREAM, "Cache samples");
+  _debug("Creating streams...\n");
+  sample_spec.format = PA_SAMPLE_S16LE; 
+  sample_spec.rate = 44100; 
+  sample_spec.channels = 1; 
+  channel_map.channels = 1; 
+  flag = PA_STREAM_AUTO_TIMING_UPDATE ;
+  volume = PA_VOLUME_NORM;
+
+  pa_cvolume cv;
+
+  assert(pa_sample_spec_valid(&sample_spec));
+  assert(pa_channel_map_valid(&channel_map));
+
+  if( !( playback = pa_stream_new( c, "SFLphone OUT" , &sample_spec, &channel_map ) ) ) 
+    _debug("Playback: pa_stream_new() failed : %s\n" , pa_strerror( pa_context_errno( c)));
+  assert( playback );
+  pa_stream_set_write_callback( playback , audioCallback, this);
+  pa_stream_connect_playback( playback , NULL , NULL , 
+      PA_STREAM_INTERPOLATE_TIMING,
+      pa_cvolume_set(&cv, sample_spec.channels , volume) , NULL );
+
+  if( !( record = pa_stream_new( c, "SFLphone IN" , &sample_spec, &channel_map ) ) ) 
+    _debug("Record: pa_stream_new() failed : %s\n" , pa_strerror( pa_context_errno( c)));
+  assert( record );
+  pa_stream_connect_record( record , NULL , NULL , PA_STREAM_INTERPOLATE_TIMING );
+
+  pa_stream_set_state_callback( record , stream_state_callback, NULL);
+  pa_stream_set_state_callback( playback , stream_state_callback, NULL);
+
+  //playback = new AudioStream(c, PLAYBACK_STREAM, "SFLphone out");
+  pa_stream_set_overflow_callback( playback , overflow , this);
+  //record = new AudioStream(c, CAPTURE_STREAM, "SFLphone in");
+  pa_stream_set_read_callback( record , audioCallback, this);
+  pa_stream_set_underflow_callback( record , underflow , this);
+  //cache = new AudioStream(c, UPLOAD_STREAM, "Cache samples");
 
   pa_threaded_mainloop_signal(m , 0);
 
@@ -187,9 +236,11 @@ PulseLayer::canGetMic()
 {
   if( record )
   {
-    int a = _micRingBuffer.AvailForGet();
-    _debug("available for get = %i\n" , a);
-    return a;
+    //int a = _micRingBuffer.AvailForGet();
+    //return a;
+    int a = pa_stream_readable_size( record );
+    _debug("available: %i\n" , a);
+    return 2048;
   }
   else
     return 0;
@@ -199,7 +250,8 @@ PulseLayer::canGetMic()
 PulseLayer::getMic(void *buffer, int toCopy)
 {
   if( record ){
-    return _micRingBuffer.Get(buffer, toCopy, 100);
+    //return _micRingBuffer.Get(buffer, toCopy, 100);
+    return readbuffer( buffer, toCopy );
   }
   else
     return 0;
@@ -220,6 +272,8 @@ PulseLayer::flushMic()
 PulseLayer::startStream (void) 
 {
   _debug("Start stream\n");
+  //pa_stream_cork( record, NULL, NULL, NULL);
+  //pa_stream_cork( playback, NULL, NULL, NULL);
 }
 
   void 
@@ -241,7 +295,7 @@ PulseLayer::audioCallback ( pa_stream* s, size_t bytes, void* userdata )
   assert( s && bytes );
   assert( bytes > 0 );
   pulse->write();
-  pulse->read();
+  //pulse->read();
 }
 
   void 
@@ -260,9 +314,9 @@ PulseLayer::overflow ( pa_stream* s, void* userdata )
   void
 PulseLayer::write( void )
 {
- // a =  pa_stream_get_buffer_attr(record->pulseStream());
+  // a =  pa_stream_get_buffer_attr(record->pulseStream());
   //if (a)
-    //_debug("Buffer attributes: maxlength=%u , fragsize=%u\n" , a->maxlength, a->fragsize );
+  //_debug("Buffer attributes: maxlength=%u , fragsize=%u\n" , a->maxlength, a->fragsize );
   //_debug("Device name = %s ; device index = %i\n" , pa_stream_get_device_name( playback->pulseStream()) , pa_stream_get_device_index( playback->pulseStream()));
   int toGet; 
   int urgentAvail; // number of data right and data left  
@@ -284,48 +338,50 @@ PulseLayer::write( void )
       //tone->getNext(out, framesPerBuffer, 100);
       toGet = framesPerBuffer;
     } /*else if ( (tone=_manager->getTelephoneFile()) != 0 ) {
-      tone->getNext(out, framesPerBuffer, 100);
-      toGet = framesPerBuffer;
-    } */else {
-      normalAvail = _mainSndRingBuffer.AvailForGet();
-      toGet = (normalAvail < (int)(framesPerBuffer * sizeof(SFLDataFormat))) ? normalAvail : framesPerBuffer * sizeof(SFLDataFormat);
-      if (toGet) {
-	_mainSndRingBuffer.Get(out, toGet, 100);
-	_debug("Write %i bytes\n" , toGet);
-	_mainSndRingBuffer.Discard(toGet);
-      } else {
-	bzero(out, framesPerBuffer * sizeof(SFLDataFormat));
-      }
-    }
+	tone->getNext(out, framesPerBuffer, 100);
+	toGet = framesPerBuffer;
+	} */else {
+	  normalAvail = _mainSndRingBuffer.AvailForGet();
+	  toGet = (normalAvail < (int)(framesPerBuffer * sizeof(SFLDataFormat))) ? normalAvail : framesPerBuffer * sizeof(SFLDataFormat);
+	  if (toGet) {
+	    _mainSndRingBuffer.Get(out, toGet, 100);
+	    _debug("Write %i bytes\n" , toGet);
+	    _mainSndRingBuffer.Discard(toGet);
+	  } else {
+	    bzero(out, framesPerBuffer * sizeof(SFLDataFormat));
+	  }
+	}
   }
-  pa_stream_write( playback->pulseStream() , out , toGet  , pa_xfree, 0 , PA_SEEK_RELATIVE);
+  pa_stream_write( playback , out , toGet  , pa_xfree, 0 , PA_SEEK_RELATIVE);
 }
 
   void 
 PulseLayer::read( void )
 {
-  //assert( record->pulseStream() );
-  int micAvailPut;
-  size_t toPut;
-  const void* data; 
-  // Handle the mic's audio stream 
-  micAvailPut = _micRingBuffer.AvailForPut();
-  toPut = 2048;
-  //toPut = (micAvailPut <= (size_t)(framesPerBuffer * sizeof(SFLDataFormat))) ? micAvailPut : framesPerBuffer * sizeof(SFLDataFormat);
-  //_debug("micAvailPut: %i - - toPut : %i\n", micAvailPut, toPut);
-
-  if( pa_stream_peek( record->pulseStream() , &data , &toPut ) < 0 ){
-    _debug("pa_stream_peek() failed: %s\n" , pa_strerror( pa_context_errno( context) ));
-    return;
-  }
+  if( (record) && pa_stream_get_state( record) == PA_STREAM_READY) {
 
-  _debug("- toPut : %i\n", toPut);
-  if( data != NULL )
-    _micRingBuffer.Put( (void*)data , toPut, 100);
+    size_t n = pa_stream_readable_size( record);
 
-  if( pa_stream_drop( record->pulseStream()) < 0 )
-    _debug("pa_stream_drop() failed: %s\n" , pa_strerror( pa_context_errno( context) ));
-  
+    if( n == (size_t) -1 ){
+      _debug("pa_stream_readable_size(): %s\n" , pa_strerror( pa_context_errno( context )) );
+      return;
+    }
+
+    const char* data;
+    size_t r;
+
+    /*if( pa_stream_peek( record , (const void**)&data , &r ) < 0 || !data ){
+      _debug("pa_stream_peek() failed: %s\n" , pa_strerror( pa_context_errno( context) ));
+      return;
+      }
+
+      _micRingBuffer.Put( (void*)data , r, 100);
+
+      if( pa_stream_drop( record ) < 0 ) {
+      _debug("pa_stream_drop() failed: %s\n" , pa_strerror( pa_context_errno( context) ));
+      return;
+      }*/
+  }
 }
 
   int
@@ -336,3 +392,20 @@ PulseLayer::putInCache( char code, void *buffer, int toCopy )
   //pa_stream_finish_upload( cache->pulseStream() );
 }
 
+  int
+PulseLayer::readbuffer( void* data , int bytes )
+{
+  if( (record) && pa_stream_get_state( record) == PA_STREAM_READY) {
+    size_t r = (size_t) bytes;
+    if( pa_stream_peek( record , (const void**)&data , &r ) < 0 || !data ){
+      _debug("pa_stream_peek() failed: %s\n" , pa_strerror( pa_context_errno( context) ));
+      return 0;
+
+      if( pa_stream_drop( record ) < 0 ) {
+	_debug("pa_stream_drop() failed: %s\n" , pa_strerror( pa_context_errno( context) ));
+	return 0;
+      }
+    }
+    return bytes;
+  }
+}

src/audio/pulselayer.h

@@ -21,7 +21,7 @@
 #define _PULSE_LAYER_H
 
 #include "audiolayer.h"
-#include "audiostream.h"
+//#include "audiostream.h"
 
 class RingBuffer;
 class ManagerImpl;
@@ -46,7 +46,6 @@ class PulseLayer : public AudioLayer {
      */
     bool openDevice(int indexIn, int indexOut, int sampleRate, int frameSize , int stream, std::string plugin) ;
 
-
     void startStream(void);
     void stopStream(void);
 
@@ -78,6 +77,7 @@ class PulseLayer : public AudioLayer {
     static void audioCallback ( pa_stream* s, size_t bytes, void* userdata );
     static void overflow ( pa_stream* s, void* userdata );
     static void underflow ( pa_stream* s, void* userdata );
+    static void stream_state_callback( pa_stream* s, void* user_data );	
 
     static void context_state_callback( pa_context* c, void* user_data );	
 
@@ -128,6 +128,7 @@ class PulseLayer : public AudioLayer {
 
     void write( void );
     void read( void );
+    int readbuffer( void* data, int bytes );
     void createStreams( pa_context* c );
     /**
      * Drop the pending frames and close the playback device
@@ -145,10 +146,12 @@ class PulseLayer : public AudioLayer {
     pa_context* context;
     pa_threaded_mainloop* m;
 
-    AudioStream* playback;
-    AudioStream* record;
-    AudioStream* cache;
+    //AudioStream* playback;
+    //AudioStream* record;
+    //AudioStream* cache;
 
+    pa_stream* playback;
+    pa_stream* record;
 };
 
 #endif // _PULSE_LAYER_H_