commit 5c6cb22fe2e8eb5c38dfd3df441febfd4c35a223
Author: skenneweg <skenneweg@techfak.uni-bielefeld.de>
Date:   Fri Jul 10 18:11:42 2020 +0200

    Enabled pubsub system
    Wrote Initialization functions for all subscriber types

diff --git a/inc/urt_message.h b/inc/urt_message.h
index a8e09a9..26eb772 100644
--- a/inc/urt_message.h
+++ b/inc/urt_message.h
@@ -45,7 +45,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 typedef struct urt_message
 {
-  urt_message_t* next;
+  struct urt_message* next;
   void* payload;
   urt_osTime_t originTime;
   unsigned int numHrtConsumersLeft;
diff --git a/inc/urt_subscriber.h b/inc/urt_subscriber.h
index b78095e..690b4f5 100644
--- a/inc/urt_subscriber.h
+++ b/inc/urt_subscriber.h
@@ -22,7 +22,6 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #ifndef URTWARE_SUBSCRIBER_H
 #define URTWARE_SUBSCRIBER_H
 
-#include <urt_types.h>
 #include <urtware.h>
 
 /******************************************************************************/
@@ -70,12 +69,13 @@ typedef struct urt_nrtsubscriber
 }urt_nrtsubscriber_t;
 
 /**
- * @brief  nrt subscriber
+ * @brief  srt subscriber
  */
 typedef struct urt_srtsubscriber
 {
   urt_basesubscriber_t base;
-  urt_srtusefulnessfunc_t* usefulnesscb;
+  void* cbparams;
+  urt_usefulness_f* usefulnesscb;
   #if (URT_CFG_PUBSUB_PROFILING)
     urt_delay_t minLatency;
     urt_delay_t maxLatency;
@@ -83,7 +83,7 @@ typedef struct urt_srtsubscriber
 }urt_srtsubscriber_t;
 
 /**
- * @brief  nrt subscriber
+ * @brief  frt subscriber
  */
 typedef struct urt_frtsubscriber
 {
@@ -106,7 +106,7 @@ typedef struct urt_frtsubscriber
 typedef struct urt_hrtsubscriber
 {
   urt_basesubscriber_t base;
-  urt_hrtsubscriber_t* next;
+  struct urt_hrtsubscriber* next;
   #if (URT_CFG_PUBSUB_QOS_DEADLINECHECKS)
     urt_delay_t deadlineOffset;
     urt_osTimer_t qosDeadlineTimer;
@@ -142,7 +142,7 @@ extern "C" {
   urt_status_t urtNrtSubscriberUnsubscribe(urt_nrtsubscriber_t* subscriber);
 
   void urtSrtSubscriberInit(urt_srtsubscriber_t* subscriber);
-  urt_status_t urtSrtSubscriberSubscribe(urt_srtsubscriber_t* subscriber, urt_topic_t* topic, urt_message_t* message, urt_srtusefulnessfunc_t* usefulnesscb, void* cbparams);
+  urt_status_t urtSrtSubscriberSubscribe(urt_srtsubscriber_t* subscriber, urt_topic_t* topic, urt_message_t* message, urt_usefulness_f* usefulnesscb, void* cbparams);
   urt_status_t urtSrtSubscriberFetchNextMessage(urt_srtsubscriber_t* subscriber, void* payload, size_t bytes, urt_delay_t* latency);
   urt_status_t urtSrtSubscriberFetchLatestMessage(urt_srtsubscriber_t* subscriber, void* payload, size_t bytes, urt_delay_t* latency);
   float urtSrtSubscriberCalculateUsefulness(urt_srtsubscriber_t* subscriber, urt_delay_t latency);
@@ -160,8 +160,6 @@ extern "C" {
   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);
-
-  float urt_srtusefulnessfunc_t (urt_delay_t dt, void* params);
 #if defined(__cplusplus)
 }
 #endif /* defined(__cplusplus) */
diff --git a/inc/urt_topic.h b/inc/urt_topic.h
index f3724a7..05ec46d 100644
--- a/inc/urt_topic.h
+++ b/inc/urt_topic.h
@@ -22,7 +22,6 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #ifndef URTWARE_TOPIC_H
 #define URTWARE_TOPIC_H
 
-#include <urt_types.h>
 #include <urtware.h>
 
 /******************************************************************************/
@@ -46,7 +45,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
 typedef struct urt_topic
 {
-  urt_topic_t* next;
+  struct urt_topic* next;
   urt_topicid_t id;
   urt_osMutex_t lock;
   urt_osEventSource_t evtSource;
@@ -55,7 +54,7 @@ typedef struct urt_topic
   urt_message_t mandatoryMessage;
   urt_message_t* latestMessage;
   #if (URT_CFG_PUBSUB_QOS_RATECHECKS)
-    urt_hrtsubscriber_t* hrtSubscribers;
+    struct urt_hrtsubscriber* hrtSubscribers;
     urt_osTimer_t qosRateTimer;
   #endif /* URT_CFG_PUBSUB_QOS_RATECHECKS */
   #if (URT_CFG_PUBSUB_PROFILING)
diff --git a/src/urt_core.c b/src/urt_core.c
index 6a2239f..4ef5607 100644
--- a/src/urt_core.c
+++ b/src/urt_core.c
@@ -244,7 +244,18 @@ urt_status_t urtCoreStopNodes(urt_status_t reason)
  * @return  Returns a pointer to the requested service. Returns NULL if no service matches the given ID.
  */
 #if (URT_CFG_PUBSUB_ENABLED)
-  urt_topic_t* urtCoreGetTopic(urt_topicid_t id) {return urt_topic_t;}
+  urt_topic_t* urtCoreGetTopic(urt_topicid_t id)
+  {
+      urtMutexLock(&core._lock);
+      urt_topic_t* topic = core._topics;
+      while (topic != NULL && topic->id < id)
+          topic = topic->next;
+      urtMutexUnlock(&core._lock);
+      if (topic != NULL && topic->id == id)
+          return topic;
+      else
+        return NULL;
+  }
 #endif /* URT_CFG_PUBSUB_ENABLED */
 
 
diff --git a/src/urt_subscriber.c b/src/urt_subscriber.c
index 1fdd4ee..a885295 100644
--- a/src/urt_subscriber.c
+++ b/src/urt_subscriber.c
@@ -50,7 +50,18 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
  *
  * @param[in] subscriber  The NRT subscriber to initialize. Must not be NULL.
  */
-void urtNrtSubscriberInit (urt_nrtsubscriber_t* subscriber) {return;}
+void urtNrtSubscriberInit (urt_nrtsubscriber_t* subscriber)
+{
+  subscriber->base.topic = NULL;
+  urtEventListenerInit(subscriber->base.evtListener);
+  subscriber->base.lastMessage = NULL;
+  subscriber->base.lastMessageTime = 0;
+  #if (URT_CFG_PUBSUB_PROFILING)
+    subscriber->minLatency = URT_DELAY_INFINITE;
+    subscriber->maxLatency = URT_DELAY_IMMEDIATE;
+  #endif /* URT_CFG_PUBSUB_PROFILING */
+  return;
+}
 
 /**
  * @brief  Subscribes the subscriber to a topic.
@@ -119,7 +130,22 @@ urt_status_t urtNrtSubscriberUnsubscribe (urt_nrtsubscriber_t* subscriber) {
  *
  * @param[in] subscriber  The SRT subscriber to initialize. Must not be NULL.
  */
-void urtSrtSubscriberInit(urt_srtsubscriber_t* subscriber) {return;}
+void urtSrtSubscriberInit(urt_srtsubscriber_t* subscriber)
+{
+  subscriber->base.topic = NULL;
+  urtEventListenerInit(subscriber->base.evtListener);
+  subscriber->base.lastMessage = NULL;
+  subscriber->base.lastMessageTime = 0;
+  #if (URT_CFG_PUBSUB_PROFILING)
+    subscriber->base.sumLatencies = 0;
+    subscriber->base.numMessagesReceived = 0;
+    subscriber->usefulnesscb = NULL;
+    subscriber->cbparams = NULL;
+    subscriber->minLatency = URT_DELAY_INFINITE;
+    subscriber->maxLatency = URT_DELAY_IMMEDIATE;
+  #endif /* URT_CFG_PUBSUB_PROFILING */
+  return;
+}
 
 /**
  * @brief  Subscribes the subscriber to a topic.
@@ -196,7 +222,33 @@ urt_status_t urtSrtSubscriberUnsubscribe(urt_srtsubscriber_t* subscriber){return
  *
  * @param[in] subscriber  The SRT subscriber to initialize. Must not be NULL.
  */
-void urtFrtSubscriberInit(urt_frtsubscriber_t* subscriber){return URT_STATUS_OK;}
+void urtFrtSubscriberInit(urt_frtsubscriber_t* subscriber)
+{
+  subscriber->base.topic = NULL;
+  urtEventListenerInit(subscriber->base.evtListener);
+  subscriber->base.lastMessage = NULL;
+  subscriber->base.lastMessageTime = 0;
+
+  #if (URT_CFG_PUBSUB_PROFILING)
+    subscriber->base.sumLatencies = 0;
+    subscriber->base.numMessagesReceived = 0;
+  #endif /* URT_CFG_PUBSUB_PROFILING */
+
+  #if (URT_CFG_PUBSUB_QOS_DEADLINECHECKS)
+    subscriber->deadlineOffset = 0;
+  #endif /* URT_CFG_PUBSUB_QOS_DEADLINECHECKS */
+
+  #if (URT_CFG_PUBSUB_QOS_JITTERCHECKS)
+    subscriber->maxJitter = 0;
+  #endif /* URT_CFG_PUBSUB_QOS_JITTERCHECKS */
+
+  #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 */
+  return;
+}
+
 
 /**
  * @brief  Subscribes the subscriber to a topic.
@@ -260,7 +312,7 @@ bool urtFrtSubscriberCalculateValidity(urt_frtsubscriber_t* subscriber, urt_dela
 /**
  * @brief  Unsubscribes from a subscriber.
  *
- * @param[in] subscriber  The NRT subscriber to be unsubscribed. Must not be NULL.
+ * @param[in] subscriber  The FRT subscriber to be unsubscribed. Must not be NULL.
  *
  * @return  Returns URT_STATUS_OK on sucess.
  *          Returns URT_STATUS_UNSUBSCRIBE_NOTOPIC if the subscriber is not associated to a topic.
@@ -273,12 +325,45 @@ urt_status_t urtFrtSubscriberUnsubscribe(urt_frtsubscriber_t* subscriber){return
  *
  * @param[in] subscriber  The HRT subscriber to initialize. Must not be NULL.
  */
-void urtHrtSubscriberInit(urt_hrtsubscriber_t* subscriber){return;}
+void urtHrtSubscriberInit(urt_hrtsubscriber_t* subscriber)
+{
+  subscriber->base.topic = NULL;
+  urtEventListenerInit(subscriber->base.evtListener);
+  subscriber->base.lastMessage = NULL;
+  subscriber->base.lastMessageTime = 0;
+
+  #if (URT_CFG_PUBSUB_PROFILING)
+    subscriber->base.sumLatencies = 0;
+    subscriber->base.numMessagesReceived = 0;
+  #endif /* URT_CFG_PUBSUB_PROFILING */
+
+  subscriber->next = NULL;
+
+  #if (URT_CFG_PUBSUB_QOS_DEADLINECHECKS)
+    subscriber->deadlineOffset = 0;
+    urtTimerInit(subscriber->qodDeadlineTimer);
+  #endif /* URT_CFG_PUBSUB_QOS_DEADLINECHECKS */
+
+  #if (URT_CFG_PUBSUB_QOS_RATECHECKS)
+    subscriber->expectedRate = 0;
+  #endif /* URT_CFG_PUBSUB_QOS_RATECHECKS */
+
+  #if (URT_CFG_PUBSUB_QOS_JITTERCHECKS)
+    subscriber->maxJitter = 0;
+  #endif /* URT_CFG_PUBSUB_QOS_JITTERCHECKS */
+
+  #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 */
+  return;
+}
+
 
 /**
  * @brief  Subscribes the subscriber to a topic.
  *
- * @param[in] subscriber  The NRT subscriber which shall subscribe to a topic. Must not be NULL.
+ * @param[in] subscriber  The HRT subscriber which shall subscribe to a topic. Must not be NULL.
  * @param[in] topic  The topic to subscribe to. Must not be NULL.
  * @param[in] messages  NULL terminated list of messages to contribute to the topic.
  *                      Messages must not be associated to another topic.
diff --git a/src/urt_topic.c b/src/urt_topic.c
index 862e101..21273ac 100644
--- a/src/urt_topic.c
+++ b/src/urt_topic.c
@@ -58,9 +58,13 @@ urt_status_t urtTopicInit(urt_topic_t* topic, urt_topicid_t id)
 {
   urt_topic_t.next = NULL;
   urt_topic_t.id = id;
-  //add later: urtmutexinit, urteventsourceinit
+  urtMutexInit(topic->lock);
+  urtEventSourceInit(topic->evtSource);
   urt_topic_t.numHrtSubscribers = 0;
-  //add later: condvar, message init, latest message
+  urtCondvarInit(topic->hrtReleased);
+  void* payload = NULL; //TODO: Replace
+  urtMessageInit(topic->mandatoryMessage, payload);
+  topic->latestMessage = &topic->mandatoryMessage;
   #if (URT_CFG_PUBSUB_QOS_RATECHECKS)
     urt_topic_t.hrtSubscribers = nullptr;
     //add later: timer init;
@@ -70,6 +74,27 @@ urt_status_t urtTopicInit(urt_topic_t* topic, urt_topicid_t id)
     urt_topic_t.numMessagesDiscarded = 0;
     urt_topic_t.numSubscribers = 0;
   #endif /* URT_CFG_PUBSUB_PROFILING */
-  //add later: close circular message buffer,...
-  return URT_STATUS_OK;
+  topic->latestMessage->next = topic->latestMessage;
+  urtMutexLock(topic->lock);
+  urt_topic_t* topicTemp = topic;
+  while (topicTemp != NULL && topicTemp->id < id)
+      topicTemp = topicTemp->next;
+
+  if (topicTemp == NULL)
+  {
+    //TODO: Append self to core's list of topic
+    urtMutexUnlock(topic->lock);
+    return URT_STATUS_OK;
+  }
+  else if (topicTemp->id > id)
+  {
+    topicTemp->next = topic;
+    urtMutexUnlock(topic->lock);
+    return URT_STATUS_OK;
+  }
+  else
+  {
+    urtMutexUnlock(topic->lock);
+    return URT_STATUS_TOPIC_DUPLICATE;
+  }
 }
diff --git a/urtware.h b/urtware.h
index ebf58da..46c38f8 100644
--- a/urtware.h
+++ b/urtware.h
@@ -96,6 +96,9 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
  * TODO: Add further µRtWare includes here (e.g. urt_core.h).
  */
 #include <apps_urtosal.h>
+#include <urt_message.h>
+#include <urt_topic.h>
+#include <urt_subscriber.h>
 #include <urt_node.h>
 #include <urt_core.h>
 
