From a6c55c090c52202f1bf8c34867cfc09c02661f95 Mon Sep 17 00:00:00 2001 From: emanuel Date: Tue, 3 Nov 2020 19:48:30 +0000 Subject: [PATCH] Packet cancellation, IVIMs --- src/CMakeLists.txt | 1 + src/denm.c | 113 +++++++------ src/denm.h | 15 +- src/facilities.c | 117 ++++++++++--- src/facilities.h | 5 + src/infrastructure.c | 393 +++++++++++++++++++++++++++++++++++++++++++ src/infrastructure.h | 64 +++++++ src/queue.h | 2 +- 8 files changed, 627 insertions(+), 83 deletions(-) create mode 100644 src/infrastructure.c create mode 100644 src/infrastructure.h diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index e6fbdbf..6f54c83 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -3,6 +3,7 @@ ADD_EXECUTABLE(it2s-itss-facilities queue.c cam.c denm.c + infrastructure.c facilities.c ) diff --git a/src/denm.c b/src/denm.c index b335c3f..d5e65a9 100644 --- a/src/denm.c +++ b/src/denm.c @@ -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); diff --git a/src/denm.h b/src/denm.h index 3397517..28e26d5 100644 --- a/src/denm.h +++ b/src/denm.h @@ -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 diff --git a/src/facilities.c b/src/facilities.c index 2e89a63..173b718 100644 --- a/src/facilities.c +++ b/src/facilities.c @@ -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"); diff --git a/src/facilities.h b/src/facilities.h index fbfbb24..aae3359 100644 --- a/src/facilities.h +++ b/src/facilities.h @@ -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; diff --git a/src/infrastructure.c b/src/infrastructure.c new file mode 100644 index 0000000..8c74041 --- /dev/null +++ b/src/infrastructure.c @@ -0,0 +1,393 @@ +#include "infrastructure.h" +#include "facilities.h" + +#include +#include + +#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, ×tamp); + 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, ×tamp); + 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, ×tamp); + 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; +} diff --git a/src/infrastructure.h b/src/infrastructure.h new file mode 100644 index 0000000..e53fe0c --- /dev/null +++ b/src/infrastructure.h @@ -0,0 +1,64 @@ +#ifndef FACILITIES_INFRASTRUCTURE_H +#define FACILITIES_INFRASTRUCTURE_H + +#include +#include +#include +#include + +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 diff --git a/src/queue.h b/src/queue.h index a3afecb..4fd05b1 100644 --- a/src/queue.h +++ b/src/queue.h @@ -5,7 +5,7 @@ #include #define QUEUE_MAX_LEN 32 -#define PACKET_MAX_LEN 512 +#define PACKET_MAX_LEN 8192 typedef struct queue { uint8_t **packet;