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