gistfile1.c

#include "stdlib.h"
#include "stdio.h"
#include "assert.h"
#include "portaudio.h"
#include "uv.h"

/* Define PA constants */
#define SAMPLE_RATE 44100
#define SAMPLE_SIZE 1
#define SAMPLE_SILENCE 128
#define FRAMES_PER_BUFFER 128
#define NUM_CHANNELS 2
#define PA_SAMPLE_TYPE paUInt8

typedef struct {
	char* outgoingSample;
	uv_udp_t socket;
	struct sockaddr_in peer;
	uv_async_t comm;
	uv_mutex_t mutex;
	PaStream* stream;
	PaStreamParameters parameters;
} Send;

typedef struct {
	char* incomingSample;
	uv_udp_t socket;
	uv_async_t comm;
	uv_mutex_t mutex;
	PaStream* stream;
	PaStreamParameters parameters;
} Receive;

typedef struct {
	Send* sender;
	Receive* receiver;
} Duplex;

uv_buf_t on_alloc(uv_handle_t* handle, size_t suggested_size) {
	uv_buf_t buf;
	buf.base = malloc(suggested_size);
	buf.len = suggested_size;

	return buf;
}

void do_record(uv_work_t* req) {
	PaError err;
	int r = 0;

	Duplex* duplexer = (Duplex*) req->data;

	while (1) {
		uv_mutex_lock(&duplexer->sender->mutex);

		err = Pa_ReadStream(duplexer->sender->stream, duplexer->sender->outgoingSample, FRAMES_PER_BUFFER);
		if (err) {
			fprintf(stderr, "%s\n", Pa_GetErrorText(err));
		}

		fprintf(stderr, "Recording\n");

		r = uv_async_send(&duplexer->sender->comm);
		if (r) {
			fprintf(stderr, "Error async send\n");	
		}
	}
}

void on_send(uv_udp_send_t* req, int status) {
	Duplex* duplexer = (Duplex*) req->data;

	uv_mutex_unlock(&duplexer->sender->mutex);
	free(req);
}

void do_send(uv_async_t* handle, int status) {
	int r;

	fprintf(stderr, "Sending voice packet\n");

	Duplex* duplexer = (Duplex*) handle->data;

	uv_udp_send_t* udpSendReq = (uv_udp_send_t*) malloc(sizeof(uv_udp_send_t));
	udpSendReq->data = (void*) duplexer;

	uv_buf_t buf = uv_buf_init(duplexer->sender->outgoingSample, FRAMES_PER_BUFFER * NUM_CHANNELS * SAMPLE_SIZE);

	r = uv_udp_send(udpSendReq, &duplexer->sender->socket, &buf, 1, duplexer->sender->peer, (uv_udp_send_cb) on_send);
	if (r)
	{
		fprintf(stderr, "Error sending voice\n");
	}
}

void do_play(uv_async_t* handle, int status) {
	Duplex* duplexer = (Duplex*) handle->data;

	PaError err = Pa_WriteStream(duplexer->receiver->stream, duplexer->receiver->incomingSample, FRAMES_PER_BUFFER);
	if (err) {
		fprintf(stderr, "Error writing streaming\n");
	}
}

void on_receive_voice(uv_udp_t* req, ssize_t nread, uv_buf_t buf, struct sockaddr* addr, unsigned flags) {
	int r;

	if (nread == -1) {
		fprintf(stderr, "Error reading voice\n");
	}

	assert(nread == FRAMES_PER_BUFFER * NUM_CHANNELS * SAMPLE_SIZE);

	Duplex* duplexer = (Duplex*) req->data;

	duplexer->receiver->incomingSample = buf.base;

	r = uv_async_send(&duplexer->receiver->comm);
	if (r) {
		fprintf(stderr, "Error receiving voice\n");
	}
}

void helper_clear_sample(char* sampleBlock) {
    int i;

    for (i=0; i < FRAMES_PER_BUFFER * NUM_CHANNELS; i++ ) {
        ((unsigned char *)sampleBlock)[i] = SAMPLE_SILENCE;
     }
}

int main(int argc, char const *argv[]) {
	int r;
	PaStream* sendStream = NULL;
	PaStream* receiveStream = NULL;

	uv_loop_t* loop = uv_default_loop();

	PaError err = Pa_Initialize();
	if (err) {
		fprintf(stderr, "PortAudio failed to initialize\n");
	}

	Duplex* duplexer = (Duplex*) malloc(sizeof(Duplex));

	Send* sendVoice = (Send*) malloc(sizeof(Send));

	sendVoice->outgoingSample = (char*) malloc(FRAMES_PER_BUFFER * NUM_CHANNELS * SAMPLE_SIZE);
	helper_clear_sample(sendVoice->outgoingSample);

	sendVoice->parameters.device = Pa_GetDefaultInputDevice();
	sendVoice->parameters.channelCount = NUM_CHANNELS;
	sendVoice->parameters.sampleFormat = PA_SAMPLE_TYPE;
	sendVoice->parameters.suggestedLatency = Pa_GetDeviceInfo(sendVoice->parameters.device)->defaultLowInputLatency;
	sendVoice->parameters.hostApiSpecificStreamInfo = NULL;

	r = uv_udp_init(loop, &sendVoice->socket);
	r = uv_udp_bind(&sendVoice->socket, uv_ip4_addr("0.0.0.0", 0), 0);

	sendVoice->peer = uv_ip4_addr(argv[1], atoi(argv[2]));

	r = uv_mutex_init(&sendVoice->mutex);

	Receive* receiveVoice = (Receive*) malloc(sizeof(Receive));

	receiveVoice->incomingSample = (char*) malloc(FRAMES_PER_BUFFER * NUM_CHANNELS * SAMPLE_SIZE);
	helper_clear_sample(receiveVoice->incomingSample);

	receiveVoice->parameters.device = Pa_GetDefaultOutputDevice();
	receiveVoice->parameters.channelCount = NUM_CHANNELS;
	receiveVoice->parameters.sampleFormat = PA_SAMPLE_TYPE;
	receiveVoice->parameters.suggestedLatency = Pa_GetDeviceInfo(receiveVoice->parameters.device)->defaultLowInputLatency;
	receiveVoice->parameters.hostApiSpecificStreamInfo = NULL;

	r = uv_mutex_init(&receiveVoice->mutex);

	r = uv_udp_init(loop, &receiveVoice->socket);
	r = uv_udp_bind(&receiveVoice->socket, uv_ip4_addr("0.0.0.0", 8000), 0);

	err = Pa_OpenStream(
		&sendStream,
		&sendVoice->parameters,
		NULL,
		SAMPLE_RATE,
		FRAMES_PER_BUFFER,
		paClipOff,
		NULL,
		NULL
	);
	if (err) {
		fprintf(stderr, "Error opening stream\n");
		return -1;
	}

	err = Pa_OpenStream(
		&receiveStream,
		NULL,
		&receiveVoice->parameters,
		SAMPLE_RATE,
		FRAMES_PER_BUFFER,
		paClipOff,
		NULL,
		NULL
	);
	if (err) {
		fprintf(stderr, "Error opening stream\n");
		return -1;
	}

	sendVoice->stream = sendStream;
	receiveVoice->stream = receiveStream;

	err = Pa_StartStream(sendStream);
	if (err) {
		fprintf(stderr, "error starting stream\n");
		return -1;
	}

	err = Pa_StartStream(receiveStream);
	if (err) {
		fprintf(stderr, "error starting stream\n");
		return -1;
	}

	duplexer->sender = sendVoice;
	duplexer->receiver = receiveVoice;

	receiveVoice->socket.data = (void*) duplexer;
	sendVoice->socket.data = (void*) duplexer;

	r = uv_async_init(loop, &sendVoice->comm, (uv_async_cb) do_send);
	uv_work_t recordAndSendReq;
	recordAndSendReq.data = (void*) duplexer;
	r = uv_queue_work(loop, &recordAndSendReq, (uv_work_cb) do_record, NULL);

	r = uv_async_init(loop, &receiveVoice->comm, (uv_async_cb) do_play);

	r = uv_udp_recv_start(&duplexer->receiver->socket, on_alloc, (uv_udp_recv_cb) on_receive_voice);

	r = uv_run(loop);

	/* Won't actually get here */
	err = Pa_StopStream(sendStream);
	err = Pa_StopStream(receiveStream);

	if (r) {
		fprintf(stderr, "uv_run failed to run\n");
	}

	return 0;
}