TiagoRG
/
it2s-itss-facilities
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
it2s-itss-facilities/src/infrastructure.c

437 lines
15 KiB
C
Raw Blame History

#include "infrastructure.h"
#include "facilities.h"
#include <unistd.h>
#include <syslog.h>
#define syslog_info(msg, ...) syslog(LOG_INFO, msg, ##__VA_ARGS__)
#define syslog_emerg(msg, ...) syslog(LOG_EMERG, "%s:%d [" msg "]", __func__, __LINE__, ##__VA_ARGS__)
#define syslog_err(msg, ...) syslog(LOG_ERR, "%s:%d [" msg "]", __func__, __LINE__, ##__VA_ARGS__)
#ifndef NDEBUG
#define syslog_debug(msg, ...) syslog(LOG_DEBUG, "%s:%d " msg "", __func__, __LINE__, ##__VA_ARGS__)
#else
#define syslog_debug(msg, ...)
#endif
static enum SERVICE_EVAL_RESULT service_check(infrastructure_t* infrastructure, enum SERVICE_TYPE type, void* its_msg) {
int rv = 0;
struct timespec systemtime;
clock_gettime(CLOCK_REALTIME, &systemtime);
long now = (long) (systemtime.tv_sec * 1000 + systemtime.tv_nsec / 1E6);
now = now - 1072915200000; // Convert EPOCH to 2004/01/01 00:00:000
switch (type) {
case SERVICE_IVI:
break;
default:
return -1;
}
IVIM_t* ivim = (IVIM_t*) its_msg;
uint64_t timestamp, valid_to, valid_from;
if (!ivim->ivi.mandatory.timeStamp) {
timestamp = now;
} else {
asn_INTEGER2ulong((INTEGER_t*) ivim->ivi.mandatory.timeStamp, &timestamp);
}
if (!ivim->ivi.mandatory.validFrom) {
valid_from = now;
} else {
asn_INTEGER2ulong((INTEGER_t*) ivim->ivi.mandatory.validFrom, &valid_from);
}
if (!ivim->ivi.mandatory.validTo) {
valid_to = valid_from + infrastructure->default_service_duration;
} else {
asn_INTEGER2ulong((INTEGER_t*) ivim->ivi.mandatory.validTo, &valid_to);
}
if (now >= valid_to) {
return SERVICE_PASSED;
}
uint32_t validity_duration = valid_to - valid_from;
bool is_new = true;
bool is_update = false;
bool exists_vacancy = false;
pthread_mutex_lock(&infrastructure->lock);
for (int i = 0; i < infrastructure->n_max_services; ++i) {
if (infrastructure->services[i]->enabled) {
if (infrastructure->services[i]->ivi_id_number == ivim->ivi.mandatory.iviIdentificationNumber) {
is_new = false;
if (timestamp > infrastructure->services[i]->timestamp) is_update = true;
break;
}
}
}
if (is_new) {
for (int i = 0; i < infrastructure->n_max_services; ++i) {
if (!infrastructure->services[i]->enabled) {
exists_vacancy = true;
break;
}
}
}
pthread_mutex_unlock(&infrastructure->lock);
if (is_update) {
switch (ivim->ivi.mandatory.iviStatus) {
case 0:
break;
case 1:
return SERVICE_UPDATE;
case 2:
return SERVICE_CANCELLATION;
case 3:
return SERVICE_NEGATION;
default:
return SERVICE_INVALID;
}
}
if (!is_new) {
return SERVICE_REPEATED;
}
if (!exists_vacancy) {
return SERVICE_NUMBER_EXCEEDED;
}
return SERVICE_NEW;
}
static int service_add(infrastructure_t* infrastructure, enum SERVICE_TYPE type, void* its_msg, int64_t* id) {
switch (type) {
case SERVICE_IVI:
break;
default:
return -1;
}
IVIM_t* ivim = (IVIM_t*) its_msg;
struct timespec systemtime;
clock_gettime(CLOCK_REALTIME, &systemtime);
long now = (long) (systemtime.tv_sec * 1000 + systemtime.tv_nsec / 1E6);
now = now - 1072915200000; // Convert EPOCH to 2004/01/01 00:00:000
uint64_t timestamp, valid_to, valid_from;
if (!ivim->ivi.mandatory.timeStamp) {
timestamp = now;
} else {
asn_INTEGER2ulong((INTEGER_t*) ivim->ivi.mandatory.timeStamp, &timestamp);
}
if (!ivim->ivi.mandatory.validFrom) {
valid_from = now;
} else {
asn_INTEGER2ulong((INTEGER_t*) ivim->ivi.mandatory.validFrom, &valid_from);
}
if (!ivim->ivi.mandatory.validTo) {
valid_to = valid_from + infrastructure->default_service_duration;
} else {
asn_INTEGER2ulong((INTEGER_t*) ivim->ivi.mandatory.validTo, &valid_to);
}
pthread_mutex_lock(&infrastructure->lock);
int index = -1;
for (int i = 0; i < infrastructure->n_max_services; ++i) {
if (!infrastructure->services[i]->enabled) {
index = i;
break;
}
}
uint8_t state;
switch (ivim->ivi.mandatory.iviStatus) {
case 0:
case 1:
state = SERVICE_ACTIVE;
break;
case 2:
state = SERVICE_CANCELLED;
break;
case 3:
state = SERVICE_NEGATED;
break;
}
switch (state) {
case SERVICE_ACTIVE:
++infrastructure->n_active_services;
break;
case EVENT_CANCELLED:
++infrastructure->n_cancelled_services;
break;
case EVENT_NEGATED:
++infrastructure->n_negated_services;
break;
}
if (index != -1) {
infrastructure->services[index]->id = rand();
*id = infrastructure->services[index]->id;
infrastructure->services[index]->enabled = true;
infrastructure->services[index]->state = state;
infrastructure->services[index]->station_id = ivim->header.stationID;
infrastructure->services[index]->valid_from = valid_from;
infrastructure->services[index]->valid_to = valid_to;
infrastructure->services[index]->timestamp = timestamp;
infrastructure->services[index]->ivi_id_number = ivim->ivi.mandatory.iviIdentificationNumber;
}
pthread_mutex_unlock(&infrastructure->lock);
if (index == -1) return 1; // Max services reached
else return 0; // Services added to db
}
static int service_update(infrastructure_t* infrastructure, enum SERVICE_TYPE type, void* its_msg, int64_t* id) {
switch (type) {
case SERVICE_IVI:
break;
default:
return -1;
}
IVIM_t* ivim = (IVIM_t*) its_msg;
struct timespec systemtime;
clock_gettime(CLOCK_REALTIME, &systemtime);
long now = (long) (systemtime.tv_sec * 1000 + systemtime.tv_nsec / 1E6);
now = now - 1072915200000; // Convert EPOCH to 2004/01/01 00:00:000
uint64_t timestamp, valid_to, valid_from;
if (!ivim->ivi.mandatory.timeStamp) {
timestamp = now;
} else {
asn_INTEGER2ulong((INTEGER_t*) ivim->ivi.mandatory.timeStamp, &timestamp);
}
if (!ivim->ivi.mandatory.validFrom) {
valid_from = now;
} else {
asn_INTEGER2ulong((INTEGER_t*) ivim->ivi.mandatory.validFrom, &valid_from);
}
if (!ivim->ivi.mandatory.validTo) {
valid_to = valid_from + infrastructure->default_service_duration;
} else {
asn_INTEGER2ulong((INTEGER_t*) ivim->ivi.mandatory.validTo, &valid_to);
}
uint8_t state;
switch (ivim->ivi.mandatory.iviStatus) {
case 0:
case 1:
state = SERVICE_ACTIVE;
break;
case 2:
state = SERVICE_CANCELLED;
break;
case 3:
state = SERVICE_NEGATED;
break;
}
int index = -1;
pthread_mutex_lock(&infrastructure->lock);
for (int i = 0; i < infrastructure->n_max_services; ++i) {
if (infrastructure->services[i]->enabled) {
if (infrastructure->services[i]->ivi_id_number == ivim->ivi.mandatory.iviIdentificationNumber) {
index = i;
break;
}
}
}
switch (infrastructure->services[index]->state) {
case SERVICE_ACTIVE:
--infrastructure->n_active_services;
break;
case SERVICE_CANCELLED:
--infrastructure->n_cancelled_services;
break;
case SERVICE_NEGATED:
--infrastructure->n_negated_services;
break;
}
switch (state) {
case SERVICE_ACTIVE:
++infrastructure->n_active_services;
break;
case SERVICE_CANCELLED:
++infrastructure->n_cancelled_services;
break;
case SERVICE_NEGATED:
++infrastructure->n_negated_services;
break;
}
if (index != -1) {
*id = infrastructure->services[index]->id;
infrastructure->services[index]->enabled = true;
infrastructure->services[index]->state = state;
infrastructure->services[index]->station_id = ivim->header.stationID;
infrastructure->services[index]->valid_from = valid_from;
infrastructure->services[index]->valid_to = valid_to;
infrastructure->services[index]->timestamp = timestamp;
if (state == SERVICE_CANCELLED || state == SERVICE_NEGATED) { // Keep terminated events for 10 mins
infrastructure->services[index]->valid_to = now + 600 * 1000;
}
ASN_STRUCT_FREE(asn_DEF_IVIM, infrastructure->services[index]->ivim);
infrastructure->services[index]->ivim = ivim;
}
pthread_mutex_unlock(&infrastructure->lock);
if (index == -1) return 1; // Event not found
else return 0; // Event updated
}
enum SERVICE_EVAL_RESULT service_eval(infrastructure_t* infrastructure, enum SERVICE_TYPE type, void* its_msg, int64_t* id) {
int rv = 0;
switch (rv = service_check(infrastructure, type, its_msg)) {
case SERVICE_NEW:
syslog_debug("[facilities] [infrastructure] new service received");
if (service_add(infrastructure, type, its_msg, id)) {
syslog_debug("[facilities] [infrastructure] failed adding service, max services reached");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
rv = -1;
}
break;
case SERVICE_INVALID:
syslog_debug("[facilities] [infrastructure] invalid service received, ignoring");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
break;
case SERVICE_PASSED:
syslog_debug("[facilities] [infrastructure] old service received, ignoring");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
break;
case SERVICE_CANCELLATION:
syslog_debug("[facilities] [infrastructure] service cancellation received");
if (service_update(infrastructure, type, its_msg, id)) {
syslog_debug("[facilities] [infrastructure] failed cancelling service, event not found");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
rv = -1;
}
break;
case SERVICE_NEGATION:
syslog_debug("[facilities] [infrastructure] service negation received");
if (service_update(infrastructure, type, its_msg, id)) {
syslog_debug("[facilities] [infrastructure] failed negating service, service not found");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
rv = -1;
}
break;
case SERVICE_UPDATE:
syslog_debug("[facilities] [infrastructure] service update received");
if (service_update(infrastructure, type, its_msg, id)) {
syslog_debug("[facilities] [infrastructure] failed updating service, service not found");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
rv = -1;
}
break;
case SERVICE_REPEATED:
syslog_debug("[facilities] [infrastructure] repeated service received or timeStamp doesn't allow an update, ignoring");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
break;
case SERVICE_NUMBER_EXCEEDED:
syslog_debug("[facilities] [infrastructure] max services reached, ignoring");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
break;
}
return rv;
}
void* infrastructure_service(void *fc) {
facilities_t *facilities = (facilities_t *) fc;
struct timespec systemtime;
uint64_t now;
infrastructure_t *infrastructure = facilities->infrastructure;
pthread_mutex_init(&infrastructure->lock, NULL);
infrastructure->services = malloc(infrastructure->n_max_services * sizeof(void*));
for (int i = 0; i < infrastructure->n_max_services; ++i) {
infrastructure->services[i] = calloc(1, sizeof(service_t));
}
while (!facilities->exit) {
clock_gettime(CLOCK_REALTIME, &systemtime);
now = (long)(systemtime.tv_sec * 1000 + systemtime.tv_nsec / 1E6);
now = now - 1072915200000; // Convert EPOCH to 2004/01/01 00:00:000
int n_awaiting_services = 0;
pthread_mutex_lock(&infrastructure->lock);
for (int i = 0; i < infrastructure->n_max_services; ++i) {
if (infrastructure->services[i]->enabled) {
if (now > infrastructure->services[i]->valid_to) { // Remove service
syslog_debug("[facilities] [infrastructure] removed service %d (expiration)", i);
infrastructure->services[i]->enabled = false;
ASN_STRUCT_FREE(asn_DEF_IVIM, infrastructure->services[i]->ivim);
switch (infrastructure->services[i]->state) {
case SERVICE_ACTIVE:
--infrastructure->n_active_services;
break;
case SERVICE_CANCELLED:
--infrastructure->n_cancelled_services;
break;
case SERVICE_NEGATED:
--infrastructure->n_negated_services;
break;
}
} else {
switch (infrastructure->services[i]->state) {
case SERVICE_ACTIVE:
if (now > infrastructure->services[i]->valid_from) {
syslog_debug("[facilities] [infrastructure] service %d expiring in %ld second(s)", i, (infrastructure->services[i]->valid_to - now)/1000);
} else {
syslog_debug("[facilities] [infrastructure] service %d starting in %ld seconds with a duration of %ld seconds", i, (infrastructure->services[i]->valid_from - now)/1000, (infrastructure->services[i]->valid_to - infrastructure->services[i]->valid_from)/1000);
++n_awaiting_services;
}
break;
case SERVICE_CANCELLED:
break;
case SERVICE_NEGATED:
break;
}
}
}
}
syslog_info("[facilities] [infrastructure] services :: [ %d active | %d awaiting | %d cancelled | %d negated ]", infrastructure->n_active_services - n_awaiting_services, n_awaiting_services, infrastructure->n_cancelled_services, infrastructure->n_negated_services);
pthread_mutex_unlock(&infrastructure->lock);
usleep(1000000);
}
return NULL;
}
Reference in New Issue View Git Blame Copy Permalink