fwp/mngr/fwp_mngr.c

   1 #include <fwp.h>
   2 #include "fwp_participant_table.h"
   3
   4 #define FWP_MTU         2346
   5 #define BUFFSIZE        FWP_MTU
   6
   7 /* buffer and socket for incomming message */
   8 static unsigned char    buffer[FWP_MTU];
   9
  10 /* Admission control test */
  11 //fwp_admctrl_test_t fwp_admctrl_test = &fwp_admctrl_stupid;
  12
  13 /**
  14  * fwp_mngt_input
  15  *
  16  * Function waits for remote or local message
  17  *
  18  * @msgb  received message
  19  * \return
  20  * On success, it returns 0 and the pointer to received message in msgb parameter.
  21  * On error, it returns negative error code
  22  *
  23  */
  24 int fwp_mngr_input(struct fwp_msgb **pmsgb)
  25 {
  26         struct fwp_msgb *msgb;
  27         ssize_t size;
  28
  29         FWP_DEBUG("Waiting for messages\n");
  30         size = fwp_recv(fwp_mngt_repointd, buffer, BUFFSIZE);
  31
  32         FWP_DEBUG("Creating fwp msgb len=%d\n", size);
  33         /* For future: fwp_socket could be allocated behind data in msgb*/
  34         if (!(msgb = fwp_msgb_alloc(size))) {
  35                 perror("No memory available.\n");
  36                 return -ENOMEM;
  37         }
  38         /*memcpy(fwp_msgb_put(msgb, len), buffer, len); */
  39         msgb->data = buffer;
  40         fwp_msgb_put(msgb, size);
  41
  42         *pmsgb = msgb;
  43         return (0);
  44 }
  45
  46 void fwp_mngr_hello(fwp_msgb_t *msgb, fwp_participant_id_t participant_id)
  47 {
  48         fwp_participant_info_t participant_info, my_info;
  49         fwp_participant_t *participant;
  50
  51         /* Create a new participant */
  52         fwp_msg_hello_inflate(msgb->data, &participant_info);
  53         participant = fwp_participant_create(&participant_info);
  54         fwp_mngt_service_vres_create(&participant->vresd);
  55         fwp_send_endpoint_create(participant->id.node_id, participant->stream_id,
  56                                         0, &participant->epointd);
  57         fwp_send_endpoint_bind(participant->epointd, participant->vresd);
  58         fwp_contract_table_init(&participant->contract_table);
  59
  60         /* Insert participant into table */
  61         fwp_participant_table_insert(participant);
  62
  63         /* Send back hello msg with mngr`s info */
  64         fwp_msgb_reset_data(msgb);
  65         my_info.id = fwp_participant_this->id;
  66         my_info.stream_id = fwp_participant_this->stream_id;
  67         fwp_msg_header_deflate(msgb->tail, FWP_MSG_HELLO,
  68                                 fwp_participant_this->id);
  69         fwp_msgb_put(msgb, sizeof(struct fwp_msg_header));
  70         fwp_msg_hello_deflate(msgb->tail, &my_info);
  71         fwp_msgb_put(msgb, sizeof(struct fwp_msg_hello));
  72         fwp_send(participant->epointd, msgb->data, msgb->len);
  73 }
  74
  75 /*void fwp_mngr_negt_request(msgb, participant_id)
  76 {
  77
  78
  79 }*/
  80
  81 void fwp_mngr_msg_handler(fwp_msgb_t *msgb)
  82 {
  83         unsigned int msg_type;
  84         fwp_participant_id_t    participant_id;
  85
  86         fwp_msg_header_inflate(msgb->data, &msg_type, &participant_id);
  87         fwp_msgb_pull(msgb, sizeof(struct fwp_msg_header));
  88         FWP_DEBUG("type = %d , nodeid = %d appid = %d\n", msg_type,
  89                         participant_id.node_id, participant_id.app_id);
  90
  91         switch (msg_type) {
  92                 case  FWP_MSG_HELLO:
  93                         fwp_mngr_hello(msgb, participant_id);
  94                         break;
  95 #if 0
  96                 case  FWP_CONTNEGT_REQ:
  97                         FWP_DEBUG("Negotiation Request received\n");
  98                         fwp_mngr_negt_request(msgb, participant_id);
  99                         break;
 100 #endif
 101                 default:
 102                         printf("Invalid message\n.");
 103                         fwp_msgb_free(msgb);
 104         }
 105
 106
 107 }
 108
 109 void fwp_mngr_main_loop()
 110 {
 111         struct fwp_msgb *msgb;
 112
 113         /* start admission control thread */
 114         while (1 /*exit_flag*/){
 115                 fwp_mngr_input(&msgb);
 116                 if (msgb)
 117                         fwp_mngr_msg_handler(msgb);
 118                 FWP_DEBUG("Mngr waiting for next msg.\n");
 119         }
 120 }
 121
 122 int main()
 123 {
 124         if (fwp_init() < 0) {
 125                 fprintf(stderr,"FWP initialization failed.\n");
 126                 exit(1);
 127         }
 128
 129         /* After initialization, destroy fwp_mngt_repointd and create the new one
 130          * with specified parameters */
 131
 132         FWP_DEBUG("Management receive endpoint created\n");
 133         fwp_receive_endpoint_create(FWP_MNGR_STREAM_ID, 0, &fwp_mngt_repointd);
 134
 135         fwp_mngr_main_loop();
 136
 137         return 0;
 138 }