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