7 files changed, 319 insertions, 130 deletions

diff --git a/src/include/gnunet_sensor_service.h b/src/include/gnunet_sensor_service.h
index f81248b28..eb4fbdc67 100644
--- a/src/include/gnunet_sensor_service.h
+++ b/src/include/gnunet_sensor_service.h
@@ -71,50 +71,6 @@ struct SensorInfoShort
 
 };
 
-GNUNET_NETWORK_STRUCT_BEGIN
-
-/**
- * Used to communicate sensor readings to
- * collection points (SENSORDASHBAORD service)
- */
-struct GNUNET_SENSOR_Reading
-{
-
-  /**
-   * GNUNET general message header
-   */
-  struct GNUNET_MessageHeader *header;
-
-  /**
-   * Size of the sensor name value, allocated
-   * at position 0 after this struct
-   */
-  size_t sensorname_size;
-
-  /**
-   * First part of sensor version number
-   */
-  uint16_t sensorversion_major;
-
-  /**
-   * Second part of sensor version number
-   */
-  uint16_t sensorversion_minor;
-
-  /**
-   * Timestamp of recorded reading
-   */
-  uint64_t timestamp;
-
-  /**
-   * Size of reading value, allocation
-   * at poistion 1 after this struct
-   */
-  size_t value_size;
-
-};
-GNUNET_NETWORK_STRUCT_END
-
 /**
  * Type of an iterator over sensor definitions.
  *
diff --git a/src/include/gnunet_sensor_util_lib.h b/src/include/gnunet_sensor_util_lib.h
index 15fb61b4c..56e0bc868 100644
--- a/src/include/gnunet_sensor_util_lib.h
+++ b/src/include/gnunet_sensor_util_lib.h
@@ -188,6 +188,78 @@ struct SensorInfo
 };
 
 /**
+ * Carries a single reading from a sensor
+ */
+struct GNUNET_SENSOR_Reading
+{
+
+  /**
+   * Sensor this reading is related to
+   */
+  struct SensorInfo *sensor;
+
+  /**
+   * Timestamp of taking the reading
+   */
+  uint64_t timestamp;
+
+  /**
+   * Reading value
+   */
+  void *value;
+
+  /**
+   * Size of @value
+   */
+  uint16_t value_size;
+
+};
+
+GNUNET_NETWORK_STRUCT_BEGIN
+
+/**
+ * Used to communicate sensor readings to
+ * collection points (SENSORDASHBAORD service)
+ */
+struct GNUNET_SENSOR_ReadingMessage
+{
+
+  /**
+   * GNUNET general message header
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Size of the sensor name value, allocated
+   * at position 0 after this struct
+   */
+  uint16_t sensorname_size;
+
+  /**
+   * First part of sensor version number
+   */
+  uint16_t sensorversion_major;
+
+  /**
+   * Second part of sensor version number
+   */
+  uint16_t sensorversion_minor;
+
+  /**
+   * Timestamp of recorded reading
+   */
+  uint64_t timestamp;
+
+  /**
+   * Size of reading value, allocation
+   * at poistion 1 after this struct
+   */
+  uint16_t value_size;
+
+};
+GNUNET_NETWORK_STRUCT_END
+
+/**
  * Reads sensor definitions from local data files
  *
  * @return a multihashmap of loaded sensors
@@ -203,6 +275,25 @@ GNUNET_SENSOR_load_all_sensors ();
 char *
 GNUNET_SENSOR_get_sensor_dir ();
 
+/**
+ * Parses a sensor reading message struct
+ *
+ * @param msg message header received
+ * @param sensors multihashmap of loaded sensors
+ * @return sensor reading struct or NULL if error
+ */
+struct GNUNET_SENSOR_Reading *
+GNUNET_SENSOR_parse_reading_message (const struct GNUNET_MessageHeader *msg,
+    struct GNUNET_CONTAINER_MultiHashMap *sensors);
+
+/**
+ * Destroys a group of sensors in a hashmap and the hashmap itself
+ *
+ * @param sensors hashmap containing the sensors
+ */
+void
+GNUNET_SENSOR_destroy_sensors (struct GNUNET_CONTAINER_MultiHashMap *sensors);
+
 #if 0                           /* keep Emacsens' auto-indent happy */
 {
 #endif
diff --git a/src/sensor/Makefile.am b/src/sensor/Makefile.am
index a9b703cbd..1ddff32be 100644
--- a/src/sensor/Makefile.am
+++ b/src/sensor/Makefile.am
@@ -56,11 +56,13 @@ libgnunetsensor_la_LDFLAGS = \
 libgnunetsensorutil_la_SOURCES = \
   sensor_util_lib.c
 libgnunetsensorutil_la_LIBADD = \
-  $(top_builddir)/src/util/libgnunetutil.la
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/statistics/libgnunetstatistics.la
 libgnunetsensorutil_la_LDFLAGS = \
   $(GN_LIB_LDFLAGS)
 libgnunetsensorutil_la_DEPENDENCIES = \
-  $(top_builddir)/src/util/libgnunetutil.la
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/statistics/libgnunetstatistics.la
 
 plugin_LTLIBRARIES = \
   libgnunet_plugin_sensor_model_gaussian.la
diff --git a/src/sensor/gnunet-service-sensor-reporting.c b/src/sensor/gnunet-service-sensor-reporting.c
index 7b32c7f6c..94f53e4ec 100644
--- a/src/sensor/gnunet-service-sensor-reporting.c
+++ b/src/sensor/gnunet-service-sensor-reporting.c
@@ -23,6 +23,7 @@
  * @brief sensor service reporting functionality
  * @author Omar Tarabai
  */
+#include <inttypes.h>
 #include "platform.h"
 #include "gnunet_util_lib.h"
 #include "sensor.h"
@@ -33,6 +34,11 @@
 #define LOG(kind,...) GNUNET_log_from (kind, "sensor-reporting",__VA_ARGS__)
 
 /**
+ * Retry interval (seconds) in case channel to collection point is busy
+ */
+#define COLLECTION_RETRY 1
+
+/**
  * Context of reporting operations
  */
 struct ReportingContext
@@ -129,6 +135,11 @@ struct CadetChannelContext
    */
   struct GNUNET_CADET_TransmitHandle *th;
 
+  /**
+   * Are we currently destroying the channel and its context?
+   */
+  int destroying;
+
 };
 
 /**
@@ -202,6 +213,7 @@ destroy_reporting_context (struct ReportingContext *rc)
 static void
 destroy_cadet_channel_context (struct CadetChannelContext *cc)
 {
+  cc->destroying = GNUNET_YES;
   if (NULL != cc->th)
   {
     GNUNET_CADET_notify_transmit_ready_cancel (cc->th);
@@ -280,6 +292,7 @@ get_cadet_channel (struct GNUNET_PeerIdentity pid)
       GNUNET_CADET_OPTION_DEFAULT);
   cc->pid = pid;
   cc->sending = GNUNET_NO;
+  cc->destroying = GNUNET_NO;
   GNUNET_CONTAINER_DLL_insert (cc_head, cc_tail, cc);
   return cc;
 }
@@ -293,25 +306,25 @@ get_cadet_channel (struct GNUNET_PeerIdentity pid)
  */
 static size_t
 construct_reading_message (struct ReportingContext *rc,
-    struct GNUNET_SENSOR_Reading **msg)
+    struct GNUNET_SENSOR_ReadingMessage **msg)
 {
-  struct GNUNET_SENSOR_Reading *ret;
-  size_t sensorname_size;
-  size_t total_size;
+  struct GNUNET_SENSOR_ReadingMessage *ret;
+  uint16_t sensorname_size;
+  uint16_t total_size;
   void *dummy;
 
   sensorname_size = strlen (rc->sensor->name) + 1;
-  total_size = sizeof(struct GNUNET_SENSOR_Reading) +
+  total_size = sizeof(struct GNUNET_SENSOR_ReadingMessage) +
       sensorname_size +
       rc->last_value_size;
   ret = GNUNET_malloc (total_size);
-  ret->header->size = htons (total_size);
-  ret->header->type = htons (GNUNET_MESSAGE_TYPE_SENSOR_READING);
-  ret->sensorname_size = GNUNET_htobe64 (sensorname_size);
+  ret->header.size = htons (total_size);
+  ret->header.type = htons (GNUNET_MESSAGE_TYPE_SENSOR_READING);
+  ret->sensorname_size = htons (sensorname_size);
   ret->sensorversion_major = htons (rc->sensor->version_major);
   ret->sensorversion_minor = htons (rc->sensor->version_minor);
   ret->timestamp = GNUNET_htobe64 (rc->timestamp);
-  ret->value_size = GNUNET_htobe64 (rc->last_value_size);
+  ret->value_size = htons (rc->last_value_size);
   dummy = &ret[1];
   memcpy (dummy, rc->sensor->name, sensorname_size);
   dummy += sensorname_size;
@@ -336,12 +349,19 @@ do_report_collection_point (void *cls, size_t size, void *buf)
   struct CadetChannelContext *cc = cls;
   size_t written = 0;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "Copying to CADET transmit buffer.\n");
+  cc->th = NULL;
   cc->sending = GNUNET_NO;
-  if (NULL == buf || size != cc->pending_msg_size)
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Copying to CADET transmit buffer.\n");
+  if (NULL == buf)
   {
     LOG (GNUNET_ERROR_TYPE_WARNING,
-        "CADET failed to transmit message to collection point, discarding.");
+        "CADET failed to transmit message (NULL buf), discarding.\n");
+  }
+  else if (size < cc->pending_msg_size)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+        "CADET failed to transmit message (small size, expected: %u, got: %u)"
+        ", discarding.\n", cc->pending_msg_size, size);
   }
   else
   {
@@ -349,6 +369,7 @@ do_report_collection_point (void *cls, size_t size, void *buf)
     written = cc->pending_msg_size;
   }
   GNUNET_free (cc->pending_msg);
+  cc->pending_msg = NULL;
   cc->pending_msg_size = 0;
   return written;
 }
@@ -365,7 +386,7 @@ static void report_collection_point
   struct ReportingContext *rc = cls;
   struct SensorInfo *sensor = rc->sensor;
   struct CadetChannelContext *cc;
-  struct GNUNET_SENSOR_Reading *msg;
+  struct GNUNET_SENSOR_ReadingMessage *msg;
   size_t msg_size;
 
   rc->cp_task = GNUNET_SCHEDULER_NO_TASK;
@@ -385,9 +406,11 @@ static void report_collection_point
   {
     LOG (GNUNET_ERROR_TYPE_DEBUG,
         "Cadet channel to collection point busy, "
-        "trying again for sensor `%s' on next interval.\n", rc->sensor->name);
-    rc->cp_task = GNUNET_SCHEDULER_add_delayed (sensor->collection_interval,
-            &report_collection_point, rc);
+        "trying again for sensor `%s' after %d seconds.\n", rc->sensor->name,
+        COLLECTION_RETRY);
+    rc->cp_task = GNUNET_SCHEDULER_add_delayed (
+      GNUNET_TIME_relative_multiply(GNUNET_TIME_UNIT_SECONDS, COLLECTION_RETRY),
+      &report_collection_point, rc);
     return;
   }
   msg_size = construct_reading_message (rc, &msg);
@@ -414,8 +437,6 @@ sensor_watch_cb (void *cls,
 {
   struct ReportingContext *rc = cls;
 
-  LOG (GNUNET_ERROR_TYPE_DEBUG, "Received a sensor `%s' watch value, "
-      "updating notification last_value.\n", rc->sensor->name);
   if (NULL != emsg)
     return GNUNET_YES;
   if (NULL != rc->last_value)
@@ -427,6 +448,9 @@ sensor_watch_cb (void *cls,
   memcpy (rc->last_value, record->value, record->value_size);
   rc->last_value_size = record->value_size;
   rc->timestamp = GNUNET_TIME_absolute_get().abs_value_us;
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Received a sensor `%s' watch value at "
+      "timestamp %" PRIu64 ", updating notification last_value.\n",
+      rc->sensor->name, rc->timestamp);
   return GNUNET_YES;
 }
 
@@ -502,6 +526,8 @@ static void cadet_channel_destroyed (void *cls,
 {
   struct CadetChannelContext *cc = channel_ctx;
 
+  if (GNUNET_YES == cc->destroying)
+    return;
   LOG (GNUNET_ERROR_TYPE_DEBUG,
       "Received a `channel destroyed' notification from CADET, "
       "cleaning up.\n");
diff --git a/src/sensor/gnunet-service-sensor.c b/src/sensor/gnunet-service-sensor.c
index b41d8906c..20466cec3 100644
--- a/src/sensor/gnunet-service-sensor.c
+++ b/src/sensor/gnunet-service-sensor.c
@@ -61,61 +61,6 @@ char *subsystem = "sensor";
 struct GNUNET_PeerIdentity peerid;
 
 /**
- * Remove sensor execution from scheduler
- *
- * @param cls unused
- * @param key hash of sensor name, key to hashmap
- * @param value a 'struct SensorInfo *'
- * @return #GNUNET_YES if we should continue to
- *         iterate,
- *         #GNUNET_NO if not.
- */
-static int destroy_sensor(void *cls,
-    const struct GNUNET_HashCode *key, void *value)
-{
-  struct SensorInfo *sensorinfo = value;
-
-  GNUNET_log(GNUNET_ERROR_TYPE_DEBUG, "Destroying sensor `%s'\n", sensorinfo->name);
-  if(GNUNET_SCHEDULER_NO_TASK != sensorinfo->execution_task)
-  {
-    GNUNET_SCHEDULER_cancel(sensorinfo->execution_task);
-    sensorinfo->execution_task = GNUNET_SCHEDULER_NO_TASK;
-  }
-  if(NULL != sensorinfo->gnunet_stat_get_handle)
-  {
-    GNUNET_STATISTICS_get_cancel(sensorinfo->gnunet_stat_get_handle);
-    sensorinfo->gnunet_stat_get_handle = NULL;
-  }
-  if(NULL != sensorinfo->ext_cmd)
-  {
-    GNUNET_OS_command_stop(sensorinfo->ext_cmd);
-    sensorinfo->ext_cmd = NULL;
-  }
-  if(NULL != sensorinfo->cfg)
-    GNUNET_CONFIGURATION_destroy(sensorinfo->cfg);
-  if(NULL != sensorinfo->name)
-    GNUNET_free(sensorinfo->name);
-  if(NULL != sensorinfo->def_file)
-    GNUNET_free(sensorinfo->def_file);
-  if(NULL != sensorinfo->description)
-    GNUNET_free(sensorinfo->description);
-  if(NULL != sensorinfo->category)
-    GNUNET_free(sensorinfo->category);
-  if(NULL != sensorinfo->capabilities)
-    GNUNET_free(sensorinfo->capabilities);
-  if(NULL != sensorinfo->gnunet_stat_service)
-    GNUNET_free(sensorinfo->gnunet_stat_service);
-  if(NULL != sensorinfo->gnunet_stat_name)
-    GNUNET_free(sensorinfo->gnunet_stat_name);
-  if(NULL != sensorinfo->ext_process)
-    GNUNET_free(sensorinfo->ext_process);
-  if(NULL != sensorinfo->ext_args)
-    GNUNET_free(sensorinfo->ext_args);
-  GNUNET_free(sensorinfo);
-  return GNUNET_YES;
-}
-
-/**
  * Disable a sensor
  * Sensor will not run again unless
  * explicitly enabled or reloaded
@@ -146,8 +91,7 @@ shutdown_task (void *cls,
 {
   SENSOR_reporting_stop();
   SENSOR_analysis_stop();
-  GNUNET_CONTAINER_multihashmap_iterate(sensors, &destroy_sensor, NULL);
-  GNUNET_CONTAINER_multihashmap_destroy(sensors);
+  GNUNET_SENSOR_destroy_sensors (sensors);
   if(NULL != statistics)
   {
     GNUNET_STATISTICS_destroy(statistics, GNUNET_YES);
@@ -464,6 +408,7 @@ void sensor_process_callback (void *cls, const char *line)
         GNUNET_PEERSTORE_STOREOPTION_MULTIPLE,
         NULL,
         NULL);
+    GNUNET_free (value);
   }
 }
 
diff --git a/src/sensor/sensor_util_lib.c b/src/sensor/sensor_util_lib.c
index 90befe6af..9e36f7ebe 100644
--- a/src/sensor/sensor_util_lib.c
+++ b/src/sensor/sensor_util_lib.c
@@ -27,6 +27,7 @@
 #include "platform.h"
 #include "gnunet_util_lib.h"
 #include "gnunet_sensor_util_lib.h"
+#include "gnunet_statistics_service.h"
 
 #define LOG(kind,...) GNUNET_log_from (kind, "sensor-util",__VA_ARGS__)
 
@@ -215,6 +216,7 @@ load_sensor_from_cfg(struct GNUNET_CONFIGURATION_Handle *cfg, const char *sectio
     return NULL;
   }
   //reporting mechanism
+  sensor->collection_point = NULL;
   if (GNUNET_OK == GNUNET_CONFIGURATION_get_value_string(cfg, sectionname, "COLLECTION_POINT", &dummy))
   {
     if(GNUNET_OK != GNUNET_CONFIGURATION_get_value_number(cfg, sectionname, "COLLECTION_INTERVAL", &time_sec))
@@ -230,6 +232,7 @@ load_sensor_from_cfg(struct GNUNET_CONFIGURATION_Handle *cfg, const char *sectio
         sensor->collection_point->public_key = public_key;
       }
     }
+    GNUNET_free (dummy);
   }
   sensor->p2p_report = GNUNET_NO;
   if (GNUNET_YES == GNUNET_CONFIGURATION_get_value_yesno(cfg, sectionname, "P2P_REPORT"))
@@ -322,7 +325,7 @@ add_sensor_to_hashmap(struct SensorInfo *sensor, struct GNUNET_CONTAINER_MultiHa
   struct GNUNET_HashCode key;
   struct SensorInfo *existing;
 
-  GNUNET_CRYPTO_hash(sensor->name, strlen(sensor->name), &key);
+  GNUNET_CRYPTO_hash(sensor->name, strlen(sensor->name) + 1, &key);
   existing = GNUNET_CONTAINER_multihashmap_get(map, &key);
   if(NULL != existing) //sensor with same name already exists
   {
@@ -369,10 +372,7 @@ reload_sensors_dir_cb(void *cls, const char *filename)
         _("Error loading sensor from file: %s\n"), filename);
     return GNUNET_OK;
   }
-  if(GNUNET_YES == add_sensor_to_hashmap(sensor, sensors))
-    LOG (GNUNET_ERROR_TYPE_DEBUG,
-        "Sensor `%s' added to global hashmap\n", sensor->name);
-  else
+  if(GNUNET_YES != add_sensor_to_hashmap(sensor, sensors))
     LOG (GNUNET_ERROR_TYPE_WARNING,
         "Could not add sensor `%s' to global hashmap\n", sensor->name);
 
@@ -422,3 +422,145 @@ GNUNET_SENSOR_load_all_sensors ()
   GNUNET_free(sensordir);
   return sensors;
 }
+
+/**
+ * Parses a sensor reading message struct
+ *
+ * @param msg message header received
+ * @param sensors multihashmap of loaded sensors
+ * @return sensor reading struct or NULL if error
+ */
+struct GNUNET_SENSOR_Reading *
+GNUNET_SENSOR_parse_reading_message (const struct GNUNET_MessageHeader *msg,
+    struct GNUNET_CONTAINER_MultiHashMap *sensors)
+{
+  uint16_t msg_size;
+  struct GNUNET_SENSOR_ReadingMessage *rm;
+  uint16_t sensorname_size;
+  uint16_t value_size;
+  void *dummy;
+  char *sensorname;
+  struct GNUNET_HashCode key;
+  struct SensorInfo *sensor;
+  struct GNUNET_SENSOR_Reading *reading;
+
+  msg_size = ntohs (msg->size);
+  if (msg_size < sizeof (struct GNUNET_SENSOR_ReadingMessage))
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING, "Invalid reading message size.\n");
+    return NULL;
+  }
+  rm = (struct GNUNET_SENSOR_ReadingMessage *)msg;
+  sensorname_size = ntohs (rm->sensorname_size);
+  value_size = ntohs (rm->value_size);
+  if ((sizeof (struct GNUNET_SENSOR_ReadingMessage)
+      + sensorname_size + value_size) != msg_size)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING, "Invalid reading message size.\n");
+    return NULL;
+  }
+  dummy = &rm[1];
+  sensorname = GNUNET_malloc (sensorname_size);
+  memcpy (sensorname, dummy, sensorname_size);
+  GNUNET_CRYPTO_hash(sensorname, sensorname_size, &key);
+  GNUNET_free (sensorname);
+  sensor = GNUNET_CONTAINER_multihashmap_get (sensors, &key);
+  if (NULL == sensor)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+        "Unknown sensor name in reading message.\n");
+    return NULL;
+  }
+  if ((sensor->version_minor != ntohs (rm->sensorversion_minor)) ||
+      (sensor->version_major != ntohs (rm->sensorversion_major)))
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+        "Sensor version mismatch in reading message.\n");
+    return NULL;
+  }
+  if (0 == strcmp (sensor->expected_datatype, "numeric") &&
+      sizeof (double) != value_size)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+        "Invalid value size for a numerical sensor.\n");
+    return NULL;
+  }
+  reading = GNUNET_new (struct GNUNET_SENSOR_Reading);
+  reading->sensor = sensor;
+  reading->timestamp = GNUNET_be64toh (rm->timestamp);
+  reading->value_size = value_size;
+  reading->value = GNUNET_malloc (value_size);
+  dummy += sensorname_size;
+  memcpy (reading->value, dummy, value_size);
+  return reading;
+}
+
+/**
+ * Remove sensor execution from scheduler
+ *
+ * @param cls unused
+ * @param key hash of sensor name, key to hashmap
+ * @param value a 'struct SensorInfo *'
+ * @return #GNUNET_YES if we should continue to
+ *         iterate,
+ *         #GNUNET_NO if not.
+ */
+static int destroy_sensor(void *cls,
+    const struct GNUNET_HashCode *key, void *value)
+{
+  struct SensorInfo *sensorinfo = value;
+
+  if(GNUNET_SCHEDULER_NO_TASK != sensorinfo->execution_task)
+  {
+    GNUNET_SCHEDULER_cancel(sensorinfo->execution_task);
+    sensorinfo->execution_task = GNUNET_SCHEDULER_NO_TASK;
+  }
+  if(NULL != sensorinfo->gnunet_stat_get_handle)
+  {
+    GNUNET_STATISTICS_get_cancel(sensorinfo->gnunet_stat_get_handle);
+    sensorinfo->gnunet_stat_get_handle = NULL;
+  }
+  if(NULL != sensorinfo->ext_cmd)
+  {
+    GNUNET_OS_command_stop(sensorinfo->ext_cmd);
+    sensorinfo->ext_cmd = NULL;
+  }
+  if(NULL != sensorinfo->cfg)
+    GNUNET_CONFIGURATION_destroy(sensorinfo->cfg);
+  if(NULL != sensorinfo->name)
+    GNUNET_free(sensorinfo->name);
+  if(NULL != sensorinfo->def_file)
+    GNUNET_free(sensorinfo->def_file);
+  if(NULL != sensorinfo->description)
+    GNUNET_free(sensorinfo->description);
+  if(NULL != sensorinfo->category)
+    GNUNET_free(sensorinfo->category);
+  if(NULL != sensorinfo->capabilities)
+    GNUNET_free(sensorinfo->capabilities);
+  if(NULL != sensorinfo->gnunet_stat_service)
+    GNUNET_free(sensorinfo->gnunet_stat_service);
+  if(NULL != sensorinfo->gnunet_stat_name)
+    GNUNET_free(sensorinfo->gnunet_stat_name);
+  if(NULL != sensorinfo->ext_process)
+    GNUNET_free(sensorinfo->ext_process);
+  if(NULL != sensorinfo->ext_args)
+    GNUNET_free(sensorinfo->ext_args);
+  if (NULL != sensorinfo->collection_point)
+    GNUNET_free (sensorinfo->collection_point);
+  GNUNET_free(sensorinfo);
+  return GNUNET_YES;
+}
+
+/**
+ * Destroys a group of sensors in a hashmap and the hashmap itself
+ *
+ * @param sensors hashmap containing the sensors
+ */
+void
+GNUNET_SENSOR_destroy_sensors (struct GNUNET_CONTAINER_MultiHashMap *sensors)
+{
+  LOG (GNUNET_ERROR_TYPE_DEBUG, "Destroying sensor list.\n");
+  GNUNET_CONTAINER_multihashmap_iterate(sensors, &destroy_sensor, NULL);
+  GNUNET_CONTAINER_multihashmap_destroy(sensors);
+}
+
diff --git a/src/sensordashboard/gnunet-service-sensordashboard.c b/src/sensordashboard/gnunet-service-sensordashboard.c
index 91c7e98f5..841268762 100644
--- a/src/sensordashboard/gnunet-service-sensordashboard.c
+++ b/src/sensordashboard/gnunet-service-sensordashboard.c
@@ -23,6 +23,7 @@
  * @brief Service collecting sensor readings from peers
  * @author Omar Tarabai
  */
+#include <inttypes.h>
 #include "platform.h"
 #include "gnunet_util_lib.h"
 #include "gnunet_applications.h"
@@ -51,6 +52,7 @@ cleanup_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
     GNUNET_CADET_disconnect(cadet);
     cadet = NULL;
   }
+  GNUNET_SENSOR_destroy_sensors (sensors);
   GNUNET_SCHEDULER_shutdown();
 }
 
@@ -69,7 +71,9 @@ static void cadet_channel_destroyed (void *cls,
     const struct GNUNET_CADET_Channel *channel,
     void *channel_ctx)
 {
+  struct GNUNET_PeerIdentity *peer = channel_ctx;
 
+  GNUNET_free (peer);
 }
 
 /**
@@ -95,10 +99,11 @@ static void *cadet_channel_created (void *cls,
     const struct GNUNET_PeerIdentity *initiator,
     uint32_t port, enum GNUNET_CADET_ChannelOption options)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-      "CADET channel opened by remote peer `%s'.\n",
-      GNUNET_i2s(initiator));
-  return NULL; /* FIXME */
+  struct GNUNET_PeerIdentity *peer;
+
+  peer = GNUNET_new (struct GNUNET_PeerIdentity);
+  memcpy (peer, initiator, sizeof (struct GNUNET_PeerIdentity));
+  return peer;
 }
 
 /**
@@ -118,7 +123,29 @@ static void *cadet_channel_created (void *cls,
 int sensor_reading_receiver (void *cls, struct GNUNET_CADET_Channel *channel,
     void **channel_ctx, const struct GNUNET_MessageHeader *message)
 {
+  struct GNUNET_PeerIdentity *peer = *channel_ctx;
+  struct GNUNET_SENSOR_Reading *reading;
 
+  reading = GNUNET_SENSOR_parse_reading_message (message, sensors);
+  if (NULL == reading)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+        "Received an invalid sensor reading from peer `%s'\n",
+        GNUNET_i2s (peer));
+    return GNUNET_SYSERR;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+      "Received a sensor reading from peer `%s':\n"
+      "# Sensor name: `%s'\n"
+      "# Timestamp: %" PRIu64 "\n"
+      "# Value size: %" PRIu64 ".\n",
+      GNUNET_i2s (peer),
+      reading->sensor->name,
+      reading->timestamp,
+      reading->value_size);
+  GNUNET_free (reading->value);
+  GNUNET_free (reading);
+  return GNUNET_OK;
 }
 
 /**