Fixed http responses and added miles invoker

2 files changed, 507 insertions, 0 deletions

diff --git a/src/uscxml/plugins/invoker/miles/MilesSessionInvoker.cpp b/src/uscxml/plugins/invoker/miles/MilesSessionInvoker.cpp
new file mode 100644
index 0000000..1a1416a
--- /dev/null
+++ b/src/uscxml/plugins/invoker/miles/MilesSessionInvoker.cpp
@@ -0,0 +1,407 @@
+#include "MilesSessionInvoker.h"
+#include <glog/logging.h>
+
+#ifdef BUILD_AS_PLUGINS
+#include <Pluma/Connector.hpp>
+#endif
+
+#include <inttypes.h>
+
+namespace uscxml {
+
+#ifdef BUILD_AS_PLUGINS
+PLUMA_CONNECTOR
+bool connect(pluma::Host& host) {
+	host.add( new MilesSessionInvokerProvider() );
+	return true;
+}
+#endif
+
+MilesSessionInvoker::MilesSessionInvoker() {
+	/* Initalize Miles */
+	miles_init();
+}
+
+MilesSessionInvoker::~MilesSessionInvoker() {
+};
+
+boost::shared_ptr<IOProcessorImpl> MilesSessionInvoker::create(Interpreter* interpreter) {
+	boost::shared_ptr<MilesSessionInvoker> invoker = boost::shared_ptr<MilesSessionInvoker>(new MilesSessionInvoker());
+	invoker->_interpreter = interpreter;
+	return invoker;
+}
+
+Data MilesSessionInvoker::getDataModelVariables() {
+	Data data;
+	return data;
+}
+
+void MilesSessionInvoker::send(const SendRequest& req) {
+	std::cout << req;
+	if (boost::iequals(req.name, "disconnect")) {
+		std::string reflectorIP = req.params.find("reflectorip")->second;
+		std::string problemName = req.params.find("problemname")->second;
+		int rv;
+		rv = miles_disconnect_reflector_session((char*)reflectorIP.c_str(), (char*)problemName.c_str());
+		if (!rv) {
+			LOG(ERROR) << "Could not disconnect from reflector session";
+			return;
+		}
+
+	}
+	if (boost::iequals(req.name, "connect")) {
+		std::string email = req.params.find("email")->second;
+		std::string reflectorIP = req.params.find("reflectorip")->second;
+		std::string problemName = req.params.find("problemname")->second;
+
+		int rv;
+		rv = miles_connect_reflector_session((char*)reflectorIP.c_str(), (char*)problemName.c_str());
+		if (!rv) {
+			LOG(ERROR) << "Could not setup reflector session";
+			return;
+		}
+
+		/* Set up audio and video RTP sockets */
+		video_rtp_in_socket = miles_net_setup_udp_socket((char*)reflectorIP.c_str(), video_port, video_port, 10, 16000);
+		audio_rtp_in_socket = miles_net_setup_udp_socket((char*)reflectorIP.c_str(), audio_port, audio_port, 10, 16000);
+		video_rtp_out_socket = video_rtp_in_socket; //miles_net_setup_udp_socket((char*)reflectorIP.c_str(), video_port, 0, 10, 16000);
+		audio_rtp_out_socket = audio_rtp_in_socket; //miles_net_setup_udp_socket((char*)reflectorIP.c_str(), audio_port, 0, 10, 16000);
+
+		/* Set up audio and video RTCP sockets */
+		video_rtcp_in_socket = miles_net_setup_udp_socket((char*)reflectorIP.c_str(), video_port+1, video_port+1, 10, 16000);
+		audio_rtcp_in_socket = miles_net_setup_udp_socket((char*)reflectorIP.c_str(), audio_port+1, audio_port+1, 10, 16000);
+		video_rtcp_out_socket = video_rtcp_in_socket; //miles_net_setup_udp_socket((char*)reflectorIP.c_str(), video_port+1, 0, 10, 16000);
+		audio_rtcp_out_socket = audio_rtcp_in_socket; //miles_net_setup_udp_socket((char*)reflectorIP.c_str(), audio_port+1, 0, 10, 16000);
+
+		/* Set up RTP audio and video sessions */
+		video_session = miles_rtp_setup_session(video_rtp_in_socket, MILES_RTP_MEDIA_TYPE_VIDEO);
+		audio_session = miles_rtp_setup_session(audio_rtp_in_socket, MILES_RTP_MEDIA_TYPE_AUDIO);
+
+		/* Set up RTCP audio and video sessions */
+		video_session->rtcp_session = miles_rtp_setup_rtcp_session(video_session, video_rtcp_in_socket);
+		audio_session->rtcp_session = miles_rtp_setup_rtcp_session(audio_session, audio_rtcp_in_socket);
+
+		/* Initialize and configure video encoder */
+		video_encoder = miles_video_codec_init_encoder();
+		video_encoder->codec_id = miles_video_codec_get_encoder_for_rtp_payload_type(MILES_RTP_PAYLOAD_TYPE_JPEG);
+		video_encoder->width = 320;
+		video_encoder->height = 240;
+		video_encoder->qfactor = 50;
+		rv = miles_video_codec_setup_encoder(video_encoder);
+		if (!rv) {
+			LOG(ERROR) << "Could not setup video encoder";
+			return;
+		}
+
+		/* Set up video grabber */
+		rv = miles_video_grabber_get_supported_grabbers(&supported_video_grabbers);
+		if(rv<=0) {
+			/* No video grabber available */
+			exit(-1);
+		}
+		video_grabber = miles_video_grabber_create_context(supported_video_grabbers[0]);
+		video_grabber->width = video_encoder->width;
+		video_grabber->height = video_encoder->height;
+		miles_video_grabber_setup(video_grabber);
+		free(supported_video_grabbers);
+
+		/* Set up outgoing RTP stream for video */
+		out_rtp_video_stream = miles_rtp_setup_outgoing_stream(video_session, video_rtp_out_socket, 0, MILES_RTP_PAYLOAD_TYPE_JPEG);
+
+		/* Initialize and configure audio encoder */
+		audio_encoder = miles_audio_codec_init_encoder();
+		audio_encoder->codec_id = miles_audio_codec_get_encoder_for_rtp_payload_type(MILES_RTP_PAYLOAD_TYPE_L16);
+		audio_encoder->sample_rate = 16000;
+		audio_encoder->bytes_per_sample = 2;
+		audio_encoder->chunk_size = 320; /* 20 ms */
+		audio_encoder->input_format = MILES_AUDIO_FORMAT_PCM;
+		rv = miles_audio_codec_setup_encoder(audio_encoder);
+		if(rv == 0) {
+			/* Couldn't set up audio codec */
+			exit(-1);
+		}
+
+
+		/* Set up audio grabber */
+		int n = miles_audio_device_get_supported_devices(&supported_audio_devices);
+		if(n<=0) {
+			/* No audio device available */
+			exit(-1);
+		}
+		/* Use first device that supports capture */
+		for(int i=0; i<n; i++) {
+			audio_dev = miles_audio_device_open(supported_audio_devices[i].id, MILES_AUDIO_FORMAT_PCM, 16000, 2, 1, 640, 1);
+			if(audio_dev)
+				break;
+		}
+		if(audio_dev == NULL)
+			exit(-1);
+
+		/* Find first audio device that supports playback */
+		for(int i=0; i<n; i++) {
+			audio_dev_playback = miles_audio_device_open(supported_audio_devices[i].id, MILES_AUDIO_FORMAT_PCM, 16000, 2, 1, 640, 0);
+			if(audio_dev_playback) {
+				audio_dev_playback_id = supported_audio_devices[i].id;
+				break;
+			}
+		}
+		if(audio_dev_playback == NULL)
+			exit(-1);
+
+		/* Set up outgoing RTP stream for audio */
+		out_rtp_audio_stream = miles_rtp_setup_outgoing_stream(audio_session, audio_rtp_out_socket, 0, MILES_RTP_PAYLOAD_TYPE_L16);
+
+		/* Associate RTP stream with codec context */
+		out_rtp_audio_stream->codec_ctx = audio_encoder;
+		out_rtp_video_stream->codec_ctx = video_encoder;
+
+		/* Set up outgoing RTCP streams for audio and video */
+		out_rtcp_audio_stream = miles_rtp_setup_outgoing_rtcp_stream(audio_session->rtcp_session, audio_rtcp_out_socket, out_rtp_audio_stream->ssrc);
+		out_rtcp_video_stream = miles_rtp_setup_outgoing_rtcp_stream(video_session->rtcp_session, video_rtcp_out_socket, out_rtp_video_stream->ssrc);
+
+		_isRunning = true;
+
+//		while(true) {
+//			rtp_video_receiver(video_session);
+//			video_transmitter(video_grabber, video_encoder, out_rtp_video_stream, out_rtcp_video_stream);
+//			rtp_audio_receiver(audio_session);
+//			audio_transmitter(audio_dev, audio_encoder, out_rtp_audio_stream, out_rtcp_audio_stream);
+//		}
+
+		_audioThread = new tthread::thread(MilesSessionInvoker::runAudio, this);
+		_videoThread = new tthread::thread(MilesSessionInvoker::runVideo, this);
+	}
+}
+
+void MilesSessionInvoker::runAudio(void* instance) {
+	((MilesSessionInvoker*)instance)->processAudio();
+}
+
+void MilesSessionInvoker::runVideo(void* instance) {
+	((MilesSessionInvoker*)instance)->processVideo();
+}
+
+void MilesSessionInvoker::processVideo() {
+	while(_isRunning) {
+		rtp_video_receiver(video_session);
+		video_transmitter(video_grabber, video_encoder, out_rtp_video_stream, out_rtcp_video_stream);
+	}
+}
+
+void MilesSessionInvoker::processAudio() {
+	while(_isRunning) {
+		rtp_audio_receiver(audio_session);
+		audio_transmitter(audio_dev, audio_encoder, out_rtp_audio_stream, out_rtcp_audio_stream);
+	}
+}
+
+void MilesSessionInvoker::cancel(const std::string sendId) {
+}
+
+void MilesSessionInvoker::invoke(const InvokeRequest& req) {
+	video_port = 5566;
+	audio_port = 5568;
+}
+
+/**
+ * Do something with an image decoded from an RTP stream (e.g. render to screen)
+ */
+void MilesSessionInvoker::render_video_image(u_int32_t ssrc, char *img, int width, int height, int img_format) {
+
+	if(img_format != MILES_IMAGE_RGBA) {
+		miles_image_convert(img, render_img, img_format, MILES_IMAGE_RGBA, width, height);
+		stbi_write_png("/Users/sradomski/Desktop/image.png", width, height, 3, render_img, width * 3);
+	} else {
+		stbi_write_png("/Users/sradomski/Desktop/image.png", width, height, 3, img, width * 3);
+	}
+
+	/* render image... */
+}
+
+/**
+ * Send an audio chunk decoded from an RTP stream to an audio device
+ */
+void MilesSessionInvoker::playback_audio(u_int32_t ssrc, char *buf, int sample_rate, int bps, int audio_format, int size) {
+	int n;
+
+	if(size<0)
+		return;
+
+	/* re-configure audio device, if needed */
+	if(audio_dev_playback == NULL || audio_dev_playback->chunk_size != size || audio_dev_playback->sample_rate != sample_rate ||
+	        audio_dev_playback->format != audio_format ||	audio_dev_playback->bytes_per_sample != bps) {
+		if(audio_dev_playback)
+			miles_audio_device_close(audio_dev_playback, 0);
+		audio_dev_playback = miles_audio_device_open(audio_dev_playback_id, audio_format, sample_rate, bps, 1, size, 0);
+		if(audio_dev_playback == NULL)
+			return;
+	}
+
+	/* play audio */
+	n = miles_audio_device_write(audio_dev_playback, buf, size);
+}
+
+/**
+ * Handle incoming video streams
+ */
+
+int MilesSessionInvoker::video_receiver(struct miles_rtp_in_stream *rtp_stream, char *data, int bytes_read) {
+	int status, n;
+	struct miles_video_codec_decode_context *codec_ctx;
+
+	codec_ctx = (struct miles_video_codec_decode_context *)rtp_stream->codec_ctx;
+
+	if(codec_ctx == NULL || !miles_video_codec_decoder_supports_rtp_payload_type(codec_ctx, rtp_stream->payload_type)) {
+		if(codec_ctx)
+			miles_video_codec_destroy_decoder(codec_ctx);
+		codec_ctx = miles_video_codec_init_decoder();
+		codec_ctx->codec_id = miles_video_codec_get_decoder_for_rtp_payload_type(rtp_stream->payload_type);
+		if(codec_ctx->codec_id == MILES_VIDEO_CODEC_UNKNOWN) {
+			/* Cannot decode the video stream */
+			return 0;
+		}
+
+		status = miles_video_codec_setup_decoder(codec_ctx);
+		if(status == 0) {
+			/* Cannot decode the video stream */
+			return 0;
+		}
+		rtp_stream->codec_ctx = (void *)codec_ctx;
+		return 0;
+	}
+	n = miles_video_codec_decode(codec_ctx, data, decoded_in_img, bytes_read);
+	if(n > 0) {
+		render_video_image(rtp_stream->ssrc, decoded_in_img, codec_ctx->width, codec_ctx->height, codec_ctx->output_format);
+	}
+	return n;
+}
+
+/**
+ * Handle incoming audio streams
+ */
+
+int MilesSessionInvoker::audio_receiver(struct miles_rtp_in_stream *rtp_stream, char *data, int bytes_read) {
+	int status, size;
+	struct miles_audio_codec_decode_context *codec_ctx;
+
+	codec_ctx = (struct miles_audio_codec_decode_context *)rtp_stream->codec_ctx;
+
+	if(codec_ctx == NULL || !miles_audio_codec_decoder_supports_rtp_payload_type(codec_ctx, rtp_stream->payload_type)) {
+		if(codec_ctx)
+			miles_audio_codec_destroy_decoder(codec_ctx);
+		codec_ctx = miles_audio_codec_init_decoder();
+		codec_ctx->codec_id = miles_audio_codec_get_decoder_for_rtp_payload_type(rtp_stream->payload_type);
+		if(codec_ctx->codec_id == MILES_AUDIO_CODEC_UNKNOWN) {
+			/* Cannot decode the audio stream */
+			return 0;
+		}
+		status = miles_audio_codec_setup_decoder(codec_ctx);
+		if(status == 0) {
+			/* Cannot decode the audio stream */
+			return 0;
+		}
+		rtp_stream->codec_ctx = (void *)codec_ctx;
+	}
+	size = miles_audio_codec_decode(codec_ctx, data, audio_in_buf);
+	if(size > 0) {
+		playback_audio(rtp_stream->ssrc, audio_in_buf, codec_ctx->sample_rate, codec_ctx->bytes_per_sample, codec_ctx->output_format, size);
+	}
+	return size;
+}
+
+/**
+ * Read and depacketize incoming RTP streams
+ */
+
+void MilesSessionInvoker::rtp_audio_receiver(struct miles_rtp_session *rtp_session) {
+	int n, m=0;
+	struct miles_rtp_in_stream *rtp_stream;
+
+	/* Poll RTP socket, read all available RTP packets */
+	while (1) {
+		n = miles_net_poll_socket(rtp_session->socket);
+		if(n<=0) return;
+
+		/* Read RTP data */
+		n = miles_rtp_recv(rtp_session, &rtp_stream, audio_data);
+		if(n>0) {
+			m = audio_receiver(rtp_stream, audio_data, n);
+		}
+
+		/* Poll RTCP socket */
+		n = miles_net_poll_socket(rtp_session->rtcp_session->socket);
+		if(n>0) {
+			/* Do RTCP packet processing */
+			n = miles_rtp_recv_rtcp(rtp_session->rtcp_session);
+		}
+	}
+}
+
+void MilesSessionInvoker::rtp_video_receiver(struct miles_rtp_session *rtp_session) {
+	int n, m=0;
+	struct miles_rtp_in_stream *rtp_stream;
+
+	/* Poll RTP socket, read all available RTP packets */
+	while (1) {
+		n = miles_net_poll_socket(rtp_session->socket);
+		if(n<=0) return;
+
+		/* Read RTP data */
+		n = miles_rtp_recv(rtp_session, &rtp_stream, video_data);
+		if(n>0) {
+			m = video_receiver(rtp_stream, video_data, n);
+		}
+
+		/* Poll RTCP socket */
+		n = miles_net_poll_socket(rtp_session->rtcp_session->socket);
+		if(n>0) {
+			/* Do RTCP packet processing */
+			n = miles_rtp_recv_rtcp(rtp_session->rtcp_session);
+		}
+	}
+}
+
+/**
+ * Send RTP video stream
+ */
+int MilesSessionInvoker::video_transmitter(struct miles_video_grabber_context *grabber, struct miles_video_codec_encode_context *codec_ctx, struct miles_rtp_out_stream *rtp_stream, struct miles_rtcp_out_stream *out_rtcp_stream) {
+	int n;
+
+	/* Send RTCP packets, if due */
+	miles_rtp_send_rtcp(out_rtcp_stream);
+
+	n = miles_video_grabber_grab(grabber, video_out_buf);
+	if(n <= 0)
+		return 0;
+	if(grabber->image_format != codec_ctx->input_format) {
+		/* image conversion ... */
+	}
+	n = miles_video_codec_encode(codec_ctx, video_out_buf, encoded_out_img);
+	if(n<=0)
+		return 0;
+	return miles_rtp_send(rtp_stream, encoded_out_img, n);
+}
+
+/**
+ * Send RTP audio stream
+ */
+int MilesSessionInvoker::audio_transmitter(struct miles_audio_device *dev, struct miles_audio_codec_encode_context *codec_ctx, struct miles_rtp_out_stream *rtp_stream, struct miles_rtcp_out_stream *out_rtcp_audio_stream) {
+	int n;
+
+	/* Send RTCP packets, if due */
+	miles_rtp_send_rtcp(out_rtcp_audio_stream);
+
+	n = miles_audio_device_read(dev, audio_read_buf, codec_ctx->chunk_size);
+	if(n <= 0)
+		return 0;
+	if(dev->format != codec_ctx->input_format) {
+		/* audio conversion needed ... */
+	}
+	n = miles_audio_codec_encode(codec_ctx, audio_read_buf, encoded_out_audio);
+	if(n<=0)
+		return 0;
+	return miles_rtp_send(rtp_stream, encoded_out_audio, n);
+}
+
+
+}
\ No newline at end of file
diff --git a/src/uscxml/plugins/invoker/miles/MilesSessionInvoker.h b/src/uscxml/plugins/invoker/miles/MilesSessionInvoker.h
new file mode 100644
index 0000000..cb3e9ee
--- /dev/null
+++ b/src/uscxml/plugins/invoker/miles/MilesSessionInvoker.h
@@ -0,0 +1,100 @@
+#ifndef MILESSESIONINVOKER_H_W09J90F0
+#define MILESSESIONINVOKER_H_W09J90F0
+
+#include <uscxml/Interpreter.h>
+
+extern "C" {
+#include "miles/miles.h"
+#include "miles/network.h"
+#include "miles/rtp.h"
+#include "miles/audio_codec.h"
+#include "miles/audio_device.h"
+#include "miles/video_codec.h"
+#include "miles/video_grabber.h"
+#include "miles/session.h"
+#include "miles/image.h"
+}
+#ifdef BUILD_AS_PLUGINS
+#include "uscxml/plugins/Plugins.h"
+#endif
+
+namespace uscxml {
+
+class MilesSessionInvoker : public InvokerImpl {
+public:
+	MilesSessionInvoker();
+	virtual ~MilesSessionInvoker();
+	virtual boost::shared_ptr<IOProcessorImpl> create(Interpreter* interpreter);
+
+	virtual std::set<std::string> getNames() {
+		std::set<std::string> names;
+		names.insert("miles");
+		names.insert("http://uscxml.tk.informatik.tu-darmstadt.de/#miles");
+		return names;
+	}
+
+	virtual Data getDataModelVariables();
+	virtual void send(const SendRequest& req);
+	virtual void cancel(const std::string sendId);
+	virtual void invoke(const InvokeRequest& req);
+
+protected:
+	int video_rtp_in_socket, audio_rtp_in_socket;
+	int video_rtp_out_socket, audio_rtp_out_socket;
+	int video_rtcp_in_socket, audio_rtcp_in_socket;
+	int video_rtcp_out_socket, audio_rtcp_out_socket;
+	struct miles_rtp_session *video_session, *audio_session;
+	struct miles_video_codec_encode_context *video_encoder;
+	struct miles_audio_codec_encode_context *audio_encoder;
+	int *supported_video_grabbers;
+	struct miles_video_grabber_context *video_grabber;
+	struct miles_rtp_out_stream *out_rtp_video_stream, *out_rtp_audio_stream;
+	struct miles_rtcp_out_stream *out_rtcp_video_stream, *out_rtcp_audio_stream;
+	struct miles_audio_device *audio_dev;
+	struct miles_audio_device_description *supported_audio_devices;
+	int video_port, audio_port;
+	std::string ip_address;
+
+	char video_out_buf[1000000];
+	char encoded_out_img[1000000];
+	char decoded_in_img[1000000];
+	char audio_in_buf[1000000];
+	char render_img[1000000];
+	char audio_data[1000000];
+	char video_data[1000000];
+
+	char encoded_out_audio[1000000];
+	char audio_read_buf[1000000];
+
+	struct miles_audio_device *audio_dev_playback;
+	int audio_dev_playback_id;
+
+	static void runAudio(void* instance);
+	static void runVideo(void* instance);
+	void processVideo();
+	void processAudio();
+
+	void render_video_image(u_int32_t ssrc, char *img, int width, int height, int img_format);
+	void playback_audio(u_int32_t ssrc, char *buf, int sample_rate, int bps, int audio_format, int size);
+	int video_receiver(struct miles_rtp_in_stream *rtp_stream, char *data, int bytes_read);
+	int audio_receiver(struct miles_rtp_in_stream *rtp_stream, char *data, int bytes_read);
+	void rtp_audio_receiver(struct miles_rtp_session *rtp_session);
+	void rtp_video_receiver(struct miles_rtp_session *rtp_session);
+	int video_transmitter(struct miles_video_grabber_context *grabber, struct miles_video_codec_encode_context *codec_ctx, struct miles_rtp_out_stream *rtp_stream, struct miles_rtcp_out_stream *out_rtcp_stream);
+	int audio_transmitter(struct miles_audio_device *dev, struct miles_audio_codec_encode_context *codec_ctx, struct miles_rtp_out_stream *rtp_stream, struct miles_rtcp_out_stream *out_rtcp_audio_stream);
+
+
+	bool _isRunning;
+	tthread::thread* _videoThread;
+	tthread::thread* _audioThread;
+	tthread::recursive_mutex _mutex;
+};
+
+#ifdef BUILD_AS_PLUGINS
+PLUMA_INHERIT_PROVIDER(MilesSessionInvoker, Invoker);
+#endif
+
+}
+
+
+#endif /* end of include guard: MILESSESIONINVOKER_H_W09J90F0 */