Lines Matching refs:message
5 /* in-core message structures
18 /* QMGR_MESSAGE *qmgr_message_realloc(message)
19 /* QMGR_MESSAGE *message;
21 /* void qmgr_message_free(message)
22 /* QMGR_MESSAGE *message;
24 /* void qmgr_message_update_warn(message)
25 /* QMGR_MESSAGE *message;
27 /* void qmgr_message_kill_record(message, offset)
28 /* QMGR_MESSAGE *message;
34 /* of in-core message structures (i.e. the total size of the
35 /* `active' message queue).
39 /* in all in-core message structures).
41 /* qmgr_message_alloc() creates an in-core message structure
56 /* file, and updates the recipient list and \fIrcpt_offset\fR message
60 /* qmgr_message_free() destroys an in-core message structure and makes
62 /* a message structure that is still referenced by queue entry structures.
64 /* qmgr_message_update_warn() takes a closed message, opens it, updates
67 /* qmgr_message_kill_record() takes a closed message, opens it, updates
70 /* inside the message content.
72 /* Warnings: malformed message file. Fatal errors: out of memory.
74 /* envelope(3) message envelope parser
144 /* qmgr_message_create - create in-core message structure */
149 QMGR_MESSAGE *message;
151 message = (QMGR_MESSAGE *) mymalloc(sizeof(QMGR_MESSAGE));
153 message->flags = 0;
154 message->qflags = qflags;
155 message->tflags = 0;
156 message->tflags_offset = 0;
157 message->rflags = QMGR_READ_FLAG_DEFAULT;
158 message->fp = 0;
159 message->refcount = 0;
160 message->single_rcpt = 0;
161 message->arrival_time.tv_sec = message->arrival_time.tv_usec = 0;
162 message->create_time = 0;
163 GETTIMEOFDAY(&message->active_time);
164 message->data_offset = 0;
165 message->queue_id = mystrdup(queue_id);
166 message->queue_name = mystrdup(queue_name);
167 message->encoding = 0;
168 message->sender = 0;
169 message->dsn_envid = 0;
170 message->dsn_ret = 0;
171 message->filter_xport = 0;
172 message->inspect_xport = 0;
173 message->redirect_addr = 0;
174 message->data_size = 0;
175 message->cont_length = 0;
176 message->warn_offset = 0;
177 message->warn_time = 0;
178 message->rcpt_offset = 0;
179 message->verp_delims = 0;
180 message->client_name = 0;
181 message->client_addr = 0;
182 message->client_port = 0;
183 message->client_proto = 0;
184 message->client_helo = 0;
185 message->sasl_method = 0;
186 message->sasl_username = 0;
187 message->sasl_sender = 0;
188 message->log_ident = 0;
189 message->rewrite_context = 0;
190 recipient_list_init(&message->rcpt_list, RCPT_LIST_INIT_QUEUE);
191 return (message);
196 static void qmgr_message_close(QMGR_MESSAGE *message)
198 vstream_fclose(message->fp);
199 message->fp = 0;
204 static int qmgr_message_open(QMGR_MESSAGE *message)
210 if (message->fp)
211 msg_panic("%s: queue file is open", message->queue_id);
217 if ((message->fp = mail_queue_open(message->queue_name,
218 message->queue_id,
221 msg_fatal("open %s %s: %m", message->queue_name, message->queue_id);
222 msg_warn("open %s %s: %m", message->queue_name, message->queue_id);
230 static void qmgr_message_oldstyle_scan(QMGR_MESSAGE *message)
241 if ((orig_offset = vstream_ftell(message->fp)) < 0)
242 msg_fatal("vstream_ftell %s: %m", VSTREAM_PATH(message->fp));
247 if (vstream_fseek(message->fp, 0, SEEK_SET) < 0)
248 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
258 rec_type = rec_get(message->fp, buf, 0);
268 if (message->data_offset == 0) {
269 if ((message->data_offset = vstream_ftell(message->fp)) < 0)
270 msg_fatal("vstream_ftell %s: %m", VSTREAM_PATH(message->fp));
271 if ((extra_offset = atol(start)) <= message->data_offset)
273 start, VSTREAM_PATH(message->fp));
274 if (vstream_fseek(message->fp, extra_offset, SEEK_SET) < 0)
275 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
276 message->data_size = extra_offset - message->data_offset;
285 if (vstream_fseek(message->fp, orig_offset, SEEK_SET) < 0)
286 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
293 if (message->data_offset == 0 || rec_type != REC_TYPE_END)
294 msg_fatal("%s: envelope records out of order", message->queue_id);
299 static int qmgr_message_read(QMGR_MESSAGE *message)
304 long save_offset = message->rcpt_offset; /* save a flag */
326 if (message->rcpt_offset) {
327 if (message->rcpt_list.len)
329 message->queue_id);
330 if (vstream_fseek(message->fp, message->rcpt_offset, SEEK_SET) < 0)
331 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
332 message->rcpt_offset = 0;
339 * may appear before or after the message content, so we keep reading
344 * per-message in-core recipient limit rather than a global in-core
350 * number of in-core recipients per message would asymptotically approach
352 * delay per recipient rather than equal delay per message.
363 if ((curr_offset = vstream_ftell(message->fp)) < 0)
364 msg_fatal("vstream_ftell %s: %m", VSTREAM_PATH(message->fp));
365 if (curr_offset == message->data_offset && curr_offset > 0) {
366 if (vstream_fseek(message->fp, message->data_size, SEEK_CUR) < 0)
367 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
368 curr_offset += message->data_size;
370 rec_type = rec_get_raw(message->fp, buf, 0, REC_FLAG_NONE);
375 if ((rec_type = rec_goto(message->fp, start)) == REC_TYPE_ERROR)
381 msg_warn("%s: message rejected: missing end record",
382 message->queue_id);
386 message->rflags |= QMGR_READ_FLAG_SEEN_ALL_NON_RCPT;
399 message->queue_id, error_text, start);
415 if (message->rcpt_offset == 0) {
416 recipient_list_add(&message->rcpt_list, curr_offset,
430 if (message->rcpt_list.len >= FUDGE(var_qmgr_rcpt_limit)) {
431 if ((message->rcpt_offset = vstream_ftell(message->fp)) < 0)
433 VSTREAM_PATH(message->fp));
434 if (message->rflags & QMGR_READ_FLAG_SEEN_ALL_NON_RCPT)
437 if (message->rflags & QMGR_READ_FLAG_MIXED_RCPT_OTHER)
441 if (curr_offset > message->data_offset) {
443 message->rflags |= QMGR_READ_FLAG_SEEN_ALL_NON_RCPT;
447 if (vstream_fseek(message->fp, message->data_offset
448 + message->data_size, SEEK_SET) < 0)
449 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
456 if (message->rcpt_offset == 0) {
474 message->queue_id, dsn_orcpt);
478 if (message->rcpt_offset == 0)
486 message->queue_id, dsn_notify);
489 if (message->rcpt_offset == 0) {
492 message->queue_id, start);
502 message->queue_id, orig_rcpt);
506 if (message->rcpt_offset == 0)
514 if (message->rflags & QMGR_READ_FLAG_SEEN_ALL_NON_RCPT)
518 if (message->data_offset == 0) {
520 &message->data_size, &message->data_offset,
521 &nrcpt, &message->rflags,
522 &message->cont_length)) >= 3) {
524 if (message->data_offset <= 0 || message->data_size <= 0) {
526 message->queue_id, start);
530 if (message->rflags & ~QMGR_READ_FLAG_USER) {
532 message->queue_id, start);
538 qmgr_message_oldstyle_scan(message);
541 msg_warn("%s: message rejected: weird size record",
542 message->queue_id);
548 if (message->cont_length == 0) {
549 message->cont_length = message->data_size;
550 } else if (message->cont_length < 0) {
552 message->queue_id, start);
559 if (message->arrival_time.tv_sec == 0)
560 REC_TYPE_TIME_SCAN(start, message->arrival_time);
564 if (message->create_time == 0)
565 message->create_time = atol(start);
569 if (message->filter_xport != 0)
570 myfree(message->filter_xport);
571 message->filter_xport = mystrdup(start);
575 if (message->inspect_xport != 0)
576 myfree(message->inspect_xport);
577 message->inspect_xport = mystrdup(start);
581 if (message->redirect_addr != 0)
582 myfree(message->redirect_addr);
583 message->redirect_addr = mystrdup(start);
587 if (message->sender == 0) {
588 message->sender = mystrdup(start);
589 opened(message->queue_id, message->sender,
590 message->cont_length, nrcpt,
591 "queue %s", message->queue_name);
596 if (message->dsn_envid == 0)
597 message->dsn_envid = mystrdup(start);
600 if (message->dsn_ret == 0) {
603 message->queue_id, start);
605 message->dsn_ret = n;
611 if (message->encoding != 0)
612 myfree(message->encoding);
613 message->encoding = mystrdup(value);
631 if (have_log_client_attr == 0 && message->client_name == 0)
632 message->client_name = mystrdup(value);
634 if (have_log_client_attr == 0 && message->client_addr == 0)
635 message->client_addr = mystrdup(value);
637 if (have_log_client_attr == 0 && message->client_port == 0)
638 message->client_port = mystrdup(value);
640 if (have_log_client_attr == 0 && message->client_proto == 0)
641 message->client_proto = mystrdup(value);
643 if (have_log_client_attr == 0 && message->client_helo == 0)
644 message->client_helo = mystrdup(value);
648 if (message->client_name != 0)
649 myfree(message->client_name);
650 message->client_name = mystrdup(value);
653 if (message->client_addr != 0)
654 myfree(message->client_addr);
655 message->client_addr = mystrdup(value);
658 if (message->client_port != 0)
659 myfree(message->client_port);
660 message->client_port = mystrdup(value);
663 if (message->client_proto != 0)
664 myfree(message->client_proto);
665 message->client_proto = mystrdup(value);
668 if (message->client_helo != 0)
669 myfree(message->client_helo);
670 message->client_helo = mystrdup(value);
673 if (message->sasl_method == 0)
674 message->sasl_method = mystrdup(value);
677 message->queue_id, MAIL_ATTR_SASL_METHOD, value);
679 if (message->sasl_username == 0)
680 message->sasl_username = mystrdup(value);
683 message->queue_id, MAIL_ATTR_SASL_USERNAME, value);
685 if (message->sasl_sender == 0)
686 message->sasl_sender = mystrdup(value);
689 message->queue_id, MAIL_ATTR_SASL_SENDER, value);
691 if (message->log_ident == 0)
692 message->log_ident = mystrdup(value);
695 message->queue_id, MAIL_ATTR_LOG_IDENT, value);
697 if (message->rewrite_context == 0)
698 message->rewrite_context = mystrdup(value);
701 message->queue_id, MAIL_ATTR_RWR_CONTEXT, value);
710 message->tflags = DEL_REQ_TRACE_FLAGS(atoi(value));
711 if (message->tflags == DEL_REQ_FLAG_RECORD)
712 message->tflags_offset = curr_offset;
714 message->tflags_offset = 0;
719 if (message->warn_offset == 0) {
720 message->warn_offset = curr_offset;
721 REC_TYPE_WARN_SCAN(start, message->warn_time);
726 if (message->verp_delims == 0) {
727 if (message->sender == 0 || message->sender[0] == 0) {
729 message->queue_id);
732 message->queue_id, start);
736 message->queue_id, message->sender);
737 message->single_rcpt = 1;
738 message->verp_delims = mystrdup(start);
751 message->queue_id, dsn_orcpt);
757 message->queue_id, orig_rcpt);
766 if (message->dsn_envid == 0)
767 message->dsn_envid = mystrdup("");
768 if (message->encoding == 0)
769 message->encoding = mystrdup(MAIL_ATTR_ENC_NONE);
770 if (message->client_name == 0)
771 message->client_name = mystrdup("");
772 if (message->client_addr == 0)
773 message->client_addr = mystrdup("");
774 if (message->client_port == 0)
775 message->client_port = mystrdup("");
776 if (message->client_proto == 0)
777 message->client_proto = mystrdup("");
778 if (message->client_helo == 0)
779 message->client_helo = mystrdup("");
780 if (message->sasl_method == 0)
781 message->sasl_method = mystrdup("");
782 if (message->sasl_username == 0)
783 message->sasl_username = mystrdup("");
784 if (message->sasl_sender == 0)
785 message->sasl_sender = mystrdup("");
786 if (message->log_ident == 0)
787 message->log_ident = mystrdup("");
788 if (message->rewrite_context == 0)
789 message->rewrite_context = mystrdup(MAIL_ATTR_RWR_LOCAL);
791 if (message->create_time == 0)
792 message->create_time = message->arrival_time.tv_sec;
805 } else if (message->arrival_time.tv_sec == 0) {
806 msg_warn("%s: message rejected: missing arrival time record",
807 message->queue_id);
808 } else if (message->sender == 0) {
809 msg_warn("%s: message rejected: missing sender record",
810 message->queue_id);
811 } else if (message->data_offset == 0) {
812 msg_warn("%s: message rejected: missing size record",
813 message->queue_id);
817 message->rcpt_offset = save_offset; /* restore flag */
823 void qmgr_message_update_warn(QMGR_MESSAGE *message)
830 if (qmgr_message_open(message)
831 || vstream_fseek(message->fp, message->warn_offset, SEEK_SET) < 0
832 || rec_fprintf(message->fp, REC_TYPE_WARN, REC_TYPE_WARN_FORMAT,
834 || vstream_fflush(message->fp))
835 msg_fatal("update queue file %s: %m", VSTREAM_PATH(message->fp));
836 qmgr_message_close(message);
839 /* qmgr_message_kill_record - mark one message record as killed */
841 void qmgr_message_kill_record(QMGR_MESSAGE *message, long offset)
845 if (qmgr_message_open(message)
846 || rec_put_type(message->fp, REC_TYPE_KILL, offset) < 0
847 || vstream_fflush(message->fp))
848 msg_fatal("update queue file %s: %m", VSTREAM_PATH(message->fp));
849 qmgr_message_close(message);
879 * Compare message transport.
911 /* qmgr_message_sort - sort message recipient addresses by domain */
913 static void qmgr_message_sort(QMGR_MESSAGE *message)
915 qsort((char *) message->rcpt_list.info, message->rcpt_list.len,
916 sizeof(message->rcpt_list.info[0]), qmgr_message_sort_compare);
918 RECIPIENT_LIST list = message->rcpt_list;
930 static int qmgr_resolve_one(QMGR_MESSAGE *message, RECIPIENT *recipient,
939 if ((message->tflags & DEL_REQ_FLAG_MTA_VRFY) == 0)
940 resolve_clnt_query_from(message->sender, addr, reply);
942 resolve_clnt_verify_from(message->sender, addr, reply);
958 static void qmgr_message_resolve(QMGR_MESSAGE *message)
961 RECIPIENT_LIST list = message->rcpt_list;
985 * Redirect overrides all else. But only once (per entire message).
989 if (message->redirect_addr) {
994 message->rcpt_offset = 0;
995 rewrite_clnt_internal(REWRITE_CANON, message->redirect_addr,
998 if (qmgr_resolve_one(message, recipient,
1019 else if (message->filter_xport
1020 && (message->tflags & DEL_REQ_TRACE_ONLY_MASK) == 0) {
1022 vstring_strcpy(reply.transport, message->filter_xport);
1036 if (qmgr_resolve_one(message, recipient,
1069 status = sent(message->tflags, message->queue_id,
1070 QMGR_MSG_STATS(&stats, message), recipient,
1074 deliver_completed(message->fp, recipient->offset);
1078 message->queue_id);
1081 message->flags |= status;
1090 if (*var_defer_xports && (message->qflags & QMGR_FLUSH_DFXP) == 0) {
1112 * This message is being flushed. If need-be unthrottle the
1115 if ((message->qflags & QMGR_FLUSH_EACH) != 0
1130 qmgr_defer_recipient(message, recipient, saved_dsn);
1139 * deliveries to different recipients of the same message can happen
1196 * This message is being flushed. If need-be unthrottle the queue.
1198 if ((message->qflags & QMGR_FLUSH_EACH) != 0
1208 qmgr_defer_recipient(message, recipient, saved_dsn);
1224 static void qmgr_message_assign(QMGR_MESSAGE *message)
1226 RECIPIENT_LIST list = message->rcpt_list;
1244 if (message->single_rcpt || entry == 0 || entry->queue != queue
1247 entry = qmgr_entry_create(queue, message);
1255 recipient_list_free(&message->rcpt_list);
1256 recipient_list_init(&message->rcpt_list, RCPT_LIST_INIT_QUEUE);
1259 /* qmgr_message_free - release memory for in-core message structure */
1261 void qmgr_message_free(QMGR_MESSAGE *message)
1263 if (message->refcount != 0)
1264 msg_panic("qmgr_message_free: reference len: %d", message->refcount);
1265 if (message->fp)
1267 myfree(message->queue_id);
1268 myfree(message->queue_name);
1269 if (message->dsn_envid)
1270 myfree(message->dsn_envid);
1271 if (message->encoding)
1272 myfree(message->encoding);
1273 if (message->sender)
1274 myfree(message->sender);
1275 if (message->verp_delims)
1276 myfree(message->verp_delims);
1277 if (message->filter_xport)
1278 myfree(message->filter_xport);
1279 if (message->inspect_xport)
1280 myfree(message->inspect_xport);
1281 if (message->redirect_addr)
1282 myfree(message->redirect_addr);
1283 if (message->client_name)
1284 myfree(message->client_name);
1285 if (message->client_addr)
1286 myfree(message->client_addr);
1287 if (message->client_port)
1288 myfree(message->client_port);
1289 if (message->client_proto)
1290 myfree(message->client_proto);
1291 if (message->client_helo)
1292 myfree(message->client_helo);
1293 if (message->sasl_method)
1294 myfree(message->sasl_method);
1295 if (message->sasl_username)
1296 myfree(message->sasl_username);
1297 if (message->sasl_sender)
1298 myfree(message->sasl_sender);
1299 if (message->log_ident)
1300 myfree(message->log_ident);
1301 if (message->rewrite_context)
1302 myfree(message->rewrite_context);
1303 recipient_list_free(&message->rcpt_list);
1305 myfree((char *) message);
1308 /* qmgr_message_alloc - create in-core message structure */
1314 QMGR_MESSAGE *message;
1320 * Create an in-core message structure.
1322 message = qmgr_message_create(queue_name, queue_id, qflags);
1325 * Extract message envelope information: time of arrival, sender address,
1330 if (qmgr_message_open(message) < 0) {
1331 qmgr_message_free(message);
1334 if (myflock(vstream_fileno(message->fp), INTERNAL_LOCK, QMGR_LOCK_MODE) < 0) {
1336 qmgr_message_close(message);
1337 qmgr_message_free(message);
1340 if (qmgr_message_read(message) < 0) {
1341 qmgr_message_close(message);
1342 qmgr_message_free(message);
1350 if (mode != 0 && fchmod(vstream_fileno(message->fp), mode) < 0)
1351 msg_fatal("fchmod %s: %m", VSTREAM_PATH(message->fp));
1367 qmgr_message_sort(message);
1368 qmgr_message_resolve(message);
1369 qmgr_message_sort(message);
1370 qmgr_message_assign(message);
1371 qmgr_message_close(message);
1372 return (message);
1376 /* qmgr_message_realloc - refresh in-core message structure */
1378 QMGR_MESSAGE *qmgr_message_realloc(QMGR_MESSAGE *message)
1385 if (message->rcpt_offset <= 0)
1386 msg_panic("%s: invalid offset: %ld", myname, message->rcpt_offset);
1388 msg_info("%s: %s %s offset %ld", myname, message->queue_name,
1389 message->queue_id, message->rcpt_offset);
1395 if (qmgr_message_open(message) < 0)
1397 if (qmgr_message_read(message) < 0) {
1398 qmgr_message_close(message);
1401 qmgr_message_sort(message);
1402 qmgr_message_resolve(message);
1403 qmgr_message_sort(message);
1404 qmgr_message_assign(message);
1405 qmgr_message_close(message);
1406 return (message);