libctdb/messages.c

   1 #include "libctdb_private.h"
   2 #include "messages.h"
   3 #include "io_elem.h"
   4 #include <ctdb.h>
   5 #include <tdb.h>
   6 #include <ctdb_protocol.h>
   7 #include <stdlib.h>
   8 #include <string.h>
   9
  10 struct message_handler_info {
  11         struct message_handler_info *next, *prev;
  12         /* Callback when we're first registered. */
  13         ctdb_set_message_handler_cb callback;
  14
  15         uint64_t srvid;
  16         ctdb_message_fn_t handler;
  17         void *private_data;
  18         struct ctdb_connection *ctdb;
  19 };
  20
  21 void deliver_message(struct ctdb_connection *ctdb, struct ctdb_req_header *hdr)
  22 {
  23         struct message_handler_info *i;
  24         struct ctdb_req_message *msg = (struct ctdb_req_message *)hdr;
  25         TDB_DATA data;
  26
  27         data.dptr = msg->data;
  28         data.dsize = msg->datalen;
  29
  30         for (i = ctdb->message_handlers; i; i = i->next) {
  31                 if (i->srvid == msg->srvid) {
  32                         i->handler(ctdb, msg->srvid, data, i->private_data);
  33                 }
  34         }
  35         /* FIXME: Report unknown messages */
  36 }
  37
  38 static void set_message_handler(int status, struct message_handler_info *info)
  39 {
  40         /* If registration failed, tell callback and clean up */
  41         if (status < 0) {
  42                 info->callback(status, info->private_data);
  43                 free(info);
  44                 return;
  45         } else {
  46                 /* Put ourselves in list of handlers. */
  47                 DLIST_ADD_END(info->ctdb->message_handlers, info,
  48                               struct message_handler_info);
  49                 /* Now call callback: it could remove us in theory. */
  50                 info->callback(status, info->private_data);
  51         }
  52 }
  53
  54 struct ctdb_request *
  55 ctdb_set_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
  56                               ctdb_set_message_handler_cb callback,
  57                               ctdb_message_fn_t handler, void *private_data)
  58 {
  59         struct ctdb_request *req;
  60         struct message_handler_info *info;
  61
  62         info = malloc(sizeof(*info));
  63         if (!info) {
  64                 return NULL;
  65         }
  66         req = new_ctdb_control_request(ctdb, CTDB_CONTROL_REGISTER_SRVID,
  67                                        CTDB_CURRENT_NODE, NULL, 0);
  68         if (!req) {
  69                 free(info);
  70                 return NULL;
  71         }
  72         req->hdr.control->srvid = srvid;
  73
  74         info->srvid = srvid;
  75         info->handler = handler;
  76         info->callback = callback;
  77         info->private_data = private_data;
  78         info->ctdb = ctdb;
  79
  80         req->callback.register_srvid = set_message_handler;
  81         req->priv_data = info;
  82
  83         return req;
  84 }
  85
  86 int ctdb_send_message(struct ctdb_connection *ctdb,
  87                       uint32_t pnn, uint64_t srvid,
  88                       TDB_DATA data)
  89 {
  90         struct ctdb_request *req;
  91         struct ctdb_req_message *pkt;
  92
  93         req = new_ctdb_request(sizeof(*pkt) + data.dsize);
  94         if (!req) {
  95                 return -1;
  96         }
  97
  98         io_elem_init_req_header(req->io,
  99                                 CTDB_REQ_MESSAGE, pnn, new_reqid(ctdb));
 100
 101         /* There's no reply to this, so we mark it cancelled immediately. */
 102         req->cancelled = true;
 103
 104         pkt = req->hdr.message;
 105         pkt->srvid = srvid;
 106         pkt->datalen = data.dsize;
 107         memcpy(pkt->data, data.dptr, data.dsize);
 108         DLIST_ADD_END(ctdb->outq, req, struct ctdb_request);
 109         return 0;
 110 }