src/multicast/gnunet-service-multicast.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009 GNUnet e.V.
   4
   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.
   9
  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.
  14
  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., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20
  21 /**
  22  * @file multicast/gnunet-service-multicast.c
  23  * @brief program that does multicast
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_signatures.h"
  29 #include "gnunet_applications.h"
  30 #include "gnunet_statistics_service.h"
  31 #include "gnunet_cadet_service.h"
  32 #include "gnunet_multicast_service.h"
  33 #include "multicast.h"
  34
  35 /**
  36  * Handle to our current configuration.
  37  */
  38 static const struct GNUNET_CONFIGURATION_Handle *cfg;
  39
  40 /**
  41  * Server handle.
  42  */
  43 static struct GNUNET_SERVER_Handle *server;
  44
  45 /**
  46  * CADET handle.
  47  */
  48 static struct GNUNET_CADET_Handle *cadet;
  49
  50 /**
  51  * Identity of this peer.
  52  */
  53 static struct GNUNET_PeerIdentity this_peer;
  54
  55 /**
  56  * Handle to the statistics service.
  57  */
  58 static struct GNUNET_STATISTICS_Handle *stats;
  59
  60 /**
  61  * Notification context, simplifies client broadcasts.
  62  */
  63 static struct GNUNET_SERVER_NotificationContext *nc;
  64
  65 /**
  66  * All connected origin clients.
  67  * Group's pub_key_hash -> struct Origin * (uniq)
  68  */
  69 static struct GNUNET_CONTAINER_MultiHashMap *origins;
  70
  71 /**
  72  * All connected member clients.
  73  * Group's pub_key_hash -> struct Member * (multi)
  74  */
  75 static struct GNUNET_CONTAINER_MultiHashMap *members;
  76
  77 /**
  78  * Connected member clients per group.
  79  * Group's pub_key_hash -> Member's pub_key_hash (uniq) -> struct Member * (uniq)
  80  */
  81 static struct GNUNET_CONTAINER_MultiHashMap *group_members;
  82
  83 /**
  84  * Incoming CADET channels with connected children in the tree.
  85  * Group's pub_key_hash -> struct Channel * (multi)
  86  */
  87 static struct GNUNET_CONTAINER_MultiHashMap *channels_in;
  88
  89 /**
  90  * Outgoing CADET channels connecting to parents in the tree.
  91  * Group's pub_key_hash -> struct Channel * (multi)
  92  */
  93 static struct GNUNET_CONTAINER_MultiHashMap *channels_out;
  94
  95 /**
  96  * Incoming replay requests from CADET.
  97  * Group's pub_key_hash ->
  98  *   H(fragment_id, message_id, fragment_offset, flags) -> struct Channel *
  99  */
 100 static struct GNUNET_CONTAINER_MultiHashMap *replay_req_cadet;
 101
 102 /**
 103  * Incoming replay requests from clients.
 104  * Group's pub_key_hash ->
 105  *   H(fragment_id, message_id, fragment_offset, flags) -> struct GNUNET_SERVER_Client *
 106  */
 107 static struct GNUNET_CONTAINER_MultiHashMap *replay_req_client;
 108
 109
 110 /**
 111  * Join status of a remote peer.
 112  */
 113 enum JoinStatus
 114 {
 115   JOIN_REFUSED  = -1,
 116   JOIN_NOT_ASKED = 0,
 117   JOIN_WAITING   = 1,
 118   JOIN_ADMITTED  = 2,
 119 };
 120
 121 enum ChannelDirection
 122 {
 123   DIR_INCOMING = 0,
 124   DIR_OUTGOING = 1,
 125 };
 126
 127
 128 /**
 129  * Context for a CADET channel.
 130  */
 131 struct Channel
 132 {
 133   /**
 134    * Group the channel belongs to.
 135    *
 136    * Only set for outgoing channels.
 137    */
 138   struct Group *grp;
 139
 140   /**
 141    * CADET channel.
 142    */
 143   struct GNUNET_CADET_Channel *channel;
 144
 145   /**
 146    * CADET transmission handle.
 147    */
 148   struct GNUNET_CADET_TransmitHandle *tmit_handle;
 149
 150   /**
 151    * Public key of the target group.
 152    */
 153   struct GNUNET_CRYPTO_EddsaPublicKey group_pub_key;
 154
 155   /**
 156    * Hash of @a group_pub_key.
 157    */
 158   struct GNUNET_HashCode group_pub_hash;
 159
 160   /**
 161    * Public key of the joining member.
 162    */
 163   struct GNUNET_CRYPTO_EcdsaPublicKey member_pub_key;
 164
 165   /**
 166    * Remote peer identity.
 167    */
 168   struct GNUNET_PeerIdentity peer;
 169
 170   /**
 171    * Is the remote peer admitted to the group?
 172    * @see enum JoinStatus
 173    */
 174   int8_t join_status;
 175
 176   /**
 177    * Number of messages waiting to be sent to CADET.
 178    */
 179   uint8_t msgs_pending;
 180
 181   /**
 182    * Channel direction.
 183    * @see enum ChannelDirection
 184    */
 185   uint8_t direction;
 186 };
 187
 188
 189 /**
 190  * List of connected clients.
 191  */
 192 struct ClientList
 193 {
 194   struct ClientList *prev;
 195   struct ClientList *next;
 196   struct GNUNET_SERVER_Client *client;
 197 };
 198
 199 /**
 200  * Common part of the client context for both an origin and member.
 201  */
 202 struct Group
 203 {
 204   struct ClientList *clients_head;
 205   struct ClientList *clients_tail;
 206
 207   /**
 208    * Public key of the group.
 209    */
 210   struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
 211
 212   /**
 213    * Hash of @a pub_key.
 214    */
 215   struct GNUNET_HashCode pub_key_hash;
 216
 217   /**
 218    * CADET port hash.
 219    */
 220   struct GNUNET_HashCode cadet_port_hash;
 221
 222   /**
 223    * Is this an origin (#GNUNET_YES), or member (#GNUNET_NO)?
 224    */
 225   uint8_t is_origin;
 226
 227   /**
 228    * Is the client disconnected? #GNUNET_YES or #GNUNET_NO
 229    */
 230   uint8_t disconnected;
 231 };
 232
 233
 234 /**
 235  * Client context for a group's origin.
 236  */
 237 struct Origin
 238 {
 239   struct Group grp;
 240
 241   /**
 242    * Private key of the group.
 243    */
 244   struct GNUNET_CRYPTO_EddsaPrivateKey priv_key;
 245
 246   /**
 247    * CADET port.
 248    */
 249   struct GNUNET_CADET_Port *cadet_port;
 250
 251   /**
 252    * Last message fragment ID sent to the group.
 253    */
 254   uint64_t max_fragment_id;
 255 };
 256
 257
 258 /**
 259  * Client context for a group member.
 260  */
 261 struct Member
 262 {
 263   struct Group grp;
 264
 265   /**
 266    * Private key of the member.
 267    */
 268   struct GNUNET_CRYPTO_EcdsaPrivateKey priv_key;
 269
 270   /**
 271    * Public key of the member.
 272    */
 273   struct GNUNET_CRYPTO_EcdsaPublicKey pub_key;
 274
 275   /**
 276    * Hash of @a pub_key.
 277    */
 278   struct GNUNET_HashCode pub_key_hash;
 279
 280   /**
 281    * Join request sent to the origin / members.
 282    */
 283   struct MulticastJoinRequestMessage *join_req;
 284
 285   /**
 286    * Join decision sent in reply to our request.
 287    *
 288    * Only a positive decision is stored here, in case of a negative decision the
 289    * client is disconnected.
 290    */
 291   struct MulticastJoinDecisionMessageHeader *join_dcsn;
 292
 293   /**
 294    * CADET channel to the origin.
 295    */
 296   struct Channel *origin_channel;
 297
 298   /**
 299    * Peer identity of origin.
 300    */
 301   struct GNUNET_PeerIdentity origin;
 302
 303   /**
 304    * Peer identity of relays (other members to connect).
 305    */
 306   struct GNUNET_PeerIdentity *relays;
 307
 308   /**
 309    * Last request fragment ID sent to the origin.
 310    */
 311   uint64_t max_fragment_id;
 312
 313   /**
 314    * Number of @a relays.
 315    */
 316   uint32_t relay_count;
 317 };
 318
 319
 320 struct ReplayRequestKey
 321 {
 322   uint64_t fragment_id;
 323   uint64_t message_id;
 324   uint64_t fragment_offset;
 325   uint64_t flags;
 326 };
 327
 328
 329 /**
 330  * Task run during shutdown.
 331  *
 332  * @param cls unused
 333  */
 334 static void
 335 shutdown_task (void *cls)
 336 {
 337   if (NULL != cadet)
 338   {
 339     GNUNET_CADET_disconnect (cadet);
 340     cadet = NULL;
 341   }
 342   if (NULL != stats)
 343   {
 344     GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
 345     stats = NULL;
 346   }
 347   /* FIXME: do more clean up here */
 348 }
 349
 350
 351 /**
 352  * Clean up origin data structures after a client disconnected.
 353  */
 354 static void
 355 cleanup_origin (struct Origin *orig)
 356 {
 357   struct Group *grp = &orig->grp;
 358   GNUNET_CONTAINER_multihashmap_remove (origins, &grp->pub_key_hash, orig);
 359   if (NULL != orig->cadet_port)
 360   {
 361     GNUNET_CADET_close_port (orig->cadet_port);
 362     orig->cadet_port = NULL;
 363   }
 364   GNUNET_free (orig);
 365 }
 366
 367
 368 /**
 369  * Clean up member data structures after a client disconnected.
 370  */
 371 static void
 372 cleanup_member (struct Member *mem)
 373 {
 374   struct Group *grp = &mem->grp;
 375   struct GNUNET_CONTAINER_MultiHashMap *
 376     grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members,
 377                                                  &grp->pub_key_hash);
 378   GNUNET_assert (NULL != grp_mem);
 379   GNUNET_CONTAINER_multihashmap_remove (grp_mem, &mem->pub_key_hash, mem);
 380
 381   if (0 == GNUNET_CONTAINER_multihashmap_size (grp_mem))
 382   {
 383     GNUNET_CONTAINER_multihashmap_remove (group_members, &grp->pub_key_hash,
 384                                           grp_mem);
 385     GNUNET_CONTAINER_multihashmap_destroy (grp_mem);
 386   }
 387   if (NULL != mem->join_dcsn)
 388   {
 389     GNUNET_free (mem->join_dcsn);
 390     mem->join_dcsn = NULL;
 391   }
 392   GNUNET_CONTAINER_multihashmap_remove (members, &grp->pub_key_hash, mem);
 393   GNUNET_free (mem);
 394 }
 395
 396
 397 /**
 398  * Clean up group data structures after a client disconnected.
 399  */
 400 static void
 401 cleanup_group (struct Group *grp)
 402 {
 403   (GNUNET_YES == grp->is_origin)
 404     ? cleanup_origin ((struct Origin *) grp)
 405     : cleanup_member ((struct Member *) grp);
 406 }
 407
 408
 409 void
 410 replay_key_hash (uint64_t fragment_id, uint64_t message_id,
 411                  uint64_t fragment_offset, uint64_t flags,
 412                  struct GNUNET_HashCode *key_hash)
 413 {
 414   struct ReplayRequestKey key = {
 415     .fragment_id = fragment_id,
 416     .message_id = message_id,
 417     .fragment_offset = fragment_offset,
 418     .flags = flags,
 419   };
 420   GNUNET_CRYPTO_hash (&key, sizeof (key), key_hash);
 421 }
 422
 423
 424 /**
 425  * Remove channel from replay request hashmap.
 426  *
 427  * @param chn
 428  *        Channel to remove.
 429  *
 430  * @return #GNUNET_YES if there are more entries to process,
 431  *         #GNUNET_NO when reached end of hashmap.
 432  */
 433 static int
 434 replay_req_remove_cadet (struct Channel *chn)
 435 {
 436   struct GNUNET_CONTAINER_MultiHashMap *
 437     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
 438                                                         &chn->grp->pub_key_hash);
 439   if (NULL == grp_replay_req)
 440     return GNUNET_NO;
 441
 442   struct GNUNET_CONTAINER_MultiHashMapIterator *
 443     it = GNUNET_CONTAINER_multihashmap_iterator_create (grp_replay_req);
 444   struct GNUNET_HashCode key;
 445   const struct Channel *c;
 446   while (GNUNET_YES
 447          == GNUNET_CONTAINER_multihashmap_iterator_next (it, &key,
 448                                                          (const void **) &c))
 449   {
 450     if (c == chn)
 451     {
 452       GNUNET_CONTAINER_multihashmap_remove (grp_replay_req, &key, chn);
 453       GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
 454       return GNUNET_YES;
 455     }
 456   }
 457   GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
 458   return GNUNET_NO;
 459 }
 460
 461
 462 /**
 463  * Remove client from replay request hashmap.
 464  *
 465  * @param client
 466  *        Client to remove.
 467  *
 468  * @return #GNUNET_YES if there are more entries to process,
 469  *         #GNUNET_NO when reached end of hashmap.
 470  */
 471 static int
 472 replay_req_remove_client (struct Group *grp, struct GNUNET_SERVER_Client *client)
 473 {
 474   struct GNUNET_CONTAINER_MultiHashMap *
 475     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
 476                                                         &grp->pub_key_hash);
 477   if (NULL == grp_replay_req)
 478     return GNUNET_NO;
 479
 480   struct GNUNET_CONTAINER_MultiHashMapIterator *
 481     it = GNUNET_CONTAINER_multihashmap_iterator_create (grp_replay_req);
 482   struct GNUNET_HashCode key;
 483   const struct GNUNET_SERVER_Client *c;
 484   while (GNUNET_YES
 485          == GNUNET_CONTAINER_multihashmap_iterator_next (it, &key,
 486                                                          (const void **) &c))
 487   {
 488     if (c == client)
 489     {
 490       GNUNET_CONTAINER_multihashmap_remove (replay_req_client, &key, client);
 491       GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
 492       return GNUNET_YES;
 493     }
 494   }
 495   GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
 496   return GNUNET_NO;
 497 }
 498
 499
 500 /**
 501  * Called whenever a client is disconnected.
 502  *
 503  * Frees our resources associated with that client.
 504  *
 505  * @param cls  Closure.
 506  * @param client  Client handle.
 507  */
 508 static void
 509 client_notify_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
 510 {
 511   if (NULL == client)
 512     return;
 513
 514   struct Group *grp
 515     = GNUNET_SERVER_client_get_user_context (client, struct Group);
 516
 517   if (NULL == grp)
 518   {
 519     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 520                 "%p User context is NULL in client_disconnect()\n", grp);
 521     GNUNET_break (0);
 522     return;
 523   }
 524
 525   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 526               "%p Client (%s) disconnected from group %s\n",
 527               grp, (GNUNET_YES == grp->is_origin) ? "origin" : "member",
 528               GNUNET_h2s (&grp->pub_key_hash));
 529
 530   struct ClientList *cl = grp->clients_head;
 531   while (NULL != cl)
 532   {
 533     if (cl->client == client)
 534     {
 535       GNUNET_CONTAINER_DLL_remove (grp->clients_head, grp->clients_tail, cl);
 536       GNUNET_free (cl);
 537       break;
 538     }
 539     cl = cl->next;
 540   }
 541
 542   while (GNUNET_YES == replay_req_remove_client (grp, client));
 543
 544   if (NULL == grp->clients_head)
 545   { /* Last client disconnected. */
 546 #if FIXME
 547     if (NULL != grp->tmit_head)
 548     { /* Send pending messages via CADET before cleanup. */
 549       transmit_message (grp);
 550     }
 551     else
 552 #endif
 553     {
 554       cleanup_group (grp);
 555     }
 556   }
 557 }
 558
 559
 560 /**
 561  * Send message to a client.
 562  */
 563 static void
 564 client_send (struct GNUNET_SERVER_Client *client,
 565              const struct GNUNET_MessageHeader *msg)
 566 {
 567   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 568               "%p Sending message to client.\n", client);
 569
 570   GNUNET_SERVER_notification_context_add (nc, client);
 571   GNUNET_SERVER_notification_context_unicast (nc, client, msg, GNUNET_NO);
 572 }
 573
 574
 575 /**
 576  * Send message to all clients connected to the group.
 577  */
 578 static void
 579 client_send_group (const struct Group *grp,
 580                    const struct GNUNET_MessageHeader *msg)
 581 {
 582   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 583               "%p Sending message to all clients of the group.\n", grp);
 584
 585   struct ClientList *cl = grp->clients_head;
 586   while (NULL != cl)
 587   {
 588     GNUNET_SERVER_notification_context_add (nc, cl->client);
 589     GNUNET_SERVER_notification_context_unicast (nc, cl->client, msg, GNUNET_NO);
 590     cl = cl->next;
 591   }
 592 }
 593
 594
 595 /**
 596  * Iterator callback for sending a message to origin clients.
 597  */
 598 static int
 599 client_send_origin_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
 600                        void *origin)
 601 {
 602   const struct GNUNET_MessageHeader *msg = cls;
 603   struct Member *orig = origin;
 604
 605   client_send_group (&orig->grp, msg);
 606   return GNUNET_YES;
 607 }
 608
 609
 610 /**
 611  * Iterator callback for sending a message to member clients.
 612  */
 613 static int
 614 client_send_member_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
 615                        void *member)
 616 {
 617   const struct GNUNET_MessageHeader *msg = cls;
 618   struct Member *mem = member;
 619
 620   if (NULL != mem->join_dcsn)
 621   { /* Only send message to admitted members */
 622     client_send_group (&mem->grp, msg);
 623   }
 624   return GNUNET_YES;
 625 }
 626
 627
 628 /**
 629  * Send message to all origin and member clients connected to the group.
 630  *
 631  * @param pub_key_hash
 632  *        H(key_pub) of the group.
 633  * @param msg
 634  *        Message to send.
 635  */
 636 static int
 637 client_send_all (struct GNUNET_HashCode *pub_key_hash,
 638                  const struct GNUNET_MessageHeader *msg)
 639 {
 640   int n = 0;
 641   n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash,
 642                                                    client_send_origin_cb,
 643                                                    (void *) msg);
 644   n += GNUNET_CONTAINER_multihashmap_get_multiple (members, pub_key_hash,
 645                                                    client_send_member_cb,
 646                                                    (void *) msg);
 647   return n;
 648 }
 649
 650
 651 /**
 652  * Send message to a random origin client or a random member client.
 653  *
 654  * @param grp  The group to send @a msg to.
 655  * @param msg  Message to send.
 656  */
 657 static int
 658 client_send_random (struct GNUNET_HashCode *pub_key_hash,
 659                     const struct GNUNET_MessageHeader *msg)
 660 {
 661   int n = 0;
 662   n = GNUNET_CONTAINER_multihashmap_get_random (origins, client_send_origin_cb,
 663                                                  (void *) msg);
 664   if (n <= 0)
 665     n = GNUNET_CONTAINER_multihashmap_get_random (members, client_send_member_cb,
 666                                                    (void *) msg);
 667   return n;
 668 }
 669
 670
 671 /**
 672  * Send message to all origin clients connected to the group.
 673  *
 674  * @param pub_key_hash
 675  *        H(key_pub) of the group.
 676  * @param msg
 677  *        Message to send.
 678  */
 679 static int
 680 client_send_origin (struct GNUNET_HashCode *pub_key_hash,
 681                     const struct GNUNET_MessageHeader *msg)
 682 {
 683   int n = 0;
 684   n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash,
 685                                                    client_send_origin_cb,
 686                                                    (void *) msg);
 687   return n;
 688 }
 689
 690
 691 /**
 692  * Send fragment acknowledgement to all clients of the channel.
 693  *
 694  * @param pub_key_hash
 695  *        H(key_pub) of the group.
 696  */
 697 static void
 698 client_send_ack (struct GNUNET_HashCode *pub_key_hash)
 699 {
 700   static struct GNUNET_MessageHeader *msg = NULL;
 701   if (NULL == msg)
 702   {
 703     msg = GNUNET_malloc (sizeof (*msg));
 704     msg->type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_FRAGMENT_ACK);
 705     msg->size = htons (sizeof (*msg));
 706   }
 707   client_send_all (pub_key_hash, msg);
 708 }
 709
 710
 711 struct CadetTransmitClosure
 712 {
 713   struct Channel *chn;
 714   const struct GNUNET_MessageHeader *msg;
 715 };
 716
 717
 718 /**
 719  * CADET is ready to transmit a message.
 720  */
 721 size_t
 722 cadet_notify_transmit_ready (void *cls, size_t buf_size, void *buf)
 723 {
 724   if (0 == buf_size)
 725   {
 726     /* FIXME: connection closed */
 727     return 0;
 728   }
 729   struct CadetTransmitClosure *tcls = cls;
 730   struct Channel *chn = tcls->chn;
 731   uint16_t msg_size = ntohs (tcls->msg->size);
 732   GNUNET_assert (msg_size <= buf_size);
 733   GNUNET_memcpy (buf, tcls->msg, msg_size);
 734   GNUNET_free (tcls);
 735
 736   if (0 == chn->msgs_pending)
 737   {
 738     GNUNET_break (0);
 739   }
 740   else if (0 == --chn->msgs_pending)
 741   {
 742     client_send_ack (&chn->group_pub_hash);
 743   }
 744   return msg_size;
 745 }
 746
 747
 748 /**
 749  * Send a message to a CADET channel.
 750  *
 751  * @param chn  Channel.
 752  * @param msg  Message.
 753  */
 754 static void
 755 cadet_send_channel (struct Channel *chn, const struct GNUNET_MessageHeader *msg)
 756 {
 757   struct CadetTransmitClosure *tcls = GNUNET_malloc (sizeof (*tcls));
 758   tcls->chn = chn;
 759   tcls->msg = msg;
 760
 761   chn->msgs_pending++;
 762   chn->tmit_handle
 763     = GNUNET_CADET_notify_transmit_ready (chn->channel, GNUNET_NO,
 764                                           GNUNET_TIME_UNIT_FOREVER_REL,
 765                                           ntohs (msg->size),
 766                                           &cadet_notify_transmit_ready,
 767                                           tcls);
 768   GNUNET_assert (NULL != chn->tmit_handle);
 769 }
 770
 771
 772 /**
 773  * Create new outgoing CADET channel.
 774  *
 775  * @param peer
 776  *        Peer to connect to.
 777  * @param group_pub_key
 778  *        Public key of group the channel belongs to.
 779  * @param group_pub_hash
 780  *        Hash of @a group_pub_key.
 781  *
 782  * @return Channel.
 783  */
 784 static struct Channel *
 785 cadet_channel_create (struct Group *grp, struct GNUNET_PeerIdentity *peer)
 786 {
 787   struct Channel *chn = GNUNET_malloc (sizeof (*chn));
 788   chn->grp = grp;
 789   chn->group_pub_key = grp->pub_key;
 790   chn->group_pub_hash = grp->pub_key_hash;
 791   chn->peer = *peer;
 792   chn->direction = DIR_OUTGOING;
 793   chn->join_status = JOIN_WAITING;
 794   chn->channel = GNUNET_CADET_channel_create (cadet, chn, &chn->peer,
 795                                               &grp->cadet_port_hash,
 796                                               GNUNET_CADET_OPTION_RELIABLE);
 797   GNUNET_CONTAINER_multihashmap_put (channels_out, &chn->group_pub_hash, chn,
 798                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 799   return chn;
 800 }
 801
 802
 803 /**
 804  * Create CADET channel and send a join request.
 805  */
 806 static void
 807 cadet_send_join_request (struct Member *mem)
 808 {
 809   mem->origin_channel = cadet_channel_create (&mem->grp, &mem->origin);
 810   cadet_send_channel (mem->origin_channel, &mem->join_req->header);
 811
 812   uint32_t i;
 813   for (i = 0; i < mem->relay_count; i++)
 814   {
 815     struct Channel *
 816       chn = cadet_channel_create (&mem->grp, &mem->relays[i]);
 817     cadet_send_channel (chn, &mem->join_req->header);
 818   }
 819 }
 820
 821
 822 static int
 823 cadet_send_join_decision_cb (void *cls,
 824                              const struct GNUNET_HashCode *group_pub_hash,
 825                              void *channel)
 826 {
 827   const struct MulticastJoinDecisionMessageHeader *hdcsn = cls;
 828   struct Channel *chn = channel;
 829
 830   if (0 == memcmp (&hdcsn->member_pub_key, &chn->member_pub_key, sizeof (chn->member_pub_key))
 831       && 0 == memcmp (&hdcsn->peer, &chn->peer, sizeof (chn->peer)))
 832   {
 833     cadet_send_channel (chn, &hdcsn->header);
 834     return GNUNET_NO;
 835   }
 836   return GNUNET_YES;
 837 }
 838
 839
 840 /**
 841  * Send join decision to a remote peer.
 842  */
 843 static void
 844 cadet_send_join_decision (struct Group *grp,
 845                           const struct MulticastJoinDecisionMessageHeader *hdcsn)
 846 {
 847   GNUNET_CONTAINER_multihashmap_get_multiple (channels_in, &grp->pub_key_hash,
 848                                               &cadet_send_join_decision_cb,
 849                                               (void *) hdcsn);
 850 }
 851
 852
 853 /**
 854  * Iterator callback for sending a message to origin clients.
 855  */
 856 static int
 857 cadet_send_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
 858                void *channel)
 859 {
 860   const struct GNUNET_MessageHeader *msg = cls;
 861   struct Channel *chn = channel;
 862   if (JOIN_ADMITTED == chn->join_status)
 863     cadet_send_channel (chn, msg);
 864   return GNUNET_YES;
 865 }
 866
 867
 868 /**
 869  * Send message to all connected children.
 870  */
 871 static int
 872 cadet_send_children (struct GNUNET_HashCode *pub_key_hash,
 873                      const struct GNUNET_MessageHeader *msg)
 874 {
 875   int n = 0;
 876   if (channels_in != NULL)
 877     n += GNUNET_CONTAINER_multihashmap_get_multiple (channels_in, pub_key_hash,
 878                                                      cadet_send_cb, (void *) msg);
 879   return n;
 880 }
 881
 882
 883 #if 0       // unused as yet
 884 /**
 885  * Send message to all connected parents.
 886  */
 887 static int
 888 cadet_send_parents (struct GNUNET_HashCode *pub_key_hash,
 889                     const struct GNUNET_MessageHeader *msg)
 890 {
 891   int n = 0;
 892   if (channels_in != NULL)
 893     n += GNUNET_CONTAINER_multihashmap_get_multiple (channels_out, pub_key_hash,
 894                                                      cadet_send_cb, (void *) msg);
 895   return n;
 896 }
 897 #endif
 898
 899
 900 /**
 901  * New incoming CADET channel.
 902  */
 903 static void *
 904 cadet_notify_channel_new (void *cls,
 905                           struct GNUNET_CADET_Channel *channel,
 906                           const struct GNUNET_PeerIdentity *initiator,
 907                           const struct GNUNET_HashCode *port,
 908                           enum GNUNET_CADET_ChannelOption options)
 909 {
 910   return NULL;
 911 }
 912
 913
 914 /**
 915  * CADET channel is being destroyed.
 916  */
 917 static void
 918 cadet_notify_channel_end (void *cls,
 919                           const struct GNUNET_CADET_Channel *channel,
 920                           void *ctx)
 921 {
 922   if (NULL == ctx)
 923     return;
 924
 925   struct Channel *chn = ctx;
 926   if (NULL != chn->grp)
 927   {
 928     if (GNUNET_NO == chn->grp->is_origin)
 929     {
 930       struct Member *mem = (struct Member *) chn->grp;
 931       if (chn == mem->origin_channel)
 932         mem->origin_channel = NULL;
 933     }
 934   }
 935
 936   while (GNUNET_YES == replay_req_remove_cadet (chn));
 937
 938   GNUNET_free (chn);
 939 }
 940
 941
 942 static void
 943 group_set_cadet_port_hash (struct Group *grp)
 944 {
 945   struct CadetPort {
 946     struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
 947     uint32_t app_type;
 948   } port = {
 949     grp->pub_key,
 950     GNUNET_APPLICATION_TYPE_MULTICAST,
 951   };
 952   GNUNET_CRYPTO_hash (&port, sizeof (port), &grp->cadet_port_hash);
 953 }
 954
 955
 956 /**
 957  * Handle a connecting client starting an origin.
 958  */
 959 static void
 960 client_recv_origin_start (void *cls, struct GNUNET_SERVER_Client *client,
 961                           const struct GNUNET_MessageHeader *m)
 962 {
 963   const struct MulticastOriginStartMessage *
 964     msg = (const struct MulticastOriginStartMessage *) m;
 965
 966   struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
 967   struct GNUNET_HashCode pub_key_hash;
 968
 969   GNUNET_CRYPTO_eddsa_key_get_public (&msg->group_key, &pub_key);
 970   GNUNET_CRYPTO_hash (&pub_key, sizeof (pub_key), &pub_key_hash);
 971
 972   struct Origin *
 973     orig = GNUNET_CONTAINER_multihashmap_get (origins, &pub_key_hash);
 974   struct Group *grp;
 975
 976   if (NULL == orig)
 977   {
 978     orig = GNUNET_new (struct Origin);
 979     orig->priv_key = msg->group_key;
 980     orig->max_fragment_id = GNUNET_ntohll (msg->max_fragment_id);
 981     grp = &orig->grp;
 982     grp->is_origin = GNUNET_YES;
 983     grp->pub_key = pub_key;
 984     grp->pub_key_hash = pub_key_hash;
 985
 986     GNUNET_CONTAINER_multihashmap_put (origins, &grp->pub_key_hash, orig,
 987                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
 988
 989     group_set_cadet_port_hash (grp);
 990     orig->cadet_port = GNUNET_CADET_open_port (cadet, &grp->cadet_port_hash,
 991                                                cadet_notify_channel_new, NULL);
 992   }
 993   else
 994   {
 995     grp = &orig->grp;
 996   }
 997
 998   struct ClientList *cl = GNUNET_new (struct ClientList);
 999   cl->client = client;
1000   GNUNET_CONTAINER_DLL_insert (grp->clients_head, grp->clients_tail, cl);
1001
1002   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1003               "%p Client connected as origin to group %s.\n",
1004               orig, GNUNET_h2s (&grp->pub_key_hash));
1005
1006   GNUNET_SERVER_client_set_user_context (client, grp);
1007   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1008 }
1009
1010
1011 /**
1012  * Handle a connecting client joining a group.
1013  */
1014 static void
1015 client_recv_member_join (void *cls, struct GNUNET_SERVER_Client *client,
1016                          const struct GNUNET_MessageHeader *m)
1017 {
1018   const struct MulticastMemberJoinMessage *
1019     msg = (const struct MulticastMemberJoinMessage *) m;
1020   uint16_t msg_size = ntohs (msg->header.size);
1021
1022   struct GNUNET_CRYPTO_EcdsaPublicKey mem_pub_key;
1023   struct GNUNET_HashCode pub_key_hash, mem_pub_key_hash;
1024
1025   GNUNET_CRYPTO_ecdsa_key_get_public (&msg->member_key, &mem_pub_key);
1026   GNUNET_CRYPTO_hash (&mem_pub_key, sizeof (mem_pub_key), &mem_pub_key_hash);
1027   GNUNET_CRYPTO_hash (&msg->group_pub_key, sizeof (msg->group_pub_key), &pub_key_hash);
1028
1029   struct GNUNET_CONTAINER_MultiHashMap *
1030     grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members, &pub_key_hash);
1031   struct Member *mem = NULL;
1032   struct Group *grp;
1033
1034   if (NULL != grp_mem)
1035   {
1036     mem = GNUNET_CONTAINER_multihashmap_get (grp_mem, &mem_pub_key_hash);
1037   }
1038   if (NULL == mem)
1039   {
1040     mem = GNUNET_new (struct Member);
1041     mem->priv_key = msg->member_key;
1042     mem->pub_key = mem_pub_key;
1043     mem->pub_key_hash = mem_pub_key_hash;
1044     mem->max_fragment_id = 0; // FIXME
1045
1046     grp = &mem->grp;
1047     grp->is_origin = GNUNET_NO;
1048     grp->pub_key = msg->group_pub_key;
1049     grp->pub_key_hash = pub_key_hash;
1050     group_set_cadet_port_hash (grp);
1051
1052     if (NULL == grp_mem)
1053     {
1054       grp_mem = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
1055       GNUNET_CONTAINER_multihashmap_put (group_members, &grp->pub_key_hash, grp_mem,
1056                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1057     }
1058     GNUNET_CONTAINER_multihashmap_put (grp_mem, &mem->pub_key_hash, mem,
1059                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1060     GNUNET_CONTAINER_multihashmap_put (members, &grp->pub_key_hash, mem,
1061                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1062   }
1063   else
1064   {
1065     grp = &mem->grp;
1066   }
1067
1068   struct ClientList *cl = GNUNET_new (struct ClientList);
1069   cl->client = client;
1070   GNUNET_CONTAINER_DLL_insert (grp->clients_head, grp->clients_tail, cl);
1071
1072   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1073               "%p Client connected to group %s..\n",
1074               mem, GNUNET_h2s (&grp->pub_key_hash));
1075   char *str = GNUNET_CRYPTO_ecdsa_public_key_to_string (&mem->pub_key);
1076   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1077               "%p ..as member %s (%s).\n",
1078               mem, GNUNET_h2s (&mem->pub_key_hash), str);
1079   GNUNET_free (str);
1080
1081   GNUNET_SERVER_client_set_user_context (client, grp);
1082
1083   if (NULL != mem->join_dcsn)
1084   { /* Already got a join decision, send it to client. */
1085     GNUNET_SERVER_notification_context_add (nc, client);
1086     GNUNET_SERVER_notification_context_unicast (nc, client,
1087                                                 (struct GNUNET_MessageHeader *)
1088                                                 mem->join_dcsn,
1089                                                 GNUNET_NO);
1090   }
1091   else
1092   { /* First client of the group, send join request. */
1093     struct GNUNET_PeerIdentity *relays = (struct GNUNET_PeerIdentity *) &msg[1];
1094     uint32_t relay_count = ntohl (msg->relay_count);
1095     uint16_t relay_size = relay_count * sizeof (*relays);
1096     struct GNUNET_MessageHeader *join_msg = NULL;
1097     uint16_t join_msg_size = 0;
1098     if (sizeof (*msg) + relay_size + sizeof (struct GNUNET_MessageHeader)
1099         <= msg_size)
1100     {
1101       join_msg = (struct GNUNET_MessageHeader *)
1102         (((char *) &msg[1]) + relay_size);
1103       join_msg_size = ntohs (join_msg->size);
1104     }
1105     if (sizeof (*msg) + relay_size + join_msg_size != msg_size)
1106     {
1107       GNUNET_break (0);
1108       GNUNET_SERVER_client_disconnect (client);
1109       return;
1110     }
1111
1112     struct MulticastJoinRequestMessage *
1113       req = GNUNET_malloc (sizeof (*req) + join_msg_size);
1114     req->header.size = htons (sizeof (*req) + join_msg_size);
1115     req->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST);
1116     req->group_pub_key = grp->pub_key;
1117     req->peer = this_peer;
1118     GNUNET_CRYPTO_ecdsa_key_get_public (&mem->priv_key, &req->member_pub_key);
1119     if (0 < join_msg_size)
1120       GNUNET_memcpy (&req[1], join_msg, join_msg_size);
1121
1122     req->member_pub_key = mem->pub_key;
1123     req->purpose.size = htonl (msg_size
1124                                - sizeof (req->header)
1125                                - sizeof (req->reserved)
1126                                - sizeof (req->signature));
1127     req->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST);
1128
1129     if (GNUNET_OK != GNUNET_CRYPTO_ecdsa_sign (&mem->priv_key, &req->purpose,
1130                                                &req->signature))
1131     {
1132       /* FIXME: handle error */
1133       GNUNET_assert (0);
1134     }
1135
1136     if (NULL != mem->join_req)
1137       GNUNET_free (mem->join_req);
1138     mem->join_req = req;
1139
1140     if (0 == client_send_origin (&grp->pub_key_hash, &mem->join_req->header))
1141     { /* No local origins, send to remote origin */
1142       cadet_send_join_request (mem);
1143     }
1144   }
1145   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1146 }
1147
1148
1149 static void
1150 client_send_join_decision (struct Member *mem,
1151                            const struct MulticastJoinDecisionMessageHeader *hdcsn)
1152 {
1153   client_send_group (&mem->grp, &hdcsn->header);
1154
1155   const struct MulticastJoinDecisionMessage *
1156     dcsn = (const struct MulticastJoinDecisionMessage *) &hdcsn[1];
1157   if (GNUNET_YES == ntohl (dcsn->is_admitted))
1158   { /* Member admitted, store join_decision. */
1159     uint16_t dcsn_size = ntohs (dcsn->header.size);
1160     mem->join_dcsn = GNUNET_malloc (dcsn_size);
1161     GNUNET_memcpy (mem->join_dcsn, dcsn, dcsn_size);
1162   }
1163   else
1164   { /* Refused entry, but replay would be still possible for past members. */
1165   }
1166 }
1167
1168
1169 /**
1170  * Join decision from client.
1171  */
1172 static void
1173 client_recv_join_decision (void *cls, struct GNUNET_SERVER_Client *client,
1174                            const struct GNUNET_MessageHeader *m)
1175 {
1176   struct Group *
1177     grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1178   const struct MulticastJoinDecisionMessageHeader *
1179     hdcsn = (const struct MulticastJoinDecisionMessageHeader *) m;
1180
1181   if (NULL == grp)
1182   {
1183     GNUNET_break (0);
1184     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1185     return;
1186   }
1187   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1188               "%p Got join decision from client for group %s..\n",
1189               grp, GNUNET_h2s (&grp->pub_key_hash));
1190
1191   struct GNUNET_CONTAINER_MultiHashMap *
1192     grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members,
1193                                                  &grp->pub_key_hash);
1194   struct Member *mem = NULL;
1195   if (NULL != grp_mem)
1196   {
1197     struct GNUNET_HashCode member_key_hash;
1198     GNUNET_CRYPTO_hash (&hdcsn->member_pub_key, sizeof (hdcsn->member_pub_key),
1199                         &member_key_hash);
1200     mem = GNUNET_CONTAINER_multihashmap_get (grp_mem, &member_key_hash);
1201     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1202                 "%p ..and member %s: %p\n",
1203                 grp, GNUNET_h2s (&member_key_hash), mem);
1204   }
1205   if (NULL != mem)
1206   { /* Found local member */
1207     client_send_join_decision (mem, hdcsn);
1208   }
1209   else
1210   { /* Look for remote member */
1211     cadet_send_join_decision (grp, hdcsn);
1212   }
1213   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1214 }
1215
1216
1217 /**
1218  * Incoming message from a client.
1219  */
1220 static void
1221 client_recv_multicast_message (void *cls, struct GNUNET_SERVER_Client *client,
1222                                const struct GNUNET_MessageHeader *m)
1223 {
1224   struct Group *
1225     grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1226   struct GNUNET_MULTICAST_MessageHeader *out;
1227   struct Origin *orig;
1228
1229   if (NULL == grp)
1230   {
1231     GNUNET_break (0);
1232     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1233     return;
1234   }
1235   GNUNET_assert (GNUNET_YES == grp->is_origin);
1236   orig = (struct Origin *) grp;
1237
1238   /* FIXME: yucky, should use separate message structs for P2P and CS! */
1239   out = (struct GNUNET_MULTICAST_MessageHeader *) GNUNET_copy_message (m);
1240   out->fragment_id = GNUNET_htonll (++orig->max_fragment_id);
1241   out->purpose.size = htonl (ntohs (out->header.size)
1242                              - sizeof (out->header)
1243                              - sizeof (out->hop_counter)
1244                              - sizeof (out->signature));
1245   out->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_MESSAGE);
1246
1247   if (GNUNET_OK != GNUNET_CRYPTO_eddsa_sign (&orig->priv_key, &out->purpose,
1248                                              &out->signature))
1249   {
1250     GNUNET_assert (0);
1251   }
1252
1253   client_send_all (&grp->pub_key_hash, &out->header);
1254   if (0 == cadet_send_children (&grp->pub_key_hash, &out->header))
1255   {
1256     client_send_ack (&grp->pub_key_hash);
1257   }
1258   GNUNET_free (out);
1259
1260   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1261 }
1262
1263
1264 /**
1265  * Incoming request from a client.
1266  */
1267 static void
1268 client_recv_multicast_request (void *cls, struct GNUNET_SERVER_Client *client,
1269                                const struct GNUNET_MessageHeader *m)
1270 {
1271   struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1272   struct Member *mem;
1273   struct GNUNET_MULTICAST_RequestHeader *out;
1274   if (NULL == grp)
1275   {
1276     GNUNET_break (0);
1277     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1278     return;
1279   }
1280   GNUNET_assert (GNUNET_NO == grp->is_origin);
1281   mem = (struct Member *) grp;
1282
1283   /* FIXME: yucky, should use separate message structs for P2P and CS! */
1284   out = (struct GNUNET_MULTICAST_RequestHeader *) GNUNET_copy_message (m);
1285   out->member_pub_key = mem->pub_key;
1286   out->fragment_id = GNUNET_ntohll (++mem->max_fragment_id);
1287   out->purpose.size = htonl (ntohs (out->header.size)
1288                              - sizeof (out->header)
1289                              - sizeof (out->member_pub_key)
1290                              - sizeof (out->signature));
1291   out->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST);
1292
1293   if (GNUNET_OK != GNUNET_CRYPTO_ecdsa_sign (&mem->priv_key, &out->purpose,
1294                                              &out->signature))
1295   {
1296     GNUNET_assert (0);
1297   }
1298
1299   uint8_t send_ack = GNUNET_YES;
1300   if (0 == client_send_origin (&grp->pub_key_hash, &out->header))
1301   { /* No local origins, send to remote origin */
1302     if (NULL != mem->origin_channel)
1303     {
1304       cadet_send_channel (mem->origin_channel, &out->header);
1305       send_ack = GNUNET_NO;
1306     }
1307     else
1308     {
1309       /* FIXME: not yet connected to origin */
1310       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1311       GNUNET_free (out);
1312       return;
1313     }
1314   }
1315   if (GNUNET_YES == send_ack)
1316   {
1317     client_send_ack (&grp->pub_key_hash);
1318   }
1319   GNUNET_free (out);
1320   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1321 }
1322
1323
1324 /**
1325  * Incoming replay request from a client.
1326  */
1327 static void
1328 client_recv_replay_request (void *cls, struct GNUNET_SERVER_Client *client,
1329                             const struct GNUNET_MessageHeader *m)
1330 {
1331   struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1332   struct Member *mem;
1333   if (NULL == grp)
1334   {
1335     GNUNET_break (0);
1336     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1337     return;
1338   }
1339   GNUNET_assert (GNUNET_NO == grp->is_origin);
1340   mem = (struct Member *) grp;
1341
1342   struct GNUNET_CONTAINER_MultiHashMap *
1343     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
1344                                                         &grp->pub_key_hash);
1345   if (NULL == grp_replay_req)
1346   {
1347     grp_replay_req = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1348     GNUNET_CONTAINER_multihashmap_put (replay_req_client,
1349                                        &grp->pub_key_hash, grp_replay_req,
1350                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1351   }
1352   struct MulticastReplayRequestMessage *
1353     rep = (struct MulticastReplayRequestMessage *) m;
1354   struct GNUNET_HashCode key_hash;
1355   replay_key_hash (rep->fragment_id, rep->message_id, rep->fragment_offset,
1356                    rep->flags, &key_hash);
1357   GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, client,
1358                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1359
1360   if (0 == client_send_origin (&grp->pub_key_hash, m))
1361   { /* No local origin, replay from remote members / origin. */
1362     if (NULL != mem->origin_channel)
1363     {
1364       cadet_send_channel (mem->origin_channel, m);
1365     }
1366     else
1367     {
1368       /* FIXME: not yet connected to origin */
1369       GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1370       return;
1371     }
1372   }
1373   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1374 }
1375
1376
1377 static int
1378 cadet_send_replay_response_cb (void *cls,
1379                                const struct GNUNET_HashCode *key_hash,
1380                                void *value)
1381 {
1382   struct Channel *chn = value;
1383   struct GNUNET_MessageHeader *msg = cls;
1384
1385   cadet_send_channel (chn, msg);
1386   return GNUNET_OK;
1387 }
1388
1389
1390 static int
1391 client_send_replay_response_cb (void *cls,
1392                                 const struct GNUNET_HashCode *key_hash,
1393                                 void *value)
1394 {
1395   struct GNUNET_SERVER_Client *client = value;
1396   struct GNUNET_MessageHeader *msg = cls;
1397
1398   client_send (client, msg);
1399   return GNUNET_OK;
1400 }
1401
1402
1403 /**
1404  * End of replay response from a client.
1405  */
1406 static void
1407 client_recv_replay_response_end (void *cls, struct GNUNET_SERVER_Client *client,
1408                                  const struct GNUNET_MessageHeader *m)
1409 {
1410   struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1411   if (NULL == grp)
1412   {
1413     GNUNET_break (0);
1414     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1415     return;
1416   }
1417
1418   struct MulticastReplayResponseMessage *
1419     res = (struct MulticastReplayResponseMessage *) m;
1420
1421   struct GNUNET_HashCode key_hash;
1422   replay_key_hash (res->fragment_id, res->message_id, res->fragment_offset,
1423                    res->flags, &key_hash);
1424
1425   struct GNUNET_CONTAINER_MultiHashMap *
1426     grp_replay_req_cadet = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
1427                                                                 &grp->pub_key_hash);
1428   if (NULL != grp_replay_req_cadet)
1429   {
1430     GNUNET_CONTAINER_multihashmap_remove_all (grp_replay_req_cadet, &key_hash);
1431   }
1432   struct GNUNET_CONTAINER_MultiHashMap *
1433     grp_replay_req_client = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
1434                                                                &grp->pub_key_hash);
1435   if (NULL != grp_replay_req_client)
1436   {
1437     GNUNET_CONTAINER_multihashmap_remove_all (grp_replay_req_client, &key_hash);
1438   }
1439   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1440 }
1441
1442
1443 /**
1444  * Incoming replay response from a client.
1445  *
1446  * Respond with a multicast message on success, or otherwise with an error code.
1447  */
1448 static void
1449 client_recv_replay_response (void *cls, struct GNUNET_SERVER_Client *client,
1450                              const struct GNUNET_MessageHeader *m)
1451 {
1452   struct Group *grp = GNUNET_SERVER_client_get_user_context (client, struct Group);
1453   if (NULL == grp)
1454   {
1455     GNUNET_break (0);
1456     GNUNET_SERVER_receive_done (client, GNUNET_SYSERR);
1457     return;
1458   }
1459
1460   struct MulticastReplayResponseMessage *
1461     res = (struct MulticastReplayResponseMessage *) m;
1462
1463   const struct GNUNET_MessageHeader *msg = m;
1464   if (GNUNET_MULTICAST_REC_OK == res->error_code)
1465   {
1466     msg = (struct GNUNET_MessageHeader *) &res[1];
1467   }
1468
1469   struct GNUNET_HashCode key_hash;
1470   replay_key_hash (res->fragment_id, res->message_id, res->fragment_offset,
1471                    res->flags, &key_hash);
1472
1473   struct GNUNET_CONTAINER_MultiHashMap *
1474     grp_replay_req_cadet = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
1475                                                               &grp->pub_key_hash);
1476   if (NULL != grp_replay_req_cadet)
1477   {
1478     GNUNET_CONTAINER_multihashmap_get_multiple (grp_replay_req_cadet, &key_hash,
1479                                                 cadet_send_replay_response_cb,
1480                                                 (void *) msg);
1481   }
1482   if (GNUNET_MULTICAST_REC_OK == res->error_code)
1483   {
1484     struct GNUNET_CONTAINER_MultiHashMap *
1485       grp_replay_req_client = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
1486                                                                  &grp->pub_key_hash);
1487     if (NULL != grp_replay_req_client)
1488     {
1489       GNUNET_CONTAINER_multihashmap_get_multiple (grp_replay_req_client, &key_hash,
1490                                                   client_send_replay_response_cb,
1491                                                   (void *) msg);
1492     }
1493   }
1494   else
1495   {
1496     client_recv_replay_response_end (cls, client, m);
1497     return;
1498   }
1499   GNUNET_SERVER_receive_done (client, GNUNET_OK);
1500 }
1501
1502
1503 /**
1504  * A new client connected.
1505  */
1506 static void
1507 client_notify_connect (void *cls, struct GNUNET_SERVER_Client *client)
1508 {
1509   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client connected: %p\n", client);
1510   /* FIXME: send connect ACK */
1511 }
1512
1513
1514 /**
1515  * Message handlers for the server.
1516  */
1517 static const struct GNUNET_SERVER_MessageHandler server_handlers[] = {
1518   { client_recv_origin_start, NULL,
1519     GNUNET_MESSAGE_TYPE_MULTICAST_ORIGIN_START, 0 },
1520
1521   { client_recv_member_join, NULL,
1522     GNUNET_MESSAGE_TYPE_MULTICAST_MEMBER_JOIN, 0 },
1523
1524   { client_recv_join_decision, NULL,
1525     GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION, 0 },
1526
1527   { client_recv_multicast_message, NULL,
1528     GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE, 0 },
1529
1530   { client_recv_multicast_request, NULL,
1531     GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST, 0 },
1532
1533   { client_recv_replay_request, NULL,
1534     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, 0 },
1535
1536   { client_recv_replay_response, NULL,
1537     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE, 0 },
1538
1539   { client_recv_replay_response_end, NULL,
1540     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE_END, 0 },
1541
1542   { NULL, NULL, 0, 0 }
1543 };
1544
1545
1546 /**
1547  * Incoming join request message from CADET.
1548  */
1549 int
1550 cadet_recv_join_request (void *cls,
1551                          struct GNUNET_CADET_Channel *channel,
1552                          void **ctx,
1553                          const struct GNUNET_MessageHeader *m)
1554 {
1555   const struct MulticastJoinRequestMessage *
1556     req = (const struct MulticastJoinRequestMessage *) m;
1557   uint16_t size = ntohs (m->size);
1558   if (size < sizeof (*req))
1559   {
1560     GNUNET_break_op (0);
1561     return GNUNET_SYSERR;
1562   }
1563   if (NULL != *ctx)
1564   {
1565     GNUNET_break_op (0);
1566     return GNUNET_SYSERR;
1567   }
1568   if (ntohl (req->purpose.size) != (size
1569                                     - sizeof (req->header)
1570                                     - sizeof (req->reserved)
1571                                     - sizeof (req->signature)))
1572   {
1573     GNUNET_break_op (0);
1574     return GNUNET_SYSERR;
1575   }
1576   if (GNUNET_OK !=
1577       GNUNET_CRYPTO_ecdsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST,
1578                                   &req->purpose, &req->signature,
1579                                   &req->member_pub_key))
1580   {
1581     GNUNET_break_op (0);
1582     return GNUNET_SYSERR;
1583   }
1584
1585   struct GNUNET_HashCode group_pub_hash;
1586   GNUNET_CRYPTO_hash (&req->group_pub_key, sizeof (req->group_pub_key), &group_pub_hash);
1587
1588   struct Channel *chn = GNUNET_malloc (sizeof *chn);
1589   chn->channel = channel;
1590   chn->group_pub_key = req->group_pub_key;
1591   chn->group_pub_hash = group_pub_hash;
1592   chn->member_pub_key = req->member_pub_key;
1593   chn->peer = req->peer;
1594   chn->join_status = JOIN_WAITING;
1595   GNUNET_CONTAINER_multihashmap_put (channels_in, &chn->group_pub_hash, chn,
1596                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1597
1598   client_send_all (&group_pub_hash, m);
1599   return GNUNET_OK;
1600 }
1601
1602
1603 /**
1604  * Incoming join decision message from CADET.
1605  */
1606 int
1607 cadet_recv_join_decision (void *cls,
1608                           struct GNUNET_CADET_Channel *channel,
1609                           void **ctx,
1610                           const struct GNUNET_MessageHeader *m)
1611 {
1612   const struct MulticastJoinDecisionMessage *
1613     dcsn = (const struct MulticastJoinDecisionMessage *) m;
1614   uint16_t size = ntohs (m->size);
1615   if (size < sizeof (*dcsn))
1616   {
1617     GNUNET_break_op (0);
1618     return GNUNET_SYSERR;
1619   }
1620   struct Channel *chn = *ctx;
1621   if (NULL == chn)
1622   {
1623     GNUNET_break_op (0);
1624     return GNUNET_SYSERR;
1625   }
1626   if (NULL == chn->grp || GNUNET_NO != chn->grp->is_origin)
1627   {
1628     GNUNET_break_op (0);
1629     return GNUNET_SYSERR;
1630   }
1631   switch (chn->join_status)
1632   {
1633   case JOIN_REFUSED:
1634     return GNUNET_SYSERR;
1635
1636   case JOIN_ADMITTED:
1637     return GNUNET_OK;
1638
1639   case JOIN_NOT_ASKED:
1640   case JOIN_WAITING:
1641     break;
1642   }
1643
1644   struct MulticastJoinDecisionMessageHeader *
1645     hdcsn = GNUNET_malloc (sizeof (*hdcsn) + size);
1646   hdcsn->peer = chn->peer;
1647   GNUNET_memcpy (&hdcsn[1], dcsn, sizeof (*hdcsn) + size);
1648
1649   struct Member *mem = (struct Member *) chn->grp;
1650   client_send_join_decision (mem, hdcsn);
1651   GNUNET_free (hdcsn);
1652   if (GNUNET_YES == ntohs (dcsn->is_admitted))
1653   {
1654     chn->join_status = JOIN_ADMITTED;
1655     return GNUNET_OK;
1656   }
1657   else
1658   {
1659     chn->join_status = JOIN_REFUSED;
1660     return GNUNET_SYSERR;
1661   }
1662 }
1663
1664 /**
1665  * Incoming multicast message from CADET.
1666  */
1667 int
1668 cadet_recv_message (void *cls,
1669                     struct GNUNET_CADET_Channel *channel,
1670                     void **ctx,
1671                     const struct GNUNET_MessageHeader *m)
1672 {
1673   const struct GNUNET_MULTICAST_MessageHeader *
1674     msg = (const struct GNUNET_MULTICAST_MessageHeader *) m;
1675   uint16_t size = ntohs (m->size);
1676   if (size < sizeof (*msg))
1677   {
1678     GNUNET_break_op (0);
1679     return GNUNET_SYSERR;
1680   }
1681   struct Channel *chn = *ctx;
1682   if (NULL == chn)
1683   {
1684     GNUNET_break_op (0);
1685     return GNUNET_SYSERR;
1686   }
1687   if (ntohl (msg->purpose.size) != (size
1688                                     - sizeof (msg->header)
1689                                     - sizeof (msg->hop_counter)
1690                                     - sizeof (msg->signature)))
1691   {
1692     GNUNET_break_op (0);
1693     return GNUNET_SYSERR;
1694   }
1695   if (GNUNET_OK !=
1696       GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_MESSAGE,
1697                                   &msg->purpose, &msg->signature,
1698                                   &chn->group_pub_key))
1699   {
1700     GNUNET_break_op (0);
1701     return GNUNET_SYSERR;
1702   }
1703
1704   client_send_all (&chn->group_pub_hash, m);
1705   return GNUNET_OK;
1706 }
1707
1708
1709 /**
1710  * Incoming multicast request message from CADET.
1711  */
1712 int
1713 cadet_recv_request (void *cls,
1714                     struct GNUNET_CADET_Channel *channel,
1715                     void **ctx,
1716                     const struct GNUNET_MessageHeader *m)
1717 {
1718   const struct GNUNET_MULTICAST_RequestHeader *
1719     req = (const struct GNUNET_MULTICAST_RequestHeader *) m;
1720   uint16_t size = ntohs (m->size);
1721   if (size < sizeof (*req))
1722   {
1723     GNUNET_break_op (0);
1724     return GNUNET_SYSERR;
1725   }
1726   struct Channel *chn = *ctx;
1727   if (NULL == chn)
1728   {
1729     GNUNET_break_op (0);
1730     return GNUNET_SYSERR;
1731   }
1732   if (ntohl (req->purpose.size) != (size
1733                                     - sizeof (req->header)
1734                                     - sizeof (req->member_pub_key)
1735                                     - sizeof (req->signature)))
1736   {
1737     GNUNET_break_op (0);
1738     return GNUNET_SYSERR;
1739   }
1740   if (GNUNET_OK !=
1741       GNUNET_CRYPTO_ecdsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST,
1742                                   &req->purpose, &req->signature,
1743                                   &req->member_pub_key))
1744   {
1745     GNUNET_break_op (0);
1746     return GNUNET_SYSERR;
1747   }
1748
1749   client_send_origin (&chn->group_pub_hash, m);
1750   return GNUNET_OK;
1751 }
1752
1753
1754 /**
1755  * Incoming multicast replay request from CADET.
1756  */
1757 int
1758 cadet_recv_replay_request (void *cls,
1759                            struct GNUNET_CADET_Channel *channel,
1760                            void **ctx,
1761                            const struct GNUNET_MessageHeader *m)
1762 {
1763   struct MulticastReplayRequestMessage rep;
1764   uint16_t size = ntohs (m->size);
1765   if (size < sizeof (rep))
1766   {
1767     GNUNET_break_op (0);
1768     return GNUNET_SYSERR;
1769   }
1770   struct Channel *chn = *ctx;
1771
1772   GNUNET_memcpy (&rep, m, sizeof (rep));
1773   GNUNET_memcpy (&rep.member_pub_key, &chn->member_pub_key, sizeof (chn->member_pub_key));
1774
1775   struct GNUNET_CONTAINER_MultiHashMap *
1776     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
1777                                                         &chn->grp->pub_key_hash);
1778   if (NULL == grp_replay_req)
1779   {
1780     grp_replay_req = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1781     GNUNET_CONTAINER_multihashmap_put (replay_req_cadet,
1782                                        &chn->grp->pub_key_hash, grp_replay_req,
1783                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1784   }
1785   struct GNUNET_HashCode key_hash;
1786   replay_key_hash (rep.fragment_id, rep.message_id, rep.fragment_offset,
1787                    rep.flags, &key_hash);
1788   GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, chn,
1789                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1790
1791   client_send_random (&chn->group_pub_hash, &rep.header);
1792   return GNUNET_OK;
1793 }
1794
1795
1796 /**
1797  * Incoming multicast replay response from CADET.
1798  */
1799 int
1800 cadet_recv_replay_response (void *cls,
1801                             struct GNUNET_CADET_Channel *channel,
1802                             void **ctx,
1803                             const struct GNUNET_MessageHeader *m)
1804 {
1805   //struct Channel *chn = *ctx;
1806
1807   /* @todo FIXME: got replay error response, send request to other members */
1808
1809   return GNUNET_OK;
1810 }
1811
1812
1813 /**
1814  * Message handlers for CADET.
1815  */
1816 static const struct GNUNET_CADET_MessageHandler cadet_handlers[] = {
1817   { cadet_recv_join_request,
1818     GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST, 0 },
1819
1820   { cadet_recv_message,
1821     GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE, 0 },
1822
1823   { cadet_recv_request,
1824     GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST, 0 },
1825
1826   { cadet_recv_replay_request,
1827     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST, 0 },
1828
1829   { cadet_recv_replay_response,
1830     GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE, 0 },
1831
1832   { NULL, 0, 0 }
1833 };
1834
1835
1836 /**
1837  * Service started.
1838  *
1839  * @param cls closure
1840  * @param server the initialized server
1841  * @param cfg configuration to use
1842  */
1843 static void
1844 run (void *cls,
1845      struct GNUNET_SERVER_Handle *srv,
1846      const struct GNUNET_CONFIGURATION_Handle *c)
1847 {
1848   cfg = c;
1849   server = srv;
1850   GNUNET_SERVER_connect_notify (server, &client_notify_connect, NULL);
1851   GNUNET_CRYPTO_get_peer_identity (cfg, &this_peer);
1852
1853   stats = GNUNET_STATISTICS_create ("multicast", cfg);
1854   origins = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
1855   members = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
1856   group_members = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1857   channels_in = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
1858   channels_out = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
1859   replay_req_cadet = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1860   replay_req_client = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1861
1862   cadet = GNUNET_CADET_connect (cfg, NULL,
1863                                 &cadet_notify_channel_end,
1864                                 cadet_handlers);
1865   GNUNET_assert (NULL != cadet);
1866
1867   nc = GNUNET_SERVER_notification_context_create (server, 1);
1868   GNUNET_SERVER_add_handlers (server, server_handlers);
1869   GNUNET_SERVER_disconnect_notify (server,
1870                                    &client_notify_disconnect, NULL);
1871   GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
1872                                  NULL);
1873 }
1874
1875
1876 /**
1877  * The main function for the multicast service.
1878  *
1879  * @param argc number of arguments from the command line
1880  * @param argv command line arguments
1881  * @return 0 ok, 1 on error
1882  */
1883 int
1884 main (int argc, char *const *argv)
1885 {
1886   return (GNUNET_OK ==
1887           GNUNET_SERVICE_run (argc, argv, "multicast",
1888                               GNUNET_SERVICE_OPTION_NONE, &run, NULL)) ? 0 : 1;
1889 }
1890
1891 /* end of gnunet-service-multicast.c */