<commit_before>string client_id;
function cb, error;

Serialization.AtomParser parser = Serialization.AtomParser();
MMP.Utils.Queue requests = MMP.Utils.Queue();
MMP.Utils.Queue buffer = MMP.Utils.Queue();

void create(string client_id, function cb, function error) {
    this_program::client_id = client_id;
    this_program::cb = cb;
    this_program::error = error;
}

void handle(object request) {
    if (request->request_type == "POST") {
	requests->push(request);	
	parser->feed(request->body_raw);

	Serialization.Atom a;
	mixed err = catch {
	    while (a = parser->parse()) {
		call_out(cb, 0, a);
	    }
	};

	if (err) { // this is reason to disconnect
	    error(err);
	}

	if (!buffer->is_empty()) {
	    _write();
	}
    } else werror("uh: %O\n", request);
}

void _write() {
    if (!requests->is_empty() && !buffer->is_empty()) {
	String.Buffer s = String.Buffer();

	while (!buffer->is_empty()) {
	    Serialization.render_atom(buffer->shift(), s); 
	}

	object request = requests->shift();

	request->resonse_and_finish(([
	    "data" : s->get(), 
	    "type" : "text/atom",
	]));
    }
}

void send(Serialization.Atom atom) {
    buffer->push(atom);

    if (!requests->is_empty()) {
	_write();
    }
}<commit_msg>now working on Stdio.File objects and keeps one connection open for ever.<commit_after>inherit Serialization.BasicTypes;
inherit Serialization.PsycTypes;
inherit Serialization.Signature;

string client_id;
function cb, error;

Serialization.AtomParser parser = Serialization.AtomParser();
MMP.Utils.Queue buffer = MMP.Utils.Queue();

object connection;
object connection_request;
object packet;

void create(string client_id, function cb, function error) {
    this_program::client_id = client_id;
    this_program::cb = cb;
    this_program::error = error;

    ::create(Serialization.TypeCache());
    packet  = PsycPacket("_message", UTF8String(), ([ "_nick" : UTF8String() ]), ([ "_huhu" : UTF8String() ]));
}


void handle(object request) {
    if (request->request_type == "POST") {
	parser->feed(request->body_raw);

	Serialization.Atom a;
	mixed err = catch {
	    while (a = parser->parse()) {
		call_out(cb, 0, this, a);
	    }
	};

	if (err) { // this is reason to disconnect
	    error(err);
	}

	if (!connection) {
	    connection_request = request;
	    connection = request->my_fd;

	    string headers = request->make_response_header(([
		"type" : "text/atom",
		"size" : -1,
	    ]));

	    connection->write(headers);


	    void muh() {
		if (!connection) return;
		PSYC.Packet p = PSYC.Packet("_message_public", ([ "_nick" : "hanswurst\454", "_huhu" : "flupp" ]), "you sucker!");
		send(packet->encode(p));
		call_out(muh, 3);
	    };

	    call_out(muh, 2);

	    if (!buffer->is_empty()) {
		_write();
	    }
	} else {
	    request->response_and_finish(([ "data" : "ok", "type" : "text/html" ]));
	}

    } else werror("uh: %O\n", request);
}

void _write() {
    if (connection && !buffer->is_empty()) {
	String.Buffer s = String.Buffer();

	while (!buffer->is_empty()) {
	    Serialization.render_atom(buffer->shift(), s); 
	}
	
	connection->write(s->get());
	if (connection->errno()) {
	    connection  = 0;
	    connection_request = 0;
	}
    }
}

void send(Serialization.Atom atom) {
    buffer->push(atom);
    werror("sending %O\n", atom);

    if (connection) {
	_write();
    }
}
{'fraction_non_alphanumeric': 0.17200119653006282, 'fraction_numerical': 0.003888722704157942, 'mean_word_length': 3.5745554035567717, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3384304', 'n_tokens_mistral': 1215, 'n_tokens_neox': 1139, 'n_words': 245}