static int analyzer_jpeg_init(struct analyzer *analyzer) {

	struct analyzer_pload_type *pload_type = analyzer_pload_type_get_by_name(ANALYZER_JPEG_PLOAD_TYPE);
	
	if (!pload_type) {
		pomlog(POMLOG_ERR "Payload type " ANALYZER_JPEG_PLOAD_TYPE " not found");
		return POM_ERR;
	}

	static struct data_item_reg pload_jpeg_data_items[ANALYZER_JPEG_PLOAD_DATA_COUNT] = { { 0 } };
	pload_jpeg_data_items[analyzer_jpeg_pload_width].name = "width";
	pload_jpeg_data_items[analyzer_jpeg_pload_width].value_type = ptype_get_type("uint16");
	pload_jpeg_data_items[analyzer_jpeg_pload_height].name = "height";
	pload_jpeg_data_items[analyzer_jpeg_pload_height].value_type = ptype_get_type("uint16");
	pload_jpeg_data_items[analyzer_jpeg_pload_exif].name = "exif";
	pload_jpeg_data_items[analyzer_jpeg_pload_exif].flags = ANALYZER_DATA_FLAG_LIST;

	static struct data_reg pload_jpeg_data = {
		.items = pload_jpeg_data_items,
		.data_count = ANALYZER_JPEG_PLOAD_DATA_COUNT
	};

	static struct analyzer_pload_reg pload_reg;
	memset(&pload_reg, 0, sizeof(struct analyzer_pload_reg));
	pload_reg.analyzer = analyzer;
	pload_reg.analyze = analyzer_jpeg_pload_analyze;
	pload_reg.cleanup = analyzer_jpeg_pload_cleanup;
	pload_reg.data_reg = &pload_jpeg_data;
	pload_reg.flags = ANALYZER_PLOAD_PROCESS_PARTIAL;

	return analyzer_pload_register(pload_type, &pload_reg);
}

struct ptype* ptype_alloc_unit(const char* type, char* unit) {

	struct ptype_reg *reg;
	for (reg = ptype_reg_head; reg && strcmp(reg->info->name, type); reg = reg->next);

	if (!reg) {
		// This should only be needed at startup
		pomlog(POMLOG_DEBUG "Ptype of type %s not found, trying to load module", type);
		char ptype_mod_name[64] = { 0 };
		strcat(ptype_mod_name, "ptype_");
		strncat(ptype_mod_name, type, sizeof(ptype_mod_name) - 1 - strlen(ptype_mod_name));
		if (!mod_load(ptype_mod_name)) {
			pomlog(POMLOG_ERR "Ptype of type %s not found", type);
			return NULL;
		}
		for (reg = ptype_reg_head; reg && strcmp(reg->info->name, type); reg = reg->next);
		if (!reg) {
			pomlog(POMLOG_ERR "Ptype of type %s not found even after loading module", type);
			return NULL;
		}
	}
	
	struct ptype *ret = malloc(sizeof(struct ptype));
	if (!ret) {
		pom_oom(sizeof(struct ptype));
		return NULL;
	}

	memset(ret, 0, sizeof(struct ptype));
	ret->type = reg;
	if (reg->info->alloc) {
		if (reg->info->alloc(ret) != POM_OK) {
			pomlog(POMLOG_ERR "Error while allocating ptype %s", type);
			free(ret);
			return NULL;
		}
	}

	if (unit) {
		ret->unit = strdup(unit);
		if (!ret->unit)
			pom_oom(strlen(unit));
	}

	return ret;
}

int core_cleanup(int emergency_cleanup) {

	core_run = 0;


	int i;
	for (i = 0; i < CORE_PROCESS_THREAD_MAX && core_processing_threads[i]; i++) {
		struct core_processing_thread *t = core_processing_threads[i];
		int res = pthread_cond_signal(&t->pkt_queue_cond);
		if (res) {
			pomlog(POMLOG_ERR "Error while signaling the restart condition : %s", pom_strerror(res));
			abort();
		}
		pthread_join(t->thread, NULL);
		res = pthread_mutex_destroy(&t->pkt_queue_lock);
		if (res)
			pomlog(POMLOG_WARN "Error while destroying a processing thread lock : %s", pom_strerror(res));
		
		res = pthread_cond_destroy(&t->pkt_queue_cond);
		if (res)
			pomlog(POMLOG_WARN "Error while destroying a processing thread condition : %s", pom_strerror(res));


		struct core_packet_queue *tmp = NULL;
		while (t->pkt_queue_head) {
			tmp = t->pkt_queue_head;
			t->pkt_queue_head = tmp->next;
			// packet_pool_cleanup() was already called when the thread stopped
			// packet_pool_release(tmp->pkt);
			free(tmp);
			pomlog(POMLOG_WARN "A packet was still in a thread's queue");
		}

		while (t->pkt_queue_unused) {
			tmp = t->pkt_queue_unused;
			t->pkt_queue_unused = tmp->next;
			// packet_pool_cleanup() was already called when the thread stopped
			// packet_pool_release(tmp->pkt);
			free(tmp);
		}

		free(t);
	}

	return POM_OK;
}

void core_pause_processing() {

	int res = pthread_rwlock_wrlock(&core_processing_lock);
	if (res) {
		pomlog(POMLOG_ERR "Error while locking core processing lock : %s", pom_strerror(res));
		abort();
	}
}

struct event_reg *event_register(struct event_reg_info *reg_info) {

	struct event_reg *evt;

	// Check if an event with the same name has already been registered
	for (evt = event_reg_head; evt && strcmp(evt->info->name, reg_info->name); evt = evt->next);
	if (evt) {
		pomlog(POMLOG_ERR "An event named %s has already been registered", reg_info->name);
		return NULL;
	}

	// Allocate the event_reg
	evt = malloc(sizeof(struct event_reg));
	if (!evt) {
		pom_oom(sizeof(struct event_reg));
		return NULL;
	}
	memset(evt, 0, sizeof(struct event_reg));

	evt->reg_instance = registry_add_instance(event_registry_class, reg_info->name);
	if (!evt->reg_instance) {
		free(evt);
		return NULL;
	}

	evt->perf_listeners = registry_instance_add_perf(evt->reg_instance, "listeners", registry_perf_type_gauge, "Number of event listeners", "listeners");
	evt->perf_ongoing = registry_instance_add_perf(evt->reg_instance, "ongoing", registry_perf_type_gauge, "Number of ongoing events", "events");
	evt->perf_processed = registry_instance_add_perf(evt->reg_instance, "processed", registry_perf_type_counter, "Number of events fully processed", "events");
	if (!evt->perf_listeners || !evt->perf_ongoing || !evt->perf_processed) {
		registry_remove_instance(evt->reg_instance);
		free(evt);
		return NULL;
	}

	evt->info = reg_info;

	evt->next = event_reg_head;
	if (evt->next)
		evt->next->prev = evt;
	event_reg_head = evt;

	pomlog(POMLOG_DEBUG "Event %s registered", reg_info->name);

	return evt;
}

int proto_cleanup() {

    struct proto *proto;
    for (proto = proto_head; proto; proto = proto->next) {

        if (proto->info->cleanup && proto->info->cleanup(proto->priv) == POM_ERR)
            pomlog(POMLOG_WARN "Error while cleaning up protocol %s", proto->info->name);
        conntrack_table_cleanup(proto->ct);

        mod_refcount_dec(proto->info->mod);
    }

    while (proto_head) {
        proto = proto_head;
        proto_head = proto->next;

        int res = pthread_rwlock_destroy(&proto->listeners_lock);
        if (res)
            pomlog(POMLOG_ERR "Error while destroying the listners lock : %s", pom_strerror(res));
        res = pthread_rwlock_destroy(&proto->expectation_lock);
        if (res)
            pomlog(POMLOG_ERR "Error while destroying the listners lock : %s", pom_strerror(res));


        free(proto);
    }

    if (proto_registry_class)
        registry_remove_class(proto_registry_class);
    proto_registry_class = NULL;

    while (proto_number_class_head) {
        struct proto_number_class *cls = proto_number_class_head;
        proto_number_class_head = cls->next;
        while (cls->nums) {
            struct proto_number *num = cls->nums;
            cls->nums = num->next;
            free(num);
        }
        free(cls->name);
        free(cls);
    }

    return POM_OK;
}

int core_cleanup(int emergency_cleanup) {

	core_run = 0;

	if (!emergency_cleanup) {
		while (core_pkt_queue_head) {
			pomlog("Waiting for all the packets to be processed");
			if (pthread_cond_broadcast(&core_pkt_queue_restart_cond)) {
				pomlog(POMLOG_ERR "Error while signaling the restart condition : %s", pom_strerror(errno));
				return POM_ERR;
			}
			sleep(1);
		}
	}

	if (pthread_cond_broadcast(&core_pkt_queue_restart_cond)) {
		pomlog(POMLOG_ERR "Error while signaling the restart condition : %s", pom_strerror(errno));
		return POM_ERR;
	}

	int i;
	for (i = 0; i < CORE_PROCESS_THREAD_MAX && core_processing_threads[i]; i++) {
		pthread_join(core_processing_threads[i]->thread, NULL);
		free(core_processing_threads[i]);
	}

	pthread_cond_destroy(&core_pkt_queue_restart_cond);

	while (core_pkt_queue_head) {
		struct core_packet_queue *tmp = core_pkt_queue_head;
		core_pkt_queue_head = tmp->next;
		packet_pool_release(tmp->pkt);
		free(tmp);
		pomlog(POMLOG_WARN "A packet was still in the buffer");
	}

	while (core_pkt_queue_unused) {
		struct core_packet_queue *tmp = core_pkt_queue_unused;
		core_pkt_queue_unused = tmp->next;
		free(tmp);
	}

	
	return POM_OK;
}

int addon_log_xml_close(void *output_priv) {

    struct output_log_xml_priv *priv = output_priv;

    if (priv->fd == -1) {
        pomlog(POMLOG_ERR "Output already stopped");
        return POM_ERR;
    }

    if (close(priv->fd)) {
        pomlog(POMLOG_ERR "Error while closing log file : %s", pom_strerror(errno));
        return POM_ERR;
    }

    priv->fd = -1;

    return POM_OK;
}

int ptype_copy(struct ptype *dst, struct ptype *src) {

	if (dst->type != src->type) {
		pomlog(POMLOG_ERR "Error, trying to copy ptypes of different type");
		return POM_ERR;
	}

	return  src->type->info->copy(dst, src);
}

void conntrack_refcount_dec(struct conntrack_entry *ce) {
	pom_mutex_lock(&ce->lock);
	if (!ce->refcount) {
		pomlog(POMLOG_ERR "Reference count already 0 !");
		abort();
	}
	ce->refcount--;
	pom_mutex_unlock(&ce->lock);
}

void registry_perf_timeticks_restart(struct registry_perf *p) {

	if (p->type != registry_perf_type_timeticks) {
		pomlog(POMLOG_ERR "Trying to restart a non timetick performance");
		return;
	}

	if (p->value & REGISTRY_PERF_TIMETICKS_STARTED) {
		pomlog(POMLOG_ERR "Timeticks performance already started");
		return;
	}

	ptime now = pom_gettimeofday();

	// value currently holds the runtime in usec
	volatile uint64_t new_val = (now + REGISTRY_PERF_TIMETICKS_STARTED) - p->value;
	p->value = new_val;
}

int packet_info_pool_init(struct packet_info_pool *pool) {

	if (pthread_mutex_init(&pool->lock, NULL)) {
		pomlog(POMLOG_ERR "Error while initializing the pkt_info_pool lock : ", pom_strerror(errno));
		return POM_ERR;
	}

	return POM_OK;
}

int event_listener_register(struct event_reg *evt_reg, void *obj, int (*process_begin) (struct event *evt, void *obj, struct proto_process_stack *stack, unsigned int stack_index), int (*process_end) (struct event *evt, void *obj)) {

	struct event_listener *lst;
	for (lst = evt_reg->listeners; lst && lst->obj != obj; lst = lst->next);

	if (lst) {
		pomlog(POMLOG_ERR "Event %s is already being listened to by obj %p", evt_reg->info->name, obj);
		return POM_ERR;
	}
	
	
	lst = malloc(sizeof(struct event_listener));
	if (!lst) {
		pom_oom(sizeof(struct event_listener));
		return POM_ERR;

	}
	memset(lst, 0, sizeof(struct event_listener));
	
	lst->obj = obj;
	lst->process_begin = process_begin;
	lst->process_end = process_end;
	
	lst->next = evt_reg->listeners;
	if (lst->next)
		lst->next->prev = lst;

	evt_reg->listeners = lst;

	if (!lst->next) {
		// Got a listener now, notify
		if (evt_reg->info->listeners_notify && evt_reg->info->listeners_notify(evt_reg->info->source_obj, evt_reg, 1) != POM_OK) {
			pomlog(POMLOG_ERR "Error while notifying event object about new listener");
			evt_reg->listeners = NULL;
			free(lst);
			return POM_ERR;
		}
	}

	registry_perf_inc(evt_reg->perf_listeners, 1);
	

	return POM_OK;
}

void registry_perf_timeticks_stop(struct registry_perf *p) {

	if (p->type != registry_perf_type_timeticks) {
		pomlog(POMLOG_ERR "Trying to stop a non timetick performance");
		return;
	}

	if (!(p->value & REGISTRY_PERF_TIMETICKS_STARTED)) {
		pomlog(POMLOG_ERR "Timeticks performance already stopped");
		return;
	}

	ptime now = pom_gettimeofday();

	// value currently holds the absolute start time expressed in usec
	// Not sure if the below is correct ...
	volatile uint64_t new_val = (now + REGISTRY_PERF_TIMETICKS_STARTED) - p->value;
	p->value = new_val;
}

static int input_pcap_dir_open(struct input *i) {

	struct input_pcap_priv *p = i->priv;

	struct input_pcap_dir_priv *dp = &p->tpriv.dir;

	// Reset the interrupt flag
	dp->interrupt_scan = 0;

	pomlog(POMLOG_INFO "Scanning directory %s for pcap files ...", PTYPE_STRING_GETVAL(dp->p_dir));

	int found = input_pcap_dir_browse(p);

	if (dp->interrupt_scan)
		return POM_ERR;

	if (found == POM_ERR)
		return POM_ERR;

	pomlog(POMLOG_INFO "Found %u files", found);
	
	dp->cur_file = dp->files;

	// Skip files which were not read
	while (dp->cur_file && !dp->cur_file->first_pkt)
		dp->cur_file = dp->cur_file->next;

	if (!dp->cur_file) {
		pomlog(POMLOG_ERR "No useable file found");
		return POM_ERR;
	}

	char errbuf[PCAP_ERRBUF_SIZE + 1] = { 0 };
	p->p = pcap_open_offline(dp->cur_file->full_path, errbuf);
	if (!p->p) {
		pomlog(POMLOG_ERR "Error opening %s for reading", dp->cur_file->full_path);
		return POM_ERR;
	}

	pomlog("Reading file %s", dp->cur_file->filename);

	return input_pcap_common_open(i);
}

void xmlrcpcmd_serial_inc() {
	pom_mutex_lock(&xmlrpccmd_serial_lock);
	xmlrpccmd_serial++;
	if (pthread_cond_broadcast(&xmlrpccmd_serial_cond)) {
		pomlog(POMLOG_ERR "Error while signaling the serial condition. Aborting");
		abort();
	}
	pom_mutex_unlock(&xmlrpccmd_serial_lock);

}

int conntrack_timer_cleanup(struct conntrack_timer *t) {

#ifdef DEBUG_CONNTRACK
	int res = pthread_mutex_lock(&t->ce->lock);

	if (!res) {
		pomlog(POMLOG_ERR "Internal error, conntrack not locked when timer cleaned up");
		pom_mutex_unlock(&t->ce->lock);
	} else if (res != EDEADLK) {
		pomlog(POMLOG_ERR "Error while locking timer lock : %s", pom_strerror(errno));
		abort();
	}
#endif

	timer_cleanup(t->timer);
	free(t);
	return POM_OK;

}

int output_file_pload_close(void *pload_instance_priv) {

	struct output_file_pload_priv *ppriv = pload_instance_priv;
	int fd = ppriv->fd;
	pomlog(POMLOG_DEBUG "File %s closed", ppriv->filename);
	free(ppriv->filename);
	free(ppriv);

	return close(fd);
}

struct ptype_reg *ptype_get_type(char *name) {
	struct ptype_reg *tmp;
	
	for (tmp = ptype_reg_head; tmp && strcmp(tmp->info->name, name); tmp = tmp->next);

	if (!tmp)
		pomlog(POMLOG_WARN "Warning, requested ptype %s not found", name);

	return tmp;
}

int output_file_pload_write(void *pload_instance_priv, void *data, size_t len) {


	struct output_file_pload_priv *ppriv = pload_instance_priv;
	int res = pom_write(ppriv->fd, data, len);
	if (res == POM_ERR)
		pomlog(POMLOG_ERR "Error while writing to file %s : %s", ppriv->filename, pom_strerror(errno));

	return res;

}