teaql-cpp/teaql_sql/src/diagnostic.cpp at main · teaql/teaql-cpp · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
#include <teaql/sql/detail/diagnostic.hpp>
#include <exception>

namespace teaql::detail {
StatementDiagnostic::StatementDiagnostic(std::allocator_arg_t,
                                         Allocator a,
                                         std::optional<ExecutionMetadata> metadata,
                                         std::optional<ExecutionObserver> observer)
    : metadata_(move_in(a.resource(), std::move(metadata))),
      observer_(move_in(a.resource(), std::move(observer))) {
    quiescence_.attach(async_arena(a.resource()), this, [](void* p) noexcept {
        static_cast<StatementDiagnostic*>(p)->cancel();
    });
}

StatementDiagnostic::~StatementDiagnostic() {
    cancel();
}

std::optional<ExecutionMetadata> StatementDiagnostic::success() {
    auto result = std::move(metadata_);
    metadata_.reset();
    quiescence_.reset();
    if (result) {
        result->ended_at = ExecutionMetadata::SystemTime::clock::now();
        result->sql_log.execution_outcome = SqlExecutionOutcome::Success;
    }
    return result;
}

void StatementDiagnostic::notify(SqlExecutionOutcome outcome) noexcept {
    auto metadata = std::move(metadata_);
    metadata_.reset();
    quiescence_.reset();
    if (!metadata) {
        return;
    }
    metadata->ended_at = ExecutionMetadata::SystemTime::clock::now();
    metadata->sql_log.execution_outcome = outcome;
    if (observer_) {
        try {
            (*observer_)(std::move(*metadata));
        } catch (...) {
        } // Sink failures never replace provider failures.
    }
}

void StatementDiagnostic::fail() noexcept {
    notify(SqlExecutionOutcome::Failure);
}

void StatementDiagnostic::cancel() noexcept {
    notify(std::uncaught_exceptions() ? SqlExecutionOutcome::Failure
                                      : SqlExecutionOutcome::Cancelled);
}

FailureJournal::FailureJournal(std::allocator_arg_t,
                               Allocator a,
                               std::optional<ExecutionObserver> observer)
    : resource_(a.resource()), observer_(move_in(a.resource(), std::move(observer))) {
    auto& arena = async_arena(resource_);
    if (observer_) {
        entries_ = &arena.make<Vector<ExecutionMetadata>>();
    }
    quiescence_.attach(arena, this, [](void* p) noexcept {
        static_cast<FailureJournal*>(p)->flush();
    });
}

std::optional<ExecutionObserver> FailureJournal::recorder() const {
    if (!entries_) {
        return {};
    }
    auto* entries = entries_;
    return ExecutionObserver(
        std::allocator_arg, Allocator(resource_), [entries](ExecutionMetadata metadata) {
            entries->emplace_back(std::move(metadata));
        });
}

void FailureJournal::disarm() noexcept {
    observer_.reset();
    quiescence_.reset();
}

void FailureJournal::flush() noexcept {
    quiescence_.reset();
    if (!observer_ || !entries_) {
        return;
    }
    auto observer = std::move(observer_);
    observer_.reset();
    auto entries = std::move(*entries_);
    for (auto& entry : entries) {
        try {
            (*observer)(std::move(entry));
        } catch (...) {
        }
    }
}
}