2 This file is part of GNUnet.
3 (C) 2009 Christian Grothoff (and other contributing authors)
5 GNUnet is free software; you can redistribute it and/or modify
6 it under the terms of the GNU General Public License as published
7 by the Free Software Foundation; either version 3, or (at your
8 option) any later version.
10 GNUnet is distributed in the hope that it will be useful, but
11 WITHOUT ANY WARRANTY; without even the implied warranty of
12 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
13 General Public License for more details.
15 You should have received a copy of the GNU General Public License
16 along with GNUnet; see the file COPYING. If not, write to the
17 Free Software Foundation, Inc., 59 Temple Place - Suite 330,
18 Boston, MA 02111-1307, USA.
22 * @file sensordashboard/gnunet-service-sensordashboard.c
23 * @brief Service collecting sensor readings from peers
24 * @author Omar Tarabai
28 #include "gnunet_util_lib.h"
29 #include "gnunet_applications.h"
30 #include "sensordashboard.h"
31 #include "gnunet_cadet_service.h"
32 #include "gnunet_sensor_util_lib.h"
33 #include "gnunet_peerstore_service.h"
37 * Context of a connected client peer
39 struct ClientPeerContext
45 struct ClientPeerContext *prev;
50 struct ClientPeerContext *next;
53 * GNUnet Peer identity
55 struct GNUNET_PeerIdentity peerid;
58 * Handle to the cadet channel
60 struct GNUNET_CADET_Channel *ch;
63 * CADET transmit handle if we requested a transmission
65 struct GNUNET_CADET_TransmitHandle *th;
68 * Head of DLL of pending messages to be sent to client
70 struct PendingMessage *pm_head;
73 * Tail of DLL of pending messages to be sent to client
75 struct PendingMessage *pm_tail;
78 * Are we in the process of destroying this context?
85 * Message queued to be sent to a client stored in a DLL
93 struct PendingMessage *prev;
98 struct PendingMessage *next;
101 * Actual queued message
103 struct GNUNET_MessageHeader *msg;
108 * Carries a single reading from a sensor
110 struct ClientSensorReading
114 * Sensor this reading is related to
116 struct GNUNET_SENSOR_SensorInfo *sensor;
119 * Timestamp of taking the reading
137 * Path to sensor definition directory
139 static char *sensor_dir;
142 * Global hashmap of defined sensors
144 static struct GNUNET_CONTAINER_MultiHashMap *sensors;
147 * Handle to CADET service
149 static struct GNUNET_CADET_Handle *cadet;
152 * Handle to the peerstore service connection
154 static struct GNUNET_PEERSTORE_Handle *peerstore;
157 * Name of this subsystem to be used for peerstore operations
159 static char *subsystem = "sensordashboard";
162 * Head of a DLL of all connected client peers
164 static struct ClientPeerContext *cp_head;
167 * Tail of a DLL of all connected client peers
169 static struct ClientPeerContext *cp_tail;
173 * Trigger sending next pending message to the given client peer if any.
175 * @param cp client peer context struct
178 trigger_send_next_msg (struct ClientPeerContext *cp);
182 * Destroy a given client peer context
184 * @param cp client peer context
187 destroy_clientpeer (struct ClientPeerContext *cp)
189 struct PendingMessage *pm;
191 cp->destroying = GNUNET_YES;
194 GNUNET_CADET_notify_transmit_ready_cancel (cp->th);
200 GNUNET_CONTAINER_DLL_remove (cp->pm_head, cp->pm_tail, pm);
201 GNUNET_free (pm->msg);
207 GNUNET_CADET_channel_destroy (cp->ch);
215 * Task run during shutdown.
221 cleanup_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
223 struct ClientPeerContext *cp;
228 GNUNET_CONTAINER_DLL_remove (cp_head, cp_tail, cp);
229 destroy_clientpeer (cp);
234 GNUNET_CADET_disconnect (cadet);
237 if (NULL != peerstore)
239 GNUNET_PEERSTORE_disconnect (peerstore, GNUNET_YES);
242 GNUNET_SENSOR_destroy_sensors (sensors);
243 if (NULL != sensor_dir)
245 GNUNET_free (sensor_dir);
248 GNUNET_SCHEDULER_shutdown ();
253 * Function called whenever a channel is destroyed. Should clean up
254 * any associated state.
256 * It must NOT call #GNUNET_CADET_channel_destroy on the channel.
258 * @param cls closure (set from #GNUNET_CADET_connect)
259 * @param channel connection to the other end (henceforth invalid)
260 * @param channel_ctx place where local state associated
261 * with the channel is stored
264 cadet_channel_destroyed (void *cls, const struct GNUNET_CADET_Channel *channel,
267 struct ClientPeerContext *cp = channel_ctx;
269 if (GNUNET_YES == cp->destroying)
272 GNUNET_CONTAINER_DLL_remove (cp_head, cp_tail, cp);
273 destroy_clientpeer (cp);
278 * Method called whenever another peer has added us to a channel
279 * the other peer initiated.
280 * Only called (once) upon reception of data with a message type which was
281 * subscribed to in #GNUNET_CADET_connect.
283 * A call to #GNUNET_CADET_channel_destroy causes the channel to be ignored. In
284 * this case the handler MUST return NULL.
287 * @param channel new handle to the channel
288 * @param initiator peer that started the channel
289 * @param port Port this channel is for.
290 * @param options CadetOption flag field, with all active option bits set to 1.
292 * @return initial channel context for the channel
293 * (can be NULL -- that's not an error)
296 cadet_channel_created (void *cls, struct GNUNET_CADET_Channel *channel,
297 const struct GNUNET_PeerIdentity *initiator,
298 uint32_t port, enum GNUNET_CADET_ChannelOption options)
300 struct ClientPeerContext *cp;
302 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
303 "Received a channel connection from peer `%s'.\n",
304 GNUNET_i2s (initiator));
305 cp = GNUNET_new (struct ClientPeerContext);
307 cp->peerid = *initiator;
309 cp->destroying = GNUNET_NO;
310 GNUNET_CONTAINER_DLL_insert (cp_head, cp_tail, cp);
316 * Function called to notify a client about the connection begin ready
317 * to queue more data. @a buf will be NULL and @a size zero if the
318 * connection was closed for writing in the meantime.
320 * Perform the actual sending of the message to client peer.
322 * @param cls closure, a `struct ClientPeerContext *`
323 * @param size number of bytes available in @a buf
324 * @param buf where the callee should write the message
325 * @return number of bytes written to @a buf
328 do_send_msg (void *cls, size_t size, void *buf)
330 struct ClientPeerContext *cp = cls;
331 struct PendingMessage *pm;
336 msg_size = ntohs (pm->msg->size);
337 GNUNET_CONTAINER_DLL_remove (cp->pm_head, cp->pm_tail, pm);
338 if (NULL == buf || size < msg_size)
340 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
341 _("Error trying to send a message to peer `%s'.\n"),
342 GNUNET_i2s (&cp->peerid));
345 memcpy (buf, pm->msg, msg_size);
346 GNUNET_free (pm->msg);
348 trigger_send_next_msg (cp);
354 * Trigger sending next pending message to the given client peer if any.
356 * @param cp client peer context struct
359 trigger_send_next_msg (struct ClientPeerContext *cp)
361 struct PendingMessage *pm;
363 if (NULL == cp->pm_head)
369 GNUNET_CADET_notify_transmit_ready (cp->ch, GNUNET_YES,
370 GNUNET_TIME_UNIT_FOREVER_REL,
371 ntohs (pm->msg->size), &do_send_msg,
377 * Add a new message to the queue to be sent to the given client peer.
379 * @param msg Message to be queued
380 * @param cp Client peer context
383 queue_msg (struct GNUNET_MessageHeader *msg, struct ClientPeerContext *cp)
385 struct PendingMessage *pm;
387 pm = GNUNET_new (struct PendingMessage);
390 GNUNET_CONTAINER_DLL_insert_tail (cp->pm_head, cp->pm_tail, pm);
391 trigger_send_next_msg (cp);
396 * Iterate over defined sensors, creates and sends brief sensor information to
397 * given client peer over CADET.
399 * @param cls closure, the client peer
400 * @param key sensor key
401 * @param value sensor value
402 * @return #GNUNET_YES to continue iteration
405 send_sensor_brief (void *cls, const struct GNUNET_HashCode *key, void *value)
407 struct ClientPeerContext *cp = cls;
408 struct GNUNET_SENSOR_SensorInfo *sensor = value;
409 struct GNUNET_SENSOR_SensorBriefMessage *msg;
410 uint16_t sensorname_size;
413 /* Create message struct */
414 sensorname_size = strlen (sensor->name) + 1;
416 sizeof (struct GNUNET_SENSOR_SensorBriefMessage) + sensorname_size;
417 msg = GNUNET_malloc (total_size);
418 msg->header.size = htons (total_size);
419 msg->header.type = htons (GNUNET_MESSAGE_TYPE_SENSOR_BRIEF);
420 msg->name_size = htons (sensorname_size);
421 msg->version_major = htons (sensor->version_major);
422 msg->version_minor = htons (sensor->version_minor);
423 memcpy (&msg[1], sensor->name, sensorname_size);
425 queue_msg ((struct GNUNET_MessageHeader *) msg, cp);
431 * Called with any sensor list request received.
433 * Each time the function must call #GNUNET_CADET_receive_done on the channel
434 * in order to receive the next message. This doesn't need to be immediate:
435 * can be delayed if some processing is done on the message.
437 * @param cls Closure (set from #GNUNET_CADET_connect).
438 * @param channel Connection to the other end.
439 * @param channel_ctx Place to store local state associated with the channel.
440 * @param message The actual message.
441 * @return #GNUNET_OK to keep the channel open,
442 * #GNUNET_SYSERR to close it (signal serious error).
445 handle_sensor_list_req (void *cls, struct GNUNET_CADET_Channel *channel,
447 const struct GNUNET_MessageHeader *message)
449 struct ClientPeerContext *cp = *channel_ctx;
450 struct GNUNET_MessageHeader *end_msg;
452 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
453 "Received a sensor list request from peer `%s'.\n",
454 GNUNET_i2s (&cp->peerid));
455 GNUNET_CONTAINER_multihashmap_iterate (sensors, &send_sensor_brief, cp);
456 end_msg = GNUNET_new (struct GNUNET_MessageHeader);
458 end_msg->size = htons (sizeof (struct GNUNET_MessageHeader));
459 end_msg->type = htons (GNUNET_MESSAGE_TYPE_SENSOR_END);
460 queue_msg (end_msg, cp);
461 GNUNET_CADET_receive_done (channel);
467 * Parses a sensor reading message struct
469 * @param msg message header received
470 * @param sensors multihashmap of loaded sensors
471 * @return sensor reading struct or NULL if error
473 static struct ClientSensorReading *
474 parse_reading_message (const struct GNUNET_MessageHeader *msg,
475 struct GNUNET_CONTAINER_MultiHashMap *sensors)
478 struct GNUNET_SENSOR_ReadingMessage *rm;
479 uint16_t sensorname_size;
483 struct GNUNET_HashCode key;
484 struct GNUNET_SENSOR_SensorInfo *sensor;
485 struct ClientSensorReading *reading;
487 msg_size = ntohs (msg->size);
488 if (msg_size < sizeof (struct GNUNET_SENSOR_ReadingMessage))
490 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Invalid reading message size.\n");
493 rm = (struct GNUNET_SENSOR_ReadingMessage *) msg;
494 sensorname_size = ntohs (rm->sensorname_size);
495 value_size = ntohs (rm->value_size);
496 if ((sizeof (struct GNUNET_SENSOR_ReadingMessage) + sensorname_size +
497 value_size) != msg_size)
499 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Invalid reading message size.\n");
503 sensorname = GNUNET_malloc (sensorname_size);
504 memcpy (sensorname, dummy, sensorname_size);
505 GNUNET_CRYPTO_hash (sensorname, sensorname_size, &key);
506 GNUNET_free (sensorname);
507 sensor = GNUNET_CONTAINER_multihashmap_get (sensors, &key);
510 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
511 "Unknown sensor name in reading message.\n");
514 if ((sensor->version_minor != ntohs (rm->sensorversion_minor)) ||
515 (sensor->version_major != ntohs (rm->sensorversion_major)))
517 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
518 "Sensor version mismatch in reading message.\n");
521 if (0 == strcmp (sensor->expected_datatype, "numeric") &&
522 sizeof (double) != value_size)
524 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
525 "Invalid value size for a numerical sensor.\n");
528 reading = GNUNET_new (struct ClientSensorReading);
529 reading->sensor = sensor;
530 reading->timestamp = GNUNET_be64toh (rm->timestamp);
531 reading->value_size = value_size;
532 reading->value = GNUNET_malloc (value_size);
533 dummy += sensorname_size;
534 memcpy (reading->value, dummy, value_size);
540 * Called with any sensor reading messages received from CADET.
542 * Each time the function must call #GNUNET_CADET_receive_done on the channel
543 * in order to receive the next message. This doesn't need to be immediate:
544 * can be delayed if some processing is done on the message.
546 * @param cls Closure (set from #GNUNET_CADET_connect).
547 * @param channel Connection to the other end.
548 * @param channel_ctx Place to store local state associated with the channel.
549 * @param message The actual message.
550 * @return #GNUNET_OK to keep the channel open,
551 * #GNUNET_SYSERR to close it (signal serious error).
554 handle_sensor_reading (void *cls, struct GNUNET_CADET_Channel *channel,
556 const struct GNUNET_MessageHeader *message)
558 struct ClientPeerContext *cp = *channel_ctx;
559 struct ClientSensorReading *reading;
561 reading = parse_reading_message (message, sensors);
564 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
565 "Received an invalid sensor reading from peer `%s'.\n",
566 GNUNET_i2s (&cp->peerid));
567 return GNUNET_SYSERR;
569 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
570 "Received a sensor reading from peer `%s':\n"
571 "# Sensor name: `%s'\n" "# Timestamp: %" PRIu64 "\n"
572 "# Value size: %" PRIu64 ".\n", GNUNET_i2s (&cp->peerid),
573 reading->sensor->name, reading->timestamp, reading->value_size);
574 GNUNET_PEERSTORE_store (peerstore, subsystem, &cp->peerid,
575 reading->sensor->name, reading->value,
576 reading->value_size, GNUNET_TIME_UNIT_FOREVER_ABS,
577 GNUNET_PEERSTORE_STOREOPTION_MULTIPLE, NULL, NULL);
578 GNUNET_free (reading->value);
579 GNUNET_free (reading);
580 GNUNET_CADET_receive_done (channel);
586 * Create a message with full information about sensor
588 * @param sensorname Name of sensor requested
589 * @return Message ready to be sent to client or NULL on error
591 static struct GNUNET_SENSOR_SensorFullMessage *
592 create_full_sensor_msg (char *sensorname)
594 struct GNUNET_HashCode key;
595 struct GNUNET_SENSOR_SensorInfo *sensor;
596 struct GNUNET_SENSOR_SensorFullMessage *msg;
598 char *sensorscript_path;
599 uint64_t sensorname_size;
600 uint64_t sensorfile_size;
601 uint64_t sensorscriptname_size;
602 uint64_t sensorscript_size;
606 GNUNET_CRYPTO_hash (sensorname, strlen (sensorname) + 1, &key);
607 sensor = GNUNET_CONTAINER_multihashmap_get (sensors, &key);
610 GNUNET_asprintf (&sensor_path, "%s%s", sensor_dir, sensorname);
612 GNUNET_DISK_file_size (sensor_path, &sensorfile_size, GNUNET_NO,
615 GNUNET_free (sensor_dir);
616 GNUNET_free (sensor_path);
619 sensorname_size = strlen (sensorname) + 1;
620 sensorscript_size = 0;
621 sensorscriptname_size = 0;
622 /* Test if there is an associated script */
623 if (NULL != sensor->ext_process)
625 GNUNET_asprintf (&sensorscript_path, "%s%s-files%s%s", sensor_dir,
626 sensor->name, DIR_SEPARATOR_STR, sensor->ext_process);
628 GNUNET_DISK_file_size (sensorscript_path, &sensorscript_size, GNUNET_NO,
631 sensorscriptname_size = strlen (sensor->ext_process) + 1;
634 /* Construct the msg */
636 sizeof (struct GNUNET_SENSOR_SensorFullMessage) + sensorname_size +
637 sensorfile_size + sensorscriptname_size + sensorscript_size;
638 msg = GNUNET_malloc (total_size);
639 msg->header.size = htons (total_size);
640 msg->header.type = htons (GNUNET_MESSAGE_TYPE_SENSOR_FULL);
641 msg->sensorname_size = htons (sensorname_size);
642 msg->sensorfile_size = htons (sensorfile_size);
643 msg->scriptname_size = htons (sensorscriptname_size);
644 msg->scriptfile_size = htons (sensorscript_size);
646 memcpy (dummy, sensorname, sensorname_size);
647 dummy += sensorname_size;
648 GNUNET_DISK_fn_read (sensor_path, dummy, sensorfile_size);
649 dummy += sensorfile_size;
650 if (sensorscriptname_size > 0)
652 memcpy (dummy, sensor->ext_process, sensorscriptname_size);
653 dummy += sensorscriptname_size;
654 GNUNET_DISK_fn_read (sensorscript_path, dummy, sensorscript_size);
655 GNUNET_free (sensorscript_path);
657 GNUNET_free (sensor_path);
663 * Called with any request for full sensor information.
665 * Each time the function must call #GNUNET_CADET_receive_done on the channel
666 * in order to receive the next message. This doesn't need to be immediate:
667 * can be delayed if some processing is done on the message.
669 * @param cls Closure (set from #GNUNET_CADET_connect).
670 * @param channel Connection to the other end.
671 * @param channel_ctx Place to store local state associated with the channel.
672 * @param message The actual message.
673 * @return #GNUNET_OK to keep the channel open,
674 * #GNUNET_SYSERR to close it (signal serious error).
677 handle_sensor_full_req (void *cls, struct GNUNET_CADET_Channel *channel,
679 const struct GNUNET_MessageHeader *message)
681 struct ClientPeerContext *cp = *channel_ctx;
682 struct GNUNET_SENSOR_SensorBriefMessage *sbm = NULL;
683 struct GNUNET_SENSOR_SensorFullMessage *sfm;
685 uint16_t sensorname_size;
687 msg_size = ntohs (message->size);
688 /* parse & error check */
689 if (msg_size > sizeof (struct GNUNET_SENSOR_SensorBriefMessage))
691 sbm = (struct GNUNET_SENSOR_SensorBriefMessage *) message;
692 sensorname_size = ntohs (sbm->name_size);
694 sizeof (struct GNUNET_SENSOR_SensorBriefMessage) + sensorname_size)
699 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
700 "Received an invalid full sensor request from peer `%s'.\n",
701 GNUNET_i2s (&cp->peerid));
702 return GNUNET_SYSERR;
704 /* Create and send msg with full sensor info */
705 sfm = create_full_sensor_msg ((char *) &sbm[1]);
708 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
709 "Error creating full sensor info msg for sensor `%s'.\n",
711 return GNUNET_SYSERR;
713 queue_msg ((struct GNUNET_MessageHeader *) sfm, cp);
714 GNUNET_CADET_receive_done (channel);
720 * Process sensordashboard requests.
723 * @param server the initialized server
724 * @param cfg configuration to use
727 run (void *cls, struct GNUNET_SERVER_Handle *server,
728 const struct GNUNET_CONFIGURATION_Handle *cfg)
730 static const struct GNUNET_SERVER_MessageHandler handlers[] = {
733 static struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
734 {&handle_sensor_reading,
735 GNUNET_MESSAGE_TYPE_SENSOR_READING, 0},
736 {&handle_sensor_list_req,
737 GNUNET_MESSAGE_TYPE_SENSOR_LIST_REQ,
738 sizeof (struct GNUNET_MessageHeader)},
739 {&handle_sensor_full_req,
740 GNUNET_MESSAGE_TYPE_SENSOR_FULL_REQ,
741 sizeof (struct GNUNET_MessageHeader)},
744 static uint32_t cadet_ports[] = {
745 GNUNET_APPLICATION_TYPE_SENSORDASHBOARD,
746 GNUNET_APPLICATION_TYPE_SENSORUPDATE,
747 GNUNET_APPLICATION_TYPE_END
751 GNUNET_CONFIGURATION_get_value_filename (cfg, "SENSOR", "SENSOR_DIR",
753 sensor_dir = GNUNET_SENSOR_get_default_sensor_dir ();
754 sensors = GNUNET_SENSOR_load_all_sensors (sensor_dir);
755 GNUNET_assert (NULL != sensors);
757 GNUNET_CADET_connect (cfg, NULL, &cadet_channel_created,
758 &cadet_channel_destroyed, cadet_handlers,
762 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
763 _("Failed to connect to `%s' service.\n"), "CADET");
764 GNUNET_SCHEDULER_add_now (&cleanup_task, NULL);
767 peerstore = GNUNET_PEERSTORE_connect (cfg);
768 if (NULL == peerstore)
770 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
771 _("Failed to connect to `%s' service.\n"), "PEERSTORE");
772 GNUNET_SCHEDULER_add_now (&cleanup_task, NULL);
775 GNUNET_SERVER_add_handlers (server, handlers);
776 GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &cleanup_task,
782 * The main function for the sensordashboard service.
784 * @param argc number of arguments from the command line
785 * @param argv command line arguments
786 * @return 0 ok, 1 on error
789 main (int argc, char *const *argv)
792 GNUNET_SERVICE_run (argc, argv, "sensordashboard",
793 GNUNET_SERVICE_OPTION_NONE, &run, NULL)) ? 0 : 1;
796 /* end of gnunet-service-sensordashboard.c */