path: root/instr-daemon.c

#include "socket-common.h"
#include "response.h"
#include "lcd.h"
#include <stdlib.h>
#include <ctype.h>

#define STDIN_BUF_SIZE 1024

int port=3333; //port to listen
int connections=0;
int maxConn=16; //max connections - 16

guint signalMyCb; //signal to register which is used in cbClientInput(), step 10 from requirements
GAsyncQueue** stdinQueue=NULL;
GThread **peers; //actual connected peers


//GIOChannel callback for stdin
static gboolean
stdinAvailable  (GIOChannel *source, GIOCondition condition, gpointer data)
{
	static char* buffer=NULL;
	static gint64 allocated=0;

	//temporary buffer to read from stdin
	char tmp[STDIN_BUF_SIZE];
	memset(tmp,0,STDIN_BUF_SIZE);

	int size = read(0, tmp, STDIN_BUF_SIZE);
	if(size<=0) {
		return TRUE;
	}

	//fix for \n or \r sending
	if(size ==1)
		if(tmp[0] == '\r' || tmp[0] == '\n') {
			return TRUE;
		}

	//the termination characters
	if(!strncmp(tmp, "..", size-1)) {
		//final reallocation of the buffer to accomodate the whole string
		buffer=realloc(buffer,allocated+1);
		buffer[allocated] = '\0';

		g_print_debug("Got the buffer: \"%s\", tid: %p\n", buffer, g_thread_self());

		if (tmp[0] == '.' && tmp[1] == '.') { //fix for single dot sending
			int count=0;

			//send the final buffer to the queue
			for(count=0; count<maxConn; count++) {
				//allocate on the heap and deallocate in response.c, cbClientOutput()
				char *toSend = realloc(NULL, allocated+1);
				memcpy(toSend, buffer, allocated+1);

				//make sure we send to only valid peers
				if(peers[count] != 0) {
					g_async_queue_push(stdinQueue[count], toSend);
				}
			}

			//free the buffer
			free(buffer);
			buffer=NULL;
			allocated=0;
			return TRUE;
		}
	}

	//allocate on the heap. buffer is NULL the first time
	buffer=realloc(buffer, allocated+size);
	memcpy(buffer+allocated, tmp, size);
	allocated += size; //keep track of allocation


	return TRUE;
}

/** pulls an integer position from the vector which contains the
	position to use for the stdinQueue vector
	@param id - the thread id
*/
static int pullIndex(GThread* id)
{
	int i=0,ret=-1;
	static GStaticMutex mutex = G_STATIC_MUTEX_INIT;
	g_static_mutex_lock (&mutex);
	for(i=0; i<maxConn; i++)
		if(peers[i] == id) {
			peers[i] = (GThread*)0;
			ret=i;
			break;
		}
	g_static_mutex_unlock (&mutex);
	return ret;
}

/**
 * pushed in the peers vector the thread id for future reference
 * @param id
 */
static int pushIndex(GThread* id)
{
	int i=0,ret=-1;
	static GStaticMutex mutex = G_STATIC_MUTEX_INIT;
	g_static_mutex_lock (&mutex);
	for(i=0; i<maxConn; i++)
		if(peers[i] == 0) {
			peers[i] = id;
			ret=i;
			break;
		}

	g_static_mutex_unlock (&mutex);
	return ret;
}

//handler for incoming connection, closes the connection if maxCon connections are already handled
static gboolean
incomingConnection (GSocketService *service,
                    GSocketConnection      *connection,
                    GSocketListener        *listener,
                    gpointer                user_data)
{
	if(connections +1 > maxConn) {
		g_print_debug("Connection closed. Max reached\n");
		return TRUE;
	}
	connections++;
	g_print_debug("Incoming connection\n");
	return FALSE;
}


//thread connection handler
static gboolean
handler (GThreadedSocketService *service,
         GSocketConnection      *connection,
         GSocketListener        *listener,
         gpointer                user_data)
{


	GOutputStream *out;
	GInputStream *in;

	out = g_io_stream_get_output_stream (G_IO_STREAM (connection));
	in = g_io_stream_get_input_stream (G_IO_STREAM (connection));

	g_print_debug("Handling, connections: %d\n", connections);

	//register ourselves in the peers vector, use the index obtained in the stdinQueue
	//should not get -1
	GThread* self = g_thread_self();
	int index=pushIndex(self);

	GSource *outSource = NULL;
	GSource *inSource = g_pollable_input_stream_create_source((struct GPollableInputStream*)in, NULL);

	//get a reference to the async queue
	GAsyncQueue *queue = g_async_queue_ref(stdinQueue[index]);

	//input from client
	g_source_set_callback(inSource, cbClientInput, (GPollableOutputStream*)out, NULL);
	g_source_attach(inSource, NULL);

	//keep thread alive, every 1000 microseconds
	while(g_source_is_destroyed(inSource)==FALSE) {
		g_usleep(1000);
		gint elems;

		//verify our queue length and activate/deactivate the "out" Source for this connection
		//if we don't do this, the out Source will be scheduled frequently and will busy loop
		if((elems=g_async_queue_length(queue))>0) {
			if(outSource == NULL) {

				outSource = g_pollable_output_stream_create_source((struct GPollableOutputStream*)out, NULL);
				g_source_set_callback(outSource, cbClientOutput, queue, NULL);
				g_source_attach(outSource, NULL);
			} else {
				g_source_destroy(outSource);
				g_source_unref(outSource);
				outSource = NULL;
			}

		} else {
			if(outSource!= NULL && !g_source_is_destroyed(outSource)) {
				g_print_debug("Destroy source\n");
				g_source_destroy(outSource);
				g_source_unref(outSource);
				outSource = NULL; //added
			}
		}
		//end of activate/deactivate
	}

	//reached the end of the thread
	if (g_output_stream_close(out, NULL, NULL) == FALSE) {
		g_print_debug("out not closed\n");
	}
	if (g_input_stream_close(in, NULL, NULL) == FALSE) {
		g_print_debug("in not closed\n");
	}

	g_print_debug("Thread end\n");
	connections--; //keep track of connections
	g_async_queue_unref(queue); //unreference the queue
	pullIndex(g_thread_self()); //unregister from the peers vector
	return TRUE;
}

int main(int argc, char **argv)
{
	GSocketService *service = NULL;
	GError *error = NULL;
	GIOChannel* stdinChannel = NULL;

	g_type_init ();

	LCD_initialize();

	//register stdin channel
	stdinChannel = g_io_channel_unix_new(0);
	if(stdinChannel == NULL) {
		g_printerr("No io channel\n");
		exit(-1);
	}
	g_io_add_watch(stdinChannel, G_IO_IN, stdinAvailable, NULL);
	int idx=0;

	//allocate a maxConn queue on the heap
	stdinQueue = malloc(sizeof(struct GAsyncQueue*) * maxConn);
	//allocate peers vector on the heap
	peers = malloc(sizeof(GThread*) * maxConn);
	for(idx=0; idx<maxConn; idx++) {
		peers[idx] = 0;
		stdinQueue[idx] = g_async_queue_new();
	}

	//create a threaded service
	service = g_threaded_socket_service_new (maxConn+1);
	if (!g_socket_listener_add_inet_port (G_SOCKET_LISTENER (service),
	                                      port,
	                                      NULL,
	                                      &error)) {
		g_printerr ("%s: %s\n", argv[0], error->message);
		free(stdinQueue);
		free(peers);
		return 1;
	}

	//init the signal signalMyCb
	initSignals(&signalMyCb);

	g_print_debug("Server listening on port %d\n", port);
	g_signal_connect (service, "run", G_CALLBACK (handler), NULL);
	g_signal_connect (service, "incoming", G_CALLBACK(incomingConnection), NULL);

	g_main_loop_run (g_main_loop_new (NULL, FALSE));
	free(stdinQueue);
	free(peers);

	return 0;

}