Mercurial > dovecot > core-2.2
annotate src/doveadm/dsync/dsync-slave-io.c @ 14691:3945a3646c67
Changed i_close_fd() API to set the fd to -1 after closing.
author | Timo Sirainen <tss@iki.fi> |
---|---|
date | Thu, 28 Jun 2012 00:27:13 +0300 |
parents | 7c058aa05b0a |
children | 0af20585964d |
rev | line source |
---|---|
14584 | 1 /* Copyright (c) 2012 Dovecot authors, see the included COPYING file */ |
2 | |
3 #include "lib.h" | |
4 #include "array.h" | |
5 #include "fd-set-nonblock.h" | |
6 #include "safe-mkstemp.h" | |
7 #include "ioloop.h" | |
8 #include "istream.h" | |
9 #include "istream-seekable.h" | |
10 #include "istream-dot.h" | |
11 #include "ostream.h" | |
12 #include "str.h" | |
13 #include "strescape.h" | |
14 #include "master-service.h" | |
15 #include "mail-cache.h" | |
16 #include "mail-storage-private.h" | |
17 #include "dsync-serializer.h" | |
18 #include "dsync-deserializer.h" | |
19 #include "dsync-mail.h" | |
20 #include "dsync-mailbox.h" | |
21 #include "dsync-mailbox-state.h" | |
22 #include "dsync-mailbox-tree.h" | |
23 #include "dsync-slave-private.h" | |
24 | |
25 #include <stdlib.h> | |
26 | |
27 #define DSYNC_SLAVE_IO_TIMEOUT_MSECS (60*10*1000) | |
28 #define DSYNC_SLAVE_IO_OUTBUF_THROTTLE_SIZE (1024*128) | |
29 | |
30 #define DSYNC_PROTOCOL_VERSION_MAJOR 3 | |
31 #define DSYNC_HANDSHAKE_VERSION "VERSION\tdsync\t3\t0\n" | |
32 | |
33 enum item_type { | |
34 ITEM_NONE, | |
35 | |
36 ITEM_HANDSHAKE, | |
37 ITEM_MAILBOX_STATE, | |
38 ITEM_MAILBOX_TREE_NODE, | |
39 ITEM_MAILBOX_DELETE, | |
40 ITEM_MAILBOX, | |
41 | |
42 ITEM_MAIL_CHANGE, | |
43 ITEM_MAIL_REQUEST, | |
44 ITEM_MAIL, | |
45 | |
46 ITEM_MAILBOX_CACHE_FIELD, | |
47 | |
48 ITEM_END_OF_LIST | |
49 }; | |
50 | |
51 #define END_OF_LIST_LINE "." | |
52 static const struct { | |
53 /* full human readable name of the item */ | |
54 const char *name; | |
55 /* unique character identifying the item */ | |
56 char chr; | |
57 const char *required_keys; | |
58 const char *optional_keys; | |
59 } items[ITEM_END_OF_LIST+1] = { | |
60 { NULL, '\0', NULL, NULL }, | |
61 { .name = "handshake", | |
62 .chr = 'H', | |
63 .optional_keys = "sync_ns_prefix sync_type " | |
64 "guid_requests mails_have_guids" | |
65 }, | |
66 { .name = "mailbox_state", | |
67 .chr = 'S', | |
68 .required_keys = "mailbox_guid last_uidvalidity last_common_uid " | |
69 "last_common_modseq" | |
70 }, | |
71 { .name = "mailbox_tree_node", | |
72 .chr = 'N', | |
73 .required_keys = "name existence", | |
74 .optional_keys = "mailbox_guid uid_validity " | |
75 "last_renamed subscribed last_subscription_change" | |
76 }, | |
77 { .name = "mailbox_delete", | |
78 .chr = 'D', | |
79 .required_keys = "hierarchy_sep", | |
80 .optional_keys = "mailboxes dirs" | |
81 }, | |
82 { .name = "mailbox", | |
83 .chr = 'B', | |
84 .required_keys = "mailbox_guid uid_validity uid_next " | |
85 "messages_count first_recent_uid highest_modseq", | |
86 .optional_keys = "cache_fields" | |
87 }, | |
88 { .name = "mail_change", | |
89 .chr = 'C', | |
90 .required_keys = "type uid", | |
91 .optional_keys = "guid hdr_hash modseq save_timestamp " | |
92 "add_flags remove_flags final_flags " | |
93 "keywords_reset keyword_changes" | |
94 }, | |
95 { .name = "mail_request", | |
96 .chr = 'R', | |
97 .optional_keys = "guid uid" | |
98 }, | |
99 { .name = "mail", | |
100 .chr = 'M', | |
101 .optional_keys = "guid uid pop3_uidl pop3_order received_date stream" | |
102 }, | |
103 { .name = "mailbox_cache_field", | |
104 .chr = 'c', | |
105 .required_keys = "name decision", | |
106 .optional_keys = "last_used" | |
107 }, | |
108 | |
109 { "end_of_list", '\0', NULL, NULL } | |
110 }; | |
111 | |
112 struct dsync_slave_io { | |
113 struct dsync_slave slave; | |
114 | |
115 char *name, *temp_path_prefix; | |
116 int fd_in, fd_out; | |
117 struct istream *input; | |
118 struct ostream *output; | |
119 struct io *io; | |
120 struct timeout *to; | |
121 | |
122 struct dsync_serializer *serializers[ITEM_END_OF_LIST]; | |
123 struct dsync_deserializer *deserializers[ITEM_END_OF_LIST]; | |
124 | |
125 pool_t ret_pool; | |
126 struct dsync_deserializer_decoder *cur_decoder; | |
127 | |
128 struct istream *mail_output, *mail_input; | |
129 struct dsync_mail *cur_mail; | |
130 char mail_output_last; | |
131 | |
132 unsigned int version_received:1; | |
133 unsigned int handshake_received:1; | |
134 unsigned int has_pending_data:1; | |
135 }; | |
136 | |
137 static void dsync_slave_io_stop(struct dsync_slave_io *slave) | |
138 { | |
139 i_stream_close(slave->input); | |
140 o_stream_close(slave->output); | |
141 io_loop_stop(current_ioloop); | |
142 } | |
143 | |
144 static int dsync_slave_io_read_mail_stream(struct dsync_slave_io *slave) | |
145 { | |
146 if (i_stream_read(slave->mail_input) < 0) { | |
147 if (slave->mail_input->stream_errno != 0) { | |
148 errno = slave->mail_input->stream_errno; | |
149 i_error("dsync(%s): read() failed: %m", slave->name); | |
150 dsync_slave_io_stop(slave); | |
151 return -1; | |
152 } | |
153 /* finished reading the mail stream */ | |
154 i_assert(slave->mail_input->eof); | |
155 i_stream_seek(slave->mail_input, 0); | |
156 slave->mail_input = NULL; | |
157 return 1; | |
158 } | |
14683
9646f80ac3e9
Added i_stream_get_data_size(). Used it where possible.
Timo Sirainen <tss@iki.fi>
parents:
14681
diff
changeset
|
159 i_stream_skip(slave->mail_input, |
9646f80ac3e9
Added i_stream_get_data_size(). Used it where possible.
Timo Sirainen <tss@iki.fi>
parents:
14681
diff
changeset
|
160 i_stream_get_data_size(slave->mail_input)); |
14584 | 161 return 0; |
162 } | |
163 | |
164 static void dsync_slave_io_input(struct dsync_slave_io *slave) | |
165 { | |
166 if (slave->mail_input != NULL) { | |
167 if (dsync_slave_io_read_mail_stream(slave) == 0) | |
168 return; | |
169 } | |
170 slave->slave.io_callback(slave->slave.io_context); | |
171 } | |
172 | |
173 static int dsync_slave_io_send_mail_stream(struct dsync_slave_io *slave) | |
174 { | |
175 const unsigned char *data; | |
176 unsigned char add; | |
177 size_t i, size; | |
178 int ret; | |
179 | |
180 while ((ret = i_stream_read_data(slave->mail_output, | |
181 &data, &size, 0)) > 0) { | |
182 add = '\0'; | |
183 for (i = 0; i < size; i++) { | |
184 if (data[i] == '\n') { | |
185 if ((i == 0 && slave->mail_output_last != '\r') || | |
186 (i > 0 && data[i-1] != '\r')) { | |
187 /* missing CR */ | |
188 add = '\r'; | |
189 break; | |
190 } | |
191 } else if (data[i] == '.' && | |
192 ((i == 0 && slave->mail_output_last == '\n') || | |
193 (i > 0 && data[i-1] == '\n'))) { | |
194 /* escape the dot */ | |
195 add = '.'; | |
196 break; | |
197 } | |
198 } | |
199 | |
200 if (i > 0) { | |
14681
ca37d1577291
Added o_stream_nsend*() and related functions to make delayed error handling safer.
Timo Sirainen <tss@iki.fi>
parents:
14629
diff
changeset
|
201 o_stream_nsend(slave->output, data, i); |
14584 | 202 slave->mail_output_last = data[i-1]; |
203 i_stream_skip(slave->mail_output, i); | |
204 } | |
205 | |
206 if (o_stream_get_buffer_used_size(slave->output) >= 4096) { | |
207 if ((ret = o_stream_flush(slave->output)) < 0) { | |
208 dsync_slave_io_stop(slave); | |
209 return -1; | |
210 } | |
211 if (ret == 0) { | |
212 /* continue later */ | |
213 o_stream_set_flush_pending(slave->output, TRUE); | |
214 return 0; | |
215 } | |
216 } | |
217 | |
218 if (add != '\0') { | |
14681
ca37d1577291
Added o_stream_nsend*() and related functions to make delayed error handling safer.
Timo Sirainen <tss@iki.fi>
parents:
14629
diff
changeset
|
219 o_stream_nsend(slave->output, &add, 1); |
14584 | 220 slave->mail_output_last = add; |
221 } | |
222 } | |
223 i_assert(ret == -1); | |
224 | |
225 if (slave->mail_output->stream_errno != 0) { | |
226 i_error("dsync(%s): read(%s) failed: %m", | |
227 slave->name, i_stream_get_name(slave->mail_output)); | |
228 dsync_slave_io_stop(slave); | |
229 return -1; | |
230 } | |
231 | |
232 /* finished sending the stream */ | |
14681
ca37d1577291
Added o_stream_nsend*() and related functions to make delayed error handling safer.
Timo Sirainen <tss@iki.fi>
parents:
14629
diff
changeset
|
233 o_stream_nsend_str(slave->output, "\r\n.\r\n"); |
14584 | 234 i_stream_unref(&slave->mail_output); |
235 return 1; | |
236 } | |
237 | |
238 static int dsync_slave_io_output(struct dsync_slave_io *slave) | |
239 { | |
240 struct ostream *output = slave->output; | |
241 int ret; | |
242 | |
243 if ((ret = o_stream_flush(output)) < 0) | |
244 ret = 1; | |
245 else if (slave->mail_output != NULL) { | |
246 if (dsync_slave_io_send_mail_stream(slave) < 0) | |
247 ret = 1; | |
248 } | |
249 timeout_reset(slave->to); | |
250 | |
251 if (!dsync_slave_is_send_queue_full(&slave->slave)) | |
252 slave->slave.io_callback(slave->slave.io_context); | |
253 return ret; | |
254 } | |
255 | |
256 static void dsync_slave_io_timeout(struct dsync_slave_io *slave) | |
257 { | |
258 i_error("dsync(%s): I/O has stalled, no activity for %u seconds", | |
259 slave->name, DSYNC_SLAVE_IO_TIMEOUT_MSECS/1000); | |
260 dsync_slave_io_stop(slave); | |
261 } | |
262 | |
263 static void dsync_slave_io_init(struct dsync_slave_io *slave) | |
264 { | |
265 unsigned int i; | |
266 | |
267 fd_set_nonblock(slave->fd_in, TRUE); | |
268 fd_set_nonblock(slave->fd_out, TRUE); | |
269 | |
270 slave->input = i_stream_create_fd(slave->fd_in, (size_t)-1, FALSE); | |
271 slave->output = o_stream_create_fd(slave->fd_out, (size_t)-1, FALSE); | |
272 slave->io = io_add(slave->fd_in, IO_READ, dsync_slave_io_input, slave); | |
14681
ca37d1577291
Added o_stream_nsend*() and related functions to make delayed error handling safer.
Timo Sirainen <tss@iki.fi>
parents:
14629
diff
changeset
|
273 o_stream_set_no_error_handling(slave->output, TRUE); |
14584 | 274 o_stream_set_flush_callback(slave->output, dsync_slave_io_output, slave); |
275 slave->to = timeout_add(DSYNC_SLAVE_IO_TIMEOUT_MSECS, | |
14629
c93ca5e46a8a
Marked functions parameters that are allowed to be NULL. Some APIs were also changed.
Timo Sirainen <tss@iki.fi>
parents:
14584
diff
changeset
|
276 dsync_slave_io_timeout, slave); |
14584 | 277 o_stream_cork(slave->output); |
14681
ca37d1577291
Added o_stream_nsend*() and related functions to make delayed error handling safer.
Timo Sirainen <tss@iki.fi>
parents:
14629
diff
changeset
|
278 o_stream_nsend_str(slave->output, DSYNC_HANDSHAKE_VERSION); |
14584 | 279 |
280 /* initialize serializers and send their headers to remote */ | |
281 for (i = 1; i < ITEM_END_OF_LIST; i++) T_BEGIN { | |
282 const char *keys; | |
283 | |
284 keys = items[i].required_keys == NULL ? items[i].optional_keys : | |
285 t_strconcat(items[i].required_keys, " ", | |
286 items[i].optional_keys, NULL); | |
287 if (keys != NULL) { | |
288 i_assert(items[i].chr != '\0'); | |
289 | |
290 slave->serializers[i] = | |
291 dsync_serializer_init(t_strsplit_spaces(keys, " ")); | |
14681
ca37d1577291
Added o_stream_nsend*() and related functions to make delayed error handling safer.
Timo Sirainen <tss@iki.fi>
parents:
14629
diff
changeset
|
292 o_stream_nsend(slave->output, &items[i].chr, 1); |
ca37d1577291
Added o_stream_nsend*() and related functions to make delayed error handling safer.
Timo Sirainen <tss@iki.fi>
parents:
14629
diff
changeset
|
293 o_stream_nsend_str(slave->output, |
14584 | 294 dsync_serializer_encode_header_line(slave->serializers[i])); |
295 } | |
296 } T_END; | |
14681
ca37d1577291
Added o_stream_nsend*() and related functions to make delayed error handling safer.
Timo Sirainen <tss@iki.fi>
parents:
14629
diff
changeset
|
297 o_stream_nsend_str(slave->output, ".\n"); |
14584 | 298 |
299 dsync_slave_flush(&slave->slave); | |
300 } | |
301 | |
302 static void dsync_slave_io_deinit(struct dsync_slave *_slave) | |
303 { | |
304 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
305 | |
306 if (slave->cur_decoder != NULL) | |
307 dsync_deserializer_decode_finish(&slave->cur_decoder); | |
308 if (slave->mail_output != NULL) | |
309 i_stream_unref(&slave->mail_output); | |
310 | |
311 timeout_remove(&slave->to); | |
312 if (slave->io != NULL) | |
313 io_remove(&slave->io); | |
314 i_stream_destroy(&slave->input); | |
315 o_stream_destroy(&slave->output); | |
316 if (close(slave->fd_in) < 0) | |
317 i_error("close(%s) failed: %m", slave->name); | |
318 if (slave->fd_in != slave->fd_out) { | |
319 if (close(slave->fd_out) < 0) | |
320 i_error("close(%s) failed: %m", slave->name); | |
321 } | |
322 pool_unref(&slave->ret_pool); | |
323 i_free(slave->temp_path_prefix); | |
324 i_free(slave->name); | |
325 i_free(slave); | |
326 } | |
327 | |
328 static int dsync_slave_io_next_line(struct dsync_slave_io *slave, | |
329 const char **line_r) | |
330 { | |
331 const char *line; | |
332 | |
333 line = i_stream_next_line(slave->input); | |
334 if (line != NULL) { | |
335 *line_r = line; | |
336 return 1; | |
337 } | |
338 | |
339 /* try reading some */ | |
340 switch (i_stream_read(slave->input)) { | |
341 case -1: | |
342 if (slave->input->stream_errno != 0) { | |
343 errno = slave->input->stream_errno; | |
344 i_error("read(%s) failed: %m", slave->name); | |
345 } else { | |
346 i_assert(slave->input->eof); | |
347 i_error("read(%s) failed: EOF", slave->name); | |
348 } | |
349 dsync_slave_io_stop(slave); | |
350 return -1; | |
351 case 0: | |
352 return 0; | |
353 } | |
354 *line_r = i_stream_next_line(slave->input); | |
355 if (*line_r == NULL) { | |
356 slave->has_pending_data = FALSE; | |
357 return 0; | |
358 } | |
359 slave->has_pending_data = TRUE; | |
360 return 1; | |
361 } | |
362 | |
14629
c93ca5e46a8a
Marked functions parameters that are allowed to be NULL. Some APIs were also changed.
Timo Sirainen <tss@iki.fi>
parents:
14584
diff
changeset
|
363 static void ATTR_FORMAT(3, 4) ATTR_NULL(2) |
14584 | 364 dsync_slave_input_error(struct dsync_slave_io *slave, |
365 struct dsync_deserializer_decoder *decoder, | |
366 const char *fmt, ...) | |
367 { | |
368 va_list args; | |
369 const char *error; | |
370 | |
371 va_start(args, fmt); | |
372 error = t_strdup_vprintf(fmt, args); | |
373 if (decoder == NULL) | |
374 i_error("dsync(%s): %s", slave->name, error); | |
375 else { | |
376 i_error("dsync(%s): %s: %s", slave->name, | |
377 dsync_deserializer_decoder_get_name(decoder), error); | |
378 } | |
379 va_end(args); | |
380 | |
381 dsync_slave_io_stop(slave); | |
382 } | |
383 | |
384 static void | |
385 dsync_slave_io_send_string(struct dsync_slave_io *slave, const string_t *str) | |
386 { | |
387 i_assert(slave->mail_output == NULL); | |
14681
ca37d1577291
Added o_stream_nsend*() and related functions to make delayed error handling safer.
Timo Sirainen <tss@iki.fi>
parents:
14629
diff
changeset
|
388 o_stream_nsend(slave->output, str_data(str), str_len(str)); |
14584 | 389 } |
390 | |
391 static int dsync_slave_check_missing_deserializers(struct dsync_slave_io *slave) | |
392 { | |
393 unsigned int i; | |
394 int ret = 0; | |
395 | |
396 for (i = 1; i < ITEM_END_OF_LIST; i++) { | |
397 if (slave->deserializers[i] == NULL && | |
398 (items[i].required_keys != NULL || | |
399 items[i].optional_keys != NULL)) { | |
400 dsync_slave_input_error(slave, NULL, | |
401 "Remote didn't handshake deserializer for %s", | |
402 items[i].name); | |
403 ret = -1; | |
404 } | |
405 } | |
406 return ret; | |
407 } | |
408 | |
409 static bool | |
410 dsync_slave_io_handshake(struct dsync_slave_io *slave, const char *line) | |
411 { | |
412 enum item_type item = ITEM_NONE; | |
413 const char *const *required_keys, *error; | |
414 unsigned int i; | |
415 | |
416 if (slave->handshake_received) | |
417 return TRUE; | |
418 | |
419 if (!slave->version_received) { | |
420 if (!version_string_verify(line, "dsync", | |
421 DSYNC_PROTOCOL_VERSION_MAJOR)) { | |
422 dsync_slave_input_error(slave, NULL, | |
423 "Remote dsync doesn't use compatible protocol"); | |
424 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
425 } | |
426 slave->version_received = TRUE; | |
427 return FALSE; | |
428 } | |
429 | |
430 if (strcmp(line, END_OF_LIST_LINE) == 0) { | |
431 /* finished handshaking */ | |
432 if (dsync_slave_check_missing_deserializers(slave) < 0) | |
433 return FALSE; | |
434 slave->handshake_received = TRUE; | |
435 return FALSE; | |
436 } | |
437 | |
438 for (i = 1; i < ITEM_END_OF_LIST; i++) { | |
439 if (items[i].chr == line[0]) { | |
440 item = i; | |
441 break; | |
442 } | |
443 } | |
444 if (item == ITEM_NONE) { | |
445 /* unknown deserializer, ignore */ | |
446 return FALSE; | |
447 } | |
448 | |
449 required_keys = items[item].required_keys == NULL ? NULL : | |
450 t_strsplit(items[item].required_keys, " "); | |
451 if (dsync_deserializer_init(items[item].name, | |
452 required_keys, line + 1, | |
453 &slave->deserializers[item], &error) < 0) { | |
454 dsync_slave_input_error(slave, NULL, | |
455 "Remote sent invalid handshake for %s: %s", | |
456 items[item].name, error); | |
457 } | |
458 return FALSE; | |
459 } | |
460 | |
461 static enum dsync_slave_recv_ret | |
462 dsync_slave_io_input_next(struct dsync_slave_io *slave, enum item_type item, | |
463 struct dsync_deserializer_decoder **decoder_r) | |
464 { | |
465 enum item_type line_item = ITEM_NONE; | |
466 const char *line, *error; | |
467 unsigned int i; | |
468 | |
469 i_assert(slave->mail_input == NULL); | |
470 | |
471 timeout_reset(slave->to); | |
472 | |
473 do { | |
474 if (dsync_slave_io_next_line(slave, &line) <= 0) | |
475 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
476 } while (!dsync_slave_io_handshake(slave, line)); | |
477 | |
478 if (strcmp(line, END_OF_LIST_LINE) == 0) { | |
479 /* end of this list */ | |
480 return DSYNC_SLAVE_RECV_RET_FINISHED; | |
481 } | |
482 for (i = 1; i < ITEM_END_OF_LIST; i++) { | |
483 if (*line == items[i].chr) { | |
484 line_item = i; | |
485 break; | |
486 } | |
487 } | |
488 if (line_item != item) { | |
489 dsync_slave_input_error(slave, NULL, | |
490 "Received unexpected input %c != %c", | |
491 *line, items[item].chr); | |
492 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
493 } | |
494 | |
495 if (slave->cur_decoder != NULL) | |
496 dsync_deserializer_decode_finish(&slave->cur_decoder); | |
497 if (dsync_deserializer_decode_begin(slave->deserializers[item], | |
498 line+1, &slave->cur_decoder, | |
499 &error) < 0) { | |
500 dsync_slave_input_error(slave, NULL, "Invalid input to %s: %s", | |
501 items[item].name, error); | |
502 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
503 } | |
504 *decoder_r = slave->cur_decoder; | |
505 return DSYNC_SLAVE_RECV_RET_OK; | |
506 } | |
507 | |
508 static void | |
509 dsync_slave_io_send_handshake(struct dsync_slave *_slave, | |
510 const struct dsync_slave_settings *set) | |
511 { | |
512 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
513 struct dsync_serializer_encoder *encoder; | |
514 string_t *str = t_str_new(128); | |
515 char sync_type[2]; | |
516 | |
517 str_append_c(str, items[ITEM_HANDSHAKE].chr); | |
518 encoder = dsync_serializer_encode_begin(slave->serializers[ITEM_HANDSHAKE]); | |
519 if (set->sync_ns_prefix != NULL) { | |
520 dsync_serializer_encode_add(encoder, "sync_ns_prefix", | |
521 set->sync_ns_prefix); | |
522 } | |
523 | |
524 sync_type[0] = sync_type[1] = '\0'; | |
525 switch (set->sync_type) { | |
526 case DSYNC_BRAIN_SYNC_TYPE_UNKNOWN: | |
527 break; | |
528 case DSYNC_BRAIN_SYNC_TYPE_FULL: | |
529 sync_type[0] = 'f'; | |
530 break; | |
531 case DSYNC_BRAIN_SYNC_TYPE_CHANGED: | |
532 sync_type[0] = 'c'; | |
533 break; | |
534 case DSYNC_BRAIN_SYNC_TYPE_STATE: | |
535 sync_type[0] = 's'; | |
536 break; | |
537 } | |
538 i_assert(sync_type[0] != '\0'); | |
539 dsync_serializer_encode_add(encoder, "sync_type", sync_type); | |
540 if (set->guid_requests) | |
541 dsync_serializer_encode_add(encoder, "guid_requests", ""); | |
542 if (set->mails_have_guids) | |
543 dsync_serializer_encode_add(encoder, "mails_have_guids", ""); | |
544 | |
545 dsync_serializer_encode_finish(&encoder, str); | |
546 dsync_slave_io_send_string(slave, str); | |
547 } | |
548 | |
549 static enum dsync_slave_recv_ret | |
550 dsync_slave_io_recv_handshake(struct dsync_slave *_slave, | |
551 const struct dsync_slave_settings **set_r) | |
552 { | |
553 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
554 struct dsync_deserializer_decoder *decoder; | |
555 struct dsync_slave_settings *set; | |
556 const char *value; | |
557 pool_t pool = slave->ret_pool; | |
558 enum dsync_slave_recv_ret ret; | |
559 | |
560 ret = dsync_slave_io_input_next(slave, ITEM_HANDSHAKE, &decoder); | |
561 if (ret != DSYNC_SLAVE_RECV_RET_OK) { | |
562 if (ret != DSYNC_SLAVE_RECV_RET_TRYAGAIN) { | |
563 i_error("dsync(%s): Unexpected input in handshake", | |
564 slave->name); | |
565 dsync_slave_io_stop(slave); | |
566 } | |
567 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
568 } | |
569 | |
570 p_clear(pool); | |
571 set = p_new(pool, struct dsync_slave_settings, 1); | |
572 | |
573 if (dsync_deserializer_decode_try(decoder, "sync_ns_prefix", &value)) | |
574 set->sync_ns_prefix = p_strdup(pool, value); | |
575 if (dsync_deserializer_decode_try(decoder, "sync_type", &value)) { | |
576 switch (value[0]) { | |
577 case 'f': | |
578 set->sync_type = DSYNC_BRAIN_SYNC_TYPE_FULL; | |
579 break; | |
580 case 'c': | |
581 set->sync_type = DSYNC_BRAIN_SYNC_TYPE_CHANGED; | |
582 break; | |
583 case 's': | |
584 set->sync_type = DSYNC_BRAIN_SYNC_TYPE_STATE; | |
585 break; | |
586 default: | |
587 dsync_slave_input_error(slave, decoder, | |
588 "Unknown sync_type: %s", value); | |
589 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
590 } | |
591 } | |
592 if (dsync_deserializer_decode_try(decoder, "guid_requests", &value)) | |
593 set->guid_requests = TRUE; | |
594 if (dsync_deserializer_decode_try(decoder, "mails_have_guids", &value)) | |
595 set->mails_have_guids = TRUE; | |
596 | |
597 *set_r = set; | |
598 return DSYNC_SLAVE_RECV_RET_OK; | |
599 } | |
600 | |
601 static void | |
602 dsync_slave_io_send_end_of_list(struct dsync_slave *_slave) | |
603 { | |
604 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
605 | |
606 i_assert(slave->mail_output == NULL); | |
607 | |
14681
ca37d1577291
Added o_stream_nsend*() and related functions to make delayed error handling safer.
Timo Sirainen <tss@iki.fi>
parents:
14629
diff
changeset
|
608 o_stream_nsend_str(slave->output, END_OF_LIST_LINE"\n"); |
14584 | 609 } |
610 | |
611 static void | |
612 dsync_slave_io_send_mailbox_state(struct dsync_slave *_slave, | |
613 const struct dsync_mailbox_state *state) | |
614 { | |
615 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
616 struct dsync_serializer_encoder *encoder; | |
617 string_t *str = t_str_new(128); | |
618 | |
619 str_append_c(str, items[ITEM_MAILBOX_STATE].chr); | |
620 encoder = dsync_serializer_encode_begin(slave->serializers[ITEM_MAILBOX_STATE]); | |
621 dsync_serializer_encode_add(encoder, "mailbox_guid", | |
622 guid_128_to_string(state->mailbox_guid)); | |
623 dsync_serializer_encode_add(encoder, "last_uidvalidity", | |
624 dec2str(state->last_uidvalidity)); | |
625 dsync_serializer_encode_add(encoder, "last_common_uid", | |
626 dec2str(state->last_common_uid)); | |
627 dsync_serializer_encode_add(encoder, "last_common_modseq", | |
628 dec2str(state->last_common_modseq)); | |
629 | |
630 dsync_serializer_encode_finish(&encoder, str); | |
631 dsync_slave_io_send_string(slave, str); | |
632 } | |
633 | |
634 static enum dsync_slave_recv_ret | |
635 dsync_slave_io_recv_mailbox_state(struct dsync_slave *_slave, | |
636 struct dsync_mailbox_state *state_r) | |
637 { | |
638 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
639 struct dsync_deserializer_decoder *decoder; | |
640 const char *value; | |
641 enum dsync_slave_recv_ret ret; | |
642 | |
643 memset(state_r, 0, sizeof(*state_r)); | |
644 | |
645 ret = dsync_slave_io_input_next(slave, ITEM_MAILBOX_STATE, &decoder); | |
646 if (ret != DSYNC_SLAVE_RECV_RET_OK) | |
647 return ret; | |
648 | |
649 value = dsync_deserializer_decode_get(decoder, "mailbox_guid"); | |
650 if (guid_128_from_string(value, state_r->mailbox_guid) < 0) { | |
651 dsync_slave_input_error(slave, decoder, "Invalid mailbox_guid"); | |
652 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
653 } | |
654 value = dsync_deserializer_decode_get(decoder, "last_uidvalidity"); | |
655 if (str_to_uint32(value, &state_r->last_uidvalidity) < 0) { | |
656 dsync_slave_input_error(slave, decoder, "Invalid last_uidvalidity"); | |
657 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
658 } | |
659 value = dsync_deserializer_decode_get(decoder, "last_common_uid"); | |
660 if (str_to_uint32(value, &state_r->last_uidvalidity) < 0) { | |
661 dsync_slave_input_error(slave, decoder, "Invalid last_common_uid"); | |
662 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
663 } | |
664 value = dsync_deserializer_decode_get(decoder, "last_common_modseq"); | |
665 if (str_to_uint32(value, &state_r->last_uidvalidity) < 0) { | |
666 dsync_slave_input_error(slave, decoder, "Invalid last_common_modseq"); | |
667 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
668 } | |
669 return DSYNC_SLAVE_RECV_RET_OK; | |
670 } | |
671 | |
672 static void | |
673 dsync_slave_io_send_mailbox_tree_node(struct dsync_slave *_slave, | |
674 const char *const *name, | |
675 const struct dsync_mailbox_node *node) | |
676 { | |
677 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
678 struct dsync_serializer_encoder *encoder; | |
679 string_t *str, *namestr; | |
680 | |
681 i_assert(*name != NULL); | |
682 | |
683 str = t_str_new(128); | |
684 str_append_c(str, items[ITEM_MAILBOX_TREE_NODE].chr); | |
685 | |
686 /* convert all hierarchy separators to tabs. mailbox names really | |
687 aren't supposed to have any tabs, but escape them anyway if there | |
688 are. */ | |
689 namestr = t_str_new(128); | |
690 for (; *name != NULL; name++) { | |
691 str_tabescape_write(namestr, *name); | |
692 str_append_c(namestr, '\t'); | |
693 } | |
694 str_truncate(namestr, str_len(namestr)-1); | |
695 | |
696 encoder = dsync_serializer_encode_begin(slave->serializers[ITEM_MAILBOX_TREE_NODE]); | |
697 dsync_serializer_encode_add(encoder, "name", str_c(namestr)); | |
698 switch (node->existence) { | |
699 case DSYNC_MAILBOX_NODE_NONEXISTENT: | |
700 dsync_serializer_encode_add(encoder, "existence", "n"); | |
701 break; | |
702 case DSYNC_MAILBOX_NODE_EXISTS: | |
703 dsync_serializer_encode_add(encoder, "existence", "y"); | |
704 break; | |
705 case DSYNC_MAILBOX_NODE_DELETED: | |
706 dsync_serializer_encode_add(encoder, "existence", "d"); | |
707 break; | |
708 } | |
709 | |
710 if (!guid_128_is_empty(node->mailbox_guid)) { | |
711 dsync_serializer_encode_add(encoder, "mailbox_guid", | |
712 guid_128_to_string(node->mailbox_guid)); | |
713 } | |
714 if (node->uid_validity != 0) { | |
715 dsync_serializer_encode_add(encoder, "uid_validity", | |
716 dec2str(node->uid_validity)); | |
717 } | |
718 if (node->last_renamed != 0) { | |
719 dsync_serializer_encode_add(encoder, "last_renamed", | |
720 dec2str(node->last_renamed)); | |
721 } | |
722 if (node->last_subscription_change != 0) { | |
723 dsync_serializer_encode_add(encoder, "last_subscription_change", | |
724 dec2str(node->last_subscription_change)); | |
725 } | |
726 if (node->subscribed) | |
727 dsync_serializer_encode_add(encoder, "subscribed", ""); | |
728 dsync_serializer_encode_finish(&encoder, str); | |
729 dsync_slave_io_send_string(slave, str); | |
730 } | |
731 | |
732 static enum dsync_slave_recv_ret | |
733 dsync_slave_io_recv_mailbox_tree_node(struct dsync_slave *_slave, | |
734 const char *const **name_r, | |
735 const struct dsync_mailbox_node **node_r) | |
736 { | |
737 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
738 struct dsync_deserializer_decoder *decoder; | |
739 struct dsync_mailbox_node *node; | |
740 const char *value; | |
741 enum dsync_slave_recv_ret ret; | |
742 | |
743 ret = dsync_slave_io_input_next(slave, ITEM_MAILBOX_TREE_NODE, &decoder); | |
744 if (ret != DSYNC_SLAVE_RECV_RET_OK) | |
745 return ret; | |
746 | |
747 p_clear(slave->ret_pool); | |
748 node = p_new(slave->ret_pool, struct dsync_mailbox_node, 1); | |
749 | |
750 value = dsync_deserializer_decode_get(decoder, "name"); | |
751 if (*value == '\0') { | |
752 dsync_slave_input_error(slave, decoder, "Empty name"); | |
753 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
754 } | |
755 *name_r = (void *)p_strsplit_tabescaped(slave->ret_pool, value); | |
756 | |
757 value = dsync_deserializer_decode_get(decoder, "existence"); | |
758 switch (*value) { | |
759 case 'n': | |
760 node->existence = DSYNC_MAILBOX_NODE_NONEXISTENT; | |
761 break; | |
762 case 'y': | |
763 node->existence = DSYNC_MAILBOX_NODE_EXISTS; | |
764 break; | |
765 case 'd': | |
766 node->existence = DSYNC_MAILBOX_NODE_DELETED; | |
767 break; | |
768 } | |
769 | |
770 if (dsync_deserializer_decode_try(decoder, "mailbox_guid", &value) && | |
771 guid_128_from_string(value, node->mailbox_guid) < 0) { | |
772 dsync_slave_input_error(slave, decoder, "Invalid mailbox_guid"); | |
773 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
774 } | |
775 if (dsync_deserializer_decode_try(decoder, "uid_validity", &value) && | |
776 str_to_uint32(value, &node->uid_validity) < 0) { | |
777 dsync_slave_input_error(slave, decoder, "Invalid uid_validity"); | |
778 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
779 } | |
780 if (dsync_deserializer_decode_try(decoder, "last_renamed", &value) && | |
781 str_to_time(value, &node->last_renamed) < 0) { | |
782 dsync_slave_input_error(slave, decoder, "Invalid last_renamed"); | |
783 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
784 } | |
785 if (dsync_deserializer_decode_try(decoder, "last_subscription_change", &value) && | |
786 str_to_time(value, &node->last_subscription_change) < 0) { | |
787 dsync_slave_input_error(slave, decoder, "Invalid last_subscription_change"); | |
788 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
789 } | |
790 if (dsync_deserializer_decode_try(decoder, "subscribed", &value)) | |
791 node->subscribed = TRUE; | |
792 | |
793 *node_r = node; | |
794 return DSYNC_SLAVE_RECV_RET_OK; | |
795 } | |
796 | |
797 static void | |
798 dsync_slave_io_send_mailbox_deletes(struct dsync_slave *_slave, | |
799 const struct dsync_mailbox_delete *deletes, | |
800 unsigned int count, char hierarchy_sep) | |
801 { | |
802 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
803 struct dsync_serializer_encoder *encoder; | |
804 string_t *str, *guidstr; | |
805 char sep[2]; | |
806 unsigned int i; | |
807 | |
808 str = t_str_new(128); | |
809 str_append_c(str, items[ITEM_MAILBOX_DELETE].chr); | |
810 | |
811 encoder = dsync_serializer_encode_begin(slave->serializers[ITEM_MAILBOX_DELETE]); | |
812 sep[0] = hierarchy_sep; sep[1] = '\0'; | |
813 dsync_serializer_encode_add(encoder, "hierarchy_sep", sep); | |
814 | |
815 guidstr = t_str_new(128); | |
816 for (i = 0; i < count; i++) { | |
817 if (deletes[i].delete_mailbox) { | |
818 str_append(guidstr, guid_128_to_string(deletes[i].guid)); | |
819 str_append_c(guidstr, ' '); | |
820 } | |
821 } | |
822 if (str_len(guidstr) > 0) { | |
823 str_truncate(guidstr, str_len(guidstr)-1); | |
824 dsync_serializer_encode_add(encoder, "mailboxes", | |
825 str_c(guidstr)); | |
826 } | |
827 | |
828 str_truncate(guidstr, 0); | |
829 for (i = 0; i < count; i++) { | |
830 if (!deletes[i].delete_mailbox) { | |
831 str_append(guidstr, guid_128_to_string(deletes[i].guid)); | |
832 str_append_c(guidstr, ' '); | |
833 } | |
834 } | |
835 if (str_len(guidstr) > 0) { | |
836 str_truncate(guidstr, str_len(guidstr)-1); | |
837 dsync_serializer_encode_add(encoder, "dirs", str_c(guidstr)); | |
838 } | |
839 dsync_serializer_encode_finish(&encoder, str); | |
840 dsync_slave_io_send_string(slave, str); | |
841 } | |
842 | |
843 ARRAY_DEFINE_TYPE(dsync_mailbox_delete, struct dsync_mailbox_delete); | |
844 static int | |
845 decode_mailbox_deletes(ARRAY_TYPE(dsync_mailbox_delete) *deletes, | |
846 const char *value, bool delete_mailbox) | |
847 { | |
848 struct dsync_mailbox_delete *del; | |
849 const char *const *guid_strings; | |
850 unsigned int i; | |
851 | |
852 guid_strings = t_strsplit(value, " "); | |
853 for (i = 0; guid_strings[i] != NULL; i++) { | |
854 del = array_append_space(deletes); | |
855 del->delete_mailbox = delete_mailbox; | |
856 if (guid_128_from_string(guid_strings[i], del->guid) < 0) | |
857 return -1; | |
858 } | |
859 return 0; | |
860 } | |
861 | |
862 static enum dsync_slave_recv_ret | |
863 dsync_slave_io_recv_mailbox_deletes(struct dsync_slave *_slave, | |
864 const struct dsync_mailbox_delete **deletes_r, | |
865 unsigned int *count_r, char *hierarchy_sep_r) | |
866 { | |
867 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
868 struct dsync_deserializer_decoder *decoder; | |
869 ARRAY_TYPE(dsync_mailbox_delete) deletes; | |
870 const char *value; | |
871 enum dsync_slave_recv_ret ret; | |
872 | |
873 ret = dsync_slave_io_input_next(slave, ITEM_MAILBOX_DELETE, &decoder); | |
874 if (ret != DSYNC_SLAVE_RECV_RET_OK) | |
875 return ret; | |
876 | |
877 p_clear(slave->ret_pool); | |
878 p_array_init(&deletes, slave->ret_pool, 16); | |
879 | |
880 value = dsync_deserializer_decode_get(decoder, "hierarchy_sep"); | |
881 if (strlen(value) != 1) { | |
882 dsync_slave_input_error(slave, decoder, "Invalid hierarchy_sep"); | |
883 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
884 } | |
885 *hierarchy_sep_r = value[0]; | |
886 | |
887 if (dsync_deserializer_decode_try(decoder, "mailboxes", &value) && | |
888 decode_mailbox_deletes(&deletes, value, TRUE) < 0) { | |
889 dsync_slave_input_error(slave, decoder, "Invalid mailboxes"); | |
890 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
891 } | |
892 if (dsync_deserializer_decode_try(decoder, "dirs", &value) && | |
893 decode_mailbox_deletes(&deletes, value, FALSE) < 0) { | |
894 dsync_slave_input_error(slave, decoder, "Invalid dirs"); | |
895 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
896 } | |
897 *deletes_r = array_get(&deletes, count_r); | |
898 return DSYNC_SLAVE_RECV_RET_OK; | |
899 } | |
900 | |
901 static const char * | |
902 get_cache_fields(struct dsync_slave_io *slave, | |
903 const struct dsync_mailbox *dsync_box) | |
904 { | |
905 struct dsync_serializer_encoder *encoder; | |
906 string_t *str; | |
907 const struct mailbox_cache_field *cache_fields; | |
908 unsigned int i, count; | |
909 char decision[3]; | |
910 | |
911 cache_fields = array_get(&dsync_box->cache_fields, &count); | |
912 if (count == 0) | |
913 return ""; | |
914 | |
915 str = t_str_new(128); | |
916 for (i = 0; i < count; i++) { | |
917 const struct mailbox_cache_field *field = &cache_fields[i]; | |
918 | |
919 encoder = dsync_serializer_encode_begin(slave->serializers[ITEM_MAILBOX_CACHE_FIELD]); | |
920 dsync_serializer_encode_add(encoder, "name", field->name); | |
921 | |
922 memset(decision, 0, sizeof(decision)); | |
923 switch (field->decision & ~MAIL_CACHE_DECISION_FORCED) { | |
924 case MAIL_CACHE_DECISION_NO: | |
925 decision[0] = 'n'; | |
926 break; | |
927 case MAIL_CACHE_DECISION_TEMP: | |
928 decision[0] = 't'; | |
929 break; | |
930 case MAIL_CACHE_DECISION_YES: | |
931 decision[0] = 'y'; | |
932 break; | |
933 } | |
934 i_assert(decision[0] != '\0'); | |
935 if ((field->decision & MAIL_CACHE_DECISION_FORCED) != 0) | |
936 decision[1] = 'F'; | |
937 dsync_serializer_encode_add(encoder, "decision", decision); | |
938 if (field->last_used != 0) { | |
939 dsync_serializer_encode_add(encoder, "last_used", | |
940 dec2str(field->last_used)); | |
941 } | |
942 dsync_serializer_encode_finish(&encoder, str); | |
943 } | |
944 if (i > 0) { | |
945 /* remove the trailing LF */ | |
946 str_truncate(str, str_len(str)-1); | |
947 } | |
948 return str_c(str); | |
949 } | |
950 | |
951 static void | |
952 dsync_slave_io_send_mailbox(struct dsync_slave *_slave, | |
953 const struct dsync_mailbox *dsync_box) | |
954 { | |
955 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
956 struct dsync_serializer_encoder *encoder; | |
957 string_t *str = t_str_new(128); | |
958 const char *value; | |
959 | |
960 str_append_c(str, items[ITEM_MAILBOX].chr); | |
961 encoder = dsync_serializer_encode_begin(slave->serializers[ITEM_MAILBOX]); | |
962 dsync_serializer_encode_add(encoder, "mailbox_guid", | |
963 guid_128_to_string(dsync_box->mailbox_guid)); | |
964 | |
965 if (dsync_box->mailbox_lost) | |
966 dsync_serializer_encode_add(encoder, "mailbox_lost", ""); | |
967 dsync_serializer_encode_add(encoder, "uid_validity", | |
968 dec2str(dsync_box->uid_validity)); | |
969 dsync_serializer_encode_add(encoder, "uid_next", | |
970 dec2str(dsync_box->uid_next)); | |
971 dsync_serializer_encode_add(encoder, "messages_count", | |
972 dec2str(dsync_box->messages_count)); | |
973 dsync_serializer_encode_add(encoder, "first_recent_uid", | |
974 dec2str(dsync_box->first_recent_uid)); | |
975 dsync_serializer_encode_add(encoder, "highest_modseq", | |
976 dec2str(dsync_box->highest_modseq)); | |
977 | |
978 value = get_cache_fields(slave, dsync_box); | |
979 if (value != NULL) | |
980 dsync_serializer_encode_add(encoder, "cache_fields", value); | |
981 | |
982 dsync_serializer_encode_finish(&encoder, str); | |
983 dsync_slave_io_send_string(slave, str); | |
984 } | |
985 | |
986 static int | |
987 parse_cache_field(struct dsync_slave_io *slave, struct dsync_mailbox *box, | |
988 const char *value) | |
989 { | |
990 struct dsync_deserializer_decoder *decoder; | |
991 struct mailbox_cache_field field; | |
992 const char *error; | |
993 int ret = 0; | |
994 | |
995 if (dsync_deserializer_decode_begin(slave->deserializers[ITEM_MAILBOX_CACHE_FIELD], | |
996 value, &decoder, &error) < 0) { | |
997 dsync_slave_input_error(slave, NULL, | |
998 "cache_field: Invalid input: %s", error); | |
999 return -1; | |
1000 } | |
1001 | |
1002 memset(&field, 0, sizeof(field)); | |
1003 value = dsync_deserializer_decode_get(decoder, "name"); | |
1004 field.name = p_strdup(slave->ret_pool, value); | |
1005 | |
1006 value = dsync_deserializer_decode_get(decoder, "decision"); | |
1007 switch (*value) { | |
1008 case 'n': | |
1009 field.decision = MAIL_CACHE_DECISION_NO; | |
1010 break; | |
1011 case 't': | |
1012 field.decision = MAIL_CACHE_DECISION_TEMP; | |
1013 break; | |
1014 case 'y': | |
1015 field.decision = MAIL_CACHE_DECISION_YES; | |
1016 break; | |
1017 default: | |
1018 dsync_slave_input_error(slave, decoder, "Invalid decision: %s", | |
1019 value); | |
1020 ret = -1; | |
1021 break; | |
1022 } | |
1023 if (value[1] == 'F') | |
1024 field.decision |= MAIL_CACHE_DECISION_FORCED; | |
1025 | |
1026 if (dsync_deserializer_decode_try(decoder, "last_used", &value) && | |
1027 str_to_time(value, &field.last_used) < 0) { | |
1028 dsync_slave_input_error(slave, decoder, "Invalid last_used"); | |
1029 ret = -1; | |
1030 } | |
1031 array_append(&box->cache_fields, &field, 1); | |
1032 | |
1033 dsync_deserializer_decode_finish(&decoder); | |
1034 return ret; | |
1035 } | |
1036 | |
1037 static enum dsync_slave_recv_ret | |
1038 dsync_slave_io_recv_mailbox(struct dsync_slave *_slave, | |
1039 const struct dsync_mailbox **dsync_box_r) | |
1040 { | |
1041 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
1042 pool_t pool = slave->ret_pool; | |
1043 struct dsync_deserializer_decoder *decoder; | |
1044 struct dsync_mailbox *box; | |
1045 const char *value; | |
1046 enum dsync_slave_recv_ret ret; | |
1047 | |
1048 p_clear(pool); | |
1049 box = p_new(pool, struct dsync_mailbox, 1); | |
1050 | |
1051 ret = dsync_slave_io_input_next(slave, ITEM_MAILBOX, &decoder); | |
1052 if (ret != DSYNC_SLAVE_RECV_RET_OK) | |
1053 return ret; | |
1054 | |
1055 value = dsync_deserializer_decode_get(decoder, "mailbox_guid"); | |
1056 if (guid_128_from_string(value, box->mailbox_guid) < 0) { | |
1057 dsync_slave_input_error(slave, decoder, "Invalid mailbox_guid"); | |
1058 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1059 } | |
1060 | |
1061 if (dsync_deserializer_decode_try(decoder, "mailbox_lost", &value)) | |
1062 box->mailbox_lost = TRUE; | |
1063 value = dsync_deserializer_decode_get(decoder, "uid_validity"); | |
1064 if (str_to_uint32(value, &box->uid_validity) < 0) { | |
1065 dsync_slave_input_error(slave, decoder, "Invalid uid_validity"); | |
1066 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1067 } | |
1068 value = dsync_deserializer_decode_get(decoder, "uid_next"); | |
1069 if (str_to_uint32(value, &box->uid_next) < 0) { | |
1070 dsync_slave_input_error(slave, decoder, "Invalid uid_next"); | |
1071 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1072 } | |
1073 value = dsync_deserializer_decode_get(decoder, "messages_count"); | |
1074 if (str_to_uint32(value, &box->messages_count) < 0) { | |
1075 dsync_slave_input_error(slave, decoder, "Invalid messages_count"); | |
1076 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1077 } | |
1078 value = dsync_deserializer_decode_get(decoder, "first_recent_uid"); | |
1079 if (str_to_uint32(value, &box->first_recent_uid) < 0) { | |
1080 dsync_slave_input_error(slave, decoder, "Invalid first_recent_uid"); | |
1081 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1082 } | |
1083 value = dsync_deserializer_decode_get(decoder, "highest_modseq"); | |
1084 if (str_to_uint64(value, &box->highest_modseq) < 0) { | |
1085 dsync_slave_input_error(slave, decoder, "Invalid highest_modseq"); | |
1086 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1087 } | |
1088 | |
1089 p_array_init(&box->cache_fields, pool, 32); | |
1090 if (dsync_deserializer_decode_try(decoder, "cache_fields", &value)) { | |
1091 const char *const *fields = t_strsplit(value, "\n"); | |
1092 for (; *fields != NULL; fields++) { | |
1093 if (parse_cache_field(slave, box, *fields) < 0) | |
1094 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1095 } | |
1096 } | |
1097 | |
1098 *dsync_box_r = box; | |
1099 return DSYNC_SLAVE_RECV_RET_OK; | |
1100 } | |
1101 | |
1102 static void | |
1103 dsync_slave_io_send_change(struct dsync_slave *_slave, | |
1104 const struct dsync_mail_change *change) | |
1105 { | |
1106 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
1107 struct dsync_serializer_encoder *encoder; | |
1108 string_t *str = t_str_new(128); | |
1109 char type[2]; | |
1110 | |
1111 str_append_c(str, items[ITEM_MAIL_CHANGE].chr); | |
1112 encoder = dsync_serializer_encode_begin(slave->serializers[ITEM_MAIL_CHANGE]); | |
1113 | |
1114 type[0] = type[1] = '\0'; | |
1115 switch (change->type) { | |
1116 case DSYNC_MAIL_CHANGE_TYPE_SAVE: | |
1117 type[0] = 's'; | |
1118 break; | |
1119 case DSYNC_MAIL_CHANGE_TYPE_EXPUNGE: | |
1120 type[0] = 'e'; | |
1121 break; | |
1122 case DSYNC_MAIL_CHANGE_TYPE_FLAG_CHANGE: | |
1123 type[0] = 'f'; | |
1124 break; | |
1125 } | |
1126 i_assert(type[0] != '\0'); | |
1127 dsync_serializer_encode_add(encoder, "type", type); | |
1128 dsync_serializer_encode_add(encoder, "uid", dec2str(change->uid)); | |
1129 if (change->guid != NULL) | |
1130 dsync_serializer_encode_add(encoder, "guid", change->guid); | |
1131 if (change->hdr_hash != NULL) { | |
1132 dsync_serializer_encode_add(encoder, "hdr_hash", | |
1133 change->hdr_hash); | |
1134 } | |
1135 if (change->modseq != 0) { | |
1136 dsync_serializer_encode_add(encoder, "modseq", | |
1137 dec2str(change->modseq)); | |
1138 } | |
1139 if (change->save_timestamp != 0) { | |
1140 dsync_serializer_encode_add(encoder, "save_timestamp", | |
1141 dec2str(change->save_timestamp)); | |
1142 } | |
1143 if (change->add_flags != 0) { | |
1144 dsync_serializer_encode_add(encoder, "add_flags", | |
1145 t_strdup_printf("%x", change->add_flags)); | |
1146 } | |
1147 if (change->remove_flags != 0) { | |
1148 dsync_serializer_encode_add(encoder, "remove_flags", | |
1149 t_strdup_printf("%x", change->remove_flags)); | |
1150 } | |
1151 if (change->final_flags != 0) { | |
1152 dsync_serializer_encode_add(encoder, "final_flags", | |
1153 t_strdup_printf("%x", change->final_flags)); | |
1154 } | |
1155 if (change->keywords_reset) | |
1156 dsync_serializer_encode_add(encoder, "keywords_reset", ""); | |
1157 | |
1158 if (array_is_created(&change->keyword_changes) && | |
1159 array_count(&change->keyword_changes) > 0) { | |
1160 string_t *kw_str = t_str_new(128); | |
1161 const char *const *changes; | |
1162 unsigned int i, count; | |
1163 | |
1164 changes = array_get(&change->keyword_changes, &count); | |
1165 str_tabescape_write(kw_str, changes[0]); | |
1166 for (i = 1; i < count; i++) { | |
1167 str_append_c(kw_str, '\t'); | |
1168 str_tabescape_write(kw_str, changes[i]); | |
1169 } | |
1170 dsync_serializer_encode_add(encoder, "keyword_changes", | |
1171 str_c(kw_str)); | |
1172 } | |
1173 | |
1174 dsync_serializer_encode_finish(&encoder, str); | |
1175 dsync_slave_io_send_string(slave, str); | |
1176 } | |
1177 | |
1178 static enum dsync_slave_recv_ret | |
1179 dsync_slave_io_recv_change(struct dsync_slave *_slave, | |
1180 const struct dsync_mail_change **change_r) | |
1181 { | |
1182 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
1183 pool_t pool = slave->ret_pool; | |
1184 struct dsync_deserializer_decoder *decoder; | |
1185 struct dsync_mail_change *change; | |
1186 const char *value; | |
1187 enum dsync_slave_recv_ret ret; | |
1188 | |
1189 p_clear(pool); | |
1190 change = p_new(pool, struct dsync_mail_change, 1); | |
1191 | |
1192 ret = dsync_slave_io_input_next(slave, ITEM_MAIL_CHANGE, &decoder); | |
1193 if (ret != DSYNC_SLAVE_RECV_RET_OK) | |
1194 return ret; | |
1195 | |
1196 value = dsync_deserializer_decode_get(decoder, "type"); | |
1197 switch (*value) { | |
1198 case 's': | |
1199 change->type = DSYNC_MAIL_CHANGE_TYPE_SAVE; | |
1200 break; | |
1201 case 'e': | |
1202 change->type = DSYNC_MAIL_CHANGE_TYPE_EXPUNGE; | |
1203 break; | |
1204 case 'f': | |
1205 change->type = DSYNC_MAIL_CHANGE_TYPE_FLAG_CHANGE; | |
1206 break; | |
1207 default: | |
1208 dsync_slave_input_error(slave, decoder, | |
1209 "Invalid type: %s", value); | |
1210 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1211 } | |
1212 | |
1213 value = dsync_deserializer_decode_get(decoder, "uid"); | |
1214 if (str_to_uint32(value, &change->uid) < 0) { | |
1215 dsync_slave_input_error(slave, decoder, "Invalid uid"); | |
1216 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1217 } | |
1218 | |
1219 if (dsync_deserializer_decode_try(decoder, "guid", &value)) | |
1220 change->guid = p_strdup(pool, value); | |
1221 if (dsync_deserializer_decode_try(decoder, "hdr_hash", &value)) | |
1222 change->hdr_hash = p_strdup(pool, value); | |
1223 if (dsync_deserializer_decode_try(decoder, "modseq", &value) && | |
1224 str_to_uint64(value, &change->modseq) < 0) { | |
1225 dsync_slave_input_error(slave, decoder, "Invalid modseq"); | |
1226 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1227 } | |
1228 if (dsync_deserializer_decode_try(decoder, "save_timestamp", &value) && | |
1229 str_to_time(value, &change->save_timestamp) < 0) { | |
1230 dsync_slave_input_error(slave, decoder, "Invalid save_timestamp"); | |
1231 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1232 } | |
1233 | |
1234 if (dsync_deserializer_decode_try(decoder, "add_flags", &value)) | |
1235 change->add_flags = strtoul(value, NULL, 16); | |
1236 if (dsync_deserializer_decode_try(decoder, "remove_flags", &value)) | |
1237 change->remove_flags = strtoul(value, NULL, 16); | |
1238 if (dsync_deserializer_decode_try(decoder, "final_flags", &value)) | |
1239 change->final_flags = strtoul(value, NULL, 16); | |
1240 if (dsync_deserializer_decode_try(decoder, "keywords_reset", &value)) | |
1241 change->keywords_reset = TRUE; | |
1242 | |
1243 if (dsync_deserializer_decode_try(decoder, "keyword_changes", &value) && | |
1244 *value != '\0') { | |
1245 const char *const *changes = t_strsplit_tab(value); | |
1246 unsigned int i, count = str_array_length(changes); | |
1247 | |
1248 p_array_init(&change->keyword_changes, pool, count); | |
1249 for (i = 0; i < count; i++) { | |
1250 value = p_strdup(pool, changes[i]); | |
1251 array_append(&change->keyword_changes, &value, 1); | |
1252 } | |
1253 } | |
1254 | |
1255 *change_r = change; | |
1256 return DSYNC_SLAVE_RECV_RET_OK; | |
1257 } | |
1258 | |
1259 static void | |
1260 dsync_slave_io_send_mail_request(struct dsync_slave *_slave, | |
1261 const struct dsync_mail_request *request) | |
1262 { | |
1263 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
1264 struct dsync_serializer_encoder *encoder; | |
1265 string_t *str = t_str_new(128); | |
1266 | |
1267 str_append_c(str, items[ITEM_MAIL_REQUEST].chr); | |
1268 encoder = dsync_serializer_encode_begin(slave->serializers[ITEM_MAIL_REQUEST]); | |
1269 if (request->guid != NULL) | |
1270 dsync_serializer_encode_add(encoder, "guid", request->guid); | |
1271 if (request->uid != 0) { | |
1272 dsync_serializer_encode_add(encoder, "uid", | |
1273 dec2str(request->uid)); | |
1274 } | |
1275 dsync_serializer_encode_finish(&encoder, str); | |
1276 dsync_slave_io_send_string(slave, str); | |
1277 } | |
1278 | |
1279 static enum dsync_slave_recv_ret | |
1280 dsync_slave_io_recv_mail_request(struct dsync_slave *_slave, | |
1281 const struct dsync_mail_request **request_r) | |
1282 { | |
1283 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
1284 struct dsync_deserializer_decoder *decoder; | |
1285 struct dsync_mail_request *request; | |
1286 const char *value; | |
1287 enum dsync_slave_recv_ret ret; | |
1288 | |
1289 p_clear(slave->ret_pool); | |
1290 request = p_new(slave->ret_pool, struct dsync_mail_request, 1); | |
1291 | |
1292 ret = dsync_slave_io_input_next(slave, ITEM_MAIL_REQUEST, &decoder); | |
1293 if (ret != DSYNC_SLAVE_RECV_RET_OK) | |
1294 return ret; | |
1295 | |
1296 if (dsync_deserializer_decode_try(decoder, "guid", &value)) | |
1297 request->guid = p_strdup(slave->ret_pool, value); | |
1298 if (dsync_deserializer_decode_try(decoder, "uid", &value) && | |
1299 str_to_uint32(value, &request->uid) < 0) { | |
1300 dsync_slave_input_error(slave, decoder, "Invalid uid"); | |
1301 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1302 } | |
1303 | |
1304 *request_r = request; | |
1305 return DSYNC_SLAVE_RECV_RET_OK; | |
1306 } | |
1307 | |
1308 static void | |
1309 dsync_slave_io_send_mail(struct dsync_slave *_slave, | |
1310 const struct dsync_mail *mail) | |
1311 { | |
1312 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
1313 struct dsync_serializer_encoder *encoder; | |
1314 string_t *str = t_str_new(128); | |
1315 | |
1316 i_assert(slave->mail_output == NULL); | |
1317 | |
1318 str_append_c(str, items[ITEM_MAIL].chr); | |
1319 encoder = dsync_serializer_encode_begin(slave->serializers[ITEM_MAIL]); | |
1320 if (mail->guid != NULL) | |
1321 dsync_serializer_encode_add(encoder, "guid", mail->guid); | |
1322 if (mail->uid != 0) | |
1323 dsync_serializer_encode_add(encoder, "uid", dec2str(mail->uid)); | |
1324 if (mail->pop3_uidl != NULL) { | |
1325 dsync_serializer_encode_add(encoder, "pop3_uidl", | |
1326 mail->pop3_uidl); | |
1327 } | |
1328 if (mail->pop3_order > 0) { | |
1329 dsync_serializer_encode_add(encoder, "pop3_order", | |
1330 dec2str(mail->pop3_order)); | |
1331 } | |
1332 if (mail->received_date > 0) { | |
1333 dsync_serializer_encode_add(encoder, "received_date", | |
1334 dec2str(mail->received_date)); | |
1335 } | |
1336 if (mail->input != NULL) | |
1337 dsync_serializer_encode_add(encoder, "stream", ""); | |
1338 | |
1339 dsync_serializer_encode_finish(&encoder, str); | |
1340 dsync_slave_io_send_string(slave, str); | |
1341 | |
1342 if (mail->input != NULL) { | |
1343 slave->mail_output_last = '\0'; | |
1344 slave->mail_output = mail->input; | |
1345 i_stream_ref(slave->mail_output); | |
1346 (void)dsync_slave_io_send_mail_stream(slave); | |
1347 } | |
1348 } | |
1349 | |
1350 static int seekable_fd_callback(const char **path_r, void *context) | |
1351 { | |
1352 struct dsync_slave_io *slave = context; | |
1353 string_t *path; | |
1354 int fd; | |
1355 | |
1356 path = t_str_new(128); | |
1357 str_append(path, slave->temp_path_prefix); | |
1358 fd = safe_mkstemp(path, 0600, (uid_t)-1, (gid_t)-1); | |
1359 if (fd == -1) { | |
1360 i_error("safe_mkstemp(%s) failed: %m", str_c(path)); | |
1361 return -1; | |
1362 } | |
1363 | |
1364 /* we just want the fd, unlink it */ | |
1365 if (unlink(str_c(path)) < 0) { | |
1366 /* shouldn't happen.. */ | |
1367 i_error("unlink(%s) failed: %m", str_c(path)); | |
14691
3945a3646c67
Changed i_close_fd() API to set the fd to -1 after closing.
Timo Sirainen <tss@iki.fi>
parents:
14687
diff
changeset
|
1368 i_close_fd(&fd); |
14584 | 1369 return -1; |
1370 } | |
1371 | |
1372 *path_r = str_c(path); | |
1373 return fd; | |
1374 } | |
1375 | |
1376 static enum dsync_slave_recv_ret | |
1377 dsync_slave_io_recv_mail(struct dsync_slave *_slave, | |
1378 struct dsync_mail **mail_r) | |
1379 { | |
1380 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
1381 pool_t pool = slave->ret_pool; | |
1382 struct dsync_deserializer_decoder *decoder; | |
1383 struct dsync_mail *mail; | |
1384 struct istream *inputs[2]; | |
1385 const char *value; | |
1386 enum dsync_slave_recv_ret ret; | |
1387 | |
1388 if (slave->mail_input != NULL) { | |
1389 /* wait until the mail's stream has been read */ | |
1390 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1391 } | |
1392 if (slave->cur_mail != NULL) { | |
1393 /* finished reading the stream, return the mail now */ | |
1394 *mail_r = slave->cur_mail; | |
1395 slave->cur_mail = NULL; | |
1396 return DSYNC_SLAVE_RECV_RET_OK; | |
1397 } | |
1398 | |
1399 p_clear(pool); | |
1400 mail = p_new(pool, struct dsync_mail, 1); | |
1401 | |
1402 ret = dsync_slave_io_input_next(slave, ITEM_MAIL, &decoder); | |
1403 if (ret != DSYNC_SLAVE_RECV_RET_OK) | |
1404 return ret; | |
1405 | |
1406 if (dsync_deserializer_decode_try(decoder, "guid", &value)) | |
1407 mail->guid = p_strdup(pool, value); | |
1408 if (dsync_deserializer_decode_try(decoder, "uid", &value) && | |
1409 str_to_uint32(value, &mail->uid) < 0) { | |
1410 dsync_slave_input_error(slave, decoder, "Invalid uid"); | |
1411 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1412 } | |
1413 if (dsync_deserializer_decode_try(decoder, "pop3_uidl", &value)) | |
1414 mail->pop3_uidl = p_strdup(pool, value); | |
1415 if (dsync_deserializer_decode_try(decoder, "pop3_order", &value) && | |
1416 str_to_uint(value, &mail->pop3_order) < 0) { | |
1417 dsync_slave_input_error(slave, decoder, "Invalid pop3_order"); | |
1418 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1419 } | |
1420 if (dsync_deserializer_decode_try(decoder, "received_date", &value) && | |
1421 str_to_time(value, &mail->received_date) < 0) { | |
1422 dsync_slave_input_error(slave, decoder, "Invalid received_date"); | |
1423 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1424 } | |
1425 if (dsync_deserializer_decode_try(decoder, "stream", &value)) { | |
1426 inputs[0] = i_stream_create_dot(slave->input, FALSE); | |
1427 inputs[1] = NULL; | |
1428 mail->input = i_stream_create_seekable(inputs, | |
1429 MAIL_READ_FULL_BLOCK_SIZE, seekable_fd_callback, slave); | |
1430 i_stream_unref(&inputs[0]); | |
1431 | |
1432 slave->mail_input = mail->input; | |
1433 if (dsync_slave_io_read_mail_stream(slave) <= 0) { | |
1434 slave->cur_mail = mail; | |
1435 return DSYNC_SLAVE_RECV_RET_TRYAGAIN; | |
1436 } | |
1437 /* already finished reading the stream */ | |
1438 i_assert(slave->mail_input == NULL); | |
1439 } | |
1440 | |
1441 *mail_r = mail; | |
1442 return DSYNC_SLAVE_RECV_RET_OK; | |
1443 } | |
1444 | |
1445 static void dsync_slave_io_flush(struct dsync_slave *_slave) | |
1446 { | |
1447 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
1448 | |
1449 o_stream_uncork(slave->output); | |
1450 o_stream_cork(slave->output); | |
1451 } | |
1452 | |
1453 static bool dsync_slave_io_is_send_queue_full(struct dsync_slave *_slave) | |
1454 { | |
1455 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
1456 size_t bytes; | |
1457 | |
1458 if (slave->mail_output != NULL) | |
1459 return TRUE; | |
1460 | |
1461 bytes = o_stream_get_buffer_used_size(slave->output); | |
1462 if (bytes < DSYNC_SLAVE_IO_OUTBUF_THROTTLE_SIZE) | |
1463 return FALSE; | |
1464 | |
1465 o_stream_set_flush_pending(slave->output, TRUE); | |
1466 return TRUE; | |
1467 } | |
1468 | |
1469 static bool dsync_slave_io_has_pending_data(struct dsync_slave *_slave) | |
1470 { | |
1471 struct dsync_slave_io *slave = (struct dsync_slave_io *)_slave; | |
1472 | |
1473 return slave->has_pending_data; | |
1474 } | |
1475 | |
1476 static const struct dsync_slave_vfuncs dsync_slave_io_vfuncs = { | |
1477 dsync_slave_io_deinit, | |
1478 dsync_slave_io_send_handshake, | |
1479 dsync_slave_io_recv_handshake, | |
1480 dsync_slave_io_send_end_of_list, | |
1481 dsync_slave_io_send_mailbox_state, | |
1482 dsync_slave_io_recv_mailbox_state, | |
1483 dsync_slave_io_send_mailbox_tree_node, | |
1484 dsync_slave_io_recv_mailbox_tree_node, | |
1485 dsync_slave_io_send_mailbox_deletes, | |
1486 dsync_slave_io_recv_mailbox_deletes, | |
1487 dsync_slave_io_send_mailbox, | |
1488 dsync_slave_io_recv_mailbox, | |
1489 dsync_slave_io_send_change, | |
1490 dsync_slave_io_recv_change, | |
1491 dsync_slave_io_send_mail_request, | |
1492 dsync_slave_io_recv_mail_request, | |
1493 dsync_slave_io_send_mail, | |
1494 dsync_slave_io_recv_mail, | |
1495 dsync_slave_io_flush, | |
1496 dsync_slave_io_is_send_queue_full, | |
1497 dsync_slave_io_has_pending_data | |
1498 }; | |
1499 | |
1500 struct dsync_slave * | |
1501 dsync_slave_init_io(int fd_in, int fd_out, const char *name, | |
1502 const char *temp_path_prefix) | |
1503 { | |
1504 struct dsync_slave_io *slave; | |
1505 | |
1506 slave = i_new(struct dsync_slave_io, 1); | |
1507 slave->slave.v = dsync_slave_io_vfuncs; | |
1508 slave->fd_in = fd_in; | |
1509 slave->fd_out = fd_out; | |
1510 slave->name = i_strdup(name); | |
1511 slave->temp_path_prefix = i_strdup(temp_path_prefix); | |
1512 slave->ret_pool = pool_alloconly_create("slave io data", 2048); | |
1513 dsync_slave_io_init(slave); | |
1514 return &slave->slave; | |
1515 } |