zeek/src/input/ReaderBackend.cc

// See the file "COPYING" in the main distribution directory for copyright.

#include "zeek/input/ReaderBackend.h"

#include "zeek/input/Manager.h"
#include "zeek/input/ReaderFrontend.h"

using zeek::threading::Field;
using zeek::threading::Value;

namespace zeek::input {

class PutMessage final : public threading::OutputMessage<ReaderFrontend> {
public:
    PutMessage(ReaderFrontend* reader, Value** val)
        : threading::OutputMessage<ReaderFrontend>("Put", reader), val(val) {}

    bool Process() override {
        input_mgr->Put(Object(), val);
        return true;
    }

private:
    Value** val;
};

class DeleteMessage final : public threading::OutputMessage<ReaderFrontend> {
public:
    DeleteMessage(ReaderFrontend* reader, Value** val)
        : threading::OutputMessage<ReaderFrontend>("Delete", reader), val(val) {}

    bool Process() override { return input_mgr->Delete(Object(), val); }

private:
    Value** val;
};

class ClearMessage final : public threading::OutputMessage<ReaderFrontend> {
public:
    ClearMessage(ReaderFrontend* reader) : threading::OutputMessage<ReaderFrontend>("Clear", reader) {}

    bool Process() override {
        input_mgr->Clear(Object());
        return true;
    }

private:
};

class ReaderErrorMessage final : public threading::OutputMessage<ReaderFrontend> {
public:
    enum Type : uint8_t { INFO, WARNING, ERROR };

    ReaderErrorMessage(ReaderFrontend* reader, Type arg_type, const char* arg_msg)
        : threading::OutputMessage<ReaderFrontend>("ReaderErrorMessage", reader) {
        type = arg_type;
        msg = util::copy_string(arg_msg);
    }

    ~ReaderErrorMessage() override { delete[] msg; }

    bool Process() override;

private:
    const char* msg;
    Type type;
};

class SendEntryMessage final : public threading::OutputMessage<ReaderFrontend> {
public:
    SendEntryMessage(ReaderFrontend* reader, Value** val)
        : threading::OutputMessage<ReaderFrontend>("SendEntry", reader), val(val) {}

    bool Process() override {
        input_mgr->SendEntry(Object(), val);
        return true;
    }

private:
    Value** val;
};

class EndCurrentSendMessage final : public threading::OutputMessage<ReaderFrontend> {
public:
    EndCurrentSendMessage(ReaderFrontend* reader)
        : threading::OutputMessage<ReaderFrontend>("EndCurrentSend", reader) {}

    bool Process() override {
        input_mgr->EndCurrentSend(Object());
        return true;
    }

private:
};

class EndOfDataMessage final : public threading::OutputMessage<ReaderFrontend> {
public:
    EndOfDataMessage(ReaderFrontend* reader) : threading::OutputMessage<ReaderFrontend>("EndOfData", reader) {}

    bool Process() override {
        input_mgr->SendEndOfData(Object());
        return true;
    }

private:
};

class ReaderClosedMessage final : public threading::OutputMessage<ReaderFrontend> {
public:
    ReaderClosedMessage(ReaderFrontend* reader) : threading::OutputMessage<ReaderFrontend>("ReaderClosed", reader) {}

    bool Process() override {
        Object()->SetDisable();
        return input_mgr->RemoveStreamContinuation(Object());
    }

private:
};

class DisableMessage final : public threading::OutputMessage<ReaderFrontend> {
public:
    DisableMessage(ReaderFrontend* writer) : threading::OutputMessage<ReaderFrontend>("Disable", writer) {}

    bool Process() override {
        Object()->SetDisable();
        // And - because we do not need disabled objects any more -
        // there is no way to re-enable them, so simply delete them.
        // This avoids the problem of having to periodically check if
        // there are any disabled readers out there. As soon as a
        // reader disables itself, it deletes itself.
        input_mgr->RemoveStream(Object());
        return true;
    }
};

bool ReaderErrorMessage::Process() {
    switch ( type ) {
        case INFO: input_mgr->Info(Object(), msg); break;

        case WARNING: input_mgr->Warning(Object(), msg); break;

        case ERROR: input_mgr->Error(Object(), msg); break;
    }

    return true;
}

using namespace input;

ReaderBackend::ReaderBackend(ReaderFrontend* arg_frontend) : MsgThread() {
    disabled = true; // disabled will be set correctly in init.
    frontend = arg_frontend;
    info = new ReaderInfo(frontend->Info());
    num_fields = 0;
    fields = nullptr;

    SetName(frontend->Name());
}

ReaderBackend::~ReaderBackend() { delete info; }

void ReaderBackend::Put(Value** val) { SendOut(new PutMessage(frontend, val)); }

void ReaderBackend::Delete(Value** val) { SendOut(new DeleteMessage(frontend, val)); }

void ReaderBackend::Clear() { SendOut(new ClearMessage(frontend)); }

void ReaderBackend::EndCurrentSend() { SendOut(new EndCurrentSendMessage(frontend)); }

void ReaderBackend::EndOfData() { SendOut(new EndOfDataMessage(frontend)); }

void ReaderBackend::SendEntry(Value** vals) { SendOut(new SendEntryMessage(frontend, vals)); }

bool ReaderBackend::Init(const int arg_num_fields, const threading::Field* const* arg_fields) {
    if ( Failed() )
        return true;

    disabled = false;

    SetOSName(Fmt("zk.%s", Name()));

    num_fields = arg_num_fields;
    fields = arg_fields;

    // disable if DoInit returns error.
    int success = DoInit(*info, arg_num_fields, arg_fields);

    if ( ! success ) {
        Error("Init failed");
        DisableFrontend();
    }

    return success;
}

bool ReaderBackend::OnFinish(double network_time) {
    if ( ! Failed() )
        DoClose();

    disabled = true; // frontend disables itself when it gets the Close-message.
    SendOut(new ReaderClosedMessage(frontend));

    if ( fields ) {
        for ( unsigned int i = 0; i < num_fields; i++ )
            delete (fields[i]);

        delete[] (fields);
        fields = nullptr;
    }

    return true;
}

bool ReaderBackend::Update() {
    if ( disabled )
        return false;

    if ( Failed() )
        return true;

    bool success = DoUpdate();
    if ( ! success )
        DisableFrontend();

    return ! disabled; // always return failure if we have been disabled in the meantime
}

void ReaderBackend::DisableFrontend() {
    // We might already have been disabled - e.g., due to a call to
    // error. In that case, ignore this...
    if ( disabled )
        return;

    // We also set disabled here, because there still may be other
    // messages queued and we will dutifully ignore these from now.
    disabled = true;
    SendOut(new DisableMessage(frontend));
}

bool ReaderBackend::OnHeartbeat(double network_time, double current_time) {
    if ( Failed() )
        return true;

    return DoHeartbeat(network_time, current_time);
}

void ReaderBackend::Info(const char* msg) {
    SendOut(new ReaderErrorMessage(frontend, ReaderErrorMessage::INFO, msg));
    MsgThread::Info(msg);
}

void ReaderBackend::StopWarningSuppression() {
    suppress_warnings = false;
    if ( warnings_suppressed > 0 )
        Warning(Fmt("Suppressed %zu warning(s)", warnings_suppressed));
    warnings_suppressed = 0;
}

void ReaderBackend::FailWarn(bool is_error, const char* msg, bool suppress_future) {
    if ( is_error )
        Error(msg);
    else {
        Warning(msg);

        if ( suppress_future )
            suppress_warnings = true;
    }
}
void ReaderBackend::Warning(const char* msg) {
    if ( suppress_warnings ) {
        warnings_suppressed++;
        return;
    }

    SendOut(new ReaderErrorMessage(frontend, ReaderErrorMessage::WARNING, msg));
    MsgThread::Warning(msg);
}

void ReaderBackend::Error(const char* msg) {
    SendOut(new ReaderErrorMessage(frontend, ReaderErrorMessage::ERROR, msg));
    MsgThread::Error(msg);

    // Force errors to be fatal.
    DisableFrontend();
}

} // namespace zeek::input