<commit_before>local smtp_session = require "smtp_session"
local smtp_data = require "smtp_data"

local smtp_context = {}
smtp_context.__index = smtp_context

-- {{{ smtp_context.new()
function smtp_context.new(nexthop, results_channel)
    local self = {}
    setmetatable(self, smtp_context)

    for i, msg in ipairs(nexthop.messages) do
        msg.contents.loader = smtp_data.new(msg.contents.storage, msg.contents.data)
        kernel:attach(msg.contents.loader)
    end

    self.session = smtp_session.new(nexthop, results_channel)
    self.host = nexthop.destination
    self.port = nexthop.port
    self.buffer = ""

    return self
end
-- }}}

-- {{{ smtp_context:queue_send()
function smtp_context:queue_send(socket, data, more_coming)
    self.pipeline = self.pipeline .. data
    while #self.pipeline > self.send_size do
        local to_send = self.pipeline:sub(1, self.send_size)
        socket:send(to_send)
        io.stderr:write("C: ["..to_send.."]\n")
        self.pipeline = self.pipeline:sub(self.send_size+1)
    end
    if not more_coming then
        socket:send(self.pipeline)
        io.stderr:write("C: ["..self.pipeline.."]\n")
        self.pipeline = ""
    end
end
-- }}}

-- {{{ smtp_context:on_error()
function smtp_context:on_error(err)
    print("ERROR: "..err)
    self.session:shutdown("softfail", "[[socket]]", "", err)
end
-- }}}

-- {{{ smtp_context:__call()
function smtp_context:__call()
    kernel:set_error_handler(smtp_context.on_error, self)

    local rec = kernel:resolve_dns(self.host, self.port)
    local socket = ratchet.socket.new(rec.family, rec.socktype, rec.protocol)
    if not socket:connect(rec.addr) then
        self.session:shutdown("softfail", "[[socket]]", "", "Connection refused.")
        return
    end

    self.send_size = get_conf.number(socket_send_size, socket) or 102400

    while not self.session.is_finished do
        if self.session:is_waiting() then
            local data = socket:recv()
            io.stderr:write("S: ["..data.."]\n")
            if data == '' then
                break
            end

            self.buffer = self.session:process_from_buffer(self.buffer .. data)
        else
            self.pipeline = ""
            repeat
                local data, more_coming = self.session:send_more()<commit_msg>connection failures can be timeouts too<commit_after>local smtp_session = require "smtp_session"
local smtp_data = require "smtp_data"

local smtp_context = {}
smtp_context.__index = smtp_context

-- {{{ smtp_context.new()
function smtp_context.new(nexthop, results_channel)
    local self = {}
    setmetatable(self, smtp_context)

    for i, msg in ipairs(nexthop.messages) do
        msg.contents.loader = smtp_data.new(msg.contents.storage, msg.contents.data)
        kernel:attach(msg.contents.loader)
    end

    self.session = smtp_session.new(nexthop, results_channel)
    self.host = nexthop.destination
    self.port = nexthop.port
    self.buffer = ""

    return self
end
-- }}}

-- {{{ smtp_context:queue_send()
function smtp_context:queue_send(socket, data, more_coming)
    self.pipeline = self.pipeline .. data
    while #self.pipeline > self.send_size do
        local to_send = self.pipeline:sub(1, self.send_size)
        socket:send(to_send)
        io.stderr:write("C: ["..to_send.."]\n")
        self.pipeline = self.pipeline:sub(self.send_size+1)
    end
    if not more_coming then
        socket:send(self.pipeline)
        io.stderr:write("C: ["..self.pipeline.."]\n")
        self.pipeline = ""
    end
end
-- }}}

-- {{{ smtp_context:on_error()
function smtp_context:on_error(err)
    print("ERROR: "..err)
    self.session:shutdown("softfail", "[[socket]]", "", err)
end
-- }}}

-- {{{ smtp_context:__call()
function smtp_context:__call()
    kernel:set_error_handler(smtp_context.on_error, self)

    local rec = kernel:resolve_dns(self.host, self.port)
    local socket = ratchet.socket.new(rec.family, rec.socktype, rec.protocol)
    if not socket:connect(rec.addr) then
        self.session:shutdown("softfail", "[[socket]]", "", "Connection failed.")
        return
    end

    self.send_size = get_conf.number(socket_send_size, socket) or 102400

    while not self.session.is_finished do
        if self.session:is_waiting() then
            local data = socket:recv()
            io.stderr:write("S: ["..data.."]\n")
            if data == '' then
                break
            end

            self.buffer = self.session:process_from_buffer(self.buffer .. data)
        else
            self.pipeline = ""
            repeat
                local data, more_coming = self.session:send_more()
{'fraction_non_alphanumeric': 0.13390742734122713, 'fraction_numerical': 0.003444564047362756, 'mean_word_length': 2.846026490066225, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '2861670', 'n_tokens_mistral': 1535, 'n_tokens_neox': 1508, 'n_words': 282}