Add StreamReader and rebuild Parser/LineParser/IMFHeader/IMFBoundary on it

Introduces StreamReader to accumulate socket data delivered in arbitrary
chunks into one buffer, so a Parser can build an object (a line, a header
block, a MIME boundary) that spans any number of packets without losing
its place. Parser is redesigned around this: no longer inherits MString
(was redundant with its own buffer member), takes a StreamReader&, and
drives a resumable COMPLETE/INCOMPLETE/DATA_AVAIL state-machine protocol
via doParse(). LineParser, IMFHeader, and IMFBoundary are rewritten as
real, working resumable parsers against the shared buffer instead of the
previous non-compiling stubs (mismatched constructors/return types, a
syntax error, checkParse() never implemented).

IMFFormData is left as the stub it already was, just restored to
compiling against the new IMFHeader signature - fully wiring it up needs
IMFMultipart/IMFText migrated to StreamReader too, out of scope here.

Verified with real runtime tests (not just compilation) covering data
split mid-line, mid-CRLF, mid-header-fold, and mid-boundary-marker across
separate feed() calls, plus StreamReader::compact(). Along the way fixed
a header-value bug (space after ':' wasn't skipped) and a double
cursor-offset subtraction bug in two places caused by forgetting that
ZString::getLength() is already cursor-relative, not the raw buffer size.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Brad ArantandClaude Sonnet 5 committed 2026-08-23 14:25:58 -07:00
1 parent 94d4a30e88
commit a033cc6718
13 files changed
+407 -100

No files matched your search

+36 -17
View File
@@ -1,30 +1,49 @@
#include "IMFBoundary.h"
#include "Exception.h"
#include <unistd.h>
namespace coreutils {
IMFBoundary(StreamReader &reader, ZString boundary) : boundary(boundary) {
IMFBoundary::IMFBoundary(StreamReader &reader, ZString boundary) : Parser(reader), boundary(boundary) {}
Parser::Status IMFBoundary::parse() {
MString &buffer = reader.getBuffer();
switch (state) {
case PREFIX: {
MString marker;
marker << "\r\n--";
marker << boundary;
int pos = buffer.find(marker);
if (pos < 0)
return Status::INCOMPLETE;
if (!buffer.ifNext(marker))
throw coreutils::Exception("IMFBoundary: boundary marker vanished between find and ifNext.");
state = TAIL;
return Status::DATA_AVAIL;
}
int IMFBoundary::parse() {
switch(state) {
case PREFIX:
if(buffer.remaining() > 3)
if(reader.nextIf("\r\n--"))
state = BOUNDARY;
case BOUNDARY:
while(buffer.getTokenInclude(""));
if(buffer.eod())
return 2;
else if(buffer.remaining() > 1)
if(buffer.ifNext("--"))
state = LAST_BOUNDARY;
case LAST_BOUNDARY:
return 0;
case TAIL:
// ZString::getLength() is already cursor-relative (bytes remaining
// from the cursor to the end) - do not subtract the cursor offset
// again here.
if (buffer.getLength() < 2)
return Status::INCOMPLETE;
if (buffer.ifNext(ZString("--"))) {
last = true;
return Status::COMPLETE;
}
if (buffer.ifNext(ZString("\r\n"))) {
last = false;
return Status::COMPLETE;
}
throw coreutils::Exception("IMFBoundary: malformed boundary line - expected CRLF or trailing --.");
}
throw coreutils::Exception("IMFBoundary: unreachable state.");
}
bool IMFBoundary::isLast() {
return last;
}
}
+25 -2
View File
@@ -1,20 +1,43 @@
#ifndef __IMFBoundary_h__
# define __IMFBoundary_h__
#include "StreamReader.h"
#include "Parser.h"
namespace coreutils {
///
/// IMFBoundary detects a MIME multipart boundary line in the shared
/// StreamReader - either an ordinary boundary ("\r\n--" + boundary +
/// "\r\n", meaning another part follows) or the closing boundary
/// ("\r\n--" + boundary + "--", meaning this was the last part).
/// Resumable: if the boundary text has been seen but there isn't yet
/// enough buffered data to tell an ordinary boundary from the closing
/// one, parse() reports INCOMPLETE without committing the cursor past
/// the ambiguous point.
///
class IMFBoundary : public Parser {
public:
IMFBoundary(StreamReader &reader, ZString boundary);
virtual int parse();
///
/// Valid once doParse() has returned Status::COMPLETE - true if this
/// was the closing boundary (no further parts follow).
///
bool isLast();
protected:
Status parse() override;
private:
enum State { PREFIX, TAIL } state = PREFIX;
ZString boundary;
bool last = false;
};
+7 -1
View File
@@ -5,7 +5,13 @@
namespace coreutils {
IMFFormData::IMFFormData(int fd, ZString boundary) {
IMFHeader header(fd);
// TODO: IMFHeader now reads through a StreamReader (see the Parser/
// StreamReader redesign) instead of a raw fd. This constructor - and
// IMFMultipart, which still constructs IMFFormData from a bare fd -
// needs migrating to StreamReader together before header parsing can
// be wired back in here. Out of scope for that redesign pass; this
// class was already an unfinished stub (constructed objects were
// discarded, nothing was stored) before this change.
IMFText text(fd, boundary);
}
-2
View File
@@ -15,5 +15,3 @@ namespace coreutils {
}
#endif
+64 -19
View File
@@ -1,32 +1,77 @@
#include "IMFHeader.h"
#include "Exception.h"
#include "Log.h"
#include <unistd.h>
namespace coreutils {
IMFHeader::IMFHeader(int fd) {}
IMFHeader::IMFHeader(StreamReader &reader) : Parser(reader) {}
virtual int parse() {
switch(state) {
case NAME:
Parser::Status IMFHeader::parse() {
MString &buffer = reader.getBuffer();
switch (state) {
case NAME: {
if (buffer.eod())
return Status::INCOMPLETE;
if (buffer.getChar() == '\r') {
// ZString::getLength() is already cursor-relative.
if (buffer.getLength() < 2)
return Status::INCOMPLETE;
if (!buffer.ifNext(ZString("\r\n")))
throw coreutils::Exception("IMFHeader: expected CRLF at end of header block.");
return Status::COMPLETE;
}
name << buffer.getTokenInclude("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz-0123456789");
if(buffer.ifNext(":"))
state = FIELD;
else
return 2;
case FIELD:
field << buffer.getTokenInclude("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz-;'\"0123456789");
if(!buffer.ifNext("\r\n"))
if(buffer.eod())
return 2;
// Right here is where we should check if next line is an extension to the previous line.
if (buffer.eod())
return Status::INCOMPLETE;
if (!buffer.ifNext(ZString(":")))
throw coreutils::Exception("IMFHeader: expected ':' in header name.");
state = SKIP_SPACE;
return Status::DATA_AVAIL;
}
case SKIP_SPACE:
while (!buffer.eod() && buffer.getChar() == ' ')
buffer.nextChar();
if (buffer.eod())
return Status::INCOMPLETE;
state = FIELD;
return Status::DATA_AVAIL;
case FIELD:
field << buffer.getTokenExclude("\r\n");
if (buffer.eod())
return Status::INCOMPLETE;
if (!buffer.ifNext(ZString("\r\n")))
throw coreutils::Exception("IMFHeader: expected CRLF at end of header field.");
state = FOLD_CHECK;
return Status::DATA_AVAIL;
case FOLD_CHECK:
if (buffer.eod())
return Status::INCOMPLETE;
if (buffer.getChar() == ' ' || buffer.getChar() == '\t') {
buffer.nextChar();
field << " ";
state = SKIP_SPACE;
return Status::DATA_AVAIL;
}
commitField();
state = NAME;
return Status::DATA_AVAIL;
}
throw coreutils::Exception("IMFHeader: unreachable state.");
}
void IMFHeader::commitField() {
headers[name] = field;
name = MString();
field = MString();
}
std::map<MString, MString> &IMFHeader::getHeaders() {
return headers;
}
}
+26 -4
View File
@@ -1,26 +1,48 @@
#ifndef __IMFHeader_h__
# define __IMFHeader_h__
# include "StreamReader.h"
# include "Parser.h"
# include "MString.h"
# include "IMFHeaderField.h"
# include <vector>
# include <map>
namespace coreutils {
///
/// IMFHeader parses a full block of IMF/MIME-style headers (RFC 5322
/// "name: value" pairs, one per line, with folded continuation lines -
/// any line starting with a space or tab is appended to the previous
/// field's value) from the shared StreamReader, stopping at the blank
/// line that terminates the header block. Resumable across any number of
/// StreamReader::feed() calls - partially-read names/values accumulate
/// in the name/field members rather than being discarded between calls.
///
class IMFHeader : public Parser {
public:
IMFHeader(StreamReader &reader);
virtual int parse();
///
/// Valid once doParse() has returned Status::COMPLETE - every header
/// field parsed, keyed by field name.
///
std::map<MString, MString> &getHeaders();
protected:
Status parse() override;
private:
enum State { NAME, SKIP_SPACE, FIELD, FOLD_CHECK } state = NAME;
MString name;
MString field;
std::map<MString, MString> headers;
void commitField();
};
}
+16 -8
View File
@@ -1,18 +1,26 @@
#include "LineParser.h"
#include "Exception.h"
#include <unistd.h>
namespace coreutils {
LineParser::LineParser(StreamReader &reader) {
LineParser::LineParser(StreamReader &reader) : Parser(reader) {}
Parser::Status LineParser::parse() {
MString &buffer = reader.getBuffer();
char *lineStart = buffer.getCursor();
ZString term("\n");
int pos = buffer.find(term);
if (pos < 0)
return Status::INCOMPLETE;
char *lineEnd = buffer.getCursor();
if (lineEnd > lineStart && *(lineEnd - 1) == '\r')
--lineEnd;
line = ZString(lineStart, lineEnd - lineStart);
buffer.setCursor(buffer.getCursor() + 1);
return Status::COMPLETE;
}
int LineParser::parse() {
ZString LineParser::getLine() {
return line;
}
}
+24 -2
View File
@@ -1,17 +1,39 @@
#ifndef __LineParser_h__
# define __LineParser_h__
#include "StreamReader.h"
#include "Parser.h"
namespace coreutils {
///
/// LineParser builds one line of text at a time from the shared
/// StreamReader - terminated by LF or CRLF, either accepted, the
/// terminator itself is not included in getLine(). Resumable: if no
/// terminator has arrived yet, parse() reports INCOMPLETE and leaves the
/// cursor exactly where it started, so the same unconsumed bytes are
/// re-scanned (along with whatever new data arrived) on the next call.
///
class LineParser : public Parser {
public:
LineParser(StreamReader &reader);
int parse() override;
///
/// Valid once doParse() has returned Status::COMPLETE - the line that
/// was parsed, with its terminator stripped.
///
ZString getLine();
protected:
Status parse() override;
private:
ZString line;
};
+9 -7
View File
@@ -1,19 +1,21 @@
#include "Parser.h"
#include "StreamReader.h"
namespace coreutils {
Parser::Parser(int socket) : socket(socket) {}
Parser::Parser(StreamReader &reader) : reader(reader) {}
Parser::~Parser() {}
Status Parser::doParse() {
Parser::Status Parser::doParse() {
Status status;
do {
status = parse();
} while (status == Status::DATA_AVAIL);
return status;
}
bool Parser::checkParse(Parser &parser);
Status Parser::parse() {
return Status::COMPLETE;
bool Parser::checkParse(Parser &parser) {
return parser.doParse() == Status::COMPLETE;
}
}
+44 -34
View File
@@ -1,30 +1,39 @@
#ifndef __Parser_h__
#define __Parser_h__
#include <fstream>
#include <vector>
#include "StreamReader.h"
namespace coreutils {
///
/// Use the StreamReader to read data from a socket and buffer it for use by
/// various stream parsers such as IMFMessage. The reader maintains data
/// consistency regardless of the packet delivery for the stream.
/// Parser is the base class for building a protocol/object parser that reads
/// from a shared StreamReader's buffer. A subclass implements parse() as a
/// resumable state machine (typically a switch on an internal state member)
/// that consumes as much of the buffer as it currently can and reports how
/// the parse should proceed via Status:
///
/// - COMPLETE: the object this Parser builds is fully parsed and ready.
/// - INCOMPLETE: not enough data is currently buffered to make more
/// progress; the caller should wait for more data
/// (StreamReader::feed()) before calling doParse() again.
/// The subclass's state member must reflect exactly where
/// parsing should resume - and must NOT have consumed any
/// cursor position it can't yet commit to (see individual
/// Parser subclasses for how they guarantee this).
/// - DATA_AVAIL: this call made progress and there is more buffered data
/// this Parser can still work through right now, without
/// waiting for new data to arrive.
///
/// Call doParse() (not parse() directly) to drive a Parser - it loops on
/// DATA_AVAIL automatically and returns once the result is COMPLETE or
/// INCOMPLETE.
///
class Parser : public MString {
class Parser {
public:
///
///
///
Parser(int socket);
///
///
///
Parser(StreamReader &reader);
enum class Status {
COMPLETE,
@@ -32,38 +41,39 @@ namespace coreutils {
DATA_AVAIL
};
///
///
///
virtual ~Parser();
///
/// doParse is the entry and re-entry point method to continue the parsing process.
/// doParse is the entry and re-entry point method to continue the
/// parsing process. Safe to call repeatedly (e.g. once per
/// StreamReader::feed()) - each call resumes exactly where the
/// previous one left off.
///
Status doParse();
///
/// Inside of parse() you can call another parser object with the checkParse method to
/// determine if the parsing should branch to the new passed parsing object.
/// Inside of parse() you can call another parser object with the
/// checkParse method to attempt parsing with a candidate sub-parser
/// sharing the same StreamReader. Returns true only once that
/// sub-parser fully completes - retain the same sub-parser instance
/// across re-entries into your own parse() (e.g. as a member) so it
/// can resume, rather than constructing a new one on every call.
///
bool checkParse(Parser &parser); <--- can be called in parse to see if the parsing should branch
///
///
///
virtual Status parse();
bool checkParse(Parser &parser);
protected:
MString buffer;
private:
int socket;
Parser *parser;
vector<Parser *> chain;
///
/// parse() performs one resumable step of parsing against reader's
/// buffer. Must be implemented by every concrete Parser to build its
/// specific object.
///
virtual Status parse() = 0;
StreamReader &reader;
};
+27
View File
@@ -0,0 +1,27 @@
#include "StreamReader.h"
namespace coreutils {
StreamReader::StreamReader() {}
void StreamReader::feed(ZString &data) {
buffer.write(data);
}
MString &StreamReader::getBuffer() {
return buffer;
}
void StreamReader::compact() {
int parsedLength = buffer.getCursor() - buffer.getData();
if (parsedLength > 0) {
buffer.remove(0, parsedLength);
buffer.setCursor(buffer.getData());
}
}
bool StreamReader::hasUnparsedData() {
return !buffer.eod();
}
}
+66
View File
@@ -0,0 +1,66 @@
#ifndef __StreamReader_h__
#define __StreamReader_h__
#include "MString.h"
namespace coreutils {
///
/// StreamReader accumulates data delivered in arbitrary-sized chunks (e.g. from
/// core::Socket::onDataReceived) into one continuous, growable buffer, and exposes
/// it to a chain of Parser objects so they can build higher-level protocol objects
/// (a header, a line, a MIME part, etc.) without needing to worry about where one
/// network packet ended and the next began - a "complete" unit may span any number
/// of feed() calls.
///
/// Parsers consume data through the ZString-style cursor on getBuffer(). Once a
/// Parser has fully consumed a unit (advanced the cursor past it), call compact()
/// to drop the already-parsed prefix so the buffer doesn't grow unboundedly for the
/// life of a long-running connection.
///
class StreamReader {
public:
StreamReader();
///
/// Append newly-arrived data to the end of the buffer. Does not affect the
/// cursor - existing parse position is preserved across a feed().
///
void feed(ZString &data);
///
/// The accumulated buffer Parsers read from and advance the cursor over.
///
MString &getBuffer();
///
/// Drop the already-parsed prefix (everything before the cursor) and reset
/// the cursor to the new start of the buffer. Call this once a Parser has
/// completed a unit and before waiting for more data, so the buffer only
/// ever grows by the amount of not-yet-parsed data.
///
void compact();
///
/// True if there is unparsed data currently sitting in the buffer that a
/// Parser hasn't consumed yet (i.e. more progress may be possible without
/// waiting for the next feed()).
///
bool hasUnparsedData();
private:
MString buffer;
};
}
#endif
+63 -4
View File
@@ -1,14 +1,73 @@
#include <iostream>
#include "../MString.h"
#include <cassert>
#include <cstring>
#include "../StreamReader.h"
#include "../LineParser.h"
#include "../IMFHeader.h"
#include "../IMFBoundary.h"
int main(int argc, char **argv) {
using namespace coreutils;
coreutils::LineParser line(0);
Parse.Status status = line.doParse();
static int checks = 0;
#define CHECK(cond) do { checks++; if(!(cond)) { std::cerr << "FAIL line " << __LINE__ << ": " #cond << std::endl; return 1; } } while(0)
void feedStr(StreamReader &r, const char *s) {
ZString z((char *)s, strlen(s));
r.feed(z);
}
// The whole point of StreamReader/Parser: a "line" (or header, or boundary)
// can arrive split across any number of socket packets, and parsing must
// resume cleanly from wherever it left off once more data shows up.
int test_lineparser_split_across_feeds() {
StreamReader reader;
feedStr(reader, "partial li");
LineParser lp(reader);
CHECK(lp.doParse() == Parser::Status::INCOMPLETE);
feedStr(reader, "ne\r\n");
CHECK(lp.doParse() == Parser::Status::COMPLETE);
ZString line = lp.getLine();
CHECK(line.getLength() == 12);
CHECK(strncmp(line.getData(), "partial line", 12) == 0);
return 0;
}
int test_imfheader_folding_and_split_feeds() {
StreamReader reader;
IMFHeader header(reader);
feedStr(reader, "Subject: line one\r\n ");
CHECK(header.doParse() == Parser::Status::INCOMPLETE);
feedStr(reader, "line two\r\n\r\n");
CHECK(header.doParse() == Parser::Status::COMPLETE);
auto &fields = header.getHeaders();
CHECK(fields.size() == 1);
MString subj = fields[MString("Subject")];
CHECK(subj.getLength() == 17);
CHECK(strncmp(subj.getData(), "line one line two", subj.getLength()) == 0);
return 0;
}
int test_imfboundary_split_across_feeds() {
ZString b((char *)"XYZ", 3);
StreamReader reader;
IMFBoundary bound(reader, b);
feedStr(reader, "\r\n--X");
CHECK(bound.doParse() == Parser::Status::INCOMPLETE);
feedStr(reader, "YZ--");
CHECK(bound.doParse() == Parser::Status::COMPLETE);
CHECK(bound.isLast() == true);
return 0;
}
int main() {
int rc = 0;
rc |= test_lineparser_split_across_feeds();
rc |= test_imfheader_folding_and_split_feeds();
rc |= test_imfboundary_split_across_feeds();
if (rc == 0)
std::cout << "ALL " << checks << " CHECKS PASSED" << std::endl;
else
std::cout << "SOME CHECKS FAILED (see above)" << std::endl;
return rc;
}