Initialized ndt queue in the same manner with the event queue

For reusing the pqueue.c, save ndt_queue stores dummy events that has
the tag of NDTs. Nonetheless, current pqueue.c can only use time as the
priority. So new pqueue should be implemented.

core/environment.c

@@ -107,8 +107,10 @@ static void environment_init_modes(environment_t* env, int num_modes, int num_st
  */
 static void environment_init_federated(environment_t* env, int num_is_present_fields) {
 #ifdef FEDERATED_CENTRALIZED
-    // TODO: init ndt_queue with proper functions.
-    // env->ndt_queue = pqueue_init(10, tag_in_reverse_order, )
+    // FIXME: Create a queue saving tags instead of events. For now, ndt_q stores 
+    // dummy events.
+    env->ndt_q = pqueue_init(INITIAL_EVENT_QUEUE_SIZE, in_reverse_order, get_event_time,
+        get_event_position, set_event_position, event_matches, print_event);
 #endif
 #ifdef FEDERATED_DECENTRALIZED
     env->_lf_intended_tag_fields = (tag_t**) calloc(num_is_present_fields, sizeof(tag_t*));

core/federated/federate.c

@@ -2321,6 +2321,16 @@ void handle_next_downstream_tag() {
 
     LF_PRINT_LOG("Received from RTI a MSG_TYPE_NEXT_DOWNSTREAM_TAG message with elapsed tag " PRINTF_TAG ".",
             NDT.time - start_time, NDT.microstep);
+    environment_t* env;
+    _lf_get_environments(&env);
+
+    if (lf_tag_compare(env->current_tag, NDT) < 0) {
+        // The current tag is less than NDT. Push NDT to ndt_q.
+        event_t* dummy = _lf_create_dummy_events(env, NULL, NDT.time, NULL, NDT.microstep);
+        pqueue_insert(env->ndt_q, dummy);
+    } else {
+        // The current tag is greater than or equal to NDT. Send LTC, NET, and ABS messages.
+    }
 }
 
 /**

core/reactor_common.c

@@ -292,6 +292,12 @@ void _lf_start_time_step(environment_t *env) {
         }
     }
 
+#ifdef FEDERATED_CENTRALIZED
+    while (lf_tag_compare(((event_t*) pqueue_peek(env->ndt_q))->time, env->current_tag.time) < 0) {
+        // Remove elements of ndt_q with tag less than the current tag.
+        pqueue_remove(env->ndt_q, pqueue_peek(env->ndt_q));
+    }
+#endif
 #ifdef FEDERATED_DECENTRALIZED
     for (int i = 0; i < env->is_present_fields_size; i++) {
         // FIXME: For now, an intended tag of (NEVER, 0)

include/core/environment.h

@@ -106,7 +106,7 @@ typedef struct environment_t {
     int _lf_intended_tag_fields_size;
 #endif // FEDERATED
 #ifdef FEDERATED_CENTRALIZED
-    pqueue_t* ndt_queue;
+    pqueue_t* ndt_q;
 #endif // FEDERATED_CENTRALIZED
 } environment_t;