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 * Global hashmap of defined sensors
139 static struct GNUNET_CONTAINER_MultiHashMap *sensors;
142 * Handle to CADET service
144 static struct GNUNET_CADET_Handle *cadet;
147 * Handle to the peerstore service connection
149 static struct GNUNET_PEERSTORE_Handle *peerstore;
152 * Name of this subsystem to be used for peerstore operations
154 static char *subsystem = "sensordashboard";
157 * Head of a DLL of all connected client peers
159 static struct ClientPeerContext *cp_head;
162 * Tail of a DLL of all connected client peers
164 static struct ClientPeerContext *cp_tail;
168 * Trigger sending next pending message to the given client peer if any.
170 * @param cp client peer context struct
173 trigger_send_next_msg (struct ClientPeerContext *cp);
177 * Destroy a given client peer context
179 * @param cp client peer context
182 destroy_clientpeer (struct ClientPeerContext *cp)
184 struct PendingMessage *pm;
186 cp->destroying = GNUNET_YES;
189 GNUNET_CADET_notify_transmit_ready_cancel (cp->th);
195 GNUNET_CONTAINER_DLL_remove (cp->pm_head, cp->pm_tail, pm);
196 GNUNET_free (pm->msg);
202 GNUNET_CADET_channel_destroy (cp->ch);
210 * Task run during shutdown.
216 cleanup_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
218 struct ClientPeerContext *cp;
223 GNUNET_CONTAINER_DLL_remove (cp_head, cp_tail, cp);
224 destroy_clientpeer (cp);
229 GNUNET_CADET_disconnect (cadet);
232 if (NULL != peerstore)
234 GNUNET_PEERSTORE_disconnect (peerstore, GNUNET_YES);
237 GNUNET_SENSOR_destroy_sensors (sensors);
238 GNUNET_SCHEDULER_shutdown ();
243 * Function called whenever a channel is destroyed. Should clean up
244 * any associated state.
246 * It must NOT call #GNUNET_CADET_channel_destroy on the channel.
248 * @param cls closure (set from #GNUNET_CADET_connect)
249 * @param channel connection to the other end (henceforth invalid)
250 * @param channel_ctx place where local state associated
251 * with the channel is stored
254 cadet_channel_destroyed (void *cls, const struct GNUNET_CADET_Channel *channel,
257 struct ClientPeerContext *cp = channel_ctx;
259 if (GNUNET_YES == cp->destroying)
262 GNUNET_CONTAINER_DLL_remove (cp_head, cp_tail, cp);
263 destroy_clientpeer (cp);
268 * Method called whenever another peer has added us to a channel
269 * the other peer initiated.
270 * Only called (once) upon reception of data with a message type which was
271 * subscribed to in #GNUNET_CADET_connect.
273 * A call to #GNUNET_CADET_channel_destroy causes the channel to be ignored. In
274 * this case the handler MUST return NULL.
277 * @param channel new handle to the channel
278 * @param initiator peer that started the channel
279 * @param port Port this channel is for.
280 * @param options CadetOption flag field, with all active option bits set to 1.
282 * @return initial channel context for the channel
283 * (can be NULL -- that's not an error)
286 cadet_channel_created (void *cls, struct GNUNET_CADET_Channel *channel,
287 const struct GNUNET_PeerIdentity *initiator,
288 uint32_t port, enum GNUNET_CADET_ChannelOption options)
290 struct ClientPeerContext *cp;
292 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
293 "Received a channel connection from peer `%s'.\n",
294 GNUNET_i2s (initiator));
295 cp = GNUNET_new (struct ClientPeerContext);
297 cp->peerid = *initiator;
299 cp->destroying = GNUNET_NO;
300 GNUNET_CONTAINER_DLL_insert (cp_head, cp_tail, cp);
306 * Function called to notify a client about the connection begin ready
307 * to queue more data. @a buf will be NULL and @a size zero if the
308 * connection was closed for writing in the meantime.
310 * Perform the actual sending of the message to client peer.
312 * @param cls closure, a `struct ClientPeerContext *`
313 * @param size number of bytes available in @a buf
314 * @param buf where the callee should write the message
315 * @return number of bytes written to @a buf
318 do_send_msg (void *cls, size_t size, void *buf)
320 struct ClientPeerContext *cp = cls;
321 struct PendingMessage *pm;
326 msg_size = ntohs (pm->msg->size);
327 GNUNET_CONTAINER_DLL_remove (cp->pm_head, cp->pm_tail, pm);
328 if (NULL == buf || size < msg_size)
330 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
331 _("Error trying to send a message to peer `%s'.\n"),
332 GNUNET_i2s (&cp->peerid));
335 memcpy (buf, pm->msg, msg_size);
336 GNUNET_free (pm->msg);
338 trigger_send_next_msg (cp);
344 * Trigger sending next pending message to the given client peer if any.
346 * @param cp client peer context struct
349 trigger_send_next_msg (struct ClientPeerContext *cp)
351 struct PendingMessage *pm;
353 if (NULL == cp->pm_head)
359 GNUNET_CADET_notify_transmit_ready (cp->ch, GNUNET_YES,
360 GNUNET_TIME_UNIT_FOREVER_REL,
361 ntohs (pm->msg->size), &do_send_msg,
367 * Add a new message to the queue to be sent to the given client peer.
369 * @param msg Message to be queued
370 * @param cp Client peer context
373 queue_msg (struct GNUNET_MessageHeader *msg, struct ClientPeerContext *cp)
375 struct PendingMessage *pm;
377 pm = GNUNET_new (struct PendingMessage);
380 GNUNET_CONTAINER_DLL_insert_tail (cp->pm_head, cp->pm_tail, pm);
381 trigger_send_next_msg (cp);
386 * Iterate over defined sensors, creates and sends brief sensor information to
387 * given client peer over CADET.
389 * @param cls closure, the client peer
390 * @param key sensor key
391 * @param value sensor value
392 * @return #GNUNET_YES to continue iteration
395 send_sensor_brief (void *cls, const struct GNUNET_HashCode *key, void *value)
397 struct ClientPeerContext *cp = cls;
398 struct GNUNET_SENSOR_SensorInfo *sensor = value;
399 struct GNUNET_SENSOR_SensorBriefMessage *msg;
400 uint16_t sensorname_size;
403 /* Create message struct */
404 sensorname_size = strlen (sensor->name) + 1;
406 sizeof (struct GNUNET_SENSOR_SensorBriefMessage) + sensorname_size;
407 msg = GNUNET_malloc (total_size);
408 msg->header.size = htons (total_size);
409 msg->header.type = htons (GNUNET_MESSAGE_TYPE_SENSOR_BRIEF);
410 msg->name_size = htons (sensorname_size);
411 msg->version_major = htons (sensor->version_major);
412 msg->version_minor = htons (sensor->version_minor);
413 memcpy (&msg[1], sensor->name, sensorname_size);
415 queue_msg ((struct GNUNET_MessageHeader *) msg, cp);
421 * Called with any sensor list request received.
423 * Each time the function must call #GNUNET_CADET_receive_done on the channel
424 * in order to receive the next message. This doesn't need to be immediate:
425 * can be delayed if some processing is done on the message.
427 * @param cls Closure (set from #GNUNET_CADET_connect).
428 * @param channel Connection to the other end.
429 * @param channel_ctx Place to store local state associated with the channel.
430 * @param message The actual message.
431 * @return #GNUNET_OK to keep the channel open,
432 * #GNUNET_SYSERR to close it (signal serious error).
435 handle_sensor_list_req (void *cls, struct GNUNET_CADET_Channel *channel,
437 const struct GNUNET_MessageHeader *message)
439 struct ClientPeerContext *cp = *channel_ctx;
440 struct GNUNET_MessageHeader *end_msg;
442 GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
443 "Received a sensor list request from peer `%s'.\n",
444 GNUNET_i2s (&cp->peerid));
445 GNUNET_CONTAINER_multihashmap_iterate (sensors, &send_sensor_brief, cp);
446 end_msg = GNUNET_new (struct GNUNET_MessageHeader);
448 end_msg->size = htons (sizeof (struct GNUNET_MessageHeader));
449 end_msg->type = htons (GNUNET_MESSAGE_TYPE_SENSOR_END);
450 queue_msg (end_msg, cp);
451 GNUNET_CADET_receive_done (channel);
457 * Parses a sensor reading message struct
459 * @param msg message header received
460 * @param sensors multihashmap of loaded sensors
461 * @return sensor reading struct or NULL if error
463 static struct ClientSensorReading *
464 parse_reading_message (const struct GNUNET_MessageHeader *msg,
465 struct GNUNET_CONTAINER_MultiHashMap *sensors)
468 struct GNUNET_SENSOR_ReadingMessage *rm;
469 uint16_t sensorname_size;
473 struct GNUNET_HashCode key;
474 struct GNUNET_SENSOR_SensorInfo *sensor;
475 struct ClientSensorReading *reading;
477 msg_size = ntohs (msg->size);
478 if (msg_size < sizeof (struct GNUNET_SENSOR_ReadingMessage))
480 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Invalid reading message size.\n");
483 rm = (struct GNUNET_SENSOR_ReadingMessage *) msg;
484 sensorname_size = ntohs (rm->sensorname_size);
485 value_size = ntohs (rm->value_size);
486 if ((sizeof (struct GNUNET_SENSOR_ReadingMessage) + sensorname_size +
487 value_size) != msg_size)
489 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Invalid reading message size.\n");
493 sensorname = GNUNET_malloc (sensorname_size);
494 memcpy (sensorname, dummy, sensorname_size);
495 GNUNET_CRYPTO_hash (sensorname, sensorname_size, &key);
496 GNUNET_free (sensorname);
497 sensor = GNUNET_CONTAINER_multihashmap_get (sensors, &key);
500 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
501 "Unknown sensor name in reading message.\n");
504 if ((sensor->version_minor != ntohs (rm->sensorversion_minor)) ||
505 (sensor->version_major != ntohs (rm->sensorversion_major)))
507 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
508 "Sensor version mismatch in reading message.\n");
511 if (0 == strcmp (sensor->expected_datatype, "numeric") &&
512 sizeof (double) != value_size)
514 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
515 "Invalid value size for a numerical sensor.\n");
518 reading = GNUNET_new (struct ClientSensorReading);
519 reading->sensor = sensor;
520 reading->timestamp = GNUNET_be64toh (rm->timestamp);
521 reading->value_size = value_size;
522 reading->value = GNUNET_malloc (value_size);
523 dummy += sensorname_size;
524 memcpy (reading->value, dummy, value_size);
530 * Called with any sensor reading messages received from CADET.
532 * Each time the function must call #GNUNET_CADET_receive_done on the channel
533 * in order to receive the next message. This doesn't need to be immediate:
534 * can be delayed if some processing is done on the message.
536 * @param cls Closure (set from #GNUNET_CADET_connect).
537 * @param channel Connection to the other end.
538 * @param channel_ctx Place to store local state associated with the channel.
539 * @param message The actual message.
540 * @return #GNUNET_OK to keep the channel open,
541 * #GNUNET_SYSERR to close it (signal serious error).
544 handle_sensor_reading (void *cls, struct GNUNET_CADET_Channel *channel,
546 const struct GNUNET_MessageHeader *message)
548 struct ClientPeerContext *cp = *channel_ctx;
549 struct ClientSensorReading *reading;
551 reading = parse_reading_message (message, sensors);
554 GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
555 "Received an invalid sensor reading from peer `%s'.\n",
556 GNUNET_i2s (&cp->peerid));
557 return GNUNET_SYSERR;
559 GNUNET_log (GNUNET_ERROR_TYPE_INFO,
560 "Received a sensor reading from peer `%s':\n"
561 "# Sensor name: `%s'\n" "# Timestamp: %" PRIu64 "\n"
562 "# Value size: %" PRIu64 ".\n", GNUNET_i2s (&cp->peerid),
563 reading->sensor->name, reading->timestamp, reading->value_size);
564 GNUNET_PEERSTORE_store (peerstore, subsystem, &cp->peerid,
565 reading->sensor->name, reading->value,
566 reading->value_size, GNUNET_TIME_UNIT_FOREVER_ABS,
567 GNUNET_PEERSTORE_STOREOPTION_MULTIPLE, NULL, NULL);
568 GNUNET_free (reading->value);
569 GNUNET_free (reading);
570 GNUNET_CADET_receive_done (channel);
576 * Create a message with full information about sensor
578 * @param sensorname Name of sensor requested
579 * @return Message ready to be sent to client or NULL on error
581 static struct GNUNET_SENSOR_SensorFullMessage *
582 create_full_sensor_msg (char *sensorname)
584 struct GNUNET_HashCode key;
585 struct GNUNET_SENSOR_SensorInfo *sensor;
586 struct GNUNET_SENSOR_SensorFullMessage *msg;
589 char *sensorscript_path;
590 uint64_t sensorname_size;
591 uint64_t sensorfile_size;
592 uint64_t sensorscriptname_size;
593 uint64_t sensorscript_size;
597 GNUNET_CRYPTO_hash (sensorname, strlen (sensorname) + 1, &key);
598 sensor = GNUNET_CONTAINER_multihashmap_get (sensors, &key);
601 sensor_dir = GNUNET_SENSOR_get_sensor_dir ();
602 GNUNET_asprintf (&sensor_path, "%s%s", sensor_dir, sensorname);
604 GNUNET_DISK_file_size (sensor_path, &sensorfile_size, GNUNET_NO,
607 GNUNET_free (sensor_dir);
608 GNUNET_free (sensor_path);
611 sensorname_size = strlen (sensorname) + 1;
612 sensorscript_size = 0;
613 sensorscriptname_size = 0;
614 /* Test if there is an associated script */
615 if (NULL != sensor->ext_process)
617 GNUNET_asprintf (&sensorscript_path, "%s%s-files%s%s", sensor_dir,
618 sensor->name, DIR_SEPARATOR_STR, sensor->ext_process);
620 GNUNET_DISK_file_size (sensorscript_path, &sensorscript_size, GNUNET_NO,
623 sensorscriptname_size = strlen (sensor->ext_process) + 1;
626 /* Construct the msg */
628 sizeof (struct GNUNET_SENSOR_SensorFullMessage) + sensorname_size +
629 sensorfile_size + sensorscriptname_size + sensorscript_size;
630 msg = GNUNET_malloc (total_size);
631 msg->header.size = htons (total_size);
632 msg->header.type = htons (GNUNET_MESSAGE_TYPE_SENSOR_FULL);
633 msg->sensorname_size = htons (sensorname_size);
634 msg->sensorfile_size = htons (sensorfile_size);
635 msg->scriptname_size = htons (sensorscriptname_size);
636 msg->scriptfile_size = htons (sensorscript_size);
638 memcpy (dummy, sensorname, sensorname_size);
639 dummy += sensorname_size;
640 GNUNET_DISK_fn_read (sensor_path, dummy, sensorfile_size);
641 dummy += sensorfile_size;
642 if (sensorscriptname_size > 0)
644 memcpy (dummy, sensor->ext_process, sensorscriptname_size);
645 dummy += sensorscriptname_size;
646 GNUNET_DISK_fn_read (sensorscript_path, dummy, sensorscript_size);
647 GNUNET_free (sensorscript_path);
649 GNUNET_free (sensor_dir);
650 GNUNET_free (sensor_path);
656 * Called with any request for full sensor information.
658 * Each time the function must call #GNUNET_CADET_receive_done on the channel
659 * in order to receive the next message. This doesn't need to be immediate:
660 * can be delayed if some processing is done on the message.
662 * @param cls Closure (set from #GNUNET_CADET_connect).
663 * @param channel Connection to the other end.
664 * @param channel_ctx Place to store local state associated with the channel.
665 * @param message The actual message.
666 * @return #GNUNET_OK to keep the channel open,
667 * #GNUNET_SYSERR to close it (signal serious error).
670 handle_sensor_full_req (void *cls, struct GNUNET_CADET_Channel *channel,
672 const struct GNUNET_MessageHeader *message)
674 struct ClientPeerContext *cp = *channel_ctx;
675 struct GNUNET_SENSOR_SensorBriefMessage *sbm = NULL;
676 struct GNUNET_SENSOR_SensorFullMessage *sfm;
678 uint16_t sensorname_size;
680 msg_size = ntohs (message->size);
681 /* parse & error check */
682 if (msg_size > sizeof (struct GNUNET_SENSOR_SensorBriefMessage))
684 sbm = (struct GNUNET_SENSOR_SensorBriefMessage *) message;
685 sensorname_size = ntohs (sbm->name_size);
687 sizeof (struct GNUNET_SENSOR_SensorBriefMessage) + sensorname_size)
692 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
693 "Received an invalid full sensor request from peer `%s'.\n",
694 GNUNET_i2s (&cp->peerid));
695 return GNUNET_SYSERR;
697 /* Create and send msg with full sensor info */
698 sfm = create_full_sensor_msg ((char *) &sbm[1]);
701 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
702 "Error creating full sensor info msg for sensor `%s'.\n",
704 return GNUNET_SYSERR;
706 queue_msg ((struct GNUNET_MessageHeader *) sfm, cp);
707 GNUNET_CADET_receive_done (channel);
713 * Process sensordashboard requests.
716 * @param server the initialized server
717 * @param cfg configuration to use
720 run (void *cls, struct GNUNET_SERVER_Handle *server,
721 const struct GNUNET_CONFIGURATION_Handle *cfg)
723 static const struct GNUNET_SERVER_MessageHandler handlers[] = {
726 static struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
727 {&handle_sensor_reading,
728 GNUNET_MESSAGE_TYPE_SENSOR_READING, 0},
729 {&handle_sensor_list_req,
730 GNUNET_MESSAGE_TYPE_SENSOR_LIST_REQ,
731 sizeof (struct GNUNET_MessageHeader)},
732 {&handle_sensor_full_req,
733 GNUNET_MESSAGE_TYPE_SENSOR_FULL_REQ,
734 sizeof (struct GNUNET_MessageHeader)},
737 static uint32_t cadet_ports[] = {
738 GNUNET_APPLICATION_TYPE_SENSORDASHBOARD,
739 GNUNET_APPLICATION_TYPE_SENSORUPDATE,
740 GNUNET_APPLICATION_TYPE_END
742 sensors = GNUNET_SENSOR_load_all_sensors ();
743 GNUNET_assert (NULL != sensors);
745 GNUNET_CADET_connect (cfg, NULL, &cadet_channel_created,
746 &cadet_channel_destroyed, cadet_handlers,
750 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
751 _("Failed to connect to `%s' service.\n"), "CADET");
752 GNUNET_SCHEDULER_add_now (&cleanup_task, NULL);
755 peerstore = GNUNET_PEERSTORE_connect (cfg);
756 if (NULL == peerstore)
758 GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
759 _("Failed to connect to `%s' service.\n"), "PEERSTORE");
760 GNUNET_SCHEDULER_add_now (&cleanup_task, NULL);
763 GNUNET_SERVER_add_handlers (server, handlers);
764 GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &cleanup_task,
770 * The main function for the sensordashboard service.
772 * @param argc number of arguments from the command line
773 * @param argv command line arguments
774 * @return 0 ok, 1 on error
777 main (int argc, char *const *argv)
780 GNUNET_SERVICE_run (argc, argv, "sensordashboard",
781 GNUNET_SERVICE_OPTION_NONE, &run, NULL)) ? 0 : 1;
784 /* end of gnunet-service-sensordashboard.c */