upload files

README

+./streamRecorder http://7329.live.streamtheworld.com:3690/XHMVSFM_SC 5
+

StreamRecorder.cpp

+////////////////////////////////////////////////////////////////////////////////
+/**
+* @file StreamRecorder.cpp
+* @author Mario Chirinos
+* @date 2014-02-12
+* @brief Streaming recorder with GStreamer
+* @note streamRecorder http://streamingmovil.radioformula.com:8000/m1033 3000
+*/
+////////////////////////////////////////////////////////////////////////////////
+#include "StreamRecorder.h"
+#include <fstream>
+#include <iostream>
+#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)
+{
+	int nFrames = ceil(time*STREAMRECORDER_SAMPLERATE/READSIZE);
+	recordTime = nFrames*READSIZE/STREAMRECORDER_SAMPLERATE;
+	cout << "record time: " << recordTime << endl;
+	bufferSize=nFrames*READSIZE*STREAMRECORDER_BYTESPERSAMPLE;
+	audioBuffer = new unsigned char[bufferSize];
+	memset (audioBuffer, 0, bufferSize);
+	audioBufferPosition=audioBuffer;
+	nBytes=0;
+	createMainPipeline();
+	connect(source);
+}
+//------------------------------------------------------------------------------
+/**
+*	Connect to the stream
+*	@param uri streaing uri
+*	@return unimplemented
+*/
+int StreamRecorder::connect(const char *uri)
+{
+	disconnect();
+	cout << "connecting to " << uri << endl;
+	gst_element_set_state (mainPipeline, GST_STATE_NULL);
+	g_object_set (G_OBJECT (streamSrc), "uri", uri, NULL);
+	gst_element_link (streamSrc, audioConvert);
+	gst_element_set_state (mainPipeline, GST_STATE_PLAYING);
+	return 0;
+}
+//------------------------------------------------------------------------------
+/**
+*	disconnect from the stream
+*	@return unimplemented
+*/
+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
+*/
+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_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");
+	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");
+	filter = gst_element_factory_make("identity", "audio_filter");
+	g_signal_connect(filter, "handoff", G_CALLBACK (filter_handoff_callback), this);
+
+	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");
+	
+	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;
+}
+//------------------------------------------------------------------------------
+/**
+*	disconnect from the stream
+*	@param the GstBus that sent the message
+*	@param message the GstMessage
+*	@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)) 
+	{
+		case GST_MESSAGE_EOS:
+			cout << "End of stream" << endl;
+			//&g_main_loop_quit (loop);
+			break;
+
+		case GST_MESSAGE_ERROR:
+			gchar  *debug;
+			GError *error;
+			gst_message_parse_error (message, &error, &debug);
+			g_free (debug);
+			cerr << "Error: "<<  error->message << endl;
+			g_error_free (error);
+			//g_main_loop_quit (loop);
+			break;
+		default:
+			break;
+	}
+	return TRUE;
+}
+//------------------------------------------------------------------------------
+/**
+*	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)
+{
+	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;
+		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;
+			exit(0);
+		} 
+		else 
+		{
+			cout <<"\tLink 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);
+	stringstream ss;
+	string fileNameStr, currentTimeStr;
+	ss << currentTime << endl;
+	getline(ss, currentTimeStr);
+
+	ss << currentTimeStr << ".flac" << endl;
+	getline(ss, fileNameStr);
+
+	int readsize = READSIZE;
+	FLAC__bool ok = true;
+	FLAC__StreamEncoder *encoder = 0;
+	FLAC__StreamMetadata *metadata[2];
+	FLAC__StreamMetadata_VorbisComment_Entry entry;
+	FLAC__StreamEncoderInitStatus init_status;
+
+
+	/* allocate the encoder */
+	if((encoder = FLAC__stream_encoder_new()) == NULL)
+	{
+		fprintf(stderr, "ERROR: allocating encoder\n");
+		return 1;
+	}
+	ok &= FLAC__stream_encoder_set_verify(encoder, true);
+	ok &= FLAC__stream_encoder_set_compression_level(encoder, 5);
+	ok &= FLAC__stream_encoder_set_channels(encoder, 1);
+	ok &= FLAC__stream_encoder_set_bits_per_sample(encoder, STREAMRECORDER_BYTESPERSAMPLE*8);
+	ok &= FLAC__stream_encoder_set_sample_rate(encoder, STREAMRECORDER_SAMPLERATE);
+	ok &= FLAC__stream_encoder_set_total_samples_estimate(encoder, nBytes/STREAMRECORDER_BYTESPERSAMPLE);
+
+	/* now add some metadata; we'll add some tags and a padding block */
+	if(ok)
+	{
+		if(
+			(metadata[0] = FLAC__metadata_object_new(FLAC__METADATA_TYPE_VORBIS_COMMENT)) == NULL ||
+			(metadata[1] = FLAC__metadata_object_new(FLAC__METADATA_TYPE_PADDING)) == NULL ||
+			/* there are many tag (vorbiscomment) functions but these are convenient for this particular use: */
+			!FLAC__metadata_object_vorbiscomment_entry_from_name_value_pair(&entry, "ARTIST", "artist") ||
+			!FLAC__metadata_object_vorbiscomment_append_comment(metadata[0], entry, /*copy=*/false) || /* copy=false: let metadata object take control of entry's allocated string */
+			!FLAC__metadata_object_vorbiscomment_entry_from_name_value_pair(&entry, "YEAR", "year") ||
+			!FLAC__metadata_object_vorbiscomment_append_comment(metadata[0], entry, /*copy=*/false)
+		)
+		{
+			fprintf(stderr, "ERROR: out of memory or tag error\n");
+			ok = false;
+		}
+
+		metadata[1]->length = 1234; /* set the padding length */
+
+		ok = FLAC__stream_encoder_set_metadata(encoder, metadata, 2);
+	}
+
+	/* initialize encoder */
+	if(ok)
+	{
+		init_status = FLAC__stream_encoder_init_file(encoder, fileNameStr.c_str(), NULL, /*client_data=*/NULL);
+		if(init_status != FLAC__STREAM_ENCODER_INIT_STATUS_OK)
+		{
+			fprintf(stderr, "ERROR: initializing encoder: %s\n", FLAC__StreamEncoderInitStatusString[init_status]);
+			ok = false;
+		}
+	}
+
+	int channels = 1;
+//	int bps = 16;
+	unsigned char* audioBufferTmp = audioBuffer;
+	unsigned char* buffer = new unsigned char[readsize*STREAMRECORDER_BYTESPERSAMPLE*channels];
+	int* pcm = new int[readsize*channels];
+	/* read blocks of samples from WAVE file and feed to encoder */
+
+	if(ok)
+	{
+		int left = nBytes;
+
+		while(left&&ok)
+		{
+
+			int need = (left>readsize? readsize : left);
+			memcpy(buffer, audioBufferTmp, need);//*sizeof(short));
+			audioBufferTmp+=need;//*sizeof(short);
+			/* convert the packed little-endian 16-bit PCM samples from WAVE into an interleaved FLAC__int32 buffer for libFLAC */
+			for(int i=0; i < need*channels; i++)
+			{
+				/* inefficient but simple and works on big- or little-endian machines */
+				pcm[i] = (FLAC__int32)(((FLAC__int16)(FLAC__int8)buffer[2*i+1] << 8) | (FLAC__int16)buffer[2*i]);
+			}
+			/* feed samples to encoder */
+			ok = FLAC__stream_encoder_process_interleaved(encoder, pcm, need/STREAMRECORDER_BYTESPERSAMPLE);
+			left -= need;
+		}
+	}
+	else
+	{
+		cout << "-ERROR-" << endl;
+	}
+
+	ok &= FLAC__stream_encoder_finish(encoder);
+
+	fprintf(stderr, "encoding: %s\n", ok? "succeeded" : "FAILED");
+	fprintf(stderr, "   state: %s\n", FLAC__StreamEncoderStateString[FLAC__stream_encoder_get_state(encoder)]);
+
+	/* now that encoding is finished, the metadata can be freed */
+	FLAC__metadata_object_delete(metadata[0]);
+	FLAC__metadata_object_delete(metadata[1]);
+	FLAC__stream_encoder_delete(encoder);
+
+	delete[] buffer;
+	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 bytesRead = length;// READSIZE*STREAMRECORDER_BYTESPERSAMPLE;
+	memcpy((char*)audioBufferPosition, (char*)data, bytesRead);
+	audioBufferPosition+=bytesRead;
+	nBytes+=bytesRead;//READSIZE;
+//	if(nBytes*STREAMRECORDER_BYTESPERSAMPLE >= bufferSize)
+	if(nBytes >= bufferSize)
+	{
+////		cout << " send to compress:" << nBytes << endl;
+		//saveWav(audioBuffer, 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
+*	@param buffer The buffer that just has been received
+*	@param user_data this
+*	@return unimplemented
+*/
+int StreamRecorder::filter_handoff_callback(GstElement* filter, GstBuffer* buffer, void* user_data)
+{
+//	cout << "filter_handoff_callback" << endl;
+	GstMapInfo info;
+	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));
+	((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

+////////////////////////////////////////////////////////////////////////////////
+// StreamRecorder.h
+// Mario Chirinos
+// 2014-02-12
+//------------------------------------------------------------------------------
+// Notes:
+//	Streaming recorder with GStreamer
+//
+////////////////////////////////////////////////////////////////////////////////
+
+#ifndef STREAMRECORDER_H
+#define STREAMRECORDER_H
+
+// your public header include
+//------------------------------------------------------------------------------
+#include <gst/gst.h>
+//------------------------------------------------------------------------------
+#define STREAMRECORDER_SAMPLERATE 44100
+#define READSIZE 1152 //For MPEG1, frame_size = 1152 samples/frame
+#define STREAMRECORDER_BYTESPERSAMPLE 2
+// the declaration of your class...
+//------------------------------------------------------------------------------
+class StreamRecorder
+{
+ private:
+	unsigned int nBytes;
+	unsigned int bufferSize;
+	unsigned char* audioBuffer;
+	unsigned char* audioBufferPosition;
+	int recordTime;
+	char* sourceName;
+	GstElement* streamSrc;
+	GstElement* audioConvert;
+	GstElement* audioResample;
+        GstElement* filterCaps;
+	GstElement* queue0;
+	GstElement* queue1;
+	GstElement* filter;
+//	GstElement* compressor;
+//	GstElement* queue2;
+//	GstElement* muxer;
+//        GstElement* fileSink; 
+        GstElement* fakeSink; 
+        GstElement* audioSink;            
+	GstElement* mainPipeline;
+//	GstElement* sinkPipeline;
+
+        GstElement* tempBin;       
+
+	int createMainPipeline();
+//	int createSinkPipeline();
+	int connect(const char *uri);
+	int disconnect();
+	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);
+	int addToBuffer(unsigned char* data, int length);
+	int compressBuffer();
+	int saveWav(unsigned char* data, int size) const;
+	void printCaps(GstPad *pad);
+ public:
+	StreamRecorder(const char* source, int time);
+};
+//------------------------------------------------------------------------------
+#endif

main.cpp

+////////////////////////////////////////////////////////////////////////////////
+//
+//
+//
+////////////////////////////////////////////////////////////////////////////////
+#include <cstdlib>
+#include <iostream>
+#include "StreamRecorder.h"
+using namespace std;
+//------------------------------------------------------------------------------
+int main(int argc, char *argv[])
+{
+	if(argc < 2)
+	{
+		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;
+}
+//------------------------------------------------------------------------------

makefile

+#-------------------------------------------------------------------------------
+#
+#
+#	Notes:
+#-------------------------------------------------------------------------------
+APPNAME = streamRecorder
+#Compiler:
+	CC=g++ -std=c++11
+
+#Compiler flags
+	CFLAGS=-c -g -Wall
+
+#Directories
+	DIRlib= /usr/local/lib	
+	incDIR= /usr/local/include
+	INCLUIDES=$(shell pkg-config --cflags glib-2.0 gstreamer-1.0 gstreamer-plugins-base-1.0 flac)
+	LIBS=$(shell pkg-config --libs glib-2.0 gstreamer-1.0 gstreamer-plugins-base-1.0 flac)
+#main function
+	mainP= main
+#-------------------------------------------------------------------------------
+all: Project
+
+Project: mainP.o StreamRecorder.o
+	$(CC) -o $(APPNAME) \
+	$(mainP).o \
+	StreamRecorder.o \
+	-L $(DIRlib) \
+	-I $(incDIR) \
+	$(LIBS)
+
+StreamRecorder.o: StreamRecorder.cpp
+	$(CC) $(INCLUIDES) $(CFLAGS) \
+	StreamRecorder.cpp
+	
+mainP.o: $(mainP).cpp
+	$(CC) $(INCLUIDES) $(CFLAGS) \
+	$(mainP).cpp
+