commit a5e142deb4483dd7f4ef39a0c3d329bc58ec416a
Author: skenneweg <skenneweg@techfak.uni-bielefeld.de>
Date:   Mon Jul 13 23:41:02 2020 +0200

    Work in the subscriber functions

diff --git a/src/urt_subscriber.c b/src/urt_subscriber.c
index a885295..1386d51 100644
--- a/src/urt_subscriber.c
+++ b/src/urt_subscriber.c
@@ -52,14 +52,16 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 void urtNrtSubscriberInit (urt_nrtsubscriber_t* subscriber)
 {
+  urtDebugAssert(subscriber);
+
   subscriber->base.topic = NULL;
   urtEventListenerInit(subscriber->base.evtListener);
   subscriber->base.lastMessage = NULL;
   subscriber->base.lastMessageTime = 0;
-  #if (URT_CFG_PUBSUB_PROFILING)
+#if (URT_CFG_PUBSUB_PROFILING == true)
     subscriber->minLatency = URT_DELAY_INFINITE;
     subscriber->maxLatency = URT_DELAY_IMMEDIATE;
-  #endif /* URT_CFG_PUBSUB_PROFILING */
+#endif /* URT_CFG_PUBSUB_PROFILING */
   return;
 }
 
@@ -76,7 +78,38 @@ void urtNrtSubscriberInit (urt_nrtsubscriber_t* subscriber)
  * @return  Returns URT_STATUS_OK on success.
  *          Returns URT_STATUS_SUBSCRIBE_TOPICSET if the subscriber is already associated to a topic.
  */
-urt_status_t urtNrtSubscriberSubscribe (urt_nrtsubscriber_t* subscriber, urt_topic_t* topic, urt_message_t* messages) {
+urt_status_t urtNrtSubscriberSubscribe (urt_nrtsubscriber_t* subscriber, urt_topic_t* topic, urt_message_t* messages)
+{
+  urtDebugAssert(subscriber);
+  urtDebugAssert(topic);
+
+  if (!subscriber->base.topic)
+      return URT_STATUS_SUBSCRIBE_TOPICSET;
+
+  subscriber->base.topic = topic;
+  //TODO: Lock topic
+
+  if (messages)
+  {
+    urt_message_t* lastMessageContribute = messages;
+    while (lastMessageContribute->next)
+    {
+        lastMessageContribute = lastMessageContribute->next;
+    }
+    lastMessageContribute->next = topic->latestMessage->next;
+    topic->latestMessage->next = messages;
+  }
+
+  subscriber->base.lastMessage = topic->latestMessage;
+  subscriber->base.lastMessageTime = topic->latestMessage->originTime;
+
+  urtEventRegister(topic->evtSource, subscriber->base.evtListener, URT_EVENTMASK_MAXPRIO, URT_EVENTFLAG_PROCEED); //TODO: Correct mask, flag?
+
+#if (URT_CFG_PUBSUB_PROFILING == true)
+    topic->numHrtSubscribers--;
+#endif /* URT_CFG_PUBSUB_PROFILING */
+
+  //TODO: Unlock topic
   return URT_STATUS_OK;
 }
 
@@ -92,12 +125,50 @@ urt_status_t urtNrtSubscriberSubscribe (urt_nrtsubscriber_t* subscriber, urt_top
  *          Returns URT_STATUS_FETCH_NOTOPIC if the subscriber is not associated to a topic.
  *          Retruns URT_STATUS_FETCH_NOMESSAGE if there is no new message to fetch.
  */
-urt_status_t urtNrtSubscriberFetchNextMessage (urt_nrtsubscriber_t* subscriber, void* payload, size_t bytes, urt_delay_t* latency) {
+urt_status_t urtNrtSubscriberFetchNextMessage (urt_nrtsubscriber_t* subscriber, void* payload, size_t bytes, urt_delay_t* latency)
+{
+  urtDebugAssert(subscriber);
+
+  urt_message_t youngestMessage;
+  if (subscriber->base.topic)
+  {
+    //TODO: Lock Topic
+    urt_osTime_t localCopy; //TODO: replace with local copy
+    if (subscriber->base.lastMessageTime == localCopy)
+    {
+      if(subscriber->base.lastMessage->next->originTime < localCopy)
+      {
+        youngestMessage = subscriber->base.lastMessage->next;
+      }
+      else
+      {
+        //TODO: Unlock Topic
+        return URT_STATUS_FETCH_NOMESSAGE;
+      }
+    }
+    else
+    {
+      youngestMessage = subscriber->base.lastMessage->next;
+      while (youngestMessage.originTime < localCopy)
+      {
+        youngestMessage = youngestMessage.next;
+      }
+    }
+  }
+  else
+  {
+    return URT_STATUS_FETCH_NOTOPIC;
+  }
+
+  latency = subscriber->base.lastMessageTime - youngestMessage.originTime;
+
+  //TODO: Other cases
+  //TODO: Unlock topic
   return URT_STATUS_OK;
 }
 
 /**
- * @brief Fetches the lates message.
+ * @brief Fetches the latest message.
  *
  * @param[in] subscriber  The NRT subscriber that shall fetch the message. Must not be NULL.
  * @param[in] payload  Pointer where to copy the payload to. May be NULL for messages without payload.
@@ -121,7 +192,23 @@ urt_status_t urtNrtSubscriberFetchLatestMessage (urt_nrtsubscriber_t* subscriber
  *          Returns URT_STATUS_UNSUBSCRIBE_NOTOPIC if the subscriber is not associated to a topic.
  */
 urt_status_t urtNrtSubscriberUnsubscribe (urt_nrtsubscriber_t* subscriber) {
-  return URT_STATUS_OK;
+  if (subscriber->base.topic)
+  {
+# if(URT_CFG_PUBSUB_PROFILING == true)
+      //TODO: LOCK TOPIC
+      subscriber->base.topic->numHrtSubscribers--;
+# endif /* URT_CFG_PUBSUB_PROFILING */
+    urtEventUnregister(subscriber->base.topic->evtSource, subscriber->base.evtListener);
+# if(URT_CFG_PUBSUB_PROFILING == true)
+      //TODO: Unlock TOPIC
+      subscriber->base.topic = NULL;
+      subscriber->base.lastMessage = NULL;
+      subscriber->base.lastMessageTime = 0;
+#endif /* URT_CFG_PUBSUB_PROFILING */
+    return URT_STATUS_OK;
+  }
+  else
+      return URT_STATUS_UNSUBSCRIBE_NOTOPIC;
 }
 
 
@@ -132,6 +219,8 @@ urt_status_t urtNrtSubscriberUnsubscribe (urt_nrtsubscriber_t* subscriber) {
  */
 void urtSrtSubscriberInit(urt_srtsubscriber_t* subscriber)
 {
+  urtDebugAssert(subscriber);
+
   subscriber->base.topic = NULL;
   urtEventListenerInit(subscriber->base.evtListener);
   subscriber->base.lastMessage = NULL;
@@ -164,7 +253,52 @@ void urtSrtSubscriberInit(urt_srtsubscriber_t* subscriber)
  *          Returns URT_STATUS_SUBSCRIBE_TOPICSET if the subscriber is already associated to a topic.
  */
 urt_status_t urtSrtSubscriberSubscribe(urt_srtsubscriber_t* subscriber, urt_topic_t* topic,
-                                       urt_message_t* message, urt_srtusefulnessfunc_t* usefulnesscb, void* cbparams) {return URT_STATUS_OK;}
+                                       urt_message_t* messages, urt_srtusefulnessfunc_t* usefulnesscb, void* cbparams)
+{
+  urtDebugAssert(subscriber);
+  urtDebugAssert(topic);
+
+  if (subscriber->base.topic)
+  {
+    return URT_STATUS_SUBSCRIBE_TOPICSET;
+  }
+  else
+  {
+    subscriber->base.topic = topic;
+    subscriber->usefulnesscb = usefulnesscb;
+    subscriber->cbparams = cbparams;
+# if (URT_CFG_PUBSUB_PROFILING == true)
+      subscriber->base.sumLatencies = 0;
+      subscriber->base.numMessagesReceived = 0;
+      subscriber->minLatency = URT_DELAY_INFINITE;
+      subscriber->maxLatency = URT_DELAY_IMMEDIATE;
+# endif  /* URT_CFG_PUBSUB_PROFILING */
+  }
+
+  //TODO: Lock topic
+  if (messages)
+  {
+    urt_message_t* lastMessageContribute = messages;
+    while (lastMessageContribute->next)
+    {
+        lastMessageContribute = lastMessageContribute->next;
+    }
+    lastMessageContribute->next = topic->latestMessage->next;
+    topic->latestMessage->next = messages;
+  }
+
+  subscriber->base.lastMessage = topic->latestMessage;
+  subscriber->base.lastMessageTime = topic->latestMessage->originTime;
+
+  urtEventRegister(topic->evtSource, subscriber->base.evtListener, URT_EVENTMASK_MAXPRIO, URT_EVENTFLAG_PROCEED); //TODO: Correct mask, flag?
+
+# if (URT_CFG_PUBSUB_PROFILING == true)
+    topic->numHrtSubscribers--;
+# endif /* URT_CFG_PUBSUB_PROFILING */
+
+  //TODO: Unlock topic
+  return URT_STATUS_OK;
+}
 
 /**
  * @brief  Fetches the next message.
@@ -204,7 +338,12 @@ urt_status_t urtSrtSubscriberFetchLatestMessage(urt_srtsubscriber_t* subscriber,
  *
  * @return  Returns the usefulness as a value within [0,1].
  */
-float urtSrtSubscriberCalculateUsefulness(urt_srtsubscriber_t* subscriber, urt_delay_t latency){return 0;}
+float urtSrtSubscriberCalculateUsefulness(urt_srtsubscriber_t* subscriber, urt_delay_t latency)
+{
+  urtDebugAssert(subscriber);
+
+  return subscriber->usefulnesscb(latency);
+}
 
 /**
  * @brief  Unsubscribes from a subscriber.
@@ -214,7 +353,29 @@ float urtSrtSubscriberCalculateUsefulness(urt_srtsubscriber_t* subscriber, urt_d
  * @return  Returns URT_STATUS_OK on sucess.
  *          Returns URT_STATUS_UNSUBSCRIBE_NOTOPIC if the subscriber is not associated to a topic.
  */
-urt_status_t urtSrtSubscriberUnsubscribe(urt_srtsubscriber_t* subscriber){return URT_STATUS_OK;}
+urt_status_t urtSrtSubscriberUnsubscribe(urt_srtsubscriber_t* subscriber)
+{
+  urtDebugAssert(subscriber);
+
+  if (subscriber->base.topic)
+  {
+    if (URT_CFG_PUBSUB_PROFILING == true)
+    {
+      //TODO: lock topic
+      subscriber->base.topic->numHrtSubscribers--;
+    }
+    urtEventUnregister(subscriber->base.topic->evtSource, subscriber->base.evtListener);
+# if (URT_CFG_PUBSUB_PROFILING == true)
+      //TODO: unlock topic
+# endif /* URT_CFG_PUBSUB_PROFILING */
+    subscriber->base.topic = NULL;
+    subscriber->base.lastMessage = NULL;
+    subscriber->base.lastMessageTime = 0;
+    return URT_STATUS_OK;
+  }
+
+  return URT_STATUS_UNSUBSCRIBE_NOTOPIC;
+}
 
 
 /**
@@ -224,6 +385,8 @@ urt_status_t urtSrtSubscriberUnsubscribe(urt_srtsubscriber_t* subscriber){return
  */
 void urtFrtSubscriberInit(urt_frtsubscriber_t* subscriber)
 {
+  urtDebugAssert(subscriber);
+
   subscriber->base.topic = NULL;
   urtEventListenerInit(subscriber->base.evtListener);
   subscriber->base.lastMessage = NULL;
@@ -267,7 +430,58 @@ void urtFrtSubscriberInit(urt_frtsubscriber_t* subscriber)
  *          Returns URT_STATUS_SUBSCRIBE_TOPICSET if the subscriber is already associated to a topic.
  */
 urt_status_t urtFrtSubscriberSubscribe(urt_frtsubscriber_t* subscriber, urt_topic_t* topic,
-                                       urt_message_t* messages, urt_delay_t deadline, urt_delay_t jitter){return URT_STATUS_OK;}
+                                       urt_message_t* messages, urt_delay_t deadline, urt_delay_t jitter)
+{
+  urtDebugAssert(subscriber);
+  urtDebugAssert(topic);
+
+    if (subscriber->base.topic)
+    {
+      return URT_STATUS_SUBSCRIBE_TOPICSET;
+    }
+    else
+    {
+      subscriber->base.topic = topic;
+# if (URT_CFG_PUBSUB_PROFILING == true)
+        subscriber->base.sumLatencies = 0;
+        subscriber->base.numMessagesReceived = 0;
+# endif /* URT_CFG_PUBSUB_PROFILING */
+# if (URT_CFG_PUBSUB_QOS_DEADLINECHECKS == true)
+        subscriber->deadlineOffset = deadline;
+# endif /* URT_CFG_PUBSUB_QOS_DEADLINECHECKS */
+# if (URT_CFG_PUBSUB_QOS_JITTERCHECKS == true)
+        subscriber->maxJitter =jitter;
+# endif /* URT_CFG_PUBSUB_QOS_JITTERCHECKS */
+# if (URT_CFG_PUBSUB_QOS_JITTERCHECKS == true || URT_CFG_PUBSUB_PROFILING == true)
+        subscriber->minLatency = URT_DELAY_INFINITE;
+        subscriber->maxLatency = URT_DELAY_IMMEDIATE;
+# endif /* URT_CFG_PUBSUB_QOS_JITTERCHECKS || URT_CFG_PUBSUB_PROFILING */
+    }
+
+    //TODO: Lock topic
+    if (messages)
+    {
+      urt_message_t* lastMessageContribute = messages;
+      while (lastMessageContribute->next)
+      {
+          lastMessageContribute = lastMessageContribute->next;
+      }
+      lastMessageContribute->next = topic->latestMessage->next;
+      topic->latestMessage->next = messages;
+    }
+
+    subscriber->base.lastMessage = topic->latestMessage;
+    subscriber->base.lastMessageTime = topic->latestMessage->originTime;
+
+    urtEventRegister(topic->evtSource, subscriber->base.evtListener, URT_EVENTMASK_MAXPRIO, URT_EVENTFLAG_PROCEED); //TODO: Correct mask, flag?
+
+# if (URT_CFG_PUBSUB_PROFILING == true)
+      topic->numHrtSubscribers--;
+# endif /* URT_CFG_PUBSUB_PROFILING */
+
+    //TODO: Unlock topic
+    return URT_STATUS_OK;
+}
 
 /**
  * @brief  Fetches the next message.
@@ -307,7 +521,14 @@ urt_status_t urtFrtSubscriberFetchLatestMessage(urt_frtsubscriber_t* subscriber,
  *
  * @return  Returns a boolean indicator whether the latency is fine.
  */
-bool urtFrtSubscriberCalculateValidity(urt_frtsubscriber_t* subscriber, urt_delay_t latency){return true;}
+bool urtFrtSubscriberCalculateValidity(urt_frtsubscriber_t* subscriber, urt_delay_t latency)
+{
+# if (URT_CFG_PUBSUB_QOS_DEADLINECHECKS == true && URT_CFG_PUBSUB_QOS_JITTERCHECKS == true)
+    if (latency < subscriber->deadlineOffset && latency < subscriber->maxJitter)
+        return true;
+# endif /* URT_CFG_PUBSUB_QOS_DEADLINECHECKS && URT_CFG_PUBSUB_QOS_JITTERCHECKS */
+  return false;
+}
 
 /**
  * @brief  Unsubscribes from a subscriber.
@@ -327,35 +548,37 @@ urt_status_t urtFrtSubscriberUnsubscribe(urt_frtsubscriber_t* subscriber){return
  */
 void urtHrtSubscriberInit(urt_hrtsubscriber_t* subscriber)
 {
+  urtDebugAssert(subscriber);
+
   subscriber->base.topic = NULL;
   urtEventListenerInit(subscriber->base.evtListener);
   subscriber->base.lastMessage = NULL;
   subscriber->base.lastMessageTime = 0;
 
-  #if (URT_CFG_PUBSUB_PROFILING)
+# if (URT_CFG_PUBSUB_PROFILING)
     subscriber->base.sumLatencies = 0;
     subscriber->base.numMessagesReceived = 0;
-  #endif /* URT_CFG_PUBSUB_PROFILING */
+# endif /* URT_CFG_PUBSUB_PROFILING */
 
   subscriber->next = NULL;
 
-  #if (URT_CFG_PUBSUB_QOS_DEADLINECHECKS)
+# if (URT_CFG_PUBSUB_QOS_DEADLINECHECKS)
     subscriber->deadlineOffset = 0;
     urtTimerInit(subscriber->qodDeadlineTimer);
-  #endif /* URT_CFG_PUBSUB_QOS_DEADLINECHECKS */
+# endif /* URT_CFG_PUBSUB_QOS_DEADLINECHECKS */
 
-  #if (URT_CFG_PUBSUB_QOS_RATECHECKS)
+# if (URT_CFG_PUBSUB_QOS_RATECHECKS)
     subscriber->expectedRate = 0;
-  #endif /* URT_CFG_PUBSUB_QOS_RATECHECKS */
+# endif /* URT_CFG_PUBSUB_QOS_RATECHECKS */
 
-  #if (URT_CFG_PUBSUB_QOS_JITTERCHECKS)
+# if (URT_CFG_PUBSUB_QOS_JITTERCHECKS)
     subscriber->maxJitter = 0;
-  #endif /* URT_CFG_PUBSUB_QOS_JITTERCHECKS */
+# endif /* URT_CFG_PUBSUB_QOS_JITTERCHECKS */
 
-  #if (URT_CFG_PUBSUB_QOS_JITTERCHECKS || URT_CFG_PUBSUB_PROFILING)
+# if (URT_CFG_PUBSUB_QOS_JITTERCHECKS || URT_CFG_PUBSUB_PROFILING)
     subscriber->minLatency = URT_DELAY_INFINITE;
     subscriber->maxLatency = URT_DELAY_IMMEDIATE;
-  #endif /* URT_CFG_PUBSUB_QOS_JITTERCHECKS || URT_CFG_PUBSUB_PROFILING */
+# endif /* URT_CFG_PUBSUB_QOS_JITTERCHECKS || URT_CFG_PUBSUB_PROFILING */
   return;
 }
 
