commit 5b7188aa231c56bc748ed85fcaf4ef31e37baee2
Author: skenneweg <skenneweg@techfak.uni-bielefeld.de>
Date:   Tue Jul 14 23:31:59 2020 +0200

    Work in the subscriber functions (mainly the subscribe/unsubscribe functions)

diff --git a/inc/urt_subscriber.h b/inc/urt_subscriber.h
index 03c2551..40ac791 100644
--- a/inc/urt_subscriber.h
+++ b/inc/urt_subscriber.h
@@ -156,7 +156,7 @@ extern "C" {
   urt_status_t urtFrtSubscriberUnsubscribe(urt_frtsubscriber_t* subscriber);
 
   void urtHrtSubscriberInit(urt_hrtsubscriber_t* subscriber);
-  urt_status_t urtHrtSubscriberSubscribe(urt_hrtsubscriber_t* subscriber, urt_topic_t* topic, urt_message_t* message, urt_delay_t deadline, urt_delay_t rate, urt_delay_t jitter);
+  urt_status_t urtHrtSubscriberSubscribe(urt_hrtsubscriber_t* subscriber, urt_topic_t* topic, urt_message_t* messages, urt_delay_t deadline, urt_delay_t rate, urt_delay_t jitter);
   urt_status_t urtHrtSubscriberFetchNextMessage(urt_hrtsubscriber_t* subscriber, void* payload, size_t bytes, urt_delay_t* latency);
   urt_status_t urtHrtSubscriberFetchLatestMessage(urt_hrtsubscriber_t* subscriber, void* payload, size_t bytes, urt_delay_t* latency);
   urt_status_t urtHrtSubscriberUnsubscribe(urt_hrtsubscriber_t* subscriber);
diff --git a/src/urt_subscriber.c b/src/urt_subscriber.c
index 1386d51..5716ba9 100644
--- a/src/urt_subscriber.c
+++ b/src/urt_subscriber.c
@@ -128,42 +128,6 @@ urt_status_t urtNrtSubscriberSubscribe (urt_nrtsubscriber_t* subscriber, urt_top
 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;
 }
 
@@ -191,12 +155,13 @@ urt_status_t urtNrtSubscriberFetchLatestMessage (urt_nrtsubscriber_t* subscriber
  * @return  Returns URT_STATUS_OK on sucess.
  *          Returns URT_STATUS_UNSUBSCRIBE_NOTOPIC if the subscriber is not associated to a topic.
  */
-urt_status_t urtNrtSubscriberUnsubscribe (urt_nrtsubscriber_t* subscriber) {
+urt_status_t urtNrtSubscriberUnsubscribe (urt_nrtsubscriber_t* subscriber)
+{
   if (subscriber->base.topic)
   {
 # if(URT_CFG_PUBSUB_PROFILING == true)
       //TODO: LOCK TOPIC
-      subscriber->base.topic->numHrtSubscribers--;
+      subscriber->base.topic->numSubscribers--;
 # endif /* URT_CFG_PUBSUB_PROFILING */
     urtEventUnregister(subscriber->base.topic->evtSource, subscriber->base.evtListener);
 # if(URT_CFG_PUBSUB_PROFILING == true)
@@ -207,8 +172,8 @@ urt_status_t urtNrtSubscriberUnsubscribe (urt_nrtsubscriber_t* subscriber) {
 #endif /* URT_CFG_PUBSUB_PROFILING */
     return URT_STATUS_OK;
   }
-  else
-      return URT_STATUS_UNSUBSCRIBE_NOTOPIC;
+
+  return URT_STATUS_UNSUBSCRIBE_NOTOPIC;
 }
 
 
@@ -262,18 +227,16 @@ urt_status_t urtSrtSubscriberSubscribe(urt_srtsubscriber_t* subscriber, urt_topi
   {
     return URT_STATUS_SUBSCRIBE_TOPICSET;
   }
-  else
-  {
-    subscriber->base.topic = topic;
-    subscriber->usefulnesscb = usefulnesscb;
-    subscriber->cbparams = cbparams;
+
+  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;
+  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)
@@ -359,11 +322,10 @@ urt_status_t urtSrtSubscriberUnsubscribe(urt_srtsubscriber_t* subscriber)
 
   if (subscriber->base.topic)
   {
-    if (URT_CFG_PUBSUB_PROFILING == true)
-    {
-      //TODO: lock topic
-      subscriber->base.topic->numHrtSubscribers--;
-    }
+# if (URT_CFG_PUBSUB_PROFILING == true)
+    //TODO: lock topic
+    subscriber->base.topic->numSubscribers--;
+# endif /* URT_CFG_PUBSUB_PROFILING */
     urtEventUnregister(subscriber->base.topic->evtSource, subscriber->base.evtListener);
 # if (URT_CFG_PUBSUB_PROFILING == true)
       //TODO: unlock topic
@@ -435,52 +397,50 @@ urt_status_t urtFrtSubscriberSubscribe(urt_frtsubscriber_t* subscriber, urt_topi
   urtDebugAssert(subscriber);
   urtDebugAssert(topic);
 
-    if (subscriber->base.topic)
-    {
-      return URT_STATUS_SUBSCRIBE_TOPICSET;
-    }
-    else
-    {
-      subscriber->base.topic = topic;
+  if (subscriber->base.topic)
+  {
+    return URT_STATUS_SUBSCRIBE_TOPICSET;
+  }
+
+  subscriber->base.topic = topic;
 # if (URT_CFG_PUBSUB_PROFILING == true)
-        subscriber->base.sumLatencies = 0;
-        subscriber->base.numMessagesReceived = 0;
+  subscriber->base.sumLatencies = 0;
+  subscriber->base.numMessagesReceived = 0;
 # endif /* URT_CFG_PUBSUB_PROFILING */
 # if (URT_CFG_PUBSUB_QOS_DEADLINECHECKS == true)
-        subscriber->deadlineOffset = deadline;
+  subscriber->deadlineOffset = deadline;
 # endif /* URT_CFG_PUBSUB_QOS_DEADLINECHECKS */
 # if (URT_CFG_PUBSUB_QOS_JITTERCHECKS == true)
-        subscriber->maxJitter =jitter;
+  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;
+  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)
+  //TODO: Lock topic
+  if (messages)
+  {
+    urt_message_t* lastMessageContribute = messages;
+    while (lastMessageContribute->next)
     {
-      urt_message_t* lastMessageContribute = messages;
-      while (lastMessageContribute->next)
-      {
-          lastMessageContribute = lastMessageContribute->next;
-      }
-      lastMessageContribute->next = topic->latestMessage->next;
-      topic->latestMessage->next = messages;
+      lastMessageContribute = lastMessageContribute->next;
     }
+    lastMessageContribute->next = topic->latestMessage->next;
+    topic->latestMessage->next = messages;
+  }
 
-    subscriber->base.lastMessage = topic->latestMessage;
-    subscriber->base.lastMessageTime = topic->latestMessage->originTime;
+  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?
+  urtEventRegister(topic->evtSource, subscriber->base.evtListener, URT_EVENTMASK_MAXPRIO, URT_EVENTFLAG_PROCEED); //TODO: Correct mask, flag?
 
 # if (URT_CFG_PUBSUB_PROFILING == true)
-      topic->numHrtSubscribers--;
+  topic->numHrtSubscribers--;
 # endif /* URT_CFG_PUBSUB_PROFILING */
 
-    //TODO: Unlock topic
-    return URT_STATUS_OK;
+  //TODO: Unlock topic
+  return URT_STATUS_OK;
 }
 
 /**
@@ -524,9 +484,10 @@ urt_status_t urtFrtSubscriberFetchLatestMessage(urt_frtsubscriber_t* subscriber,
 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;
+  if (latency < subscriber->deadlineOffset && latency < subscriber->maxJitter)
+    return true;
 # endif /* URT_CFG_PUBSUB_QOS_DEADLINECHECKS && URT_CFG_PUBSUB_QOS_JITTERCHECKS */
+
   return false;
 }
 
@@ -538,7 +499,31 @@ bool urtFrtSubscriberCalculateValidity(urt_frtsubscriber_t* subscriber, urt_dela
  * @return  Returns URT_STATUS_OK on sucess.
  *          Returns URT_STATUS_UNSUBSCRIBE_NOTOPIC if the subscriber is not associated to a topic.
  */
-urt_status_t urtFrtSubscriberUnsubscribe(urt_frtsubscriber_t* subscriber){return URT_STATUS_OK;}
+urt_status_t urtFrtSubscriberUnsubscribe(urt_frtsubscriber_t* subscriber)
+{
+  urtDebugAssert(subscriber);
+
+  if (subscriber->base.topic)
+  {
+    //TODO: lock topic
+    urtEventUnregister(subscriber->base.topic->evtSource, subscriber->base.evtListener);
+    //TODO: decrement topic's HRT counter
+# if (URT_CFG_PUBSUB_PROFILING == true)
+    subscriber->base.topic->numSubscribers--;
+# endif /* URT_CFG_PUBSUB_PROFILING */
+//Hier weiter
+
+# 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;
+}
 
 
 /**
@@ -602,7 +587,78 @@ void urtHrtSubscriberInit(urt_hrtsubscriber_t* subscriber)
  *          Returns URT_STATUS_SUBSCRIBE_TOPICSET if the subscriber is already associated to a topic.
  */
 urt_status_t urtHrtSubscriberSubscribe(urt_hrtsubscriber_t* subscriber, urt_topic_t* topic,
-                                       urt_message_t* message, urt_delay_t deadline, urt_delay_t rate, urt_delay_t jitter){return URT_STATUS_OK;}
+                                       urt_message_t* messages, urt_delay_t deadline, urt_delay_t rate, urt_delay_t jitter)
+{
+  urtDebugAssert(subscriber);
+  urtDebugAssert(topic);
+
+  if (subscriber->base.topic)
+  {
+    return URT_STATUS_SUBSCRIBE_TOPICSET;
+  }
+
+  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 */
+# if (URT_CFG_PUBSUB_QOS_RATECHECKS == true)
+  subscriber->expectedRate = rate;
+# endif /* URT_CFG_PUBSUB_QOS_RATECHECKS */
+
+  //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_QOS_RATECHECKS == true)
+  urt_hrtsubscriber_t* hrtSubscriber = subscriber->base.topic->hrtSubscribers;
+  while (!hrtSubscriber /* && expected Rate is lower */)
+  {
+      hrtSubscriber = hrtSubscriber->next;
+  }
+
+  if (!hrtSubscriber)
+  {
+    //TODO: Append self to topic's list of HRT subscribers
+  }
+  else
+  {
+    //TOOD: insert self in front of current HRT susbcriber
+    subscriber->next = hrtSubscriber;
+  }
+# endif /* URT_CFG_PUBSUB_QOS_RATECHECKS */
+
+  topic->numHrtSubscribers--;
+# if (URT_CFG_PUBSUB_PROFILING == true)
+  topic->numSubscribers--;
+# endif /* URT_CFG_PUBSUB_PROFILING */
+
+  //TODO: Unlock topic
+  return URT_STATUS_OK;
+}
 
 
 /**
@@ -644,4 +700,47 @@ urt_status_t urtHrtSubscriberFetchLatestMessage(urt_hrtsubscriber_t* subscriber,
  * @return  Returns URT_STATUS_OK on sucess.
  *          Returns URT_STATUS_UNSUBSCRIBE_NOTOPIC if the subscriber is not associated to a topic.
  */
-urt_status_t urtHrtSubscriberUnsubscribe(urt_hrtsubscriber_t* subscriber){return URT_STATUS_OK;}
+urt_status_t urtHrtSubscriberUnsubscribe(urt_hrtsubscriber_t* subscriber)
+{
+  urtDebugAssert(subscriber);
+
+  if (subscriber->base.topic)
+  {
+    //TODO: lock topic
+    urtEventUnregister(subscriber->base.topic->evtSource, subscriber->base.evtListener);
+    subscriber->base.topic->numHrtSubscribers--;
+# if (URT_CFG_PUBSUB_PROFILING == true)
+    subscriber->base.topic->numSubscribers--;
+# endif /* URT_CFG_PUBSUB_PROFILING */
+# if (URT_CFG_PUBSUB_QOS_RATECHECKS == true)
+    //TODO: remove self from topics lsit of HRT subscribers
+    //TODO: ...
+# endif /* URT_CFG_PUBSUB_QOS_RATECHECKS */
+
+    urt_message_t* messageTemp = subscriber->base.lastMessage;
+    while (messageTemp->next->originTime < messageTemp->originTime)
+    {
+        messageTemp = messageTemp->next;
+        messageTemp->numHrtConsumersLeft--;
+# if(URT_CFG_PUBSUB_PROFILING == true)
+        messageTemp->numConsumersLeft--;
+# endif /* URT_CFG_PUBSUB_PROFILING */
+    }
+    bool temp = false;
+    if (temp /*TODO: HRT counter of any message became 0?*/)
+    {
+      //TODO: signal topics condition variable
+    }
+
+    //TODO: unlock topic
+    subscriber->base.topic = NULL;
+    subscriber->base.lastMessage = NULL;
+    subscriber->base.lastMessageTime = 0;
+    return URT_STATUS_OK;
+  }
+
+  return URT_STATUS_UNSUBSCRIBE_NOTOPIC;
+}
+
+
+
