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
59 /* limit imposed this time is based on the position of the message
63 /* qmgr_message_free() destroys an in-core message structure and makes
65 /* a message structure that is still referenced by queue entry structures.
67 /* qmgr_message_update_warn() takes a closed message, opens it, updates
70 /* qmgr_message_kill_record() takes a closed message, opens it, updates
73 /* inside the message content.
75 /* Warnings: malformed message file. Fatal errors: out of memory.
77 /* envelope(3) message envelope parser
153 /* qmgr_message_create - create in-core message structure */
158 QMGR_MESSAGE *message;
160 message = (QMGR_MESSAGE *) mymalloc(sizeof(QMGR_MESSAGE));
162 message->flags = 0;
163 message->qflags = qflags;
164 message->tflags = 0;
165 message->tflags_offset = 0;
166 message->rflags = QMGR_READ_FLAG_DEFAULT;
167 message->fp = 0;
168 message->refcount = 0;
169 message->single_rcpt = 0;
170 message->arrival_time.tv_sec = message->arrival_time.tv_usec = 0;
171 message->create_time = 0;
172 GETTIMEOFDAY(&message->active_time);
173 message->queued_time = sane_time();
174 message->refill_time = 0;
175 message->data_offset = 0;
176 message->queue_id = mystrdup(queue_id);
177 message->queue_name = mystrdup(queue_name);
178 message->encoding = 0;
179 message->sender = 0;
180 message->dsn_envid = 0;
181 message->dsn_ret = 0;
182 message->filter_xport = 0;
183 message->inspect_xport = 0;
184 message->redirect_addr = 0;
185 message->data_size = 0;
186 message->cont_length = 0;
187 message->warn_offset = 0;
188 message->warn_time = 0;
189 message->rcpt_offset = 0;
190 message->verp_delims = 0;
191 message->client_name = 0;
192 message->client_addr = 0;
193 message->client_port = 0;
194 message->client_proto = 0;
195 message->client_helo = 0;
196 message->sasl_method = 0;
197 message->sasl_username = 0;
198 message->sasl_sender = 0;
199 message->log_ident = 0;
200 message->rewrite_context = 0;
201 recipient_list_init(&message->rcpt_list, RCPT_LIST_INIT_QUEUE);
202 message->rcpt_count = 0;
203 message->rcpt_limit = var_qmgr_msg_rcpt_limit;
204 message->rcpt_unread = 0;
205 QMGR_LIST_INIT(message->job_list);
206 return (message);
211 static void qmgr_message_close(QMGR_MESSAGE *message)
213 vstream_fclose(message->fp);
214 message->fp = 0;
219 static int qmgr_message_open(QMGR_MESSAGE *message)
225 if (message->fp)
226 msg_panic("%s: queue file is open", message->queue_id);
232 if ((message->fp = mail_queue_open(message->queue_name,
233 message->queue_id,
236 msg_fatal("open %s %s: %m", message->queue_name, message->queue_id);
237 msg_warn("open %s %s: %m", message->queue_name, message->queue_id);
245 static void qmgr_message_oldstyle_scan(QMGR_MESSAGE *message)
256 if ((orig_offset = vstream_ftell(message->fp)) < 0)
257 msg_fatal("vstream_ftell %s: %m", VSTREAM_PATH(message->fp));
262 if (vstream_fseek(message->fp, 0, SEEK_SET) < 0)
263 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
272 message->rcpt_unread = 0;
274 rec_type = rec_get(message->fp, buf, 0);
286 message->rcpt_unread++;
290 if (message->data_offset == 0) {
291 if ((message->data_offset = vstream_ftell(message->fp)) < 0)
292 msg_fatal("vstream_ftell %s: %m", VSTREAM_PATH(message->fp));
293 if ((extra_offset = atol(start)) <= message->data_offset)
295 start, VSTREAM_PATH(message->fp));
296 if (vstream_fseek(message->fp, extra_offset, SEEK_SET) < 0)
297 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
298 message->data_size = extra_offset - message->data_offset;
307 if (vstream_fseek(message->fp, orig_offset, SEEK_SET) < 0)
308 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
315 if (message->data_offset == 0 || rec_type != REC_TYPE_END)
316 msg_fatal("%s: envelope records out of order", message->queue_id);
321 static int qmgr_message_read(QMGR_MESSAGE *message)
326 long save_offset = message->rcpt_offset; /* save a flag */
327 int save_unread = message->rcpt_unread; /* save a count */
349 * For the first time, the message recipient limit is calculated from the
353 * because there are no message jobs yet.
355 * For the next time, the recipient limit is based solely on the message
358 if (message->rcpt_offset) {
359 if (message->rcpt_list.len)
361 message->queue_id);
362 if (vstream_fseek(message->fp, message->rcpt_offset, SEEK_SET) < 0)
363 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
364 message->rcpt_offset = 0;
365 recipient_limit = message->rcpt_limit - message->rcpt_count;
368 if (recipient_limit < message->rcpt_limit)
369 recipient_limit = message->rcpt_limit;
375 msg_panic("%s: no recipient slots available", message->queue_id);
377 msg_info("%s: recipient limit %d", message->queue_id, recipient_limit);
383 * may appear before or after the message content, so we keep reading
403 if ((curr_offset = vstream_ftell(message->fp)) < 0)
404 msg_fatal("vstream_ftell %s: %m", VSTREAM_PATH(message->fp));
405 if (curr_offset == message->data_offset && curr_offset > 0) {
406 if (vstream_fseek(message->fp, message->data_size, SEEK_CUR) < 0)
407 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
408 curr_offset += message->data_size;
410 rec_type = rec_get_raw(message->fp, buf, 0, REC_FLAG_NONE);
415 if ((rec_type = rec_goto(message->fp, start)) == REC_TYPE_ERROR)
421 msg_warn("%s: message rejected: missing end record",
422 message->queue_id);
426 message->rflags |= QMGR_READ_FLAG_SEEN_ALL_NON_RCPT;
439 message->queue_id, error_text, start);
454 if (message->rcpt_offset == 0) {
455 message->rcpt_unread--;
456 recipient_list_add(&message->rcpt_list, curr_offset,
470 if (message->rcpt_list.len >= recipient_limit) {
471 if ((message->rcpt_offset = vstream_ftell(message->fp)) < 0)
473 VSTREAM_PATH(message->fp));
474 if (message->rflags & QMGR_READ_FLAG_SEEN_ALL_NON_RCPT)
477 if (message->rflags & QMGR_READ_FLAG_MIXED_RCPT_OTHER)
481 if (curr_offset > message->data_offset) {
483 message->rflags |= QMGR_READ_FLAG_SEEN_ALL_NON_RCPT;
487 if (vstream_fseek(message->fp, message->data_offset
488 + message->data_size, SEEK_SET) < 0)
489 msg_fatal("seek file %s: %m", VSTREAM_PATH(message->fp));
496 if (message->rcpt_offset == 0) {
497 message->rcpt_unread--;
515 message->queue_id, dsn_orcpt);
519 if (message->rcpt_offset == 0)
527 message->queue_id, dsn_notify);
530 if (message->rcpt_offset == 0) {
533 message->queue_id, start);
543 message->queue_id, orig_rcpt);
547 if (message->rcpt_offset == 0)
555 if (message->rflags & QMGR_READ_FLAG_SEEN_ALL_NON_RCPT)
559 if (message->data_offset == 0) {
561 &message->data_size, &message->data_offset,
562 &message->rcpt_unread, &message->rflags,
563 &message->cont_length)) >= 3) {
565 if (message->data_offset <= 0 || message->data_size <= 0) {
567 message->queue_id, start);
571 if (message->rflags & ~QMGR_READ_FLAG_USER) {
573 message->queue_id, start);
579 qmgr_message_oldstyle_scan(message);
582 msg_warn("%s: message rejected: weird size record",
583 message->queue_id);
589 if (message->cont_length == 0) {
590 message->cont_length = message->data_size;
591 } else if (message->cont_length < 0) {
593 message->queue_id, start);
600 if (message->arrival_time.tv_sec == 0)
601 REC_TYPE_TIME_SCAN(start, message->arrival_time);
605 if (message->create_time == 0)
606 message->create_time = atol(start);
610 if (message->filter_xport != 0)
611 myfree(message->filter_xport);
612 message->filter_xport = mystrdup(start);
616 if (message->inspect_xport != 0)
617 myfree(message->inspect_xport);
618 message->inspect_xport = mystrdup(start);
622 if (message->redirect_addr != 0)
623 myfree(message->redirect_addr);
624 message->redirect_addr = mystrdup(start);
628 if (message->sender == 0) {
629 message->sender = mystrdup(start);
630 opened(message->queue_id, message->sender,
631 message->cont_length, message->rcpt_unread,
632 "queue %s", message->queue_name);
637 if (message->dsn_envid == 0)
638 message->dsn_envid = mystrdup(start);
641 if (message->dsn_ret == 0) {
644 message->queue_id, start);
646 message->dsn_ret = n;
652 if (message->encoding != 0)
653 myfree(message->encoding);
654 message->encoding = mystrdup(value);
672 if (have_log_client_attr == 0 && message->client_name == 0)
673 message->client_name = mystrdup(value);
675 if (have_log_client_attr == 0 && message->client_addr == 0)
676 message->client_addr = mystrdup(value);
678 if (have_log_client_attr == 0 && message->client_port == 0)
679 message->client_port = mystrdup(value);
681 if (have_log_client_attr == 0 && message->client_proto == 0)
682 message->client_proto = mystrdup(value);
684 if (have_log_client_attr == 0 && message->client_helo == 0)
685 message->client_helo = mystrdup(value);
689 if (message->client_name != 0)
690 myfree(message->client_name);
691 message->client_name = mystrdup(value);
694 if (message->client_addr != 0)
695 myfree(message->client_addr);
696 message->client_addr = mystrdup(value);
699 if (message->client_port != 0)
700 myfree(message->client_port);
701 message->client_port = mystrdup(value);
704 if (message->client_proto != 0)
705 myfree(message->client_proto);
706 message->client_proto = mystrdup(value);
709 if (message->client_helo != 0)
710 myfree(message->client_helo);
711 message->client_helo = mystrdup(value);
714 if (message->sasl_method == 0)
715 message->sasl_method = mystrdup(value);
718 message->queue_id, MAIL_ATTR_SASL_METHOD, value);
720 if (message->sasl_username == 0)
721 message->sasl_username = mystrdup(value);
724 message->queue_id, MAIL_ATTR_SASL_USERNAME, value);
726 if (message->sasl_sender == 0)
727 message->sasl_sender = mystrdup(value);
730 message->queue_id, MAIL_ATTR_SASL_SENDER, value);
732 if (message->log_ident == 0)
733 message->log_ident = mystrdup(value);
736 message->queue_id, MAIL_ATTR_LOG_IDENT, value);
738 if (message->rewrite_context == 0)
739 message->rewrite_context = mystrdup(value);
742 message->queue_id, MAIL_ATTR_RWR_CONTEXT, value);
751 message->tflags = DEL_REQ_TRACE_FLAGS(atoi(value));
752 if (message->tflags == DEL_REQ_FLAG_RECORD)
753 message->tflags_offset = curr_offset;
755 message->tflags_offset = 0;
760 if (message->warn_offset == 0) {
761 message->warn_offset = curr_offset;
762 REC_TYPE_WARN_SCAN(start, message->warn_time);
767 if (message->verp_delims == 0) {
768 if (message->sender == 0 || message->sender[0] == 0) {
770 message->queue_id);
773 message->queue_id, start);
777 message->queue_id, message->sender);
778 message->single_rcpt = 1;
779 message->verp_delims = mystrdup(start);
792 message->queue_id, dsn_orcpt);
798 message->queue_id, orig_rcpt);
807 message->refill_time = sane_time();
814 if (message->dsn_envid == 0)
815 message->dsn_envid = mystrdup("");
816 if (message->encoding == 0)
817 message->encoding = mystrdup(MAIL_ATTR_ENC_NONE);
818 if (message->client_name == 0)
819 message->client_name = mystrdup("");
820 if (message->client_addr == 0)
821 message->client_addr = mystrdup("");
822 if (message->client_port == 0)
823 message->client_port = mystrdup("");
824 if (message->client_proto == 0)
825 message->client_proto = mystrdup("");
826 if (message->client_helo == 0)
827 message->client_helo = mystrdup("");
828 if (message->sasl_method == 0)
829 message->sasl_method = mystrdup("");
830 if (message->sasl_username == 0)
831 message->sasl_username = mystrdup("");
832 if (message->sasl_sender == 0)
833 message->sasl_sender = mystrdup("");
834 if (message->log_ident == 0)
835 message->log_ident = mystrdup("");
836 if (message->rewrite_context == 0)
837 message->rewrite_context = mystrdup(MAIL_ATTR_RWR_LOCAL);
839 if (message->create_time == 0)
840 message->create_time = message->arrival_time.tv_sec;
851 if (message->rcpt_unread < 0
852 || (message->rcpt_offset == 0 && message->rcpt_unread != 0)) {
854 message->queue_id, message->rcpt_unread);
855 message->rcpt_unread = 0;
859 } else if (message->arrival_time.tv_sec == 0) {
860 msg_warn("%s: message rejected: missing arrival time record",
861 message->queue_id);
862 } else if (message->sender == 0) {
863 msg_warn("%s: message rejected: missing sender record",
864 message->queue_id);
865 } else if (message->data_offset == 0) {
866 msg_warn("%s: message rejected: missing size record",
867 message->queue_id);
871 message->rcpt_offset = save_offset; /* restore flag */
872 message->rcpt_unread = save_unread; /* restore count */
873 recipient_list_free(&message->rcpt_list);
874 recipient_list_init(&message->rcpt_list, RCPT_LIST_INIT_QUEUE);
880 void qmgr_message_update_warn(QMGR_MESSAGE *message)
887 if (qmgr_message_open(message)
888 || vstream_fseek(message->fp, message->warn_offset, SEEK_SET) < 0
889 || rec_fprintf(message->fp, REC_TYPE_WARN, REC_TYPE_WARN_FORMAT,
891 || vstream_fflush(message->fp))
892 msg_fatal("update queue file %s: %m", VSTREAM_PATH(message->fp));
893 qmgr_message_close(message);
896 /* qmgr_message_kill_record - mark one message record as killed */
898 void qmgr_message_kill_record(QMGR_MESSAGE *message, long offset)
902 if (qmgr_message_open(message)
903 || rec_put_type(message->fp, REC_TYPE_KILL, offset) < 0
904 || vstream_fflush(message->fp))
905 msg_fatal("update queue file %s: %m", VSTREAM_PATH(message->fp));
906 qmgr_message_close(message);
936 * Compare message transport.
968 /* qmgr_message_sort - sort message recipient addresses by domain */
970 static void qmgr_message_sort(QMGR_MESSAGE *message)
972 qsort((char *) message->rcpt_list.info, message->rcpt_list.len,
973 sizeof(message->rcpt_list.info[0]), qmgr_message_sort_compare);
975 RECIPIENT_LIST list = message->rcpt_list;
987 static int qmgr_resolve_one(QMGR_MESSAGE *message, RECIPIENT *recipient,
996 if ((message->tflags & DEL_REQ_FLAG_MTA_VRFY) == 0)
997 resolve_clnt_query_from(message->sender, addr, reply);
999 resolve_clnt_verify_from(message->sender, addr, reply);
1015 static void qmgr_message_resolve(QMGR_MESSAGE *message)
1018 RECIPIENT_LIST list = message->rcpt_list;
1042 * Redirect overrides all else. But only once (per entire message).
1046 if (message->redirect_addr) {
1051 message->rcpt_offset = 0;
1052 message->rcpt_unread = 0;
1054 rewrite_clnt_internal(REWRITE_CANON, message->redirect_addr,
1057 if (qmgr_resolve_one(message, recipient,
1078 else if (message->filter_xport
1079 && (message->tflags & DEL_REQ_TRACE_ONLY_MASK) == 0) {
1081 vstring_strcpy(reply.transport, message->filter_xport);
1095 if (qmgr_resolve_one(message, recipient,
1128 status = sent(message->tflags, message->queue_id,
1129 QMGR_MSG_STATS(&stats, message), recipient,
1133 deliver_completed(message->fp, recipient->offset);
1137 message->queue_id);
1140 message->flags |= status;
1149 if (*var_defer_xports && (message->qflags & QMGR_FLUSH_DFXP) == 0) {
1171 * This message is being flushed. If need-be unthrottle the
1174 if ((message->qflags & QMGR_FLUSH_EACH) != 0
1189 qmgr_defer_recipient(message, recipient, saved_dsn);
1198 * deliveries to different recipients of the same message can happen
1255 * This message is being flushed. If need-be unthrottle the queue.
1257 if ((message->qflags & QMGR_FLUSH_EACH) != 0
1267 qmgr_defer_recipient(message, recipient, saved_dsn);
1283 static void qmgr_message_assign(QMGR_MESSAGE *message)
1285 RECIPIENT_LIST list = message->rcpt_list;
1311 * Lookup or instantiate the message job if necessary.
1314 job = qmgr_job_obtain(message, queue->transport);
1329 if (message->single_rcpt || entry == 0
1331 entry = qmgr_entry_create(peer, message);
1341 message->rcpt_count++;
1346 * Release the message recipient list and reinitialize it for the next
1349 recipient_list_free(&message->rcpt_list);
1350 recipient_list_init(&message->rcpt_list, RCPT_LIST_INIT_QUEUE);
1354 * all transports to which we assigned new recipients, this message may
1360 for (job = message->job_list.next; job; job = job->message_peers.next)
1368 static void qmgr_message_move_limits(QMGR_MESSAGE *message)
1372 for (job = message->job_list.next; job; job = job->message_peers.next)
1376 /* qmgr_message_free - release memory for in-core message structure */
1378 void qmgr_message_free(QMGR_MESSAGE *message)
1382 if (message->refcount != 0)
1383 msg_panic("qmgr_message_free: reference len: %d", message->refcount);
1384 if (message->fp)
1386 while ((job = message->job_list.next) != 0)
1388 myfree(message->queue_id);
1389 myfree(message->queue_name);
1390 if (message->dsn_envid)
1391 myfree(message->dsn_envid);
1392 if (message->encoding)
1393 myfree(message->encoding);
1394 if (message->sender)
1395 myfree(message->sender);
1396 if (message->verp_delims)
1397 myfree(message->verp_delims);
1398 if (message->filter_xport)
1399 myfree(message->filter_xport);
1400 if (message->inspect_xport)
1401 myfree(message->inspect_xport);
1402 if (message->redirect_addr)
1403 myfree(message->redirect_addr);
1404 if (message->client_name)
1405 myfree(message->client_name);
1406 if (message->client_addr)
1407 myfree(message->client_addr);
1408 if (message->client_port)
1409 myfree(message->client_port);
1410 if (message->client_proto)
1411 myfree(message->client_proto);
1412 if (message->client_helo)
1413 myfree(message->client_helo);
1414 if (message->sasl_method)
1415 myfree(message->sasl_method);
1416 if (message->sasl_username)
1417 myfree(message->sasl_username);
1418 if (message->sasl_sender)
1419 myfree(message->sasl_sender);
1420 if (message->log_ident)
1421 myfree(message->log_ident);
1422 if (message->rewrite_context)
1423 myfree(message->rewrite_context);
1424 recipient_list_free(&message->rcpt_list);
1426 myfree((char *) message);
1429 /* qmgr_message_alloc - create in-core message structure */
1435 QMGR_MESSAGE *message;
1441 * Create an in-core message structure.
1443 message = qmgr_message_create(queue_name, queue_id, qflags);
1446 * Extract message envelope information: time of arrival, sender address,
1451 if (qmgr_message_open(message) < 0) {
1452 qmgr_message_free(message);
1455 if (myflock(vstream_fileno(message->fp), INTERNAL_LOCK, QMGR_LOCK_MODE) < 0) {
1457 qmgr_message_close(message);
1458 qmgr_message_free(message);
1461 if (qmgr_message_read(message) < 0) {
1462 qmgr_message_close(message);
1463 qmgr_message_free(message);
1471 if (mode != 0 && fchmod(vstream_fileno(message->fp), mode) < 0)
1472 msg_fatal("fchmod %s: %m", VSTREAM_PATH(message->fp));
1488 qmgr_message_sort(message);
1489 qmgr_message_resolve(message);
1490 qmgr_message_sort(message);
1491 qmgr_message_assign(message);
1492 qmgr_message_close(message);
1493 if (message->rcpt_offset == 0)
1494 qmgr_message_move_limits(message);
1495 return (message);
1499 /* qmgr_message_realloc - refresh in-core message structure */
1501 QMGR_MESSAGE *qmgr_message_realloc(QMGR_MESSAGE *message)
1508 if (message->rcpt_offset <= 0)
1509 msg_panic("%s: invalid offset: %ld", myname, message->rcpt_offset);
1511 msg_info("%s: %s %s offset %ld", myname, message->queue_name,
1512 message->queue_id, message->rcpt_offset);
1518 if (qmgr_message_open(message) < 0)
1520 if (qmgr_message_read(message) < 0) {
1521 qmgr_message_close(message);
1524 qmgr_message_sort(message);
1525 qmgr_message_resolve(message);
1526 qmgr_message_sort(message);
1527 qmgr_message_assign(message);
1528 qmgr_message_close(message);
1529 if (message->rcpt_offset == 0)
1530 qmgr_message_move_limits(message);
1531 return (message);