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 /* Remove type-safety macros. */
  11 #undef ctdb_set_message_handler_send
  12 #undef ctdb_set_message_handler_recv
  13 #undef ctdb_remove_message_handler_send
  14
  15 struct message_handler_info {
  16         struct message_handler_info *next, *prev;
  17
  18         uint64_t srvid;
  19         ctdb_message_fn_t handler;
  20         void *private_data;
  21 };
  22
  23 void deliver_message(struct ctdb_connection *ctdb, struct ctdb_req_header *hdr)
  24 {
  25         struct message_handler_info *i;
  26         struct ctdb_req_message *msg = (struct ctdb_req_message *)hdr;
  27         TDB_DATA data;
  28
  29         data.dptr = msg->data;
  30         data.dsize = msg->datalen;
  31
  32         for (i = ctdb->message_handlers; i; i = i->next) {
  33                 if (i->srvid == msg->srvid) {
  34                         i->handler(ctdb, msg->srvid, data, i->private_data);
  35                 }
  36         }
  37         /* FIXME: Report unknown messages */
  38 }
  39
  40 int ctdb_set_message_handler_recv(struct ctdb_connection *ctdb,
  41                                   struct ctdb_request *req)
  42 {
  43         struct message_handler_info *info = req->extra;
  44         struct ctdb_reply_control *reply;
  45
  46         reply = unpack_reply_control(ctdb, req, CTDB_CONTROL_REGISTER_SRVID);
  47         if (!reply || reply->status != 0) {
  48                 return -1;
  49         }
  50
  51         /* Put ourselves in list of handlers. */
  52         DLIST_ADD(ctdb->message_handlers, info);
  53         /* Keep safe from destructor */
  54         req->extra = NULL;
  55         return 0;
  56 }
  57
  58 static void free_info(struct ctdb_connection *ctdb, struct ctdb_request *req)
  59 {
  60         free(req->extra);
  61 }
  62
  63 struct ctdb_request *
  64 ctdb_set_message_handler_send(struct ctdb_connection *ctdb, uint64_t srvid,
  65                               ctdb_message_fn_t handler,
  66                               ctdb_callback_t callback, void *private_data)
  67 {
  68         struct message_handler_info *info;
  69         struct ctdb_request *req;
  70
  71         info = malloc(sizeof(*info));
  72         if (!info) {
  73                 return NULL;
  74         }
  75
  76         req = new_ctdb_control_request(ctdb, CTDB_CONTROL_REGISTER_SRVID,
  77                                        CTDB_CURRENT_NODE, NULL, 0,
  78                                        callback, private_data);
  79         if (!req) {
  80                 free(info);
  81                 return NULL;
  82         }
  83         req->extra = info;
  84         req->extra_destructor = free_info;
  85         req->hdr.control->srvid = srvid;
  86
  87         info->srvid = srvid;
  88         info->handler = handler;
  89         info->private_data = private_data;
  90
  91         return req;
  92 }
  93
  94 int ctdb_send_message(struct ctdb_connection *ctdb,
  95                       uint32_t pnn, uint64_t srvid,
  96                       TDB_DATA data)
  97 {
  98         struct ctdb_request *req;
  99         struct ctdb_req_message *pkt;
 100
 101         /* We just discard it once it's finished: no reply. */
 102         req = new_ctdb_request(offsetof(struct ctdb_req_message, data) + data.dsize,
 103                                ctdb_cancel_callback, NULL);
 104         if (!req) {
 105                 return -1;
 106         }
 107
 108         io_elem_init_req_header(req->io,
 109                                 CTDB_REQ_MESSAGE, pnn, new_reqid(ctdb));
 110
 111         pkt = req->hdr.message;
 112         pkt->srvid = srvid;
 113         pkt->datalen = data.dsize;
 114         memcpy(pkt->data, data.dptr, data.dsize);
 115         DLIST_ADD_END(ctdb->outq, req, struct ctdb_request);
 116         return 0;
 117 }