restart the connection and save partial buffer

StreamRecorder.cpp

@@ -13,87 +13,119 @@
 #include <ctime>
 #include <cmath>
 #include <sstream>
-//#include <thread> 
+
 #include "FLAC/metadata.h"
 #include "FLAC/stream_encoder.h"
 #include <cstring>
 using namespace std;
 //------------------------------------------------------------------------------
+
 /**
 *	Constructor
 *	@param source streaing url
 *	@param time split recording in segments of time seconds.
 */
-StreamRecorder::StreamRecorder(const char* source, int time)
-{
+
+StreamRecorder::StreamRecorder(const char* source, int time) {
+ 
 	int nFrames = ceil(time*STREAMRECORDER_SAMPLERATE/READSIZE);
 	recordTime = nFrames*READSIZE/STREAMRECORDER_SAMPLERATE;
 	cout << "record time: " << recordTime << endl;
+    
+    // buffer size
 	bufferSize=nFrames*READSIZE*STREAMRECORDER_BYTESPERSAMPLE;
+    
+    // buffer that ig going to contain the audio
 	audioBuffer = new unsigned char[bufferSize];
+    
+    // Puts all the elements in audio buffer in 0
 	memset (audioBuffer, 0, bufferSize);
+    
+    // set the pointer to the actual position of the buffer;
 	audioBufferPosition=audioBuffer;
+    
+    // Bytes saved counter
 	nBytes=0;
+    
+    // flag that save the conncetion state
+    isConnectionLost=false;
+    
+    // Create the elements and sets capabilities
 	createMainPipeline();
+    
+    // Connec with the URI
 	connect(source);
 }
 //------------------------------------------------------------------------------
+
 /**
 *	Connect to the stream
 *	@param uri streaing uri
 *	@return unimplemented
 */
-int StreamRecorder::connect(const char *uri)
-{
+
+int StreamRecorder::connect(const char *uri) {
+ 
 	disconnect();
 	cout << "connecting to " << uri << endl;
+    
+    // Uri destiny
+    pluginUri = (char*)uri;
+	
     gst_element_set_state (mainPipeline, GST_STATE_NULL);
+    gst_element_set_state (mainPipeline, GST_STATE_READY);
+    
+    // Assign the uri property
 	g_object_set (G_OBJECT (streamSrc), "uri", uri, NULL);
+ 
 	gst_element_link (streamSrc, audioConvert);
+    
+    // Main pipeline starts to work
 	gst_element_set_state (mainPipeline, GST_STATE_PLAYING);
 	return 0;
 }
 //------------------------------------------------------------------------------
+
 /**
 *	disconnect from the stream
 *	@return unimplemented
 */
-int StreamRecorder::disconnect()
-{
+
+int StreamRecorder::disconnect() {
 	gst_element_unlink (streamSrc, audioConvert);
 	gst_element_set_state (mainPipeline, GST_STATE_NULL);
 	return 0;
 }
 //------------------------------------------------------------------------------
+
 /**
 *	Create main pipeline
-*	@return 0 on succeess else on error
+*	@return 0 on success else on error
 */
-int StreamRecorder::createMainPipeline()
-{
+
+int StreamRecorder::createMainPipeline() {
     
     mainPipeline = gst_pipeline_new("stream-recorder");
 
 	GstBus* bus;
 	bus = gst_pipeline_get_bus(GST_PIPELINE (mainPipeline));
-	gst_bus_add_watch(bus, bus_callback, NULL);
+	gst_bus_add_watch(bus, bus_callback, this);
 	gst_object_unref(GST_OBJECT (bus));
 
 	streamSrc = gst_element_factory_make("uridecodebin", "stream_source");
 	audioConvert = gst_element_factory_make ("audioconvert", "audio_convert");
-	audioResample = gst_element_factory_make ("audioresample", "audio_resample");
+	//audioResample = gst_element_factory_make ("audioresample", "audio_resample");
 	filterCaps = gst_element_factory_make("capsfilter", "filter_cap");
 
 	GstCaps *fc = gst_caps_new_full(gst_structure_new ("audio/x-raw",
                                                        "channels", G_TYPE_INT, 1,
                                                        "rate", G_TYPE_INT, STREAMRECORDER_SAMPLERATE,
-//						"width", G_TYPE_INT, 16,	
-//						"depth", G_TYPE_INT, 16,
                                                        "format", G_TYPE_STRING, "S16LE",
                                                        "signed", G_TYPE_BOOLEAN, 1,	//MUST BE SIGNED
                                                        "endianness", G_TYPE_INT, 1234,
                                                        NULL),
                                     NULL);
+    
  	g_object_set(G_OBJECT (filterCaps), "caps", fc, NULL);
 
 	queue0 = gst_element_factory_make("queue", "queue0");
@@ -102,27 +134,33 @@ int StreamRecorder::createMainPipeline()
 
 	queue1 = gst_element_factory_make("queue", "queue1");
 	fakeSink = gst_element_factory_make("fakesink", "fake_sink");
-	audioSink = gst_element_factory_make("autoaudiosink", "speaker");
-//	g_object_set (G_OBJECT (fakeSink), "signal-handoffs", TRUE, NULL);
-//	g_signal_connect(fakeSink, "handoff", G_CALLBACK(buffer_callback), this);
-////	g_object_set (G_OBJECT (tmpFileSink), "location", str.c_str(), NULL);
-//	compressor = gst_element_factory_make("vorbisenc", "audio_compressor");
+	//audioSink = gst_element_factory_make("autoaudiosink", "speaker");
 	
-	gst_bin_add_many (GST_BIN (mainPipeline), streamSrc, audioConvert, filterCaps, queue0, filter,   queue1, fakeSink, NULL);
-	if(!gst_element_link_many(audioConvert, filterCaps, queue0, filter, queue1, fakeSink,  NULL))
-//	gst_bin_add_many (GST_BIN (mainPipeline), streamSrc, audioConvert, filterCaps, queue0, filter, queue1, audioSink, NULL);
-//	if(!gst_element_link_many(audioConvert, filterCaps, queue0, filter, queue1, audioSink,  NULL))
+	gst_bin_add_many (GST_BIN (mainPipeline), streamSrc, audioConvert, filterCaps, queue0, filter,queue1, fakeSink, NULL);
+	if(!gst_element_link_many(audioConvert, filterCaps, queue0, filter, queue1, fakeSink,  NULL)){
+    //gst_bin_add_many (GST_BIN (mainPipeline), streamSrc, audioConvert, filterCaps, queue0, filter, queue1, audioSink, NULL);
+    //if(!gst_element_link_many(audioConvert, filterCaps, queue0, filter, queue1, audioSink,  NULL)){
     
-	{
 		cerr << "mainPipeline: Failed to link elements in the pipeline" << endl;
 		exit(0);
-		return 1;
 	}
+ 
 	g_signal_connect(streamSrc, "pad-added", G_CALLBACK(srcNewPad_callback), this);
 	gst_element_set_state (mainPipeline, GST_STATE_NULL);
+  
  	return 0;
 }
+
+//------------------------------------------------------------------------------
+
+gboolean StreamRecorder::reconnectURIStream(void *instance){
+    cout << "Restarting the main pipeline" << endl;
+    ((StreamRecorder*)instance)->connect(((StreamRecorder*)instance)->pluginUri);
+    return FALSE;
+}
+
 //------------------------------------------------------------------------------
+
 /**
 *	disconnect from the stream
 *	@param the GstBus that sent the message
@@ -130,126 +168,124 @@ int StreamRecorder::createMainPipeline()
 *	@param user_data NULL
 *	@return unimplemented
 */
-int StreamRecorder::bus_callback (GstBus *bus, GstMessage *message, void *user_data)
-{
-	printf("StreamRecorder got %s message\n", GST_MESSAGE_TYPE_NAME (message));
-	switch (GST_MESSAGE_TYPE (message)) 
-	{
+
+int StreamRecorder::bus_callback (GstBus *bus, GstMessage *message, void *user_data) {
+	
+    
+    //printf("StreamRecorder got %s message\n", GST_MESSAGE_TYPE_NAME (message));
+    
+    if(GST_MESSAGE_TYPE (message) == GST_MESSAGE_EOS){
+        ((StreamRecorder*)user_data)->isConnectionLost = true;
+    }
+    
+	switch (GST_MESSAGE_TYPE (message)) {
+        
         case GST_MESSAGE_EOS:
+        
 			cout << "End of stream" << endl;
-			//&g_main_loop_quit (loop);
+            cout << "End sometimes src" << endl;
 			break;
 
 		case GST_MESSAGE_ERROR:
+   
 			gchar  *debug;
 			GError *error;
 			gst_message_parse_error (message, &error, &debug);
 			g_free (debug);
+            
+            // Print specific error
 			cerr << "Error: "<<  error->message << endl;
 			g_error_free (error);
-			//g_main_loop_quit (loop);
+    
+            // Try to reconnect with the uri stream
+            g_timeout_add(60*1000, reconnectURIStream, user_data);
+            
+            // Enter only if the connection is lost
+            if(((StreamRecorder*)user_data)->isConnectionLost){
+                savePartialBuffer(user_data);
+                ((StreamRecorder*)user_data)->isConnectionLost = false;
+                
+            }
+            
 			break;
+		
         default:
+        
 			break;
 	}
+ 
 	return TRUE;
 }
+
+//------------------------------------------------------------------------------
+
+void StreamRecorder::savePartialBuffer(void *user_data){
+    
+    // Calculate the number of bytes that are missing for reach the max buffer size
+    // The buffer size is obtained based on the maximum recording time
+    int missingBytes = ((StreamRecorder*)user_data)->bufferSize - ((StreamRecorder*)user_data)->nBytes;
+    
+    // Put the buffer position in the last position of the buffer
+    ((StreamRecorder*)user_data)->audioBufferPosition+=missingBytes;
+    
+    // Save in flac format
+    ((StreamRecorder*)user_data)->compressBuffer();
+    
+    // Restart the possition pointer
+    ((StreamRecorder*)user_data)->audioBufferPosition=((StreamRecorder*)user_data)->audioBuffer;
+    
+    // Restart the buffer default values
+    memset (((StreamRecorder*)user_data)->audioBuffer, 0, ((StreamRecorder*)user_data)->bufferSize);
+    ((StreamRecorder*)user_data)->nBytes=0;
+    
+}
+
 //------------------------------------------------------------------------------
+
 /**
 *	CallBack to link the pads created by uridecodebin
 *	@param element The uridecodebin element
 *	@param pad The pad added
 *	@param data this
 */
-void StreamRecorder::srcNewPad_callback(GstElement *element, GstPad *pad, void *data)
-{
+
+void StreamRecorder::srcNewPad_callback(GstElement *element, GstPad *pad, void *data) {
+ 
 	cout << gst_element_get_name(element)<< " adding pad.." << gst_pad_get_name (pad) << endl;
 	cout << "Pad Name: " << gst_pad_get_name (pad) << endl;
-//	 ((StreamRecorder*)data)->printCaps(pad);
+ 
 	GstPad *sinkpad;
+	
     // gst_pad_get_caps is for gst v0.1
 	//GstCaps *new_pad_caps = gst_pad_get_caps (pad);
+ 
 	GstCaps *new_pad_caps = gst_pad_query_caps(pad, NULL);
 	GstStructure *new_pad_struct = gst_caps_get_structure (new_pad_caps, 0);
 	const gchar *new_pad_type = gst_structure_get_name (new_pad_struct);
 	
-	if (g_str_has_prefix (new_pad_type, "audio/x-raw"))
-	{
-		cout << "linking " << new_pad_type <<endl;
+	if (g_str_has_prefix (new_pad_type, "audio/x-raw")) {
+  
+		cout  << "linking " << new_pad_type << endl;
 		GstElement *nextElement = ((StreamRecorder*)data)->audioConvert;
 		sinkpad = gst_element_get_static_pad (nextElement, "sink");
 
-		if (GST_PAD_LINK_FAILED (gst_pad_link (pad, sinkpad)))
-		{
-			cerr << "\tType is "<< new_pad_type <<" but link failed." << endl;
+		if (GST_PAD_LINK_FAILED (gst_pad_link (pad, sinkpad))) {
+   
+			cerr << "Type is "<< new_pad_type <<" but link failed." << endl;
 			exit(0);
 		} 
-		else 
-		{
-			cout <<"\tLink succeeded " << new_pad_type << endl;
+		else {
+			cout <<"Link succeeded " << new_pad_type << endl;
 		}
 	}
 }
 //------------------------------------------------------------------------------
-/**
-*	Save audio data in wav format
-*	@param data Audio buffer
-*	@param length Buffer length
-*	@return unimplemented	
-*/
-int StreamRecorder::saveWav(unsigned char* data, int length) const
-{
-
-	cout << "saveWav:" << length << endl;
-	
-	unsigned char* newData = new unsigned char[length];
-	memcpy(newData, data, length);
-
-	stringstream ss;
-	string fileNameStr;
-	ss << time(NULL) << ".wav"<<endl;
-	getline(ss, fileNameStr);
 
-	ofstream myFile;
-
-	int size = 36+ length;
-	int subSize = 16;
-	short format=1;
-	short channels=1;
-	int sampleRate = STREAMRECORDER_SAMPLERATE;
-	int byteRate = sampleRate*channels*STREAMRECORDER_BYTESPERSAMPLE;
-	short blockAlign = channels * STREAMRECORDER_BYTESPERSAMPLE;
-	short bitsPerSample = STREAMRECORDER_BYTESPERSAMPLE*8;
-	int dataSize= length;
-
-
-	myFile.open(fileNameStr.c_str(), std::ios::binary);
-
-	myFile.write("RIFF", 4);
-	myFile.write((char*)&size, sizeof(int));
-	myFile.write("WAVE", 4);
-
-	myFile.write("fmt ", 4);
-	myFile.write((char*)&subSize, sizeof(int));
-	myFile.write((char*)&format, sizeof(short));    // Format (1 = PCM)
-	myFile.write((char*)&channels, sizeof(short));  // Channels
-	myFile.write((char*)&sampleRate, sizeof(int));  // Sample Rate
-	myFile.write((char*)&byteRate, sizeof(int)); // Byterate
-	myFile.write((char*)&blockAlign, sizeof(short));// Frame size
-	myFile.write((char*)&bitsPerSample, sizeof(short));// Bits per sample
-	myFile.write("data", 4);
-	myFile.write((char*)&dataSize, sizeof(int));
-	myFile.write((char*)newData,  dataSize);
-	myFile.close();
-
-	delete[] newData;
-	return 0;
-}
-//------------------------------------------------------------------------------
 /**
 *	Save audio data (audioBuffer) in flac format
 *	@return unimplemented
 */
+
 int StreamRecorder::compressBuffer()
 {
 	long int currentTime = time(NULL);
@@ -268,7 +304,6 @@ int StreamRecorder::compressBuffer()
 	FLAC__StreamMetadata_VorbisComment_Entry entry;
 	FLAC__StreamEncoderInitStatus init_status;
 
-
 	/* allocate the encoder */
 	if((encoder = FLAC__stream_encoder_new()) == NULL)
 	{
@@ -362,45 +397,47 @@ int StreamRecorder::compressBuffer()
 	delete[] pcm;
 	return 0;
 }
+
+
+
 //------------------------------------------------------------------------------
+
 /**
 *	Add audio data to audioBuffer
 *	@param data Audio data to add
 *	@param length Data length
 *	@return Bytes writen
 */
-int StreamRecorder::addToBuffer(unsigned char* data, int length)
-{
-//	cout << "addToBuffer("<<length<<")" << endl;
+
+int StreamRecorder::addToBuffer(unsigned char* data, int length) {
+
+    //cout << "addToBuffer(" << length << ")" << endl;
 	int bytesRead = length;// READSIZE*STREAMRECORDER_BYTESPERSAMPLE;
+ 
 	memcpy((char*)audioBufferPosition, (char*)data, bytesRead);
+    cout << *audioBufferPosition << endl;
+	
     audioBufferPosition+=bytesRead;
 	nBytes+=bytesRead;//READSIZE;
-//	if(nBytes*STREAMRECORDER_BYTESPERSAMPLE >= bufferSize)
-	if(nBytes >= bufferSize)
-	{
-////		cout << " send to compress:" << nBytes << endl;
-		//saveWav(audioBuffer, bufferSize);
+    
+    cout << "Bytes readed " << nBytes << endl;
+    cout << "Buffer size " << bufferSize << endl;
+	
+    if(nBytes >= bufferSize) {
+    
 		compressBuffer();
 		audioBufferPosition=audioBuffer;
 		memset (audioBuffer, 0, bufferSize);
 		nBytes=0;
+  
 	}
-//	else
-//	{
-//		cout << "Size dont match: "<< nBytes << " >= " << bufferSize  << endl<< endl;
-//	}
+ 
 	return nBytes;
-}
-//------------------------------------------------------------------------------
  
-void StreamRecorder::buffer_callback(GstElement *fakesink, GstBuffer *buffer, GstPad *pad, gpointer user_data)
-{
-//	cout << "Buffer_callback" << endl;
-//	cout<<(char*)buffer<<endl;
-//	((StreamRecorder*)user_data)->addToBuffer((unsigned short*)buffer);
 }
+
 //------------------------------------------------------------------------------
+
 /**
 *	CallBack for handoff signal of identity filter
 *	@param filter Identity filter
@@ -408,53 +445,22 @@ void StreamRecorder::buffer_callback(GstElement *fakesink, GstBuffer *buffer, Gs
 *	@param user_data this
 *	@return unimplemented
 */
+
 int StreamRecorder::filter_handoff_callback(GstElement* filter, GstBuffer* buffer, void* user_data)
 {
-//	cout << "filter_handoff_callback" << endl;
+    cout << "LLEGO INFO" << endl;
 	GstMapInfo info;
-	if(!gst_buffer_map (buffer, &info, GST_MAP_READ))
-	{
+	if(!gst_buffer_map (buffer, &info, GST_MAP_READ)) {
 		cout << "ERROR: MAPPING IS NOT VALID" << endl;
 	}
+ 
 	//GST_BUFFER_DATA is for gst v0.1
 	//((StreamRecorder*)user_data)->addToBuffer((unsigned char*)GST_BUFFER_DATA (buffer));
+    
+    // user data is the class
 	((StreamRecorder*)user_data)->addToBuffer((unsigned char*)info.data, info.size);
 	gst_buffer_unmap (buffer, &info);
 	return 0;
 }
 //------------------------------------------------------------------------------
-/**
-*	Print pad capabilities
-*	@param pad pad
-*/
-void StreamRecorder::printCaps(GstPad *pad)
-{
-//	guint32 format;
-	//GstCaps *padCaps = gst_pad_get_caps(pad);
-	GstCaps *padCaps = gst_pad_query_caps(pad, NULL);
-	unsigned int nCaps = gst_caps_get_size(padCaps);
-	const GstStructure *str;
-	cout << nCaps << " capabilities detected" << endl;
-
-	cout << gst_caps_to_string(padCaps) << endl;
-	for(unsigned int i=0; i < nCaps; i++)
-	{
-		cout << " Capability: " << i << endl;
-		str = gst_caps_get_structure (padCaps, i);
-		int nFields =  gst_structure_n_fields (str);
-		cout << gst_structure_get_name (str)<< ": fields = " << nFields << endl;
-
-
-//		for(int n=0; n < nFields; n++)
-//		{
-//			const char* name = gst_structure_nth_field_name(str, n);
-//			cout << name << endl;
-//				cout << G_VALUE_TYPE(gst_structure_get_value (str, name))<< endl;
-//		}
-
-//		cout << ": "<< width << "x" << height<< " @ " << fpsNumerator <<"/" <<fpsDenominator <<" : "<< depth << " bits, " << endl;
-	}
-
-}
-//------------------------------------------------------------------------------
 

StreamRecorder.h

@@ -14,6 +14,7 @@
 // your public header include
 //------------------------------------------------------------------------------
 #include <gst/gst.h>
+#include <jmorecfg.h>
 //------------------------------------------------------------------------------
 #define STREAMRECORDER_SAMPLERATE 44100
 #define READSIZE 1152 //For MPEG1, frame_size = 1152 samples/frame
@@ -23,44 +24,49 @@
 class StreamRecorder
 {
     private:
+
         unsigned int nBytes;
 		unsigned int bufferSize;
 		unsigned char* audioBuffer;
 		unsigned char* audioBufferPosition;
      	int recordTime;
-	char* sourceName;
+        bool isConnectionLost;
+    
+        char * pluginUri;
+        //char* sourceName;
      	GstElement* streamSrc;
      	GstElement* audioConvert;
-	GstElement* audioResample;
+     	//GstElement* audioResample;
      	GstElement* filterCaps;
      	GstElement* queue0;
      	GstElement* queue1;
      	GstElement* filter;
-//	GstElement* compressor;
-//	GstElement* queue2;
-//	GstElement* muxer;
-//        GstElement* fileSink; 
      	GstElement* fakeSink;
-        GstElement* audioSink;            
+     	//GstElement* audioSink;
      	GstElement* mainPipeline;
-//	GstElement* sinkPipeline;
-
-        GstElement* tempBin;       
+     	//GstElement* tempBin;
     
 		int createMainPipeline();
-//	int createSinkPipeline();
 		int connect(const char *uri);
 		int disconnect();
+    
+        // callbacks
         static void srcNewPad_callback(GstElement *element, GstPad *pad, void *data);
         static int bus_callback(GstBus *bus, GstMessage *message, void *data);
-//	static void block_async_cb (GstPad * pad, gboolean blocked, gpointer user_data);
-	static void buffer_callback(GstElement *fakesink, GstBuffer *buffer, GstPad *pad, gpointer user_data);
         static int filter_handoff_callback(GstElement* filter, GstBuffer* buffer, void* user_data);
+		
+        // add data to buffer
         int addToBuffer(unsigned char* data, int length);
 		int compressBuffer();
-	int saveWav(unsigned char* data, int size) const;
-	void printCaps(GstPad *pad);
+    
+        // Save information when connection fails
+		static void savePartialBuffer(void *user_data);
+	    
+        // Restart the pipeline
+		static gboolean reconnectURIStream(void *data);
+    
     public:
+	    
 	    StreamRecorder(const char* source, int time);
 };
 //------------------------------------------------------------------------------

main.cpp

@@ -8,19 +8,20 @@
 #include "StreamRecorder.h"
 using namespace std;
 //------------------------------------------------------------------------------
-int main(int argc, char *argv[])
-{
-	if(argc < 2)
-	{
+int main(int argc, char *argv[]) {
+ 
+	if(argc < 3) {
 		cerr << "too few arguments" << endl;
 		return EXIT_FAILURE;
 	}
+ 
 	gst_init (&argc, &argv);
 	StreamRecorder myRecorder = StreamRecorder(argv[1],atoi(argv[2]));
 
 	GMainLoop* main_loop = NULL;
 	main_loop = g_main_loop_new (NULL, FALSE);
 	g_main_loop_run(main_loop);
+    
     return 0;
 }
 //------------------------------------------------------------------------------