<commit_before>/*
 * This istream filter catches fatal errors and attempts to ignore
 * them.
 *
 * author: <NAME> <<EMAIL>>
 */

#include "istream_catch.hxx"
#include "istream_internal.hxx"
#include "istream_forward.hxx"
#include "util/Cast.hxx"

#include <daemon/log.h>

#include <assert.h>

struct CatchIstream {
    struct istream output;
    struct istream *input;
    off_t available;

    GError *(*callback)(GError *error, void *ctx);
    void *callback_ctx;
};

static constexpr char space[] =
    "                                "
    "                                "
    "                                "
    "                                ";

static void
catch_send_whitespace(CatchIstream *c)
{
    size_t length, nbytes;

    assert(c->input == nullptr);
    assert(c->available > 0);

    do {
        if (c->available >= (off_t)sizeof(space) - 1)
            length = sizeof(space) - 1;
        else
            length = (size_t)c->available;

        nbytes = istream_invoke_data(&c->output, space, length);
        if (nbytes == 0)
            return;

        c->available -= nbytes;
        if (nbytes < length)
            return;
    } while (c->available > 0);

    istream_deinit_eof(&c->output);
}


/*
 * istream handler
 *
 */

static size_t
catch_input_data(const void *data, size_t length, void *ctx)
{
    auto *c = (CatchIstream *)ctx;

    size_t nbytes = istream_invoke_data(&c->output, data, length);
    if (nbytes > 0) {
        if ((off_t)nbytes < c->available)
            c->available -= (off_t)nbytes;
        else
            c->available = 0;
    }

    return nbytes;
}

static ssize_t
catch_input_direct(enum istream_direct type, int fd, size_t max_length,
                   void *ctx)
{
    auto *c = (CatchIstream *)ctx;

    ssize_t nbytes = istream_invoke_direct(&c->output, type, fd, max_length);
    if (nbytes > 0) {
        if ((off_t)nbytes < c->available)
            c->available -= (off_t)nbytes;
        else
            c->available = 0;
    }

    return nbytes;
}

static void
catch_input_abort(GError *error, void *ctx)
{
    auto *c = (CatchIstream *)ctx;

    error = c->callback(error, c->callback_ctx);
    if (error != nullptr) {
        /* forward error to our handler */
        istream_deinit_abort(&c->output, error);
        return;
    }

    /* the error has been handled by the callback, and he has disposed
       it */

    c->input = nullptr;

    if (c->available > 0) {
        /* according to a previous call to method "available", there
           is more data which we must provide - fill that with space
           characters */
        c->input = nullptr;
        catch_send_whitespace(c);
    } else
        istream_deinit_eof(&c->output);
}

static const struct istream_handler catch_input_handler = {
    .data = catch_input_data,
    .direct = catch_input_direct,
    .eof = istream_forward_eof,
    .abort = catch_input_abort,
};


/*
 * istream implementation
 *
 */

static inline CatchIstream *
istream_to_catch(struct istream *istream)
{
    return &ContainerCast2(*istream, &CatchIstream::output);
}

static off_t
istream_catch_available(struct istream *istream, bool partial)
{
    CatchIstream *c = istream_to_catch(istream);

    if (c->input != nullptr) {
        off_t available = istream_available(c->input, partial);
        if (available != (off_t)-1 && available > c->available)
            c->available = available;

        return available;
    } else
        return c->available;
}

static void
istream_catch_read(struct istream *istream)
{
    CatchIstream *c = istream_to_catch(istream);

    if (c->input != nullptr) {
        istream_handler_set_direct(c->input, c->output.handler_direct);
        istream_read(c->input);
    } else
        catch_send_whitespace(c);
}

static void
istream_catch_close(struct istream *istream)
{
    CatchIstream *c = istream_to_catch(istream);

    if (c->input != nullptr)
        istream_free_handler(&c->input);

    istream_deinit(&c->output);
}

static const struct istream_class istream_catch = {
    .available = istream_catch_available,
    .read = istream_catch_read,
    .close = istream_catch_close,
};


/*
 * constructor
 *
 */

struct istream *
istream_catch_new(struct pool *pool, struct istream *input,
                  GError *(*callback)(GError *error, void *ctx), void *ctx)
{
    assert(input != nullptr);<commit_msg>istream_catch: move functions into the struct<commit_after>/*
 * This istream filter catches fatal errors and attempts to ignore
 * them.
 *
 * author: <NAME> <<EMAIL>>
 */

#include "istream_catch.hxx"
#include "istream_internal.hxx"
#include "istream_forward.hxx"
#include "util/Cast.hxx"

#include <daemon/log.h>

#include <assert.h>

struct CatchIstream {
    struct istream output;
    struct istream *input;
    off_t available;

    GError *(*callback)(GError *error, void *ctx);
    void *callback_ctx;

    void SendSpace();
};

static constexpr char space[] =
    "                                "
    "                                "
    "                                "
    "                                ";

void
CatchIstream::SendSpace()
{
    assert(input == nullptr);
    assert(available > 0);

    do {
        size_t length;
        if (available >= (off_t)sizeof(space) - 1)
            length = sizeof(space) - 1;
        else
            length = (size_t)available;

        size_t nbytes = istream_invoke_data(&output, space, length);
        if (nbytes == 0)
            return;

        available -= nbytes;
        if (nbytes < length)
            return;
    } while (available > 0);

    istream_deinit_eof(&output);
}


/*
 * istream handler
 *
 */

static size_t
catch_input_data(const void *data, size_t length, void *ctx)
{
    auto *c = (CatchIstream *)ctx;

    size_t nbytes = istream_invoke_data(&c->output, data, length);
    if (nbytes > 0) {
        if ((off_t)nbytes < c->available)
            c->available -= (off_t)nbytes;
        else
            c->available = 0;
    }

    return nbytes;
}

static ssize_t
catch_input_direct(enum istream_direct type, int fd, size_t max_length,
                   void *ctx)
{
    auto *c = (CatchIstream *)ctx;

    ssize_t nbytes = istream_invoke_direct(&c->output, type, fd, max_length);
    if (nbytes > 0) {
        if ((off_t)nbytes < c->available)
            c->available -= (off_t)nbytes;
        else
            c->available = 0;
    }

    return nbytes;
}

static void
catch_input_abort(GError *error, void *ctx)
{
    auto *c = (CatchIstream *)ctx;

    error = c->callback(error, c->callback_ctx);
    if (error != nullptr) {
        /* forward error to our handler */
        istream_deinit_abort(&c->output, error);
        return;
    }

    /* the error has been handled by the callback, and he has disposed
       it */

    c->input = nullptr;

    if (c->available > 0) {
        /* according to a previous call to method "available", there
           is more data which we must provide - fill that with space
           characters */
        c->input = nullptr;
        c->SendSpace();
    } else
        istream_deinit_eof(&c->output);
}

static const struct istream_handler catch_input_handler = {
    .data = catch_input_data,
    .direct = catch_input_direct,
    .eof = istream_forward_eof,
    .abort = catch_input_abort,
};


/*
 * istream implementation
 *
 */

static inline CatchIstream *
istream_to_catch(struct istream *istream)
{
    return &ContainerCast2(*istream, &CatchIstream::output);
}

static off_t
istream_catch_available(struct istream *istream, bool partial)
{
    CatchIstream *c = istream_to_catch(istream);

    if (c->input != nullptr) {
        off_t available = istream_available(c->input, partial);
        if (available != (off_t)-1 && available > c->available)
            c->available = available;

        return available;
    } else
        return c->available;
}

static void
istream_catch_read(struct istream *istream)
{
    CatchIstream *c = istream_to_catch(istream);

    if (c->input != nullptr) {
        istream_handler_set_direct(c->input, c->output.handler_direct);
        istream_read(c->input);
    } else
        c->SendSpace();
}

static void
istream_catch_close(struct istream *istream)
{
    CatchIstream *c = istream_to_catch(istream);

    if (c->input != nullptr)
        istream_free_handler(&c->input);

    istream_deinit(&c->output);
}

static const struct istream_class istream_catch = {
    .available = istream_catch_available,
    .read = istream_catch_read,
    .close = istream_catch_close,
};


/*
 * constructor
 *
 */

struct istream *
istream_catch_new(struct pool *pool, struct istream *input,
                  GError *(*callback)(GError *error, void *ctx), void *ctx)
{
    assert(input != nullptr);
{'fraction_non_alphanumeric': 0.1180895930696455, 'fraction_numerical': 0.0027356662487176566, 'mean_word_length': 2.4007751937984496, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 91, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5750699', 'n_tokens_mistral': 2890, 'n_tokens_neox': 2817, 'n_words': 672}