/*
	geojson2ew - geoJSON to earthworm 

	Copyright (c) 2014 California Institute of Technology.
	All rights reserved, November 6, 2014.
        This program is distributed WITHOUT ANY WARRANTY whatsoever.
        Do not redistribute this program without written permission.

	Authors: Kevin Frechette & Paul Friberg, ISTI.
*/
#include <ctype.h>
#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include <stdint.h>
#include <amqp_tcp_socket.h>
#include <amqp.h>
#include <amqp_framing.h>
#include "earthworm.h"  /* need this for the logit() call */
#include "externs.h"
#include "json_conn.h"

static JSON_CONN_PARAMS *params = NULL;
static amqp_connection_state_t conn = NULL;
static struct timeval *timeout = NULL;
static amqp_rpc_reply_t res;
static amqp_envelope_t envelope;
static char *msg_buffer = NULL;
static size_t msg_buffer_size;

int check_for_error(int x, char const *context) {
	if (x < 0) {
		logit("e", "%s: %s\n", context, amqp_error_string2(x));
		return 1;
	}
	return 0;
}

int check_for_amqp_error(amqp_rpc_reply_t x, char const *context) {
	int status = 1;
	switch (x.reply_type) {
	case AMQP_RESPONSE_NORMAL:
		status = 0;
		break;

	case AMQP_RESPONSE_NONE:
		logit("e", "%s: missing RPC reply type!\n", context);
		break;

	case AMQP_RESPONSE_LIBRARY_EXCEPTION:
		logit("e", "%s: library_error=%s\n", context,
				amqp_error_string2(x.library_error));
		break;

	case AMQP_RESPONSE_SERVER_EXCEPTION:
		switch (x.reply.id) {
		case AMQP_CONNECTION_CLOSE_METHOD: {
			amqp_connection_close_t *m =
					(amqp_connection_close_t *) x.reply.decoded;
			logit("e", "%s: server connection error %d, message: %.*s\n",
					context, m->reply_code, (int) m->reply_text.len,
					(char *) m->reply_text.bytes);
			break;
		}
		case AMQP_CHANNEL_CLOSE_METHOD: {
			amqp_channel_close_t *m = (amqp_channel_close_t *) x.reply.decoded;
			logit("e", "%s: server channel error %d, message: %.*s\n",
					context, m->reply_code, (int) m->reply_text.len,
					(char *) m->reply_text.bytes);
			break;
		}
		default:
			logit("e", "%s: unknown server error, method id 0x%08X\n",
					context, x.reply.id);
			break;
		}
		break;
	}

	return status;
}

int close_json_connection() {
	int status = 0;
	if (conn != NULL) {
		if (check_for_amqp_error(
				amqp_channel_close(conn, params->channel_number,
				AMQP_REPLY_SUCCESS), "Closing channel")
				|| check_for_amqp_error(
						amqp_connection_close(conn, AMQP_REPLY_SUCCESS),
						"Closing connection")
				|| check_for_error(amqp_destroy_connection(conn),
						"Ending connection")) {
			status = 1;
		}
		conn = NULL;
	}
	params = NULL;
	return status;
}

void free_json_message() {
	amqp_destroy_envelope(&envelope);
}

int open_json_connection(JSON_CONN_PARAMS *p) {
	params = p;
	if (p->read_timeout > 0) {
		timeout = malloc(sizeof(struct timeval));
		timeout->tv_sec = p->read_timeout;
		timeout->tv_usec = 0;
	}
	conn = amqp_new_connection();
	amqp_socket_t *socket = amqp_tcp_socket_new(conn);
	if (!socket) {
		logit("e", "could not create TCP socket\n");
		return 1;
	}

	if (amqp_socket_open(socket, p->hostname, p->port)) {
		logit("e", "could not open TCP socket\n");
		return 1;
	}

	if (check_for_amqp_error(
			amqp_login(conn, p->vhost, p->max_channels, p->frame_max_size,
					p->heartbeat_seconds, AMQP_SASL_METHOD_PLAIN, p->username,
					p->password), "Logging in")) {
		return 1;
	}
	amqp_channel_open(conn, params->channel_number);
	if (check_for_amqp_error(amqp_get_rpc_reply(conn), "Opening channel")) {
		return 1;
	}

	// http://comments.gmane.org/gmane.comp.networking.rabbitmq.general/5468
	// no_ack should be true since we are not sending out acknowledge messages
	amqp_boolean_t no_local = 0;
	amqp_boolean_t no_ack = 1;
	amqp_boolean_t exclusive = 0;
	amqp_basic_consume(conn, params->channel_number,
			amqp_cstring_bytes(params->queuename), amqp_empty_bytes, no_local,
			no_ack, exclusive, amqp_empty_table);
	return check_for_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
}

char * read_json_message() {
	amqp_maybe_release_buffers(conn);
	res = amqp_consume_message(conn, &envelope, timeout, 0);
	if (AMQP_RESPONSE_NORMAL != res.reply_type) {
		free_json_message();
		return NULL;
	}
	size_t const len = envelope.message.body.len;
	if (len == 0) {
		free_json_message();
		return NULL;
	}
	char * firstbyte = (char *) envelope.message.body.bytes;
	char * lastbyte = firstbyte + len - 1;
	if (*lastbyte != 0) {
		if (*lastbyte != '\n')
			logit("e", "bytes are not null terminated (%d)\n", *lastbyte);
		*lastbyte = 0;
	}
	// if new length is greater than buffer size
	if (envelope.message.body.len > msg_buffer_size) {
		if (msg_buffer != NULL) // if message buffer exists
			free(msg_buffer); // free the message buffer
		msg_buffer_size = (envelope.message.body.len / 1024 + 1) * 1024;
		msg_buffer = malloc(msg_buffer_size);
	} else if (memcmp(msg_buffer, envelope.message.body.bytes,
			envelope.message.body.len) == 0) {
                if (Verbose & VERBOSE_DUP) {
		   logit("e", "duplicate message:\n%s\n", msg_buffer);
		}
		free_json_message();
		return NULL;
	}
	memcpy(msg_buffer, envelope.message.body.bytes, envelope.message.body.len);
	return msg_buffer;
}

void set_json_connection_params_to_defaults(JSON_CONN_PARAMS *p) {
	memset(p, 0, sizeof(JSON_CONN_PARAMS));
	p->channel_number = JSON_CONN_DEFAULT_CHANNEL_NUMBER;
	p->frame_max_size = JSON_CONN_DEFAULT_FRAME_MAX_SIZE;
	p->heartbeat_seconds = JSON_CONN_DEFAULT_HEARTBEAT_SECONDS;
	p->max_channels = JSON_CONN_DEFAULT_MAX_CHANNELS;
	p->vhost = JSON_CONN_DEFAULT_VHOST;
	p->read_timeout = JSON_CONN_DEFAULT_READ_TIMEOUT;
	p->data_timeout = JSON_CONN_DEFAULT_DATA_TIMEOUT;
}
