+/*---------- expiry, flow control and deferral ----------*/
+
+/*
+ * flow control notes
+ * to ensure articles go away eventually
+ * separate queue for each input file
+ * queue expiry
+ * every period, check head of backlog queue for expiry with SMretrieve
+ * if too old: discard, and check next article
+ * also check every backlog article as we read it
+ * flush expiry
+ * after too long in SEPARATED/DROPPING ie Separated/Finishing/Dropping
+ * one-off: eat queued articles from flushing and write them to defer
+ * one-off: connfail all connections which have any articles from flushing
+ * newly read articles from flushing go straight to defer
+ * this should take care of it and get us out of this state
+ * to avoid filling up ram needlessly
+ * input control
+ * limit number of queued articles for each ipf
+ * pause/resume inputfile tailing
+ */
+
+static void check_reading_pause_resume(InputFile *ipf) {
+ if (ipf->queue.count >= max_queue_per_ipf)
+ inputfile_reading_pause(ipf);
+ else
+ inputfile_reading_resume(ipf);
+}
+
+static void article_defer(Article *art /* not on a queue */, int whichcount) {
+ open_defer();
+ if (fprintf(defer, "%s %s\n", TokenToText(art->token), art->messageid) <0
+ || fflush(defer))
+ sysfatal("write to defer file %s",path_defer);
+ article_done(art, whichcount);
+}
+
+static int article_check_expired(Article *art /* must be queued, not conn */) {
+ ARTHANDLE *artdata= SMretrieve(art->token, RETR_STAT);
+ if (artdata) { SMfreearticle(artdata); return 0; }
+
+ LIST_REMOVE(art->ipf->queue, art);
+ art->missing= 1;
+ art->ipf->count_nooffer_missing++;
+ article_done(art,-1);
+ return 1;
+}
+
+static void inputfile_queue_check_expired(InputFile *ipf) {
+ if (!ipf) return;
+
+ for (;;) {
+ Article *art= LIST_HEAD(ipf->queue);
+ int exp= article_check_expired(art);
+ if (!exp) break;
+ }
+ check_reading_pause_resume(ipf);
+}
+
+static void article_autodefer(InputFile *ipf, Article *art) {
+ ipf->autodefer++;
+ article_defer(art,-1);
+}
+
+static int has_article_in(const ArticleList *al, InputFile *ipf) {
+ Article *art;
+ for (art=LIST_HEAD(*al); art; art=LIST_NEXT(art))
+ if (art->ipf == ipf) return 1;
+ return 0;
+}
+
+static void autodefer_input_file_articles(InputFile *ipf) {
+ Article *art;
+ while ((art= LIST_REMHEAD(ipf->queue)))
+ article_autodefer(ipf, art);
+}
+
+static void autodefer_input_file(InputFile *ipf) {
+ ipf->autodefer= 0;
+
+ autodefer_input_file_articles(ipf);
+
+ if (ipf->inprogress) {
+ Conn *walk;
+ FOR_CONN(walk) {
+ if (has_article_in(&walk->waiting, ipf) ||
+ has_article_in(&walk->priority, ipf) ||
+ has_article_in(&walk->sent, ipf))
+ walk->quitting= -1;
+ }
+ while (ipf->inprogress) {
+ FOR_CONN(walk)
+ if (walk->quitting < 0) goto found;
+ abort(); /* where are they ?? */
+
+ found:
+ connfail(walk, "connection is stuck or crawling,"
+ " and we need to finish flush");
+ autodefer_input_file_articles(ipf);
+ }
+ }
+
+ check_reading_pause_resume(ipf);
+}
+