<commit_before>#include "direct.hxx"
#include "fb_pool.hxx"
#include "istream/istream.hxx"
#include "istream/istream_byte.hxx"
#include "istream/istream_cat.hxx"
#include "istream/istream_fail.hxx"
#include "istream/istream_four.hxx"
#include "istream/istream_head.hxx"
#include "istream/istream_hold.hxx"
#include "istream/istream_inject.hxx"
#include "istream/istream_later.hxx"
#include "istream/istream_pointer.hxx"
#include "istream/istream_oo.hxx"

#include <glib.h>
#include <event.h>

#include <stdio.h>
#ifdef EXPECTED_RESULT
#include <string.h>
#endif

enum {
#ifdef NO_BLOCKING
    enable_blocking = false,
#else
    enable_blocking = true,
#endif
};

static inline GQuark
test_quark(void)
{
    return g_quark_from_static_string("test");
}

#ifndef FILTER_CLEANUP
static void
cleanup(void)
{
}
#endif

struct Context {
    IstreamPointer input;

    bool half = false;
    bool got_data;
    bool eof = false;
#ifdef EXPECTED_RESULT
    bool record = false;
    char buffer[sizeof(EXPECTED_RESULT) * 2];
    size_t buffer_length = 0;
#endif
    struct istream *abort_istream = nullptr;
    int abort_after = 0;

    int block_after = -1;

    bool block_byte = false, block_byte_state = false;

    explicit Context(struct istream &_input)
        :input(_input, MakeIstreamHandler<Context>::handler, this) {}

    /* handler */
    size_t OnData(const void *data, size_t length);
    ssize_t OnDirect(FdType type, int fd, size_t max_length);
    void OnEof();
    void OnError(GError *error);
};

/*
 * istream handler
 *
 */

size_t
Context::OnData(gcc_unused const void *data, size_t length)
{
    got_data = true;

    if (block_byte) {
        block_byte_state = !block_byte_state;
        if (block_byte_state)
            return 0;
    }

    if (abort_istream != nullptr && abort_after-- == 0) {
        GError *error = g_error_new_literal(test_quark(), 0, "abort_istream");
        istream_inject_fault(abort_istream, error);
        abort_istream = nullptr;
        return 0;
    }

    if (half && length > 8)
        length = (length + 1) / 2;

    if (block_after >= 0) {
        --block_after;
        if (block_after == -1)
            /* block once */
            return 0;
    }

#ifdef EXPECTED_RESULT
    if (record) {
#ifdef __clang__
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wstring-plus-int"
#endif

        assert(buffer_length + length < sizeof(buffer));
        assert(memcmp(EXPECTED_RESULT + buffer_length, data, length) == 0);

#ifdef __clang__
#pragma GCC diagnostic pop
#endif

        if (buffer_length + length < sizeof(buffer))
            memcpy(buffer + buffer_length, data, length);
        buffer_length += length;
    }
#endif

    return length;
}

ssize_t
Context::OnDirect(gcc_unused FdType type, gcc_unused int fd, size_t max_length)
{
    got_data = true;

    if (abort_istream != nullptr) {
        GError *error = g_error_new_literal(test_quark(), 0, "abort_istream");
        istream_inject_fault(abort_istream, error);
        abort_istream = nullptr;
        return 0;
    }

    return max_length;<commit_msg>test/t_istream_filter: postpone the istream_inject_fault() call<commit_after>#include "direct.hxx"
#include "fb_pool.hxx"
#include "istream/istream.hxx"
#include "istream/istream_byte.hxx"
#include "istream/istream_cat.hxx"
#include "istream/istream_fail.hxx"
#include "istream/istream_four.hxx"
#include "istream/istream_head.hxx"
#include "istream/istream_hold.hxx"
#include "istream/istream_inject.hxx"
#include "istream/istream_later.hxx"
#include "istream/istream_pointer.hxx"
#include "istream/istream_oo.hxx"
#include "event/DeferEvent.hxx"
#include "event/Callback.hxx"

#include <glib.h>
#include <event.h>

#include <stdio.h>
#ifdef EXPECTED_RESULT
#include <string.h>
#endif

enum {
#ifdef NO_BLOCKING
    enable_blocking = false,
#else
    enable_blocking = true,
#endif
};

static inline GQuark
test_quark(void)
{
    return g_quark_from_static_string("test");
}

#ifndef FILTER_CLEANUP
static void
cleanup(void)
{
}
#endif

struct Context {
    IstreamPointer input;

    bool half = false;
    bool got_data;
    bool eof = false;
#ifdef EXPECTED_RESULT
    bool record = false;
    char buffer[sizeof(EXPECTED_RESULT) * 2];
    size_t buffer_length = 0;
#endif
    struct istream *abort_istream = nullptr;
    int abort_after = 0;

    int block_after = -1;

    bool block_byte = false, block_byte_state = false;

    DeferEvent defer_inject_event;
    struct istream *defer_inject_istream = nullptr;
    GError *defer_inject_error = nullptr;

    explicit Context(struct istream &_input)
        :input(_input, MakeIstreamHandler<Context>::handler, this),
         defer_inject_event(MakeSimpleEventCallback(Context,
                                                    DeferredInject),
                            this) {}

    ~Context() {
        if (defer_inject_error != nullptr)
            g_error_free(defer_inject_error);
    }

    void DeferInject(struct istream *istream, GError *error) {
        assert(error != nullptr);
        assert(defer_inject_istream == nullptr);
        assert(defer_inject_error == nullptr);

        defer_inject_istream = istream;
        defer_inject_error = error;
        defer_inject_event.Add();
    }

    void DeferredInject() {
        assert(defer_inject_istream != nullptr);
        assert(defer_inject_error != nullptr);

        auto i = defer_inject_istream;
        defer_inject_istream = nullptr;
        auto e = defer_inject_error;
        defer_inject_error = nullptr;

        istream_inject_fault(i, e);
    }

    /* handler */
    size_t OnData(const void *data, size_t length);
    ssize_t OnDirect(FdType type, int fd, size_t max_length);
    void OnEof();
    void OnError(GError *error);
};

/*
 * istream handler
 *
 */

size_t
Context::OnData(gcc_unused const void *data, size_t length)
{
    got_data = true;

    if (block_byte) {
        block_byte_state = !block_byte_state;
        if (block_byte_state)
            return 0;
    }

    if (abort_istream != nullptr && abort_after-- == 0) {
        DeferInject(abort_istream,
                    g_error_new_literal(test_quark(), 0, "abort_istream"));
        abort_istream = nullptr;
        return 0;
    }

    if (half && length > 8)
        length = (length + 1) / 2;

    if (block_after >= 0) {
        --block_after;
        if (block_after == -1)
            /* block once */
            return 0;
    }

#ifdef EXPECTED_RESULT
    if (record) {
#ifdef __clang__
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wstring-plus-int"
#endif

        assert(buffer_length + length < sizeof(buffer));
        assert(memcmp(EXPECTED_RESULT + buffer_length, data, length) == 0);

#ifdef __clang__
#pragma GCC diagnostic pop
#endif

        if (buffer_length + length < sizeof(buffer))
            memcpy(buffer + buffer_length, data, length);
        buffer_length += length;
    }
#endif

    return length;
}

ssize_t
Context::OnDirect(gcc_unused FdType type, gcc_unused int fd, size_t max_length)
{
    got_data = true;

    if (abort_istream != nullptr) {
        DeferInject(abort_istream,
                    g_error_new_literal(test_quark(), 0, "abort_istream"));
        abort_istream = nullptr;
        return 0;
    }

    return max_length;
{'fraction_non_alphanumeric': 0.10773026315789473, 'fraction_numerical': 0.004660087719298245, 'mean_word_length': 3.060656649972176, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1521057', 'n_tokens_mistral': 2523, 'n_tokens_neox': 2445, 'n_words': 549}