This source file includes following definitions.
- upcast
- be_underlying_writebuf_full
- be_readbuf_full
- be_null_filter
- bufferevent_filter_new
- be_filter_destruct
- be_filter_enable
- be_filter_disable
- be_filter_process_input
- be_filter_process_output
- bufferevent_filtered_outbuf_cb
- be_filter_readcb
- be_filter_writecb
- be_filter_eventcb
- be_filter_flush
- be_filter_ctrl
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29 #include <sys/types.h>
30
31 #include "event2/event-config.h"
32
33 #ifdef _EVENT_HAVE_SYS_TIME_H
34 #include <sys/time.h>
35 #endif
36
37 #include <errno.h>
38 #include <stdio.h>
39 #include <stdlib.h>
40 #include <string.h>
41 #ifdef _EVENT_HAVE_STDARG_H
42 #include <stdarg.h>
43 #endif
44
45 #ifdef WIN32
46 #include <winsock2.h>
47 #endif
48
49 #include "event2/util.h"
50 #include "event2/bufferevent.h"
51 #include "event2/buffer.h"
52 #include "event2/bufferevent_struct.h"
53 #include "event2/event.h"
54 #include "log-internal.h"
55 #include "mm-internal.h"
56 #include "bufferevent-internal.h"
57 #include "util-internal.h"
58
59
60 static int be_filter_enable(struct bufferevent *, short);
61 static int be_filter_disable(struct bufferevent *, short);
62 static void be_filter_destruct(struct bufferevent *);
63
64 static void be_filter_readcb(struct bufferevent *, void *);
65 static void be_filter_writecb(struct bufferevent *, void *);
66 static void be_filter_eventcb(struct bufferevent *, short, void *);
67 static int be_filter_flush(struct bufferevent *bufev,
68 short iotype, enum bufferevent_flush_mode mode);
69 static int be_filter_ctrl(struct bufferevent *, enum bufferevent_ctrl_op, union bufferevent_ctrl_data *);
70
71 static void bufferevent_filtered_outbuf_cb(struct evbuffer *buf,
72 const struct evbuffer_cb_info *info, void *arg);
73
74 struct bufferevent_filtered {
75 struct bufferevent_private bev;
76
77
78 struct bufferevent *underlying;
79
80 struct evbuffer_cb_entry *outbuf_cb;
81
82
83 unsigned got_eof;
84
85
86 void (*free_context)(void *);
87
88 bufferevent_filter_cb process_in;
89
90 bufferevent_filter_cb process_out;
91
92 void *context;
93 };
94
95 const struct bufferevent_ops bufferevent_ops_filter = {
96 "filter",
97 evutil_offsetof(struct bufferevent_filtered, bev.bev),
98 be_filter_enable,
99 be_filter_disable,
100 be_filter_destruct,
101 _bufferevent_generic_adj_timeouts,
102 be_filter_flush,
103 be_filter_ctrl,
104 };
105
106
107
108 static inline struct bufferevent_filtered *
109 upcast(struct bufferevent *bev)
110 {
111 struct bufferevent_filtered *bev_f;
112 if (bev->be_ops != &bufferevent_ops_filter)
113 return NULL;
114 bev_f = (void*)( ((char*)bev) -
115 evutil_offsetof(struct bufferevent_filtered, bev.bev));
116 EVUTIL_ASSERT(bev_f->bev.bev.be_ops == &bufferevent_ops_filter);
117 return bev_f;
118 }
119
120 #define downcast(bev_f) (&(bev_f)->bev.bev)
121
122
123
124
125 static int
126 be_underlying_writebuf_full(struct bufferevent_filtered *bevf,
127 enum bufferevent_flush_mode state)
128 {
129 struct bufferevent *u = bevf->underlying;
130 return state == BEV_NORMAL &&
131 u->wm_write.high &&
132 evbuffer_get_length(u->output) >= u->wm_write.high;
133 }
134
135
136
137 static int
138 be_readbuf_full(struct bufferevent_filtered *bevf,
139 enum bufferevent_flush_mode state)
140 {
141 struct bufferevent *bufev = downcast(bevf);
142 return state == BEV_NORMAL &&
143 bufev->wm_read.high &&
144 evbuffer_get_length(bufev->input) >= bufev->wm_read.high;
145 }
146
147
148
149 static enum bufferevent_filter_result
150 be_null_filter(struct evbuffer *src, struct evbuffer *dst, ev_ssize_t lim,
151 enum bufferevent_flush_mode state, void *ctx)
152 {
153 (void)state;
154 if (evbuffer_remove_buffer(src, dst, lim) == 0)
155 return BEV_OK;
156 else
157 return BEV_ERROR;
158 }
159
160 struct bufferevent *
161 bufferevent_filter_new(struct bufferevent *underlying,
162 bufferevent_filter_cb input_filter,
163 bufferevent_filter_cb output_filter,
164 int options,
165 void (*free_context)(void *),
166 void *ctx)
167 {
168 struct bufferevent_filtered *bufev_f;
169 int tmp_options = options & ~BEV_OPT_THREADSAFE;
170
171 if (!underlying)
172 return NULL;
173
174 if (!input_filter)
175 input_filter = be_null_filter;
176 if (!output_filter)
177 output_filter = be_null_filter;
178
179 bufev_f = mm_calloc(1, sizeof(struct bufferevent_filtered));
180 if (!bufev_f)
181 return NULL;
182
183 if (bufferevent_init_common(&bufev_f->bev, underlying->ev_base,
184 &bufferevent_ops_filter, tmp_options) < 0) {
185 mm_free(bufev_f);
186 return NULL;
187 }
188 if (options & BEV_OPT_THREADSAFE) {
189 bufferevent_enable_locking(downcast(bufev_f), NULL);
190 }
191
192 bufev_f->underlying = underlying;
193
194 bufev_f->process_in = input_filter;
195 bufev_f->process_out = output_filter;
196 bufev_f->free_context = free_context;
197 bufev_f->context = ctx;
198
199 bufferevent_setcb(bufev_f->underlying,
200 be_filter_readcb, be_filter_writecb, be_filter_eventcb, bufev_f);
201
202 bufev_f->outbuf_cb = evbuffer_add_cb(downcast(bufev_f)->output,
203 bufferevent_filtered_outbuf_cb, bufev_f);
204
205 _bufferevent_init_generic_timeout_cbs(downcast(bufev_f));
206 bufferevent_incref(underlying);
207
208 bufferevent_enable(underlying, EV_READ|EV_WRITE);
209 bufferevent_suspend_read(underlying, BEV_SUSPEND_FILT_READ);
210
211 return downcast(bufev_f);
212 }
213
214 static void
215 be_filter_destruct(struct bufferevent *bev)
216 {
217 struct bufferevent_filtered *bevf = upcast(bev);
218 EVUTIL_ASSERT(bevf);
219 if (bevf->free_context)
220 bevf->free_context(bevf->context);
221
222 if (bevf->bev.options & BEV_OPT_CLOSE_ON_FREE) {
223
224
225
226
227
228 if (BEV_UPCAST(bevf->underlying)->refcnt < 2) {
229 event_warnx("BEV_OPT_CLOSE_ON_FREE set on an "
230 "bufferevent with too few references");
231 } else {
232 bufferevent_free(bevf->underlying);
233 }
234 } else {
235 if (bevf->underlying) {
236 if (bevf->underlying->errorcb == be_filter_eventcb)
237 bufferevent_setcb(bevf->underlying,
238 NULL, NULL, NULL, NULL);
239 bufferevent_unsuspend_read(bevf->underlying,
240 BEV_SUSPEND_FILT_READ);
241 }
242 }
243
244 _bufferevent_del_generic_timeout_cbs(bev);
245 }
246
247 static int
248 be_filter_enable(struct bufferevent *bev, short event)
249 {
250 struct bufferevent_filtered *bevf = upcast(bev);
251 if (event & EV_WRITE)
252 BEV_RESET_GENERIC_WRITE_TIMEOUT(bev);
253
254 if (event & EV_READ) {
255 BEV_RESET_GENERIC_READ_TIMEOUT(bev);
256 bufferevent_unsuspend_read(bevf->underlying,
257 BEV_SUSPEND_FILT_READ);
258 }
259 return 0;
260 }
261
262 static int
263 be_filter_disable(struct bufferevent *bev, short event)
264 {
265 struct bufferevent_filtered *bevf = upcast(bev);
266 if (event & EV_WRITE)
267 BEV_DEL_GENERIC_WRITE_TIMEOUT(bev);
268 if (event & EV_READ) {
269 BEV_DEL_GENERIC_READ_TIMEOUT(bev);
270 bufferevent_suspend_read(bevf->underlying,
271 BEV_SUSPEND_FILT_READ);
272 }
273 return 0;
274 }
275
276 static enum bufferevent_filter_result
277 be_filter_process_input(struct bufferevent_filtered *bevf,
278 enum bufferevent_flush_mode state,
279 int *processed_out)
280 {
281 enum bufferevent_filter_result res;
282 struct bufferevent *bev = downcast(bevf);
283
284 if (state == BEV_NORMAL) {
285
286
287 if (!(bev->enabled & EV_READ) ||
288 be_readbuf_full(bevf, state))
289 return BEV_OK;
290 }
291
292 do {
293 ev_ssize_t limit = -1;
294 if (state == BEV_NORMAL && bev->wm_read.high)
295 limit = bev->wm_read.high -
296 evbuffer_get_length(bev->input);
297
298 res = bevf->process_in(bevf->underlying->input,
299 bev->input, limit, state, bevf->context);
300
301 if (res == BEV_OK)
302 *processed_out = 1;
303 } while (res == BEV_OK &&
304 (bev->enabled & EV_READ) &&
305 evbuffer_get_length(bevf->underlying->input) &&
306 !be_readbuf_full(bevf, state));
307
308 if (*processed_out)
309 BEV_RESET_GENERIC_READ_TIMEOUT(bev);
310
311 return res;
312 }
313
314
315 static enum bufferevent_filter_result
316 be_filter_process_output(struct bufferevent_filtered *bevf,
317 enum bufferevent_flush_mode state,
318 int *processed_out)
319 {
320
321 enum bufferevent_filter_result res = BEV_OK;
322 struct bufferevent *bufev = downcast(bevf);
323 int again = 0;
324
325 if (state == BEV_NORMAL) {
326
327
328
329
330
331 if (!(bufev->enabled & EV_WRITE) ||
332 be_underlying_writebuf_full(bevf, state) ||
333 !evbuffer_get_length(bufev->output))
334 return BEV_OK;
335 }
336
337
338
339 evbuffer_cb_set_flags(bufev->output, bevf->outbuf_cb, 0);
340
341 do {
342 int processed = 0;
343 again = 0;
344
345 do {
346 ev_ssize_t limit = -1;
347 if (state == BEV_NORMAL &&
348 bevf->underlying->wm_write.high)
349 limit = bevf->underlying->wm_write.high -
350 evbuffer_get_length(bevf->underlying->output);
351
352 res = bevf->process_out(downcast(bevf)->output,
353 bevf->underlying->output,
354 limit,
355 state,
356 bevf->context);
357
358 if (res == BEV_OK)
359 processed = *processed_out = 1;
360 } while (
361 res == BEV_OK &&
362
363 (bufev->enabled & EV_WRITE) &&
364
365
366 evbuffer_get_length(bufev->output) &&
367
368 !be_underlying_writebuf_full(bevf,state));
369
370 if (processed &&
371 evbuffer_get_length(bufev->output) <= bufev->wm_write.low) {
372
373 _bufferevent_run_writecb(bufev);
374
375 if (res == BEV_OK &&
376 (bufev->enabled & EV_WRITE) &&
377 evbuffer_get_length(bufev->output) &&
378 !be_underlying_writebuf_full(bevf, state)) {
379 again = 1;
380 }
381 }
382 } while (again);
383
384
385 evbuffer_cb_set_flags(bufev->output,bevf->outbuf_cb,
386 EVBUFFER_CB_ENABLED);
387
388 if (*processed_out)
389 BEV_RESET_GENERIC_WRITE_TIMEOUT(bufev);
390
391 return res;
392 }
393
394
395 static void
396 bufferevent_filtered_outbuf_cb(struct evbuffer *buf,
397 const struct evbuffer_cb_info *cbinfo, void *arg)
398 {
399 struct bufferevent_filtered *bevf = arg;
400 struct bufferevent *bev = downcast(bevf);
401
402 if (cbinfo->n_added) {
403 int processed_any = 0;
404
405
406 _bufferevent_incref_and_lock(bev);
407 be_filter_process_output(bevf, BEV_NORMAL, &processed_any);
408 _bufferevent_decref_and_unlock(bev);
409 }
410 }
411
412
413 static void
414 be_filter_readcb(struct bufferevent *underlying, void *_me)
415 {
416 struct bufferevent_filtered *bevf = _me;
417 enum bufferevent_filter_result res;
418 enum bufferevent_flush_mode state;
419 struct bufferevent *bufev = downcast(bevf);
420 int processed_any = 0;
421
422 _bufferevent_incref_and_lock(bufev);
423
424 if (bevf->got_eof)
425 state = BEV_FINISHED;
426 else
427 state = BEV_NORMAL;
428
429
430 res = be_filter_process_input(bevf, state, &processed_any);
431 (void)res;
432
433
434
435
436 if (processed_any &&
437 evbuffer_get_length(bufev->input) >= bufev->wm_read.low)
438 _bufferevent_run_readcb(bufev);
439
440 _bufferevent_decref_and_unlock(bufev);
441 }
442
443
444
445 static void
446 be_filter_writecb(struct bufferevent *underlying, void *_me)
447 {
448 struct bufferevent_filtered *bevf = _me;
449 struct bufferevent *bev = downcast(bevf);
450 int processed_any = 0;
451
452 _bufferevent_incref_and_lock(bev);
453 be_filter_process_output(bevf, BEV_NORMAL, &processed_any);
454 _bufferevent_decref_and_unlock(bev);
455 }
456
457
458 static void
459 be_filter_eventcb(struct bufferevent *underlying, short what, void *_me)
460 {
461 struct bufferevent_filtered *bevf = _me;
462 struct bufferevent *bev = downcast(bevf);
463
464 _bufferevent_incref_and_lock(bev);
465
466 _bufferevent_run_eventcb(bev, what);
467 _bufferevent_decref_and_unlock(bev);
468 }
469
470 static int
471 be_filter_flush(struct bufferevent *bufev,
472 short iotype, enum bufferevent_flush_mode mode)
473 {
474 struct bufferevent_filtered *bevf = upcast(bufev);
475 int processed_any = 0;
476 EVUTIL_ASSERT(bevf);
477
478 _bufferevent_incref_and_lock(bufev);
479
480 if (iotype & EV_READ) {
481 be_filter_process_input(bevf, mode, &processed_any);
482 }
483 if (iotype & EV_WRITE) {
484 be_filter_process_output(bevf, mode, &processed_any);
485 }
486
487
488 bufferevent_flush(bevf->underlying, iotype, mode);
489
490 _bufferevent_decref_and_unlock(bufev);
491
492 return processed_any;
493 }
494
495 static int
496 be_filter_ctrl(struct bufferevent *bev, enum bufferevent_ctrl_op op,
497 union bufferevent_ctrl_data *data)
498 {
499 struct bufferevent_filtered *bevf;
500 switch (op) {
501 case BEV_CTRL_GET_UNDERLYING:
502 bevf = upcast(bev);
503 data->ptr = bevf->underlying;
504 return 0;
505 case BEV_CTRL_GET_FD:
506 case BEV_CTRL_SET_FD:
507 case BEV_CTRL_CANCEL_ALL:
508 default:
509 return -1;
510 }
511 }