#include "usg_common.h"
#include "usg_typedef.h"
#include "shm_queue.h"
#include "shm_allocator.h"

#include "mem_pool.h"
#include "hashtable.h"
#include "sem_util.h"
#include "socket.h"
#include <map>


enum shm_msg_type_t
{
	SHM_SOCKET_OPEN = 1,
	SHM_SOKET_CLOSE = 2,
	SHM_COMMON_MSG = 3
	
};

typedef struct shm_msg_t {
	int port;
	shm_msg_type_t type;
	size_t size;
	void * buf;

} shm_msg_t;

typedef struct shm_socket_t {
	int port;
	shm_mod_t mod;
	SHMQueue<shm_msg_t> *queue;
	SHMQueue<shm_msg_t> *remoteQueue;
	// std::map<int, SHMQueue<shm_msg_t>* > *remoteQueueMap;
	int slots;
	int items;
	int is_server;

} shm_socket_t;



void shm_init(int size) {
	mem_pool_init(size);
}

void shm_destroy() {
	mem_pool_destroy();
}

void shm_free(void *buf) {
	free(buf);
}

void *shm_open_socket(int mod) {
	shm_socket_t *socket = (shm_socket_t *)malloc(sizeof(shm_socket_t));
	socket->remoteQueueMap = new std::map<int, SHMQueue<shm_msg_t>* >;
	socket->port = -1;
	socket->mod = (shm_mod_t)mod;
printf("mod===%d\n", socket->mod);
	socket->is_server = 0;
	if (mod == REQ_REP) {
		socket->slots = SemUtil::get(IPC_PRIVATE, 1);
    	socket->items = SemUtil::get(IPC_PRIVATE, 0);
	}
	
	return (void *)socket;
}


int shm_close_socket(void *socket) {
	shm_socket_t * _socket = (shm_socket_t *) socket;
	delete _socket->queue;

	std::map<int, SHMQueue<shm_msg_t>* > *remoteQueueMap = _socket->remoteQueueMap;
	for(auto iter = remoteQueueMap->begin(); iter != remoteQueueMap->end(); iter++) {
		delete iter->second;
	}
	delete _socket->remoteQueueMap;

	if (_socket->mod == REQ_REP) {
		SemUtil::remove(_socket->slots);
    	SemUtil::remove(_socket->items);
	}
	
	free(socket);
	return 0;

}


int shm_bind(void* socket, int port) {
	shm_socket_t * _socket = (shm_socket_t *) socket;
	_socket -> port = port;
	return 0;
}

int shm_listen(void* socket) {
	shm_socket_t * _socket = (shm_socket_t *) socket;
	_socket->is_server = 1;
	int  port;
	hashtable_t *hashtable = mm_get_hashtable();
	if(_socket -> port == -1) {
		port = hashtable_alloc_key(hashtable);
		_socket -> port = port;
	} else {
		 
		if(hashtable_get(hashtable, _socket->port)!= NULL) {
			err_exit(0, "key %d has already been in used!", _socket->port);
		}
	}

	_socket->queue = new SHMQueue<shm_msg_t>(_socket->port, 16);
	return 0;
}


static int __shm_rev__(shm_socket_t* _socket) {
	shm_msg_t src;
	
	std::map<int, SHMQueue<shm_msg_t>* > *remoteQueueMap = _socket->remoteQueueMap;
	bool rv = _socket->queue->pop(src);

	if (rv) {
		if(src.type == SHM_SOCKET_OPEN) {
			_socket->remoteQueue = new SHMQueue<shm_msg_t>(src.port, 0);
		} 



		 
		
		void * _buf = malloc(src.size);
		memcpy(_buf, src.buf, src.size);
		*buf = _buf;
		*size = src.size;
		mm_free(src.buf);
		return 0;
	} else {
		return 1;
	}
}

int shm_connect(void* socket, int port) {
	shm_socket_t * _socket = (shm_socket_t *) socket;
	hashtable_t *hashtable = mm_get_hashtable();
	if(hashtable_get(hashtable, port)== NULL) {
		err_exit(0, "shm_connect：connect at port %d  failed!", port);
	}
	if(_socket -> port == -1) {
		_socket -> port = hashtable_alloc_key(hashtable);
	} else {
		 
		if(hashtable_get(hashtable, _socket->port)!= NULL) {
			err_exit(0, "key %d has already been in used!", _socket->port);
		}
	}

	_socket->queue = new SHMQueue<shm_msg_t>(_socket->port, 16);
	_socket->remoteQueueMap->insert({port,  new SHMQueue<shm_msg_t>(port, 0)});
	return 0;
}

int shm_send(void *socket, void *buf, int size) {
	shm_socket_t * _socket = (shm_socket_t *) socket;
	hashtable_t *hashtable = mm_get_hashtable();
	shm_msg_t dest;
	dest.port = _socket->port;
	dest.size = size;
	dest.buf = mm_malloc(size);
	memcpy(dest.buf, buf, size);

	std::map<int, SHMQueue<shm_msg_t>* > *remoteQueueMap = _socket->remoteQueueMap;
	for(auto iter = remoteQueueMap->begin(); iter != remoteQueueMap->end(); iter++) {
		if(hashtable_get(hashtable, iter->first)== NULL) {
			err_msg(0, "shm_send：connect at port %d  failed, the other part has been closed!", iter->first);
			delete iter->second;
			remoteQueueMap->erase(iter);
			continue;
		}
		if(_socket->mod == REQ_REP && _socket->is_server == 1)
		  	SemUtil::dec(_socket->items);

		iter->second->push(dest);

		if( _socket->mod == REQ_REP && _socket->is_server == 1) {
			delete iter->second;
			remoteQueueMap->erase(iter);
			SemUtil::inc(_socket->slots);
		}

		  	
	}
	return 0;
}

int shm_recv(void* socket, void **buf, int *size) {
	shm_socket_t * _socket = (shm_socket_t *) socket;
	shm_msg_t src;
	
	std::map<int, SHMQueue<shm_msg_t>* > *remoteQueueMap = _socket->remoteQueueMap;
	bool rv = _socket->queue->pop(src);
	if (rv) {
		if( _socket->is_server == 1 && remoteQueueMap->find(src.port) == remoteQueueMap->end()) {
		 if(_socket->mod == REQ_REP)
		  	SemUtil::dec(_socket->slots);

		  remoteQueueMap->insert({src.port,  new SHMQueue<shm_msg_t>(src.port, 0)});

		  if(_socket->mod == REQ_REP)
		  	SemUtil::inc(_socket->items);
		}
		
		void * _buf = malloc(src.size);
		memcpy(_buf, src.buf, src.size);
		*buf = _buf;
		*size = src.size;
		mm_free(src.buf);
		return 0;
	} else {
		return 1;
	}
}