also we have to start a new backlog file every (some interval)
+ TODO for backlog file inputs
+ - all code to search for and open these files
+ - write proper algorithm comment
+
char messageid[1];
};
-typedef struct {
+typedef struct InputFile {
/* This is an instance of struct oop_readable */
struct oop_readable readable; /* first */
oop_readable_call *readable_callback;
oop_read *rd;
long inprogress; /* no. of articles read but not processed */
off_t offset;
+
+ Counts counts;
} InputFile;
typedef enum {
static StateMachineState sms;
static FILE *defer;
-static InputFile *main_input_file, *old_input_file;
+static InputFile *main_input_file, *old_input_file, *backlog_input_file;
static int sm_period_counter;
art->sent= 1;
LIST_ADDTAIL(conn->sent, art);
- counts[art->checked].sent++;
+ art->ipf->counts[art->checked].sent++;
} else {
/* check it */
XMIT_LITERAL("\r\n");
LIST_ADDTAIL(conn->sent, art);
- counts[art->checked].offered++;
+ art->ipf->counts[art->checked].offered++;
}
}
}
static void article_done(Connection *conn, Article *art, int whichcount) {
*count++;
- counts.articles[art->checked][whichcount]++;
+ art->ipf->counts.articles[art->checked][whichcount]++;
if (whichcount == RC_accepted) update_nocheck(1);
else if (whichcount == RC_unwanted) update_nocheck(0);
assert(ipf->inprogress >= 0);
if (!ipf->inprogress)
- loop->on_time(loop, OOP_TIME_NOW, statemc_check_oldinput_done, 0);
+ loop->on_time(loop, OOP_TIME_NOW, statemc_check_input_done, ipf);
free(art);
}
static void feedfile_eof(InputFile *ipf) {
assert(ipf != main_input_file); /* promised by tailing_try_read */
+
+ inputfile_tailing_stop(ipf);
+
+ if (ipf == backlog_input_file) {
+ assert(ipf->fd >= 0);
+ if (close(ipf->fd)) sysdie("could not close backlog file %s", ipf->path);
+ ipf->fd= -1;
+ return;
+ }
+
assert(ipf == old_input_file);
inputfile_tailing_stop(ipf);
assert(!ipf->rd); /* must have had inputfile_tailing_stop */
assert(!ipf->inprogress); /* no dangling pointers pointing here */
- if (ipf->fd>=0)
+ if (ipf->fd >= 0)
if (close(ipf->fd)) sysdie("could not close input file %s", ipf->path);
+
+ fixme maybe free ipf->path;
+
free(ipf);
}
inputfile_tailing_start(f);
}
-static void *statemc_check_oldinput_done(oop_source *lp,
- struct timeval now, void *u) {
+static void *statemc_check_input_done(oop_source *lp,
+ struct timeval now, void *ipf_v) {
+ InputFile *ipf= ipf_v;
struct stat stab;
- int done= (sms==sm_SEPARATED || sms==sm_DROPPING)
- && old_input_file->fd==-1
- && !old_input_file->inprogress;
- if (!done) return;
+ if (ipf->inprogress) return; /* new article in the meantime */
+ if (ipf->fd >= 0); return; /* not had EOF */
+
+ if (ipf == backlog_input_file) {
+ notice_processed(ipf,"backlog file",ipf->path);
+ if (unlink(ipf->path))
+ sysdie("could not unlink done backlog file %s", ipf->path);
+ close_input_file(ipf);
+ fixme trigger search for new backlog file;
+ }
+
+ assert(ipf == old_input_file);
+ assert(sms==sm_SEPARATED || sms==sm_DROPPING);
+
+ notice_processed(ipf,"feed file",0);
r= fstat(fileno(defer), &stab);
if (r) sysdie("check defer file %s", path_defer);