source3/torture/msg_sink.c

   1 /*
   2  *  Unix SMB/CIFS implementation.
   3  *  Receive and count messages
   4  *  Copyright (C) Volker Lendecke 2014
   5  *
   6  *  This program is free software; you can redistribute it and/or modify
   7  *  it under the terms of the GNU General Public License as published by
   8  *  the Free Software Foundation; either version 3 of the License, or
   9  *  (at your option) any later version.
  10  *
  11  *  This program is distributed in the hope that it will be useful,
  12  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  *  GNU General Public License for more details.
  15  *
  16  *  You should have received a copy of the GNU General Public License
  17  *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18  */
  19
  20 #include "replace.h"
  21 #include "includes.h"
  22 #include "lib/util/server_id.h"
  23 #include "messages.h"
  24 #include "lib/util/tevent_unix.h"
  25 #include <stdio.h>
  26
  27 struct sink_state {
  28         struct tevent_context *ev;
  29         struct messaging_context *msg_ctx;
  30         int msg_type;
  31         unsigned *counter;
  32 };
  33
  34 static void sink_done(struct tevent_req *subreq);
  35
  36 static struct tevent_req *sink_send(TALLOC_CTX *mem_ctx,
  37                                     struct tevent_context *ev,
  38                                     struct messaging_context *msg_ctx,
  39                                     int msg_type, unsigned *counter)
  40 {
  41         struct tevent_req *req, *subreq;
  42         struct sink_state *state;
  43
  44         req = tevent_req_create(mem_ctx, &state, struct sink_state);
  45         if (req == NULL) {
  46                 return NULL;
  47         }
  48         state->ev = ev;
  49         state->msg_ctx = msg_ctx;
  50         state->msg_type = msg_type;
  51         state->counter = counter;
  52
  53         subreq = messaging_read_send(state, state->ev, state->msg_ctx,
  54                                      state->msg_type);
  55         if (tevent_req_nomem(subreq, req)) {
  56                 return tevent_req_post(req, ev);
  57         }
  58         tevent_req_set_callback(subreq, sink_done, req);
  59         return req;
  60 }
  61
  62 static void sink_done(struct tevent_req *subreq)
  63 {
  64         struct tevent_req *req = tevent_req_callback_data(
  65                 subreq, struct tevent_req);
  66         struct sink_state *state = tevent_req_data(
  67                 req, struct sink_state);
  68         int ret;
  69
  70         ret = messaging_read_recv(subreq, NULL, NULL);
  71         TALLOC_FREE(subreq);
  72         if (tevent_req_error(req, ret)) {
  73                 return;
  74         }
  75
  76         *state->counter += 1;
  77
  78         subreq = messaging_read_send(state, state->ev, state->msg_ctx,
  79                                      state->msg_type);
  80         if (tevent_req_nomem(subreq, req)) {
  81                 return;
  82         }
  83         tevent_req_set_callback(subreq, sink_done, req);
  84 }
  85
  86 static int sink_recv(struct tevent_req *req)
  87 {
  88         int err;
  89
  90         if (tevent_req_is_unix_error(req, &err)) {
  91                 return err;
  92         }
  93         return 0;
  94 }
  95
  96 struct prcount_state {
  97         struct tevent_context *ev;
  98         struct timeval interval;
  99         unsigned *counter;
 100 };
 101
 102 static void prcount_waited(struct tevent_req *subreq);
 103
 104 static struct tevent_req *prcount_send(TALLOC_CTX *mem_ctx,
 105                                        struct tevent_context *ev,
 106                                        struct timeval interval,
 107                                        unsigned *counter)
 108 {
 109         struct tevent_req *req, *subreq;
 110         struct prcount_state *state;
 111
 112         req = tevent_req_create(mem_ctx, &state, struct prcount_state);
 113         if (req == NULL) {
 114                 return NULL;
 115         }
 116         state->ev = ev;
 117         state->interval = interval;
 118         state->counter = counter;
 119
 120         subreq = tevent_wakeup_send(
 121                 state, state->ev,
 122                 timeval_current_ofs(state->interval.tv_sec,
 123                                     state->interval.tv_usec));
 124         if (tevent_req_nomem(subreq, req)) {
 125                 return tevent_req_post(req, ev);
 126         }
 127         tevent_req_set_callback(subreq, prcount_waited, req);
 128         return req;
 129 }
 130
 131 static void prcount_waited(struct tevent_req *subreq)
 132 {
 133         struct tevent_req *req = tevent_req_callback_data(
 134                 subreq, struct tevent_req);
 135         struct prcount_state *state = tevent_req_data(
 136                 req, struct prcount_state);
 137         bool ok;
 138
 139         ok = tevent_wakeup_recv(subreq);
 140         TALLOC_FREE(subreq);
 141         if (!ok) {
 142                 tevent_req_error(req, ENOMEM);
 143                 return;
 144         }
 145
 146         printf("%u\n", *state->counter);
 147
 148         subreq = tevent_wakeup_send(
 149                 state, state->ev,
 150                 timeval_current_ofs(state->interval.tv_sec,
 151                                     state->interval.tv_usec));
 152         if (tevent_req_nomem(subreq, req)) {
 153                 return;
 154         }
 155         tevent_req_set_callback(subreq, prcount_waited, req);
 156 }
 157
 158 static int prcount_recv(struct tevent_req *req)
 159 {
 160         int err;
 161
 162         if (tevent_req_is_unix_error(req, &err)) {
 163                 return err;
 164         }
 165         return 0;
 166 }
 167
 168 struct msgcount_state {
 169         unsigned count;
 170 };
 171
 172 static void msgcount_sunk(struct tevent_req *subreq);
 173 static void msgcount_printed(struct tevent_req *subreq);
 174
 175 static struct tevent_req *msgcount_send(TALLOC_CTX *mem_ctx,
 176                                         struct tevent_context *ev,
 177                                         struct messaging_context *msg_ctx,
 178                                         int msg_type, struct timeval interval)
 179 {
 180         struct tevent_req *req, *subreq;
 181         struct msgcount_state *state;
 182
 183         req = tevent_req_create(mem_ctx, &state, struct msgcount_state);
 184         if (req == NULL) {
 185                 return NULL;
 186         }
 187
 188         subreq = sink_send(state, ev, msg_ctx, msg_type, &state->count);
 189         if (tevent_req_nomem(subreq, req)) {
 190                 return tevent_req_post(req, ev);
 191         }
 192         tevent_req_set_callback(subreq, msgcount_sunk, req);
 193
 194         subreq = prcount_send(state, ev, interval, &state->count);
 195         if (tevent_req_nomem(subreq, req)) {
 196                 return tevent_req_post(req, ev);
 197         }
 198         tevent_req_set_callback(subreq, msgcount_printed, req);
 199
 200         return req;
 201 }
 202
 203 static void msgcount_sunk(struct tevent_req *subreq)
 204 {
 205         struct tevent_req *req = tevent_req_callback_data(
 206                 subreq, struct tevent_req);
 207         int ret;
 208
 209         ret = sink_recv(subreq);
 210         TALLOC_FREE(subreq);
 211         if (tevent_req_error(req, ret)) {
 212                 return;
 213         }
 214         tevent_req_done(req);
 215 }
 216
 217 static void msgcount_printed(struct tevent_req *subreq)
 218 {
 219         struct tevent_req *req = tevent_req_callback_data(
 220                 subreq, struct tevent_req);
 221         int ret;
 222
 223         ret = prcount_recv(subreq);
 224         TALLOC_FREE(subreq);
 225         if (tevent_req_error(req, ret)) {
 226                 return;
 227         }
 228         tevent_req_done(req);
 229 }
 230
 231 static int msgcount_recv(struct tevent_req *req)
 232 {
 233         int err;
 234
 235         if (tevent_req_is_unix_error(req, &err)) {
 236                 return err;
 237         }
 238         return 0;
 239 }
 240
 241 int main(void)
 242 {
 243         TALLOC_CTX *frame = talloc_stackframe();
 244         struct tevent_context *ev;
 245         struct messaging_context *msg_ctx;
 246         struct tevent_req *req;
 247         int ret;
 248         struct server_id id;
 249         struct server_id_buf tmp;
 250
 251         lp_load_global(get_dyn_CONFIGFILE());
 252
 253         ev = tevent_context_init(frame);
 254         if (ev == NULL) {
 255                 perror("tevent_context_init failed");
 256                 return -1;
 257         }
 258
 259         msg_ctx = messaging_init(ev, ev);
 260         if (msg_ctx == NULL) {
 261                 perror("messaging_init failed");
 262                 return -1;
 263         }
 264
 265         id = messaging_server_id(msg_ctx);
 266
 267         printf("server_id: %s\n", server_id_str_buf(id, &tmp));
 268
 269         req = msgcount_send(ev, ev, msg_ctx, MSG_SMB_NOTIFY,
 270                             timeval_set(1, 0));
 271         if (req == NULL) {
 272                 perror("msgcount_send failed");
 273                 return -1;
 274         }
 275
 276         if (!tevent_req_poll(req, ev)) {
 277                 perror("tevent_req_poll failed");
 278                 return -1;
 279         }
 280
 281         ret = msgcount_recv(req);
 282         printf("msgcount_recv returned %d\n", ret);
 283
 284         return 0;
 285 }