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 it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your 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      Affero General Public License for more details.
  14
  15      You should have received a copy of the GNU Affero General Public License
  16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 */
  18
  19 /**
  20  * @file multicast/gnunet-service-multicast.c
  21  * @brief program that does multicast
  22  * @author Christian Grothoff
  23  */
  24 #include "platform.h"
  25 #include "gnunet_util_lib.h"
  26 #include "gnunet_signatures.h"
  27 #include "gnunet_applications.h"
  28 #include "gnunet_statistics_service.h"
  29 #include "gnunet_cadet_service.h"
  30 #include "gnunet_multicast_service.h"
  31 #include "multicast.h"
  32
  33 /**
  34  * Handle to our current configuration.
  35  */
  36 static const struct GNUNET_CONFIGURATION_Handle *cfg;
  37
  38 /**
  39  * Service handle.
  40  */
  41 static struct GNUNET_SERVICE_Handle *service;
  42
  43 /**
  44  * CADET handle.
  45  */
  46 static struct GNUNET_CADET_Handle *cadet;
  47
  48 /**
  49  * Identity of this peer.
  50  */
  51 static struct GNUNET_PeerIdentity this_peer;
  52
  53 /**
  54  * Handle to the statistics service.
  55  */
  56 static struct GNUNET_STATISTICS_Handle *stats;
  57
  58 /**
  59  * All connected origin clients.
  60  * Group's pub_key_hash -> struct Origin * (uniq)
  61  */
  62 static struct GNUNET_CONTAINER_MultiHashMap *origins;
  63
  64 /**
  65  * All connected member clients.
  66  * Group's pub_key_hash -> struct Member * (multi)
  67  */
  68 static struct GNUNET_CONTAINER_MultiHashMap *members;
  69
  70 /**
  71  * Connected member clients per group.
  72  * Group's pub_key_hash -> Member's pub_key_hash (uniq) -> struct Member * (uniq)
  73  */
  74 static struct GNUNET_CONTAINER_MultiHashMap *group_members;
  75
  76 /**
  77  * Incoming CADET channels with connected children in the tree.
  78  * Group's pub_key_hash -> struct Channel * (multi)
  79  */
  80 static struct GNUNET_CONTAINER_MultiHashMap *channels_in;
  81
  82 /**
  83  * Outgoing CADET channels connecting to parents in the tree.
  84  * Group's pub_key_hash -> struct Channel * (multi)
  85  */
  86 static struct GNUNET_CONTAINER_MultiHashMap *channels_out;
  87
  88 /**
  89  * Incoming replay requests from CADET.
  90  * Group's pub_key_hash ->
  91  *   H(fragment_id, message_id, fragment_offset, flags) -> struct Channel *
  92  */
  93 static struct GNUNET_CONTAINER_MultiHashMap *replay_req_cadet;
  94
  95 /**
  96  * Incoming replay requests from clients.
  97  * Group's pub_key_hash ->
  98  *   H(fragment_id, message_id, fragment_offset, flags) -> struct GNUNET_SERVICE_Client *
  99  */
 100 static struct GNUNET_CONTAINER_MultiHashMap *replay_req_client;
 101
 102
 103 /**
 104  * Join status of a remote peer.
 105  */
 106 enum JoinStatus
 107 {
 108   JOIN_REFUSED  = -1,
 109   JOIN_NOT_ASKED = 0,
 110   JOIN_WAITING   = 1,
 111   JOIN_ADMITTED  = 2,
 112 };
 113
 114 enum ChannelDirection
 115 {
 116   DIR_INCOMING = 0,
 117   DIR_OUTGOING = 1,
 118 };
 119
 120
 121 /**
 122  * Context for a CADET channel.
 123  */
 124 struct Channel
 125 {
 126   /**
 127    * Group the channel belongs to.
 128    *
 129    * Only set for outgoing channels.
 130    */
 131   struct Group *group;
 132
 133   /**
 134    * CADET channel.
 135    */
 136   struct GNUNET_CADET_Channel *channel;
 137
 138   // FIXME: not used
 139   /**
 140    * CADET transmission handle.
 141    */
 142   struct GNUNET_CADET_TransmitHandle *tmit_handle;
 143
 144   /**
 145    * Public key of the target group.
 146    */
 147   struct GNUNET_CRYPTO_EddsaPublicKey group_pub_key;
 148
 149   /**
 150    * Hash of @a group_pub_key.
 151    */
 152   struct GNUNET_HashCode group_pub_hash;
 153
 154   /**
 155    * Public key of the joining member.
 156    */
 157   struct GNUNET_CRYPTO_EcdsaPublicKey member_pub_key;
 158
 159   /**
 160    * Remote peer identity.
 161    */
 162   struct GNUNET_PeerIdentity peer;
 163
 164   /**
 165    * Current window size, set by cadet_notify_window_change()
 166    */
 167   int32_t window_size;
 168
 169   /**
 170    * Is the connection established?
 171    */
 172   int8_t is_connected;
 173
 174   /**
 175    * Is the remote peer admitted to the group?
 176    * @see enum JoinStatus
 177    */
 178   int8_t join_status;
 179
 180   /**
 181    * Number of messages waiting to be sent to CADET.
 182    */
 183   uint8_t msgs_pending;
 184
 185   /**
 186    * Channel direction.
 187    * @see enum ChannelDirection
 188    */
 189   uint8_t direction;
 190 };
 191
 192
 193 /**
 194  * List of connected clients.
 195  */
 196 struct ClientList
 197 {
 198   struct ClientList *prev;
 199   struct ClientList *next;
 200   struct GNUNET_SERVICE_Client *client;
 201 };
 202
 203
 204 /**
 205  * Client context for an origin or member.
 206  */
 207 struct Group
 208 {
 209   struct ClientList *clients_head;
 210   struct ClientList *clients_tail;
 211
 212   /**
 213    * Public key of the group.
 214    */
 215   struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
 216
 217   /**
 218    * Hash of @a pub_key.
 219    */
 220   struct GNUNET_HashCode pub_key_hash;
 221
 222   /**
 223    * CADET port hash.
 224    */
 225   struct GNUNET_HashCode cadet_port_hash;
 226
 227   /**
 228    * Is the client disconnected? #GNUNET_YES or #GNUNET_NO
 229    */
 230   uint8_t is_disconnected;
 231
 232   /**
 233    * Is this an origin (#GNUNET_YES), or member (#GNUNET_NO)?
 234    */
 235   uint8_t is_origin;
 236
 237   union {
 238     struct Origin *origin;
 239     struct Member *member;
 240   };
 241 };
 242
 243
 244 /**
 245 * Client context for a group's origin.
 246  */
 247 struct Origin
 248 {
 249   struct Group group;
 250
 251   /**
 252    * Private key of the group.
 253    */
 254   struct GNUNET_CRYPTO_EddsaPrivateKey priv_key;
 255
 256   /**
 257    * CADET port.
 258    */
 259   struct GNUNET_CADET_Port *cadet_port;
 260
 261   /**
 262    * Last message fragment ID sent to the group.
 263    */
 264   uint64_t max_fragment_id;
 265 };
 266
 267
 268 /**
 269  * Client context for a group member.
 270  */
 271 struct Member
 272 {
 273   struct Group group;
 274
 275   /**
 276    * Private key of the member.
 277    */
 278   struct GNUNET_CRYPTO_EcdsaPrivateKey priv_key;
 279
 280   /**
 281    * Public key of the member.
 282    */
 283   struct GNUNET_CRYPTO_EcdsaPublicKey pub_key;
 284
 285   /**
 286    * Hash of @a pub_key.
 287    */
 288   struct GNUNET_HashCode pub_key_hash;
 289
 290   /**
 291    * Join request sent to the origin / members.
 292    */
 293   struct MulticastJoinRequestMessage *join_req;
 294
 295   /**
 296    * Join decision sent in reply to our request.
 297    *
 298    * Only a positive decision is stored here, in case of a negative decision the
 299    * client is disconnected.
 300    */
 301   struct MulticastJoinDecisionMessageHeader *join_dcsn;
 302
 303   /**
 304    * CADET channel to the origin.
 305    */
 306   struct Channel *origin_channel;
 307
 308   /**
 309    * Peer identity of origin.
 310    */
 311   struct GNUNET_PeerIdentity origin;
 312
 313   /**
 314    * Peer identity of relays (other members to connect).
 315    */
 316   struct GNUNET_PeerIdentity *relays;
 317
 318   /**
 319    * Last request fragment ID sent to the origin.
 320    */
 321   uint64_t max_fragment_id;
 322
 323   /**
 324    * Number of @a relays.
 325    */
 326   uint32_t relay_count;
 327 };
 328
 329
 330 /**
 331  * Client context.
 332  */
 333 struct Client {
 334   struct GNUNET_SERVICE_Client *client;
 335   struct Group *group;
 336 };
 337
 338
 339 struct ReplayRequestKey
 340 {
 341   uint64_t fragment_id;
 342   uint64_t message_id;
 343   uint64_t fragment_offset;
 344   uint64_t flags;
 345 };
 346
 347
 348 static struct Channel *
 349 cadet_channel_create (struct Group *grp, struct GNUNET_PeerIdentity *peer);
 350
 351 static void
 352 cadet_channel_destroy (struct Channel *chn);
 353
 354 static void
 355 client_send_join_decision (struct Member *mem,
 356                            const struct MulticastJoinDecisionMessageHeader *hdcsn);
 357
 358
 359 /**
 360  * Task run during shutdown.
 361  *
 362  * @param cls unused
 363  */
 364 static void
 365 shutdown_task (void *cls)
 366 {
 367   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 368               "shutting down\n");
 369   if (NULL != cadet)
 370   {
 371     GNUNET_CADET_disconnect (cadet);
 372     cadet = NULL;
 373   }
 374   if (NULL != stats)
 375   {
 376     GNUNET_STATISTICS_destroy (stats, GNUNET_YES);
 377     stats = NULL;
 378   }
 379   /* FIXME: do more clean up here */
 380 }
 381
 382
 383 /**
 384  * Clean up origin data structures after a client disconnected.
 385  */
 386 static void
 387 cleanup_origin (struct Origin *orig)
 388 {
 389   struct Group *grp = &orig->group;
 390   GNUNET_CONTAINER_multihashmap_remove (origins, &grp->pub_key_hash, orig);
 391   if (NULL != orig->cadet_port)
 392   {
 393     GNUNET_CADET_close_port (orig->cadet_port);
 394     orig->cadet_port = NULL;
 395   }
 396   GNUNET_free (orig);
 397 }
 398
 399
 400 /**
 401  * Clean up member data structures after a client disconnected.
 402  */
 403 static void
 404 cleanup_member (struct Member *mem)
 405 {
 406   struct Group *grp = &mem->group;
 407   struct GNUNET_CONTAINER_MultiHashMap *
 408     grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members,
 409                                                  &grp->pub_key_hash);
 410   GNUNET_assert (NULL != grp_mem);
 411   GNUNET_CONTAINER_multihashmap_remove (grp_mem, &mem->pub_key_hash, mem);
 412
 413   if (0 == GNUNET_CONTAINER_multihashmap_size (grp_mem))
 414   {
 415     GNUNET_CONTAINER_multihashmap_remove (group_members, &grp->pub_key_hash,
 416                                           grp_mem);
 417     GNUNET_CONTAINER_multihashmap_destroy (grp_mem);
 418   }
 419   if (NULL != mem->join_dcsn)
 420   {
 421     GNUNET_free (mem->join_dcsn);
 422     mem->join_dcsn = NULL;
 423   }
 424   if (NULL != mem->origin_channel)
 425   {
 426     GNUNET_CADET_channel_destroy (mem->origin_channel->channel);
 427     mem->origin_channel = NULL;
 428   }
 429   GNUNET_CONTAINER_multihashmap_remove (members, &grp->pub_key_hash, mem);
 430   GNUNET_free (mem);
 431 }
 432
 433
 434 /**
 435  * Clean up group data structures after a client disconnected.
 436  */
 437 static void
 438 cleanup_group (struct Group *grp)
 439 {
 440   (GNUNET_YES == grp->is_origin)
 441     ? cleanup_origin (grp->origin)
 442     : cleanup_member (grp->member);
 443 }
 444
 445
 446 void
 447 replay_key_hash (uint64_t fragment_id, uint64_t message_id,
 448                  uint64_t fragment_offset, uint64_t flags,
 449                  struct GNUNET_HashCode *key_hash)
 450 {
 451   struct ReplayRequestKey key = {
 452     .fragment_id = fragment_id,
 453     .message_id = message_id,
 454     .fragment_offset = fragment_offset,
 455     .flags = flags,
 456   };
 457   GNUNET_CRYPTO_hash (&key, sizeof (key), key_hash);
 458 }
 459
 460
 461 /**
 462  * Remove channel from replay request hashmap.
 463  *
 464  * @param chn
 465  *        Channel to remove.
 466  *
 467  * @return #GNUNET_YES if there are more entries to process,
 468  *         #GNUNET_NO when reached end of hashmap.
 469  */
 470 static int
 471 replay_req_remove_cadet (struct Channel *chn)
 472 {
 473   if (NULL == chn || NULL == chn->group)
 474     return GNUNET_SYSERR;
 475
 476   struct GNUNET_CONTAINER_MultiHashMap *
 477     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
 478                                                         &chn->group->pub_key_hash);
 479   if (NULL == grp_replay_req)
 480     return GNUNET_NO;
 481
 482   struct GNUNET_CONTAINER_MultiHashMapIterator *
 483     it = GNUNET_CONTAINER_multihashmap_iterator_create (grp_replay_req);
 484   struct GNUNET_HashCode key;
 485   const struct Channel *c;
 486   while (GNUNET_YES
 487          == GNUNET_CONTAINER_multihashmap_iterator_next (it, &key,
 488                                                          (const void **) &c))
 489   {
 490     if (c == chn)
 491     {
 492       GNUNET_CONTAINER_multihashmap_remove (grp_replay_req, &key, chn);
 493       GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
 494       return GNUNET_YES;
 495     }
 496   }
 497   GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
 498   return GNUNET_NO;
 499 }
 500
 501
 502 /**
 503  * Remove client from replay request hashmap.
 504  *
 505  * @param client
 506  *        Client to remove.
 507  *
 508  * @return #GNUNET_YES if there are more entries to process,
 509  *         #GNUNET_NO when reached end of hashmap.
 510  */
 511 static int
 512 replay_req_remove_client (struct Group *grp, struct GNUNET_SERVICE_Client *client)
 513 {
 514   struct GNUNET_CONTAINER_MultiHashMap *
 515     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
 516                                                         &grp->pub_key_hash);
 517   if (NULL == grp_replay_req)
 518     return GNUNET_NO;
 519
 520   struct GNUNET_CONTAINER_MultiHashMapIterator *
 521     it = GNUNET_CONTAINER_multihashmap_iterator_create (grp_replay_req);
 522   struct GNUNET_HashCode key;
 523   const struct GNUNET_SERVICE_Client *c;
 524   while (GNUNET_YES
 525          == GNUNET_CONTAINER_multihashmap_iterator_next (it, &key,
 526                                                          (const void **) &c))
 527   {
 528     if (c == client)
 529     {
 530       GNUNET_CONTAINER_multihashmap_remove (grp_replay_req, &key, client);
 531       GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
 532       return GNUNET_YES;
 533     }
 534   }
 535   GNUNET_CONTAINER_multihashmap_iterator_destroy (it);
 536   return GNUNET_NO;
 537 }
 538
 539
 540 /**
 541  * Send message to a client.
 542  */
 543 static void
 544 client_send (struct GNUNET_SERVICE_Client *client,
 545              const struct GNUNET_MessageHeader *msg)
 546 {
 547   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 548               "%p Sending message to client.\n", client);
 549
 550   struct GNUNET_MQ_Envelope *
 551     env = GNUNET_MQ_msg_copy (msg);
 552
 553   GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client),
 554                   env);
 555 }
 556
 557
 558 /**
 559  * Send message to all clients connected to the group.
 560  */
 561 static void
 562 client_send_group_keep_envelope (const struct Group *grp,
 563                                  struct GNUNET_MQ_Envelope *env)
 564 {
 565   struct ClientList *cli = grp->clients_head;
 566
 567   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 568               "%p Sending message to all clients of the group.\n",
 569               grp);
 570   while (NULL != cli)
 571   {
 572     GNUNET_MQ_send_copy (GNUNET_SERVICE_client_get_mq (cli->client),
 573                          env);
 574     cli = cli->next;
 575   }
 576 }
 577
 578
 579 /**
 580  * Send message to all clients connected to the group and
 581  * takes care of freeing @env.
 582  */
 583 static void
 584 client_send_group (const struct Group *grp,
 585                    struct GNUNET_MQ_Envelope *env)
 586 {
 587   client_send_group_keep_envelope (grp, env);
 588   GNUNET_MQ_discard (env);
 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   struct GNUNET_MQ_Envelope *env = cls;
 600   struct Member *orig = origin;
 601
 602   client_send_group_keep_envelope (&orig->group, env);
 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   struct GNUNET_MQ_Envelope *env = cls;
 615   struct Member *mem = member;
 616
 617   if (NULL != mem->join_dcsn)
 618   { /* Only send message to admitted members */
 619     client_send_group_keep_envelope (&mem->group, env);
 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                  struct GNUNET_MQ_Envelope *env)
 636 {
 637   int n = 0;
 638   n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash,
 639                                                    client_send_origin_cb,
 640                                                    (void *) env);
 641   n += GNUNET_CONTAINER_multihashmap_get_multiple (members, pub_key_hash,
 642                                                    client_send_member_cb,
 643                                                    (void *) env);
 644   GNUNET_MQ_discard (env);
 645   return n;
 646 }
 647
 648
 649 /**
 650  * Send message to a random origin client or a random member client.
 651  *
 652  * @param grp  The group to send @a msg to.
 653  * @param msg  Message to send.
 654  */
 655 static int
 656 client_send_random (struct GNUNET_HashCode *pub_key_hash,
 657                     struct GNUNET_MQ_Envelope *env)
 658 {
 659   int n = 0;
 660   n = GNUNET_CONTAINER_multihashmap_get_random (origins, client_send_origin_cb,
 661                                                  (void *) env);
 662   if (n <= 0)
 663     n = GNUNET_CONTAINER_multihashmap_get_random (members, client_send_member_cb,
 664                                                    (void *) env);
 665   GNUNET_MQ_discard (env);
 666   return n;
 667 }
 668
 669
 670 /**
 671  * Send message to all origin clients connected to the group.
 672  *
 673  * @param pub_key_hash
 674  *        H(key_pub) of the group.
 675  * @param msg
 676  *        Message to send.
 677  */
 678 static int
 679 client_send_origin (struct GNUNET_HashCode *pub_key_hash,
 680                     struct GNUNET_MQ_Envelope *env)
 681 {
 682   int n = 0;
 683   n += GNUNET_CONTAINER_multihashmap_get_multiple (origins, pub_key_hash,
 684                                                    client_send_origin_cb,
 685                                                    (void *) env);
 686   return n;
 687 }
 688
 689
 690 /**
 691  * Send fragment acknowledgement to all clients of the channel.
 692  *
 693  * @param pub_key_hash
 694  *        H(key_pub) of the group.
 695  */
 696 static void
 697 client_send_ack (struct GNUNET_HashCode *pub_key_hash)
 698 {
 699   struct GNUNET_MQ_Envelope *env;
 700
 701   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 702               "Sending message ACK to client.\n");
 703   env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_MULTICAST_FRAGMENT_ACK);
 704   client_send_all (pub_key_hash, env);
 705 }
 706
 707
 708 struct CadetTransmitClosure
 709 {
 710   struct Channel *chn;
 711   const struct GNUNET_MessageHeader *msg;
 712 };
 713
 714
 715 /**
 716  * Send a message to a CADET channel.
 717  *
 718  * @param chn  Channel.
 719  * @param msg  Message.
 720  */
 721 static void
 722 cadet_send_channel (struct Channel *chn, const struct GNUNET_MessageHeader *msg)
 723 {
 724   struct GNUNET_MQ_Envelope *
 725     env = GNUNET_MQ_msg_copy (msg);
 726
 727   GNUNET_MQ_send (GNUNET_CADET_get_mq (chn->channel), env);
 728
 729   if (0 < chn->window_size)
 730   {
 731     client_send_ack (&chn->group_pub_hash);
 732   }
 733   else
 734   {
 735     chn->msgs_pending++;
 736     GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 737                 "%p Queuing message. Pending messages: %u\n",
 738                 chn, chn->msgs_pending);
 739   }
 740 }
 741
 742
 743 /**
 744  * Create CADET channel and send a join request.
 745  */
 746 static void
 747 cadet_send_join_request (struct Member *mem)
 748 {
 749   mem->origin_channel = cadet_channel_create (&mem->group, &mem->origin);
 750   cadet_send_channel (mem->origin_channel, &mem->join_req->header);
 751
 752   uint32_t i;
 753   for (i = 0; i < mem->relay_count; i++)
 754   {
 755     struct Channel *
 756       chn = cadet_channel_create (&mem->group, &mem->relays[i]);
 757     cadet_send_channel (chn, &mem->join_req->header);
 758   }
 759 }
 760
 761
 762 static int
 763 cadet_send_join_decision_cb (void *cls,
 764                              const struct GNUNET_HashCode *group_pub_hash,
 765                              void *channel)
 766 {
 767   const struct MulticastJoinDecisionMessageHeader *hdcsn = cls;
 768   struct Channel *chn = channel;
 769
 770   const struct MulticastJoinDecisionMessage *dcsn =
 771     (struct MulticastJoinDecisionMessage *) &hdcsn[1];
 772
 773   if (0 == memcmp (&hdcsn->member_pub_key, &chn->member_pub_key, sizeof (chn->member_pub_key))
 774       && 0 == memcmp (&hdcsn->peer, &chn->peer, sizeof (chn->peer)))
 775   {
 776     if (GNUNET_YES == ntohl (dcsn->is_admitted))
 777     {
 778       chn->join_status = JOIN_ADMITTED;
 779     }
 780     else
 781     {
 782       chn->join_status = JOIN_REFUSED;
 783     }
 784     cadet_send_channel (chn, &hdcsn->header);
 785     return GNUNET_YES;
 786   }
 787
 788   // return GNUNET_YES to continue the multihashmap_get iteration
 789   return GNUNET_YES;
 790 }
 791
 792
 793 /**
 794  * Send join decision to a remote peer.
 795  */
 796 static void
 797 cadet_send_join_decision (struct Group *grp,
 798                           const struct MulticastJoinDecisionMessageHeader *hdcsn)
 799 {
 800   GNUNET_CONTAINER_multihashmap_get_multiple (channels_in, &grp->pub_key_hash,
 801                                               &cadet_send_join_decision_cb,
 802                                               (void *) hdcsn);
 803 }
 804
 805
 806 /**
 807  * Iterator callback for sending a message to origin clients.
 808  */
 809 static int
 810 cadet_send_cb (void *cls, const struct GNUNET_HashCode *pub_key_hash,
 811                void *channel)
 812 {
 813   const struct GNUNET_MessageHeader *msg = cls;
 814   struct Channel *chn = channel;
 815   if (JOIN_ADMITTED == chn->join_status)
 816     cadet_send_channel (chn, msg);
 817   return GNUNET_YES;
 818 }
 819
 820
 821 /**
 822  * Send message to all connected children.
 823  */
 824 static int
 825 cadet_send_children (struct GNUNET_HashCode *pub_key_hash,
 826                      const struct GNUNET_MessageHeader *msg)
 827 {
 828   int n = 0;
 829   if (channels_in != NULL)
 830     n += GNUNET_CONTAINER_multihashmap_get_multiple (channels_in, pub_key_hash,
 831                                                      cadet_send_cb, (void *) msg);
 832   return n;
 833 }
 834
 835
 836 #if 0       // unused as yet
 837 /**
 838  * Send message to all connected parents.
 839  */
 840 static int
 841 cadet_send_parents (struct GNUNET_HashCode *pub_key_hash,
 842                     const struct GNUNET_MessageHeader *msg)
 843 {
 844   int n = 0;
 845   if (channels_in != NULL)
 846     n += GNUNET_CONTAINER_multihashmap_get_multiple (channels_out, pub_key_hash,
 847                                                      cadet_send_cb, (void *) msg);
 848   return n;
 849 }
 850 #endif
 851
 852
 853 /**
 854  * CADET channel connect handler.
 855  *
 856  * @see GNUNET_CADET_ConnectEventHandler()
 857  */
 858 static void *
 859 cadet_notify_connect (void *cls,
 860                       struct GNUNET_CADET_Channel *channel,
 861                       const struct GNUNET_PeerIdentity *source)
 862 {
 863   struct Channel *chn = GNUNET_malloc (sizeof (struct Channel));
 864   chn->group = cls;
 865   chn->channel = channel;
 866   chn->direction = DIR_INCOMING;
 867   chn->join_status = JOIN_NOT_ASKED;
 868
 869   GNUNET_CONTAINER_multihashmap_put (channels_in, &chn->group->pub_key_hash, chn,
 870                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
 871   return chn;
 872 }
 873
 874
 875 /**
 876  * CADET window size change handler.
 877  *
 878  * @see GNUNET_CADET_WindowSizeEventHandler()
 879  */
 880 static void
 881 cadet_notify_window_change (void *cls,
 882                             const struct GNUNET_CADET_Channel *channel,
 883                             int window_size)
 884 {
 885   struct Channel *chn = cls;
 886
 887   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 888               "%p Window size changed to %d.  Pending messages: %u\n",
 889               chn, window_size, chn->msgs_pending);
 890
 891   chn->is_connected = GNUNET_YES;
 892   chn->window_size = (int32_t) window_size;
 893
 894   for (int i = 0; i < window_size; i++)
 895   {
 896     if (0 < chn->msgs_pending)
 897     {
 898       client_send_ack (&chn->group_pub_hash);
 899       chn->msgs_pending--;
 900     }
 901     else
 902     {
 903       break;
 904     }
 905   }
 906 }
 907
 908
 909 /**
 910  * CADET channel disconnect handler.
 911  *
 912  * @see GNUNET_CADET_DisconnectEventHandler()
 913  */
 914 static void
 915 cadet_notify_disconnect (void *cls,
 916                          const struct GNUNET_CADET_Channel *channel)
 917 {
 918   if (NULL == cls)
 919     return;
 920
 921   struct Channel *chn = cls;
 922   if (NULL != chn->group)
 923   {
 924     if (GNUNET_NO == chn->group->is_origin)
 925     {
 926       struct Member *mem = (struct Member *) chn->group;
 927       if (chn == mem->origin_channel)
 928         mem->origin_channel = NULL;
 929     }
 930   }
 931
 932   int ret;
 933   do
 934   {
 935     ret = replay_req_remove_cadet (chn);
 936   }
 937   while (GNUNET_YES == ret);
 938
 939   GNUNET_free (chn);
 940 }
 941
 942
 943 static int
 944 check_cadet_join_request (void *cls,
 945                           const struct MulticastJoinRequestMessage *req)
 946 {
 947   struct Channel *chn = cls;
 948
 949   if (NULL == chn
 950       || JOIN_NOT_ASKED != chn->join_status)
 951   {
 952     return GNUNET_SYSERR;
 953   }
 954
 955   uint16_t size = ntohs (req->header.size);
 956   if (size < sizeof (*req))
 957   {
 958     GNUNET_break_op (0);
 959     return GNUNET_SYSERR;
 960   }
 961   if (ntohl (req->purpose.size) != (size
 962                                     - sizeof (req->header)
 963                                     - sizeof (req->reserved)
 964                                     - sizeof (req->signature)))
 965   {
 966     GNUNET_break_op (0);
 967     return GNUNET_SYSERR;
 968   }
 969   if (GNUNET_OK !=
 970       GNUNET_CRYPTO_ecdsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST,
 971                                   &req->purpose, &req->signature,
 972                                   &req->member_pub_key))
 973   {
 974     GNUNET_break_op (0);
 975     return GNUNET_SYSERR;
 976   }
 977
 978   return GNUNET_OK;
 979 }
 980
 981
 982 /**
 983  * Incoming join request message from CADET.
 984  */
 985 static void
 986 handle_cadet_join_request (void *cls,
 987                            const struct MulticastJoinRequestMessage *req)
 988 {
 989   struct Channel *chn = cls;
 990   GNUNET_CADET_receive_done (chn->channel);
 991
 992   struct GNUNET_HashCode group_pub_hash;
 993   GNUNET_CRYPTO_hash (&req->group_pub_key, sizeof (req->group_pub_key), &group_pub_hash);
 994   chn->group_pub_key = req->group_pub_key;
 995   chn->group_pub_hash = group_pub_hash;
 996   chn->member_pub_key = req->member_pub_key;
 997   chn->peer = req->peer;
 998   chn->join_status = JOIN_WAITING;
 999
1000   client_send_all (&group_pub_hash,
1001                    GNUNET_MQ_msg_copy (&req->header));
1002 }
1003
1004
1005 static int
1006 check_cadet_join_decision (void *cls,
1007                            const struct MulticastJoinDecisionMessageHeader *hdcsn)
1008 {
1009   uint16_t size = ntohs (hdcsn->header.size);
1010   if (size < sizeof (struct MulticastJoinDecisionMessageHeader) +
1011              sizeof (struct MulticastJoinDecisionMessage))
1012   {
1013     GNUNET_break_op (0);
1014     return GNUNET_SYSERR;
1015   }
1016
1017   struct Channel *chn = cls;
1018   if (NULL == chn)
1019   {
1020     GNUNET_break (0);
1021     return GNUNET_SYSERR;
1022   }
1023   if (NULL == chn->group || GNUNET_NO != chn->group->is_origin)
1024   {
1025     GNUNET_break (0);
1026     return GNUNET_SYSERR;
1027   }
1028   switch (chn->join_status)
1029   {
1030   case JOIN_REFUSED:
1031     return GNUNET_SYSERR;
1032
1033   case JOIN_ADMITTED:
1034     return GNUNET_OK;
1035
1036   case JOIN_NOT_ASKED:
1037   case JOIN_WAITING:
1038     break;
1039   }
1040
1041   return GNUNET_OK;
1042 }
1043
1044
1045 /**
1046  * Incoming join decision message from CADET.
1047  */
1048 static void
1049 handle_cadet_join_decision (void *cls,
1050                             const struct MulticastJoinDecisionMessageHeader *hdcsn)
1051 {
1052   const struct MulticastJoinDecisionMessage *
1053     dcsn = (const struct MulticastJoinDecisionMessage *) &hdcsn[1];
1054
1055   struct Channel *chn = cls;
1056   GNUNET_CADET_receive_done (chn->channel);
1057
1058   // FIXME: do we need to copy chn->peer or compare it with hdcsn->peer?
1059   struct Member *mem = (struct Member *) chn->group;
1060   client_send_join_decision (mem, hdcsn);
1061   if (GNUNET_YES == ntohl (dcsn->is_admitted))
1062   {
1063     chn->join_status = JOIN_ADMITTED;
1064   }
1065   else
1066   {
1067     chn->join_status = JOIN_REFUSED;
1068     cadet_channel_destroy (chn);
1069   }
1070 }
1071
1072
1073 static int
1074 check_cadet_message (void *cls,
1075                      const struct GNUNET_MULTICAST_MessageHeader *msg)
1076 {
1077   uint16_t size = ntohs (msg->header.size);
1078   if (size < sizeof (*msg))
1079   {
1080     GNUNET_break_op (0);
1081     return GNUNET_SYSERR;
1082   }
1083
1084   struct Channel *chn = cls;
1085   if (NULL == chn)
1086   {
1087     GNUNET_break (0);
1088     return GNUNET_SYSERR;
1089   }
1090   if (ntohl (msg->purpose.size) != (size
1091                                     - sizeof (msg->header)
1092                                     - sizeof (msg->hop_counter)
1093                                     - sizeof (msg->signature)))
1094   {
1095     GNUNET_break_op (0);
1096     return GNUNET_SYSERR;
1097   }
1098   if (GNUNET_OK !=
1099       GNUNET_CRYPTO_eddsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_MESSAGE,
1100                                   &msg->purpose, &msg->signature,
1101                                   &chn->group_pub_key))
1102   {
1103     GNUNET_break_op (0);
1104     return GNUNET_SYSERR;
1105   }
1106
1107   return GNUNET_OK;
1108 }
1109
1110
1111 /**
1112  * Incoming multicast message from CADET.
1113  */
1114 static void
1115 handle_cadet_message (void *cls,
1116                       const struct GNUNET_MULTICAST_MessageHeader *msg)
1117 {
1118   struct Channel *chn = cls;
1119   GNUNET_CADET_receive_done (chn->channel);
1120   client_send_all (&chn->group_pub_hash,
1121                    GNUNET_MQ_msg_copy (&msg->header));
1122 }
1123
1124
1125 static int
1126 check_cadet_request (void *cls,
1127                      const struct GNUNET_MULTICAST_RequestHeader *req)
1128 {
1129   uint16_t size = ntohs (req->header.size);
1130   if (size < sizeof (*req))
1131   {
1132     GNUNET_break_op (0);
1133     return GNUNET_SYSERR;
1134   }
1135
1136   struct Channel *chn = cls;
1137   if (NULL == chn)
1138   {
1139     GNUNET_break (0);
1140     return GNUNET_SYSERR;
1141   }
1142   if (ntohl (req->purpose.size) != (size
1143                                     - sizeof (req->header)
1144                                     - sizeof (req->member_pub_key)
1145                                     - sizeof (req->signature)))
1146   {
1147     GNUNET_break_op (0);
1148     return GNUNET_SYSERR;
1149   }
1150   if (GNUNET_OK !=
1151       GNUNET_CRYPTO_ecdsa_verify (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST,
1152                                   &req->purpose, &req->signature,
1153                                   &req->member_pub_key))
1154   {
1155     GNUNET_break_op (0);
1156     return GNUNET_SYSERR;
1157   }
1158
1159   return GNUNET_OK;
1160 }
1161
1162
1163 /**
1164  * Incoming multicast request message from CADET.
1165  */
1166 static void
1167 handle_cadet_request (void *cls,
1168                       const struct GNUNET_MULTICAST_RequestHeader *req)
1169 {
1170   struct Channel *chn = cls;
1171   GNUNET_CADET_receive_done (chn->channel);
1172   client_send_origin (&chn->group_pub_hash,
1173                       GNUNET_MQ_msg_copy (&req->header));
1174 }
1175
1176
1177 // FIXME: do checks in handle_cadet_replay_request
1178 //static int
1179 //check_cadet_replay_request (void *cls,
1180 //                            const struct MulticastReplayRequestMessage *req)
1181 //{
1182 //  uint16_t size = ntohs (req->header.size);
1183 //  if (size < sizeof (*req))
1184 //  {
1185 //    GNUNET_break_op (0);
1186 //    return GNUNET_SYSERR;
1187 //  }
1188 //
1189 //  struct Channel *chn = cls;
1190 //  if (NULL == chn)
1191 //  {
1192 //    GNUNET_break_op (0);
1193 //    return GNUNET_SYSERR;
1194 //  }
1195 //
1196 //  return GNUNET_OK;
1197 //}
1198
1199
1200 /**
1201  * Incoming multicast replay request from CADET.
1202  */
1203 static void
1204 handle_cadet_replay_request (void *cls,
1205                              const struct MulticastReplayRequestMessage *req)
1206 {
1207   struct Channel *chn = cls;
1208
1209   GNUNET_CADET_receive_done (chn->channel);
1210
1211   struct MulticastReplayRequestMessage rep = *req;
1212   GNUNET_memcpy (&rep.member_pub_key, &chn->member_pub_key, sizeof (chn->member_pub_key));
1213
1214   struct GNUNET_CONTAINER_MultiHashMap *
1215     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
1216                                                         &chn->group->pub_key_hash);
1217   if (NULL == grp_replay_req)
1218   {
1219     grp_replay_req = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1220     GNUNET_CONTAINER_multihashmap_put (replay_req_cadet,
1221                                        &chn->group->pub_key_hash, grp_replay_req,
1222                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1223   }
1224   struct GNUNET_HashCode key_hash;
1225   replay_key_hash (rep.fragment_id,
1226                    rep.message_id,
1227                    rep.fragment_offset,
1228                    rep.flags,
1229                    &key_hash);
1230   GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, chn,
1231                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1232
1233   client_send_random (&chn->group_pub_hash,
1234                       GNUNET_MQ_msg_copy (&rep.header));
1235 }
1236
1237
1238 static int
1239 check_cadet_replay_response (void *cls,
1240                              const struct MulticastReplayResponseMessage *res)
1241 {
1242   struct Channel *chn = cls;
1243   if (NULL == chn)
1244   {
1245     GNUNET_break (0);
1246     return GNUNET_SYSERR;
1247   }
1248   return GNUNET_OK;
1249 }
1250
1251
1252 /**
1253  * Incoming multicast replay response from CADET.
1254  */
1255 static void
1256 handle_cadet_replay_response (void *cls,
1257                               const struct MulticastReplayResponseMessage *res)
1258 {
1259   struct Channel *chn = cls;
1260   GNUNET_CADET_receive_done (chn->channel);
1261
1262   /* @todo FIXME: got replay error response, send request to other members */
1263 }
1264
1265
1266 static void
1267 group_set_cadet_port_hash (struct Group *grp)
1268 {
1269   struct CadetPort {
1270     struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
1271     uint32_t app_type;
1272   } port = {
1273     grp->pub_key,
1274     GNUNET_APPLICATION_TYPE_MULTICAST,
1275   };
1276   GNUNET_CRYPTO_hash (&port, sizeof (port), &grp->cadet_port_hash);
1277 }
1278
1279
1280
1281 /**
1282  * Create new outgoing CADET channel.
1283  *
1284  * @param peer
1285  *        Peer to connect to.
1286  * @param group_pub_key
1287  *        Public key of group the channel belongs to.
1288  * @param group_pub_hash
1289  *        Hash of @a group_pub_key.
1290  *
1291  * @return Channel.
1292  */
1293 static struct Channel *
1294 cadet_channel_create (struct Group *grp, struct GNUNET_PeerIdentity *peer)
1295 {
1296   struct Channel *chn = GNUNET_malloc (sizeof (*chn));
1297   chn->group = grp;
1298   chn->group_pub_key = grp->pub_key;
1299   chn->group_pub_hash = grp->pub_key_hash;
1300   chn->peer = *peer;
1301   chn->direction = DIR_OUTGOING;
1302   chn->is_connected = GNUNET_NO;
1303   chn->join_status = JOIN_WAITING;
1304
1305   struct GNUNET_MQ_MessageHandler cadet_handlers[] = {
1306     GNUNET_MQ_hd_var_size (cadet_message,
1307                            GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE,
1308                            struct GNUNET_MULTICAST_MessageHeader,
1309                            chn),
1310
1311     GNUNET_MQ_hd_var_size (cadet_join_decision,
1312                            GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION,
1313                            struct MulticastJoinDecisionMessageHeader,
1314                            chn),
1315
1316     GNUNET_MQ_hd_fixed_size (cadet_replay_request,
1317                              GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST,
1318                              struct MulticastReplayRequestMessage,
1319                              chn),
1320
1321     GNUNET_MQ_hd_var_size (cadet_replay_response,
1322                            GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE,
1323                            struct MulticastReplayResponseMessage,
1324                            chn),
1325
1326     GNUNET_MQ_handler_end ()
1327   };
1328
1329   chn->channel = GNUNET_CADET_channel_create (cadet, chn, &chn->peer,
1330                                               &grp->cadet_port_hash,
1331                                               GNUNET_CADET_OPTION_RELIABLE,
1332                                               cadet_notify_window_change,
1333                                               cadet_notify_disconnect,
1334                                               cadet_handlers);
1335   GNUNET_CONTAINER_multihashmap_put (channels_out, &chn->group_pub_hash, chn,
1336                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1337   return chn;
1338 }
1339
1340
1341 /**
1342  * Destroy outgoing CADET channel.
1343  */
1344 static void
1345 cadet_channel_destroy (struct Channel *chn)
1346 {
1347   GNUNET_CADET_channel_destroy (chn->channel);
1348   GNUNET_CONTAINER_multihashmap_remove_all (channels_out, &chn->group_pub_hash);
1349   GNUNET_free (chn);
1350 }
1351
1352 /**
1353  * Handle a connecting client starting an origin.
1354  */
1355 static void
1356 handle_client_origin_start (void *cls,
1357                             const struct MulticastOriginStartMessage *msg)
1358 {
1359   struct Client *c = cls;
1360   struct GNUNET_SERVICE_Client *client = c->client;
1361
1362   struct GNUNET_CRYPTO_EddsaPublicKey pub_key;
1363   struct GNUNET_HashCode pub_key_hash;
1364
1365   GNUNET_CRYPTO_eddsa_key_get_public (&msg->group_key, &pub_key);
1366   GNUNET_CRYPTO_hash (&pub_key, sizeof (pub_key), &pub_key_hash);
1367
1368   struct Origin *
1369     orig = GNUNET_CONTAINER_multihashmap_get (origins, &pub_key_hash);
1370   struct Group *grp;
1371
1372   if (NULL == orig)
1373   {
1374     orig = GNUNET_new (struct Origin);
1375     orig->priv_key = msg->group_key;
1376     orig->max_fragment_id = GNUNET_ntohll (msg->max_fragment_id);
1377
1378     grp = c->group = &orig->group;
1379     grp->origin = orig;
1380     grp->is_origin = GNUNET_YES;
1381     grp->pub_key = pub_key;
1382     grp->pub_key_hash = pub_key_hash;
1383     grp->is_disconnected = GNUNET_NO;
1384
1385     GNUNET_CONTAINER_multihashmap_put (origins, &grp->pub_key_hash, orig,
1386                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1387
1388     group_set_cadet_port_hash (grp);
1389
1390     struct GNUNET_MQ_MessageHandler cadet_handlers[] = {
1391       GNUNET_MQ_hd_var_size (cadet_message,
1392                              GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE,
1393                              struct GNUNET_MULTICAST_MessageHeader,
1394                              grp),
1395
1396       GNUNET_MQ_hd_var_size (cadet_request,
1397                              GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST,
1398                              struct GNUNET_MULTICAST_RequestHeader,
1399                              grp),
1400
1401       GNUNET_MQ_hd_var_size (cadet_join_request,
1402                              GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST,
1403                              struct MulticastJoinRequestMessage,
1404                              grp),
1405
1406       GNUNET_MQ_hd_fixed_size (cadet_replay_request,
1407                                GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST,
1408                                struct MulticastReplayRequestMessage,
1409                                grp),
1410
1411       GNUNET_MQ_hd_var_size (cadet_replay_response,
1412                              GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE,
1413                              struct MulticastReplayResponseMessage,
1414                              grp),
1415
1416       GNUNET_MQ_handler_end ()
1417     };
1418
1419
1420     orig->cadet_port = GNUNET_CADET_open_port (cadet,
1421                                                &grp->cadet_port_hash,
1422                                                cadet_notify_connect,
1423                                                grp,
1424                                                cadet_notify_window_change,
1425                                                cadet_notify_disconnect,
1426                                                cadet_handlers);
1427   }
1428   else
1429   {
1430     grp = &orig->group;
1431   }
1432
1433   struct ClientList *cl = GNUNET_new (struct ClientList);
1434   cl->client = client;
1435   GNUNET_CONTAINER_DLL_insert (grp->clients_head, grp->clients_tail, cl);
1436
1437   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1438               "%p Client connected as origin to group %s.\n",
1439               orig, GNUNET_h2s (&grp->pub_key_hash));
1440   GNUNET_SERVICE_client_continue (client);
1441 }
1442
1443
1444 static int
1445 check_client_member_join (void *cls,
1446                           const struct MulticastMemberJoinMessage *msg)
1447 {
1448   uint16_t msg_size = ntohs (msg->header.size);
1449   struct GNUNET_PeerIdentity *relays = (struct GNUNET_PeerIdentity *) &msg[1];
1450   uint32_t relay_count = ntohl (msg->relay_count);
1451
1452   if (0 == relay_count)
1453   {
1454     GNUNET_break (0);
1455     return GNUNET_SYSERR;
1456   }
1457   if (UINT32_MAX / relay_count < sizeof (*relays)){
1458       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1459                     "relay_count (%lu) * sizeof (*relays)  (%lu) exceeds UINT32_MAX!\n",
1460               (unsigned long)relay_count,
1461               sizeof (*relays));
1462       return GNUNET_SYSERR;
1463   }
1464   uint32_t relay_size = relay_count * sizeof (*relays);
1465   struct GNUNET_MessageHeader *join_msg = NULL;
1466   uint16_t join_msg_size = 0;
1467   if (sizeof (*msg) + relay_size + sizeof (struct GNUNET_MessageHeader)
1468       <= msg_size)
1469   {
1470     join_msg = (struct GNUNET_MessageHeader *)
1471       (((char *) &msg[1]) + relay_size);
1472     join_msg_size = ntohs (join_msg->size);
1473     if (UINT16_MAX - join_msg_size < sizeof (struct MulticastJoinRequestMessage)){
1474         GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1475                     "join_msg_size (%u) + sizeof (struct MulticastJoinRequestMessage) (%lu) exceeds UINT16_MAX!\n",
1476                 (unsigned)join_msg_size,
1477                 (unsigned long)sizeof (struct MulticastJoinRequestMessage));
1478         return GNUNET_SYSERR;
1479     }
1480   }
1481   if (msg_size != (sizeof (*msg) + relay_size + join_msg_size)){
1482       GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1483                     "msg_size does not match real size of message!\n");
1484       return GNUNET_SYSERR;
1485   }else{
1486       return GNUNET_OK;
1487   }
1488 }
1489
1490
1491 /**
1492  * Handle a connecting client joining a group.
1493  */
1494 static void
1495 handle_client_member_join (void *cls,
1496                            const struct MulticastMemberJoinMessage *msg)
1497 {
1498   struct Client *c = cls;
1499   struct GNUNET_SERVICE_Client *client = c->client;
1500
1501   uint16_t msg_size = ntohs (msg->header.size);
1502
1503   struct GNUNET_CRYPTO_EcdsaPublicKey mem_pub_key;
1504   struct GNUNET_HashCode pub_key_hash, mem_pub_key_hash;
1505
1506   GNUNET_CRYPTO_ecdsa_key_get_public (&msg->member_key, &mem_pub_key);
1507   GNUNET_CRYPTO_hash (&mem_pub_key, sizeof (mem_pub_key), &mem_pub_key_hash);
1508   GNUNET_CRYPTO_hash (&msg->group_pub_key, sizeof (msg->group_pub_key), &pub_key_hash);
1509
1510   struct GNUNET_CONTAINER_MultiHashMap *
1511     grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members, &pub_key_hash);
1512   struct Member *mem = NULL;
1513   struct Group *grp;
1514
1515   if (NULL != grp_mem)
1516   {
1517     mem = GNUNET_CONTAINER_multihashmap_get (grp_mem, &mem_pub_key_hash);
1518   }
1519
1520   if (NULL == mem)
1521   {
1522     mem = GNUNET_new (struct Member);
1523     mem->origin = msg->origin;
1524     mem->priv_key = msg->member_key;
1525     mem->pub_key = mem_pub_key;
1526     mem->pub_key_hash = mem_pub_key_hash;
1527     mem->max_fragment_id = 0; // FIXME
1528
1529     grp = c->group = &mem->group;
1530     grp->member = mem;
1531     grp->is_origin = GNUNET_NO;
1532     grp->pub_key = msg->group_pub_key;
1533     grp->pub_key_hash = pub_key_hash;
1534     grp->is_disconnected = GNUNET_NO;
1535     group_set_cadet_port_hash (grp);
1536
1537     if (NULL == grp_mem)
1538     {
1539       grp_mem = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
1540       GNUNET_CONTAINER_multihashmap_put (group_members, &grp->pub_key_hash, grp_mem,
1541                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1542     }
1543     GNUNET_CONTAINER_multihashmap_put (grp_mem, &mem->pub_key_hash, mem,
1544                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1545
1546     // FIXME: should the members hash map have option UNIQUE_FAST?
1547     GNUNET_CONTAINER_multihashmap_put (members, &grp->pub_key_hash, mem,
1548                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1549   }
1550   else
1551   {
1552     grp = &mem->group;
1553   }
1554
1555   struct ClientList *cl = GNUNET_new (struct ClientList);
1556   cl->client = client;
1557   GNUNET_CONTAINER_DLL_insert (grp->clients_head, grp->clients_tail, cl);
1558
1559   char *str = GNUNET_CRYPTO_ecdsa_public_key_to_string (&mem->pub_key);
1560   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1561               "Client connected to group %s as member %s (%s). size = %d\n",
1562               GNUNET_h2s (&grp->pub_key_hash),
1563               GNUNET_h2s2 (&mem->pub_key_hash),
1564               str,
1565               GNUNET_CONTAINER_multihashmap_size (members));
1566   GNUNET_free (str);
1567
1568   if (NULL != mem->join_dcsn)
1569   { /* Already got a join decision, send it to client. */
1570     struct GNUNET_MQ_Envelope *
1571       env = GNUNET_MQ_msg_copy (&mem->join_dcsn->header);
1572
1573     GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client),
1574                     env);
1575   }
1576   else
1577   { /* First client of the group, send join request. */
1578     struct GNUNET_PeerIdentity *relays = (struct GNUNET_PeerIdentity *) &msg[1];
1579     uint32_t relay_count = ntohl (msg->relay_count);
1580     uint16_t relay_size = relay_count * sizeof (*relays);
1581     struct GNUNET_MessageHeader *join_msg = NULL;
1582     uint16_t join_msg_size = 0;
1583     if (sizeof (*msg) + relay_size + sizeof (struct GNUNET_MessageHeader)
1584         <= msg_size)
1585     {
1586       join_msg = (struct GNUNET_MessageHeader *)
1587         (((char *) &msg[1]) + relay_size);
1588       join_msg_size = ntohs (join_msg->size);
1589     }
1590
1591     uint16_t req_msg_size = sizeof (struct MulticastJoinRequestMessage) + join_msg_size;
1592     struct MulticastJoinRequestMessage *
1593       req = GNUNET_malloc (req_msg_size);
1594     req->header.size = htons (req_msg_size);
1595     req->header.type = htons (GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_REQUEST);
1596     req->group_pub_key = grp->pub_key;
1597     req->peer = this_peer;
1598     GNUNET_CRYPTO_ecdsa_key_get_public (&mem->priv_key, &req->member_pub_key);
1599     if (0 < join_msg_size)
1600       GNUNET_memcpy (&req[1], join_msg, join_msg_size);
1601
1602     req->member_pub_key = mem->pub_key;
1603     req->purpose.size = htonl (req_msg_size
1604                                - sizeof (req->header)
1605                                - sizeof (req->reserved)
1606                                - sizeof (req->signature));
1607     req->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST);
1608
1609     if (GNUNET_OK != GNUNET_CRYPTO_ecdsa_sign (&mem->priv_key, &req->purpose,
1610                                                &req->signature))
1611     {
1612       /* FIXME: handle error */
1613       GNUNET_assert (0);
1614     }
1615
1616     if (NULL != mem->join_req)
1617       GNUNET_free (mem->join_req);
1618     mem->join_req = req;
1619
1620     if (0 ==
1621         client_send_origin (&grp->pub_key_hash,
1622                             GNUNET_MQ_msg_copy (&mem->join_req->header)))
1623     { /* No local origins, send to remote origin */
1624       cadet_send_join_request (mem);
1625     }
1626   }
1627   GNUNET_SERVICE_client_continue (client);
1628 }
1629
1630
1631 static void
1632 client_send_join_decision (struct Member *mem,
1633                            const struct MulticastJoinDecisionMessageHeader *hdcsn)
1634 {
1635   client_send_group (&mem->group, GNUNET_MQ_msg_copy (&hdcsn->header));
1636
1637   const struct MulticastJoinDecisionMessage *
1638     dcsn = (const struct MulticastJoinDecisionMessage *) &hdcsn[1];
1639   if (GNUNET_YES == ntohl (dcsn->is_admitted))
1640   { /* Member admitted, store join_decision. */
1641     uint16_t dcsn_size = ntohs (dcsn->header.size);
1642     mem->join_dcsn = GNUNET_malloc (dcsn_size);
1643     GNUNET_memcpy (mem->join_dcsn, dcsn, dcsn_size);
1644   }
1645   else
1646   { /* Refused entry, but replay would be still possible for past members. */
1647   }
1648 }
1649
1650
1651 static int
1652 check_client_join_decision (void *cls,
1653                             const struct MulticastJoinDecisionMessageHeader *hdcsn)
1654 {
1655   return GNUNET_OK;
1656 }
1657
1658
1659 /**
1660  * Join decision from client.
1661  */
1662 static void
1663 handle_client_join_decision (void *cls,
1664                              const struct MulticastJoinDecisionMessageHeader *hdcsn)
1665 {
1666   struct Client *c = cls;
1667   struct GNUNET_SERVICE_Client *client = c->client;
1668   struct Group *grp = c->group;
1669
1670   if (NULL == grp)
1671   {
1672     GNUNET_break (0);
1673     GNUNET_SERVICE_client_drop (client);
1674     return;
1675   }
1676   GNUNET_assert (GNUNET_NO == grp->is_disconnected);
1677   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1678               "%p got join decision from client for group %s..\n",
1679               grp, GNUNET_h2s (&grp->pub_key_hash));
1680
1681   struct GNUNET_CONTAINER_MultiHashMap *
1682     grp_mem = GNUNET_CONTAINER_multihashmap_get (group_members,
1683                                                  &grp->pub_key_hash);
1684   struct Member *mem = NULL;
1685   if (NULL != grp_mem)
1686   {
1687     struct GNUNET_HashCode member_key_hash;
1688     GNUNET_CRYPTO_hash (&hdcsn->member_pub_key, sizeof (hdcsn->member_pub_key),
1689                         &member_key_hash);
1690     mem = GNUNET_CONTAINER_multihashmap_get (grp_mem, &member_key_hash);
1691     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1692                 "%p ..and member %s: %p\n",
1693                 grp, GNUNET_h2s (&member_key_hash), mem);
1694   }
1695
1696   if (NULL != mem)
1697   { /* Found local member */
1698     client_send_join_decision (mem, hdcsn);
1699   }
1700   else
1701   { /* Look for remote member */
1702     cadet_send_join_decision (grp, hdcsn);
1703   }
1704   GNUNET_SERVICE_client_continue (client);
1705 }
1706
1707
1708 static void
1709 handle_client_part_request (void *cls,
1710                             const struct GNUNET_MessageHeader *msg)
1711 {
1712   struct Client *c = cls;
1713   struct GNUNET_SERVICE_Client *client = c->client;
1714   struct Group *grp = c->group;
1715   struct GNUNET_MQ_Envelope *env;
1716
1717   if (NULL == grp)
1718   {
1719     GNUNET_break (0);
1720     GNUNET_SERVICE_client_drop (client);
1721     return;
1722   }
1723   GNUNET_assert (GNUNET_NO == grp->is_disconnected);
1724   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1725               "%p got part request from client for group %s.\n",
1726               grp, GNUNET_h2s (&grp->pub_key_hash));
1727   grp->is_disconnected = GNUNET_YES;
1728   env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_MULTICAST_PART_ACK);
1729   client_send_group (grp, env);
1730   GNUNET_SERVICE_client_continue (client);
1731 }
1732
1733
1734 static int
1735 check_client_multicast_message (void *cls,
1736                                 const struct GNUNET_MULTICAST_MessageHeader *msg)
1737 {
1738   return GNUNET_OK;
1739 }
1740
1741
1742 /**
1743  * Incoming message from a client.
1744  */
1745 static void
1746 handle_client_multicast_message (void *cls,
1747                                  const struct GNUNET_MULTICAST_MessageHeader *msg)
1748 {
1749   // FIXME: what if GNUNET_YES == grp->is_disconnected? Do we allow sending messages?
1750   struct Client *c = cls;
1751   struct GNUNET_SERVICE_Client *client = c->client;
1752   struct Group *grp = c->group;
1753
1754   if (NULL == grp)
1755   {
1756     GNUNET_break (0);
1757     GNUNET_SERVICE_client_drop (client);
1758     return;
1759   }
1760   GNUNET_assert (GNUNET_YES == grp->is_origin);
1761   struct Origin *orig = grp->origin;
1762
1763   // FIXME: use GNUNET_MQ_msg_copy
1764   /* FIXME: yucky, should use separate message structs for P2P and CS! */
1765   struct GNUNET_MULTICAST_MessageHeader *
1766     out = (struct GNUNET_MULTICAST_MessageHeader *) GNUNET_copy_message (&msg->header);
1767   out->fragment_id = GNUNET_htonll (++orig->max_fragment_id);
1768   out->purpose.size = htonl (ntohs (out->header.size)
1769                              - sizeof (out->header)
1770                              - sizeof (out->hop_counter)
1771                              - sizeof (out->signature));
1772   out->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_MESSAGE);
1773
1774   if (GNUNET_OK != GNUNET_CRYPTO_eddsa_sign (&orig->priv_key, &out->purpose,
1775                                              &out->signature))
1776   {
1777     GNUNET_assert (0);
1778   }
1779
1780   client_send_all (&grp->pub_key_hash, GNUNET_MQ_msg_copy (&out->header));
1781   cadet_send_children (&grp->pub_key_hash, &out->header);
1782   client_send_ack (&grp->pub_key_hash);
1783   GNUNET_free (out);
1784
1785   GNUNET_SERVICE_client_continue (client);
1786 }
1787
1788
1789 static int
1790 check_client_multicast_request (void *cls,
1791                                 const struct GNUNET_MULTICAST_RequestHeader *req)
1792 {
1793   return GNUNET_OK;
1794 }
1795
1796
1797 /**
1798  * Incoming request from a client.
1799  */
1800 static void
1801 handle_client_multicast_request (void *cls,
1802                                  const struct GNUNET_MULTICAST_RequestHeader *req)
1803 {
1804   struct Client *c = cls;
1805   struct GNUNET_SERVICE_Client *client = c->client;
1806   struct Group *grp = c->group;
1807
1808   if (NULL == grp)
1809   {
1810     GNUNET_break (0);
1811     GNUNET_SERVICE_client_drop (client);
1812     return;
1813   }
1814   GNUNET_assert (GNUNET_NO == grp->is_disconnected);
1815   GNUNET_assert (GNUNET_NO == grp->is_origin);
1816   struct Member *mem = grp->member;
1817
1818   /* FIXME: yucky, should use separate message structs for P2P and CS! */
1819   struct GNUNET_MULTICAST_RequestHeader *
1820     out = (struct GNUNET_MULTICAST_RequestHeader *) GNUNET_copy_message (&req->header);
1821   out->member_pub_key = mem->pub_key;
1822   out->fragment_id = GNUNET_ntohll (++mem->max_fragment_id);
1823   out->purpose.size = htonl (ntohs (out->header.size)
1824                              - sizeof (out->header)
1825                              - sizeof (out->member_pub_key)
1826                              - sizeof (out->signature));
1827   out->purpose.purpose = htonl (GNUNET_SIGNATURE_PURPOSE_MULTICAST_REQUEST);
1828
1829   if (GNUNET_OK != GNUNET_CRYPTO_ecdsa_sign (&mem->priv_key, &out->purpose,
1830                                              &out->signature))
1831   {
1832     GNUNET_assert (0);
1833   }
1834
1835   uint8_t send_ack = GNUNET_YES;
1836   if (0 ==
1837       client_send_origin (&grp->pub_key_hash,
1838                           GNUNET_MQ_msg_copy (&out->header)))
1839   { /* No local origins, send to remote origin */
1840     if (NULL != mem->origin_channel)
1841     {
1842       cadet_send_channel (mem->origin_channel, &out->header);
1843       send_ack = GNUNET_NO;
1844     }
1845     else
1846     {
1847       /* FIXME: not yet connected to origin */
1848       GNUNET_SERVICE_client_drop (client);
1849       GNUNET_free (out);
1850       return;
1851     }
1852   }
1853   if (GNUNET_YES == send_ack)
1854   {
1855     client_send_ack (&grp->pub_key_hash);
1856   }
1857   GNUNET_free (out);
1858   GNUNET_SERVICE_client_continue (client);
1859 }
1860
1861
1862 /**
1863  * Incoming replay request from a client.
1864  */
1865 static void
1866 handle_client_replay_request (void *cls,
1867                               const struct MulticastReplayRequestMessage *rep)
1868 {
1869   struct Client *c = cls;
1870   struct GNUNET_SERVICE_Client *client = c->client;
1871   struct Group *grp = c->group;
1872
1873   if (NULL == grp)
1874   {
1875     GNUNET_break (0);
1876     GNUNET_SERVICE_client_drop (client);
1877     return;
1878   }
1879   GNUNET_assert (GNUNET_NO == grp->is_disconnected);
1880   GNUNET_assert (GNUNET_NO == grp->is_origin);
1881   struct Member *mem = grp->member;
1882
1883   struct GNUNET_CONTAINER_MultiHashMap *
1884     grp_replay_req = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
1885                                                         &grp->pub_key_hash);
1886   if (NULL == grp_replay_req)
1887   {
1888     grp_replay_req = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
1889     GNUNET_CONTAINER_multihashmap_put (replay_req_client,
1890                                        &grp->pub_key_hash, grp_replay_req,
1891                                        GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
1892   }
1893
1894   struct GNUNET_HashCode key_hash;
1895   replay_key_hash (rep->fragment_id, rep->message_id, rep->fragment_offset,
1896                    rep->flags, &key_hash);
1897   GNUNET_CONTAINER_multihashmap_put (grp_replay_req, &key_hash, client,
1898                                      GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
1899
1900   if (0 ==
1901       client_send_origin (&grp->pub_key_hash,
1902                           GNUNET_MQ_msg_copy (&rep->header)))
1903   { /* No local origin, replay from remote members / origin. */
1904     if (NULL != mem->origin_channel)
1905     {
1906       cadet_send_channel (mem->origin_channel, &rep->header);
1907     }
1908     else
1909     {
1910       /* FIXME: not yet connected to origin */
1911
1912       GNUNET_assert (0);
1913       GNUNET_SERVICE_client_drop (client);
1914       return;
1915     }
1916   }
1917   GNUNET_SERVICE_client_continue (client);
1918 }
1919
1920
1921 static int
1922 cadet_send_replay_response_cb (void *cls,
1923                                const struct GNUNET_HashCode *key_hash,
1924                                void *value)
1925 {
1926   struct Channel *chn = value;
1927   struct GNUNET_MessageHeader *msg = cls;
1928
1929   cadet_send_channel (chn, msg);
1930   return GNUNET_OK;
1931 }
1932
1933
1934 static int
1935 client_send_replay_response_cb (void *cls,
1936                                 const struct GNUNET_HashCode *key_hash,
1937                                 void *value)
1938 {
1939   struct GNUNET_SERVICE_Client *client = value;
1940   struct GNUNET_MessageHeader *msg = cls;
1941
1942   client_send (client, msg);
1943   return GNUNET_OK;
1944 }
1945
1946
1947 static int
1948 check_client_replay_response_end (void *cls,
1949                                   const struct MulticastReplayResponseMessage *res)
1950 {
1951   return GNUNET_OK;
1952 }
1953
1954
1955 /**
1956  * End of replay response from a client.
1957  */
1958 static void
1959 handle_client_replay_response_end (void *cls,
1960                                    const struct MulticastReplayResponseMessage *res)
1961 {
1962   struct Client *c = cls;
1963   struct GNUNET_SERVICE_Client *client = c->client;
1964   struct Group *grp = c->group;
1965
1966   if (NULL == grp)
1967   {
1968     GNUNET_break (0);
1969     GNUNET_SERVICE_client_drop (client);
1970     return;
1971   }
1972   GNUNET_assert (GNUNET_NO == grp->is_disconnected);
1973
1974   struct GNUNET_HashCode key_hash;
1975   replay_key_hash (res->fragment_id, res->message_id, res->fragment_offset,
1976                    res->flags, &key_hash);
1977
1978   struct GNUNET_CONTAINER_MultiHashMap *
1979     grp_replay_req_cadet = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
1980                                                               &grp->pub_key_hash);
1981   if (NULL != grp_replay_req_cadet)
1982   {
1983     GNUNET_CONTAINER_multihashmap_remove_all (grp_replay_req_cadet, &key_hash);
1984   }
1985   struct GNUNET_CONTAINER_MultiHashMap *
1986     grp_replay_req_client = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
1987                                                                &grp->pub_key_hash);
1988   if (NULL != grp_replay_req_client)
1989   {
1990     GNUNET_CONTAINER_multihashmap_remove_all (grp_replay_req_client, &key_hash);
1991   }
1992   GNUNET_SERVICE_client_continue (client);
1993 }
1994
1995
1996 static int
1997 check_client_replay_response (void *cls,
1998                               const struct MulticastReplayResponseMessage *res)
1999 {
2000   const struct GNUNET_MessageHeader *msg;
2001   if (GNUNET_MULTICAST_REC_OK == res->error_code)
2002   {
2003     msg = GNUNET_MQ_extract_nested_mh (res);
2004     if (NULL == msg)
2005     {
2006       return GNUNET_SYSERR;
2007     }
2008   }
2009   return GNUNET_OK;
2010 }
2011
2012
2013 /**
2014  * Incoming replay response from a client.
2015  *
2016  * Respond with a multicast message on success, or otherwise with an error code.
2017  */
2018 static void
2019 handle_client_replay_response (void *cls,
2020                                const struct MulticastReplayResponseMessage *res)
2021 {
2022   struct Client *c = cls;
2023   struct GNUNET_SERVICE_Client *client = c->client;
2024   struct Group *grp = c->group;
2025
2026   if (NULL == grp)
2027   {
2028     GNUNET_break (0);
2029     GNUNET_SERVICE_client_drop (client);
2030     return;
2031   }
2032   GNUNET_assert (GNUNET_NO == grp->is_disconnected);
2033
2034   const struct GNUNET_MessageHeader *msg = &res->header;
2035   if (GNUNET_MULTICAST_REC_OK == res->error_code)
2036   {
2037     msg = GNUNET_MQ_extract_nested_mh (res);
2038   }
2039
2040   struct GNUNET_HashCode key_hash;
2041   replay_key_hash (res->fragment_id, res->message_id, res->fragment_offset,
2042                    res->flags, &key_hash);
2043
2044   struct GNUNET_CONTAINER_MultiHashMap *
2045     grp_replay_req_cadet = GNUNET_CONTAINER_multihashmap_get (replay_req_cadet,
2046                                                               &grp->pub_key_hash);
2047   if (NULL != grp_replay_req_cadet)
2048   {
2049     GNUNET_CONTAINER_multihashmap_get_multiple (grp_replay_req_cadet, &key_hash,
2050                                                 cadet_send_replay_response_cb,
2051                                                 (void *) msg);
2052   }
2053   if (GNUNET_MULTICAST_REC_OK == res->error_code)
2054   {
2055     struct GNUNET_CONTAINER_MultiHashMap *
2056       grp_replay_req_client = GNUNET_CONTAINER_multihashmap_get (replay_req_client,
2057                                                                  &grp->pub_key_hash);
2058     if (NULL != grp_replay_req_client)
2059     {
2060       GNUNET_CONTAINER_multihashmap_get_multiple (grp_replay_req_client, &key_hash,
2061                                                   client_send_replay_response_cb,
2062                                                   (void *) msg);
2063     }
2064   }
2065   else
2066   {
2067     handle_client_replay_response_end (c, res);
2068     return;
2069   }
2070   GNUNET_SERVICE_client_continue (client);
2071 }
2072
2073
2074 /**
2075  * A new client connected.
2076  *
2077  * @param cls NULL
2078  * @param client client to add
2079  * @param mq message queue for @a client
2080  * @return @a client
2081  */
2082 static void *
2083 client_notify_connect (void *cls,
2084                        struct GNUNET_SERVICE_Client *client,
2085                        struct GNUNET_MQ_Handle *mq)
2086 {
2087   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client connected: %p\n", client);
2088   /* FIXME: send connect ACK */
2089
2090   struct Client *c = GNUNET_new (struct Client);
2091   c->client = client;
2092
2093   return c;
2094 }
2095
2096
2097 /**
2098  * Called whenever a client is disconnected.
2099  * Frees our resources associated with that client.
2100  *
2101  * @param cls closure
2102  * @param client identification of the client
2103  * @param app_ctx must match @a client
2104  */
2105 static void
2106 client_notify_disconnect (void *cls,
2107                           struct GNUNET_SERVICE_Client *client,
2108                           void *app_ctx)
2109 {
2110   struct Client *c = app_ctx;
2111   struct Group *grp = c->group;
2112   GNUNET_free (c);
2113
2114   if (NULL == grp)
2115   {
2116     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
2117                 "%p User context is NULL in client_disconnect()\n", grp);
2118     GNUNET_break (0);
2119     return;
2120   }
2121
2122   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2123               "%p Client (%s) disconnected from group %s\n",
2124               grp, (GNUNET_YES == grp->is_origin) ? "origin" : "member",
2125               GNUNET_h2s (&grp->pub_key_hash));
2126
2127   // FIXME (due to protocol change): here we must not remove all clients,
2128   // only the one we were notified about!
2129   struct ClientList *cl = grp->clients_head;
2130   while (NULL != cl)
2131   {
2132     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
2133                 "iterating clients for group %p\n",
2134                 grp);
2135     if (cl->client == client)
2136     {
2137       GNUNET_CONTAINER_DLL_remove (grp->clients_head, grp->clients_tail, cl);
2138       GNUNET_free (cl);
2139       break;
2140     }
2141     cl = cl->next;
2142   }
2143
2144   while (GNUNET_YES == replay_req_remove_client (grp, client));
2145
2146   if (NULL == grp->clients_head)
2147   { /* Last client disconnected. */
2148     cleanup_group (grp);
2149   }
2150 }
2151
2152
2153 /**
2154  * Service started.
2155  *
2156  * @param cls closure
2157  * @param server the initialized server
2158  * @param cfg configuration to use
2159  */
2160 static void
2161 run (void *cls,
2162      const struct GNUNET_CONFIGURATION_Handle *c,
2163      struct GNUNET_SERVICE_Handle *svc)
2164 {
2165   cfg = c;
2166   service = svc;
2167   GNUNET_CRYPTO_get_peer_identity (cfg, &this_peer);
2168
2169   stats = GNUNET_STATISTICS_create ("multicast", cfg);
2170   origins = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
2171   members = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
2172   group_members = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
2173   channels_in = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
2174   channels_out = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
2175   replay_req_cadet = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
2176   replay_req_client = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_NO);
2177
2178   cadet = GNUNET_CADET_connect (cfg);
2179
2180   GNUNET_assert (NULL != cadet);
2181
2182   GNUNET_SCHEDULER_add_shutdown (&shutdown_task,
2183                                  NULL);
2184 }
2185
2186
2187 /**
2188  * Define "main" method using service macro.
2189  */
2190 GNUNET_SERVICE_MAIN
2191 ("multicast",
2192  GNUNET_SERVICE_OPTION_NONE,
2193  &run,
2194  &client_notify_connect,
2195  &client_notify_disconnect,
2196  NULL,
2197  GNUNET_MQ_hd_fixed_size (client_origin_start,
2198                           GNUNET_MESSAGE_TYPE_MULTICAST_ORIGIN_START,
2199                           struct MulticastOriginStartMessage,
2200                           NULL),
2201  GNUNET_MQ_hd_var_size (client_member_join,
2202                         GNUNET_MESSAGE_TYPE_MULTICAST_MEMBER_JOIN,
2203                         struct MulticastMemberJoinMessage,
2204                         NULL),
2205  GNUNET_MQ_hd_var_size (client_join_decision,
2206                         GNUNET_MESSAGE_TYPE_MULTICAST_JOIN_DECISION,
2207                         struct MulticastJoinDecisionMessageHeader,
2208                         NULL),
2209  GNUNET_MQ_hd_fixed_size (client_part_request,
2210                           GNUNET_MESSAGE_TYPE_MULTICAST_PART_REQUEST,
2211                           struct GNUNET_MessageHeader,
2212                           NULL),
2213  GNUNET_MQ_hd_var_size (client_multicast_message,
2214                         GNUNET_MESSAGE_TYPE_MULTICAST_MESSAGE,
2215                         struct GNUNET_MULTICAST_MessageHeader,
2216                         NULL),
2217  GNUNET_MQ_hd_var_size (client_multicast_request,
2218                         GNUNET_MESSAGE_TYPE_MULTICAST_REQUEST,
2219                         struct GNUNET_MULTICAST_RequestHeader,
2220                         NULL),
2221  GNUNET_MQ_hd_fixed_size (client_replay_request,
2222                           GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_REQUEST,
2223                           struct MulticastReplayRequestMessage,
2224                           NULL),
2225  GNUNET_MQ_hd_var_size (client_replay_response,
2226                         GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE,
2227                         struct MulticastReplayResponseMessage,
2228                         NULL),
2229  GNUNET_MQ_hd_var_size (client_replay_response_end,
2230                         GNUNET_MESSAGE_TYPE_MULTICAST_REPLAY_RESPONSE_END,
2231                         struct MulticastReplayResponseMessage,
2232                         NULL));
2233
2234 /* end of gnunet-service-multicast.c */