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

Packet cancellation, IVIMs

This commit is contained in:
emanuel 2020-11-03 19:48:30 +00:00
parent 2aa194b18f
commit a6c55c090c
8 changed files with 627 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/CMakeLists.txt
View File

@ -3,6 +3,7 @@ ADD_EXECUTABLE(it2s-itss-facilities
queue.c
cam.c
denm.c
infrastructure.c
facilities.c
)

113
src/denm.c
View File

@ -51,16 +51,14 @@ static enum EVENT_CHECK_RESULT event_check(den_t *den, DENM_t *denm) {
pthread_mutex_lock(&den->lock);
for (int i = 0; i < den->no_max_events; ++i) {
for (int i = 0; i < den->n_max_events; ++i) {
if (den->events[i]->enabled) {
if (den->events[i]->station_id == denm->denm.management.actionID.originatingStationID &&
den->events[i]->sn == denm->denm.management.actionID.sequenceNumber) {
den->events[i]->sn == denm->denm.management.actionID.sequenceNumber) {
is_new = false;
if (den->events[i]->reference_time < e_reference_time ||
den->events[i]->detection_time < e_detection_time) {
is_update = true;
}
is_new = false;
if (den->events[i]->reference_time < e_reference_time) is_update = true;
break;
}
@ -68,7 +66,7 @@ static enum EVENT_CHECK_RESULT event_check(den_t *den, DENM_t *denm) {
}
if (is_new) {
for (int i = 0; i < den->no_max_events; ++i) {
for (int i = 0; i < den->n_max_events; ++i) {
if (!den->events[i]->enabled) {
exists_vacancy = true;
break;
@ -78,15 +76,16 @@ static enum EVENT_CHECK_RESULT event_check(den_t *den, DENM_t *denm) {
pthread_mutex_unlock(&den->lock);
if (denm->denm.management.termination != NULL) {
if (*denm->denm.management.termination == 0) {
return EVENT_CANCELLATION;
} else if (*denm->denm.management.termination == 1) {
return EVENT_NEGATION;
}
}
if (is_update) {
if (denm->denm.management.termination != NULL) {
if (*denm->denm.management.termination == 0) {
return EVENT_CANCELLATION;
} else if (*denm->denm.management.termination == 1) {
return EVENT_NEGATION;
}
}
return EVENT_UPDATE;
}
@ -101,7 +100,7 @@ static enum EVENT_CHECK_RESULT event_check(den_t *den, DENM_t *denm) {
return EVENT_NEW;
}
static int event_add(den_t *den, DENM_t *denm) {
static int event_add(den_t *den, DENM_t *denm, int64_t* id) {
struct timespec systemtime;
clock_gettime(CLOCK_REALTIME, &systemtime);
@ -122,7 +121,7 @@ static int event_add(den_t *den, DENM_t *denm) {
pthread_mutex_lock(&den->lock);
int index = -1;
for (int i = 0; i < den->no_max_events; ++i) {
for (int i = 0; i < den->n_max_events; ++i) {
if (!den->events[i]->enabled) {
index = i;
break;
@ -134,23 +133,25 @@ static int event_add(den_t *den, DENM_t *denm) {
if (*(denm->denm.management.termination) == 0) {
state = EVENT_CANCELLED;
} else if (*(denm->denm.management.termination) == 1) {
state = EVENT_NEGATION;
state = EVENT_NEGATED;
}
}
switch (state) {
case EVENT_ACTIVE:
++den->no_active_events;
++den->n_active_events;
break;
case EVENT_CANCELLED:
++den->no_cancelled_events;
++den->n_cancelled_events;
break;
case EVENT_NEGATED:
++den->no_negated_events;
++den->n_negated_events;
break;
}
if (index != -1) {
den->events[index]->id = rand();
*id = den->events[index]->id;
den->events[index]->enabled = true;
den->events[index]->state = state;
den->events[index]->station_id = denm->denm.management.actionID.originatingStationID;
@ -175,7 +176,7 @@ static int event_add(den_t *den, DENM_t *denm) {
else return 0; // Event added to db
}
static int event_update(den_t *den, DENM_t *denm) {
static int event_update(den_t *den, DENM_t *denm, int64_t* id) {
struct timespec systemtime;
clock_gettime(CLOCK_REALTIME, &systemtime);
@ -191,17 +192,18 @@ static int event_update(den_t *den, DENM_t *denm) {
if (*(denm->denm.management.termination) == 0) {
state = EVENT_CANCELLED;
} else if (*(denm->denm.management.termination) == 1) {
state = EVENT_NEGATION;
state = EVENT_NEGATED;
}
}
int index = -1;
pthread_mutex_lock(&den->lock);
for (int i = 0; i < den->no_max_events; ++i) {
for (int i = 0; i < den->n_max_events; ++i) {
if (den->events[i]->enabled) {
if (den->events[i]->station_id == denm->denm.management.actionID.originatingStationID &&
den->events[i]->sn == denm->denm.management.actionID.sequenceNumber) {
den->events[i]->sn == denm->denm.management.actionID.sequenceNumber) {
index = i;
break;
}
@ -210,29 +212,31 @@ static int event_update(den_t *den, DENM_t *denm) {
switch (den->events[index]->state) {
case EVENT_ACTIVE:
--den->no_active_events;
--den->n_active_events;
break;
case EVENT_CANCELLED:
--den->no_cancelled_events;
--den->n_cancelled_events;
break;
case EVENT_NEGATED:
--den->no_negated_events;
--den->n_negated_events;
break;
}
switch (state) {
case EVENT_ACTIVE:
++den->no_active_events;
++den->n_active_events;
break;
case EVENT_CANCELLED:
++den->no_cancelled_events;
++den->n_cancelled_events;
break;
case EVENT_NEGATED:
++den->no_negated_events;
++den->n_negated_events;
break;
}
if (index != -1) {
*id = den->events[index]->id;
den->events[index]->state = state;
den->events[index]->detection_time = e_detection_time;
den->events[index]->cause = denm->denm.situation->eventType.causeCode;
@ -243,6 +247,10 @@ static int event_update(den_t *den, DENM_t *denm) {
den->events[index]->expiration_time = e_detection_time + *(uint32_t *) denm->denm.management.validityDuration * 1000;
}
if (state == EVENT_CANCELLED || state == EVENT_NEGATED) { // Keep terminated events for 10 mins
den->events[index]->expiration_time = now + 600 * 1000;
}
den->events[index]->latitude = denm->denm.management.eventPosition.latitude;
den->events[index]->longitude = denm->denm.management.eventPosition.longitude;
@ -256,16 +264,15 @@ static int event_update(den_t *den, DENM_t *denm) {
else return 0; // Event updated
}
int event_manage(den_t *den, DENM_t *denm) {
int event_manage(den_t *den, DENM_t *denm, int64_t* id) {
int rv = 0;
switch (event_check(den, denm)) {
switch (rv = event_check(den, denm)) {
case EVENT_NEW:
rv = event_add(den, denm);
syslog_debug("[facilities] [den] new event received");
if (rv) {
if (event_add(den, denm, id)) {
syslog_debug("[facilities] [den] failed adding event, max events reached");
ASN_STRUCT_FREE(asn_DEF_DENM, denm);
rv = 1;
rv = -1;
}
break;
case EVENT_INVALID:
@ -278,32 +285,30 @@ int event_manage(den_t *den, DENM_t *denm) {
break;
case EVENT_CANCELLATION:
syslog_debug("[facilities] [den] event cancellation received");
rv = event_update(den, denm);
if (rv) {
if (event_update(den, denm, id)) {
syslog_debug("[facilities] [den] failed cancelling event, event not found");
ASN_STRUCT_FREE(asn_DEF_DENM, denm);
rv = 1;
rv = -1;
}
break;
case EVENT_NEGATION:
syslog_debug("[facilities] [den] event negation received");
rv = event_update(den, denm);
if (rv) {
if (event_update(den, denm, id)) {
syslog_debug("[facilities] [den] failed negating event, event not found");
ASN_STRUCT_FREE(asn_DEF_DENM, denm);
rv = 1;
rv = -1;
}
break;
case EVENT_UPDATE:
syslog_debug("[facilities] [den] event update received");
rv = event_update(den, denm);
if (rv) {
if (event_update(den, denm, id)) {
syslog_debug("[facilities] [den] failed updating event, event not found");
ASN_STRUCT_FREE(asn_DEF_DENM, denm);
rv = -1;
}
break;
case EVENT_REPEATED:
syslog_debug("[facilities] [den] repeated event received, ignoring");
syslog_debug("[facilities] [den] repeated event received or referenceTime doesn't allow an update, ignoring");
ASN_STRUCT_FREE(asn_DEF_DENM, denm);
break;
case EVENT_NUMBER_EXCEEDED:
@ -323,10 +328,10 @@ void* den_service(void *fc) {
den_t *den = facilities->den;
pthread_mutex_init(&den->lock, NULL);
den->no_max_events = 32;
den->n_max_events = 32;
den->events = malloc(den->no_max_events * sizeof(void*));
for (int i = 0; i < den->no_max_events; ++i) {
den->events = malloc(den->n_max_events * sizeof(void*));
for (int i = 0; i < den->n_max_events; ++i) {
den->events[i] = calloc(1, sizeof(event_t));
}
@ -336,20 +341,22 @@ void* den_service(void *fc) {
now = now - 1072915200000; // Convert EPOCH to 2004/01/01 00:00:000
pthread_mutex_lock(&den->lock);
for (int i = 0; i < den->no_max_events; ++i) {
for (int i = 0; i < den->n_max_events; ++i) {
if (den->events[i]->enabled) {
if (now >= den->events[i]->expiration_time) { // Remove event
syslog_debug("[facilities] [den] removed event %d (expiration)", i);
den->events[i]->enabled = false;
ASN_STRUCT_FREE(asn_DEF_DENM, den->events[i]->denm);
switch (den->events[i]->state) {
case EVENT_ACTIVE:
--den->no_active_events;
--den->n_active_events;
break;
case EVENT_CANCELLED:
--den->no_cancelled_events;
--den->n_cancelled_events;
break;
case EVENT_NEGATED:
--den->no_negated_events;
--den->n_negated_events;
break;
}
} else {
@ -366,7 +373,7 @@ void* den_service(void *fc) {
}
}
syslog_info("[facilities] [den] events :: [ %d active | %d cancelled | %d negated ]", den->no_active_events, den->no_cancelled_events, den->no_negated_events);
syslog_info("[facilities] [den] events :: [ %d active | %d cancelled | %d negated ]", den->n_active_events, den->n_cancelled_events, den->n_negated_events);
pthread_mutex_unlock(&den->lock);

15
src/denm.h
View File

@ -13,6 +13,7 @@ enum EVENT_STATE {
};
typedef struct event {
uint64_t id;
DENM_t *denm;
uint32_t station_id;
uint32_t sn;
@ -30,11 +31,11 @@ typedef struct event {
typedef struct den {
event_t ** events;
uint32_t sn;
uint16_t no_stored_events;
uint16_t no_active_events;
uint16_t no_cancelled_events;
uint16_t no_negated_events;
uint16_t no_max_events;
uint16_t n_stored_events;
uint16_t n_active_events;
uint16_t n_cancelled_events;
uint16_t n_negated_events;
uint16_t n_max_events;
pthread_mutex_t lock;
} den_t;
@ -56,8 +57,8 @@ enum EVENT_CHECK_RESULT {
* @param denm the DENM to evaluate
* @return 0 if event OK, 1 if event NOK
*/
int event_manage(den_t *den, DENM_t *denm);
int event_manage(den_t* den, DENM_t* denm, int64_t* id);
void* den_service(void *fc);
void* den_service(void* fc);
#endif

117
src/facilities.c
View File

@ -86,13 +86,15 @@ static int transport_indication(facilities_t *facilities, void* responder, uint8
break;
case Port_denm:
;
#ifdef DEBUG
uint8_t* xml_denm = malloc(4096);
asn_enc_rval_t rve = xer_encode_to_buffer(xml_denm, 4096, 0x02, &asn_DEF_DENM, its_msg);
uint8_t* xml_denm = malloc(32768);
asn_enc_rval_t rve = xer_encode_to_buffer(xml_denm, 32768, 0x02, &asn_DEF_DENM, its_msg);
syslog_debug("DENM XER %d: %.*s", (int)rve.encoded, (int)rve.encoded , xml_denm);
free(xml_denm);
#endif
event_manage(facilities->den, its_msg);
int64_t id = -1;
event_manage(facilities->den, its_msg, &id);
break;
default:
@ -108,9 +110,9 @@ static int transport_indication(facilities_t *facilities, void* responder, uint8
fdi->data.buf = malloc(bdi->data.size);
memcpy(fdi->data.buf, bdi->data.buf, bdi->data.size);
uint8_t buffer[512];
uint8_t buffer[PACKET_MAX_LEN];
buffer[0] = 4; // Facilities
asn_enc_rval_t enc = oer_encode_to_buffer(&asn_DEF_FacilitiesDataIndication, NULL, fdi, buffer+1, 511);
asn_enc_rval_t enc = oer_encode_to_buffer(&asn_DEF_FacilitiesDataIndication, NULL, fdi, buffer+1, PACKET_MAX_LEN-1);
queue_add(facilities->tx_queue, buffer, enc.encoded+1, 5);
pthread_cond_signal(&facilities->tx_queue->trigger);
@ -133,7 +135,7 @@ static int facilities_request(facilities_t *facilities, void* responder, uint8_t
void *its_msg = NULL;
asn_TYPE_descriptor_t *its_msg_def = NULL;
bool managed_event = false;
bool managed_msg = false;
FacilitiesDataRequest_t *fdreq = calloc(1, sizeof(FacilitiesDataRequest_t));
FacilitiesDataResult_t *fdres = calloc(1, sizeof(FacilitiesDataResult_t));
@ -154,6 +156,7 @@ static int facilities_request(facilities_t *facilities, void* responder, uint8_t
uint32_t transmission_duration = 0;
uint32_t transmission_interval = 0;
uint64_t transmission_start = 0;
switch (fdreq->choice.singleMessage.itssMessageType) {
case ItssMessageType_cam:
@ -193,21 +196,66 @@ static int facilities_request(facilities_t *facilities, void* responder, uint8_t
goto cleanup;
}
int64_t id = -1;
bool is_update = false;
if (fdreq->choice.singleMessage.itssMessageType == ItssMessageType_denm) {
managed_event = true;
if (event_manage(facilities->den, its_msg)) {
managed_msg = true;
uint8_t event_type = event_manage(facilities->den, its_msg, &id);
// Do not free its_msg! event_manage takes care of the msg
// id will get set to another val if EVENT NEW or UPDATE or CANCELLATION or NEGATION
if (event_type != EVENT_NEW &&
event_type != EVENT_UPDATE &&
event_type != EVENT_CANCELLATION &&
event_type != EVENT_NEGATION) {
fwd = false;
}
if (event_type == EVENT_UPDATE ||
event_type == EVENT_CANCELLATION ||
event_type == EVENT_NEGATION) {
is_update = true;
}
if ( ((DENM_t*)its_msg)->denm.management.transmissionInterval ) {
transmission_interval = *( (uint32_t*) ((DENM_t*)its_msg)->denm.management.transmissionInterval );
if ( ((DENM_t*)its_msg)->denm.management.validityDuration ) {
transmission_duration = *( (uint32_t*) ((DENM_t*)its_msg)->denm.management.validityDuration );
transmission_duration = *( (uint32_t*) ((DENM_t*)its_msg)->denm.management.validityDuration ) * 1000;
} else {
transmission_duration = 30000;
}
}
} else if (fdreq->choice.singleMessage.itssMessageType == ItssMessageType_ivim) {
managed_msg = true;
uint8_t service_type = service_eval(facilities->infrastructure, SERVICE_IVI, its_msg, &id);
if (service_type != SERVICE_NEW &&
service_type != SERVICE_UPDATE &&
service_type != SERVICE_CANCELLATION &&
service_type != SERVICE_NEGATION) {
fwd = false;
}
if (service_type == SERVICE_UPDATE ||
service_type == SERVICE_CANCELLATION ||
service_type == SERVICE_NEGATION) {
is_update = true;
}
uint64_t valid_to, valid_from;
asn_INTEGER2ulong((INTEGER_t*) &((IVIM_t*) its_msg)->ivi.mandatory.validTo, &valid_to);
asn_INTEGER2ulong((INTEGER_t*) &((IVIM_t*) its_msg)->ivi.mandatory.validFrom, &valid_from);
transmission_start = valid_from;
transmission_interval = 1000;
transmission_duration = valid_to - valid_from;
}
// Respond to [itss]
@ -223,6 +271,12 @@ static int facilities_request(facilities_t *facilities, void* responder, uint8_t
// Forward message to [transport]
if (fwd) {
if (id != -1) {
bdr->id = id;
} else {
bdr->id = 0;
}
bdr->btpType = BTPType_btpB;
bdr->data.buf = malloc(fdreq->choice.singleMessage.data.size);
@ -233,6 +287,11 @@ static int facilities_request(facilities_t *facilities, void* responder, uint8_t
for (int i = 0; i < 6; ++i) bdr->gnDestinationAddress.buf[i] = 0xff;
bdr->gnDestinationAddress.size = 6;
if (transmission_start) {
bdr->gnRepetitionStart = malloc(sizeof(long));
*bdr->gnRepetitionStart = transmission_start;
}
if (transmission_interval) {
bdr->gnRepetitionInterval = malloc(sizeof(long));
*bdr->gnRepetitionInterval = transmission_interval;
@ -243,10 +302,15 @@ static int facilities_request(facilities_t *facilities, void* responder, uint8_t
*bdr->gnMaximumRepetitionTime = transmission_duration;
}
if (is_update) {
bdr->gnIsUpdate = malloc(sizeof(long));
*bdr->gnIsUpdate = 1;
}
// Encode ITS message into OER
uint8_t bdr_oer[384];
uint8_t bdr_oer[2048];
bdr_oer[0] = 4; // [facilities] service id
enc = oer_encode_to_buffer(&asn_DEF_BTPDataRequest, NULL, bdr, bdr_oer + 1, 383);
enc = oer_encode_to_buffer(&asn_DEF_BTPDataRequest, NULL, bdr, bdr_oer + 1, 2047);
if (enc.encoded == -1) {
syslog_err("[facilities] failed encoding BDR (%s)", enc.failed_type->name);
rv = 1;
@ -265,7 +329,7 @@ static int facilities_request(facilities_t *facilities, void* responder, uint8_t
fdres->code = ResultCode_accepted;
fdres->events = calloc(1, sizeof(Events_t));
uint16_t nae = facilities->den->no_active_events;
uint16_t nae = facilities->den->n_active_events;
switch (fdreq->choice.activeEvents) {
case EventType_active:
@ -277,8 +341,8 @@ static int facilities_request(facilities_t *facilities, void* responder, uint8_t
if (facilities->den->events[i]->state == EVENT_ACTIVE) {
fdres->events->list.array[j] = calloc(1, sizeof(ItssMessage_t));
fdres->events->list.array[j]->itssMessageType = ItssMessageType_denm;
fdres->events->list.array[j]->data.buf = malloc(512);
asn_enc_rval_t enc = uper_encode_to_buffer(&asn_DEF_DENM, NULL, facilities->den->events[i]->denm, fdres->events->list.array[j]->data.buf, 512);
fdres->events->list.array[j]->data.buf = malloc(2048);
asn_enc_rval_t enc = uper_encode_to_buffer(&asn_DEF_DENM, NULL, facilities->den->events[i]->denm, fdres->events->list.array[j]->data.buf, 2048);
if (enc.encoded == -1) {
syslog_err("[facilities] failed encoding DENM for FDResult (%s)", enc.failed_type->name);
continue;
@ -294,8 +358,8 @@ static int facilities_request(facilities_t *facilities, void* responder, uint8_t
rv = 1;
goto cleanup;
}
fdres_oer = malloc(4096);
asn_enc_rval_t enc = oer_encode_to_buffer(&asn_DEF_FacilitiesDataResult, NULL, fdres, fdres_oer, 4096);
fdres_oer = malloc(16384);
asn_enc_rval_t enc = oer_encode_to_buffer(&asn_DEF_FacilitiesDataResult, NULL, fdres, fdres_oer, 16384);
if (enc.encoded == -1) {
syslog_err("[facilities] failed encoding FDResult (%s)", enc.failed_type->name);
pthread_mutex_unlock(&facilities->den->lock);
@ -321,7 +385,7 @@ static int facilities_request(facilities_t *facilities, void* responder, uint8_t
zmq_send(responder, fdres_oer, enc.encoded, 0);
}
free(fdres_oer);
if (its_msg_def && !managed_event) ASN_STRUCT_FREE(*its_msg_def, its_msg);
if (its_msg_def && !managed_msg) ASN_STRUCT_FREE(*its_msg_def, its_msg);
ASN_STRUCT_FREE(asn_DEF_FacilitiesDataResult, fdres);
ASN_STRUCT_FREE(asn_DEF_FacilitiesDataRequest, fdreq);
ASN_STRUCT_FREE(asn_DEF_BTPDataRequest, bdr);
@ -384,6 +448,9 @@ int main() {
facilities_t facilities;
facilities.exit = false;
time_t t;
srand((unsigned) time(&t));
if (itss_config(&facilities, "/etc/it2s/itss.toml")) return 1;
void *context = zmq_ctx_new();
@ -398,6 +465,8 @@ int main() {
facilities.den = calloc(1, sizeof(den_t));
facilities.infrastructure = calloc(1, sizeof(infrastructure_t));
// Tx
pthread_create(&facilities.transmitting, NULL, tx, (void*) &facilities);
@ -407,20 +476,24 @@ int main() {
// DEN
pthread_create(&facilities.den_service, NULL, den_service, (void*) &facilities);
uint8_t buffer[512];
// Infrastructure
pthread_create(&facilities.infrastructure_service, NULL, infrastructure_service, (void*) &facilities);
uint8_t buffer[PACKET_MAX_LEN];
uint8_t code;
syslog_info("[facilities] listening");
while (!facilities.exit) {
memset(buffer, 0x00, 512);
zmq_recv(responder, buffer, 512, 0);
memset(buffer, 0x00, PACKET_MAX_LEN);
zmq_recv(responder, buffer, PACKET_MAX_LEN, 0);
switch (buffer[0]) {
case 3:
code = transport_indication(&facilities, responder, buffer+1, 511);
code = transport_indication(&facilities, responder, buffer+1, PACKET_MAX_LEN-1);
break;
case 5:
code = facilities_request(&facilities, responder, buffer+1, 511);
code = facilities_request(&facilities, responder, buffer+1, PACKET_MAX_LEN-1);
break;
default:
syslog_debug("[facilities] unrecognized service");

5
src/facilities.h
View File

@ -6,6 +6,7 @@
#include "cam.h"
#include "denm.h"
#include "infrastructure.h"
#include "queue.h"
#define FACILITIES_ADDRESS "ipc:///tmp/itss/facilities"
@ -15,6 +16,7 @@
typedef struct facilities {
pthread_t ca_service;
pthread_t den_service;
pthread_t infrastructure_service;
pthread_t transmitting;
// ZMQ
@ -29,6 +31,9 @@ typedef struct facilities {
// DEN
den_t *den;
// Infrastructure
infrastructure_t* infrastructure;
int station_type;
bool exit;

393
src/infrastructure.c Normal file
View File

@ -0,0 +1,393 @@
#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;
switch (type) {
case SERVICE_IVI:
break;
default:
return -1;
}
IVIM_t* ivim = (IVIM_t*) its_msg;
uint64_t timestamp, valid_to, valid_from;
asn_INTEGER2ulong((INTEGER_t*) &ivim->ivi.mandatory.timeStamp, &timestamp);
asn_INTEGER2ulong((INTEGER_t*) &ivim->ivi.mandatory.validTo, &valid_to);
asn_INTEGER2ulong((INTEGER_t*) &ivim->ivi.mandatory.validFrom, &valid_from);
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
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;
asn_INTEGER2ulong((INTEGER_t*) &ivim->ivi.mandatory.timeStamp, &timestamp);
asn_INTEGER2ulong((INTEGER_t*) &ivim->ivi.mandatory.validTo, &valid_to);
asn_INTEGER2ulong((INTEGER_t*) &ivim->ivi.mandatory.validFrom, &valid_from);
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;
asn_INTEGER2ulong((INTEGER_t*) &ivim->ivi.mandatory.timeStamp, &timestamp);
asn_INTEGER2ulong((INTEGER_t*) &ivim->ivi.mandatory.validTo, &valid_to);
asn_INTEGER2ulong((INTEGER_t*) &ivim->ivi.mandatory.validFrom, &valid_from);
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;
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 EVENT_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 EVENT_INVALID:
syslog_debug("[facilities] [infrastructure] invalid service received, ignoring");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
break;
case EVENT_PASSED:
syslog_debug("[facilities] [infrastructure] old service received, ignoring");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
break;
case EVENT_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 EVENT_NEGATION:
syslog_debug("[facilities] [infrastructure] service negation received");
if (service_update(infrastructure, type, its_msg, id)) {
syslog_debug("[facilities] [infrastructure] failed negating service, event not found");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
rv = -1;
}
break;
case EVENT_UPDATE:
syslog_debug("[facilities] [infrastructure] service update received");
if (service_update(infrastructure, type, its_msg, id)) {
syslog_debug("[facilities] [infrastructure] failed updating service, event not found");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
rv = -1;
}
break;
case EVENT_REPEATED:
syslog_debug("[facilities] [infrastructure] repeated service received or referenceTime doesn't allow an update, ignoring");
ASN_STRUCT_FREE(asn_DEF_IVIM, its_msg);
break;
case EVENT_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->n_max_services = 32;
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, (now - infrastructure->services[i]->valid_from)/1000, (infrastructure->services[i]->valid_to - infrastructure->services[i]->valid_to)/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;
}

64
src/infrastructure.h Normal file
View File

@ -0,0 +1,64 @@
#ifndef FACILITIES_INFRASTRUCTURE_H
#define FACILITIES_INFRASTRUCTURE_H
#include <stdint.h>
#include <ivim/IVIM.h>
#include <pthread.h>
#include <stdbool.h>
enum SERVICE_STATE {
SERVICE_ACTIVE,
SERVICE_CANCELLED,
SERVICE_NEGATED
};
enum SERVICE_TYPE {
SERVICE_TLM,
SERVICE_RLT,
SERVICE_IVI,
SERVICE_TLC
};
typedef struct service {
uint64_t id;
bool enabled;
enum SERVICE_TYPE type;
uint64_t station_id;
enum SERVICE_STATE state;
IVIM_t *ivim;
uint64_t ivi_id_number;
uint64_t timestamp;
uint64_t valid_from;
uint64_t valid_to;
} service_t;
typedef struct infrastructure {
service_t ** services;
uint16_t n_services;
uint16_t n_active_services;
uint16_t n_awaiting_services;
uint16_t n_cancelled_services;
uint16_t n_negated_services;
uint16_t n_max_services;
pthread_mutex_t lock;
} infrastructure_t;
enum SERVICE_EVAL_RESULT {
SERVICE_NEW,
SERVICE_INVALID,
SERVICE_PASSED,
SERVICE_CANCELLATION,
SERVICE_NEGATION,
SERVICE_UPDATE,
SERVICE_REPEATED,
SERVICE_NUMBER_EXCEEDED
};
enum SERVICE_EVAL_RESULT service_eval(infrastructure_t* infrastructure, enum SERVICE_TYPE type, void* its_msg, int64_t* id);
void* infrastructure_service(void* fc);
#endif

2
src/queue.h
View File

@ -5,7 +5,7 @@
#include <stdint.h>
#define QUEUE_MAX_LEN 32
#define PACKET_MAX_LEN 512
#define PACKET_MAX_LEN 8192
typedef struct queue {
uint8_t **packet;