Deferred lazy closure: re-invokes the fetcher on every read, wrapping each invocation's anchor in a per-call shared_ptr so a returned PayloadView can outlive the call without holding the fetcher. For kPureLazy sources, where holding bytes until read time costs more than re-fetching.
| 121 | // outlive the call without holding the fetcher. For kPureLazy sources, where |
| 122 | // holding bytes until read time costs more than re-fetching. |
| 123 | std::function<sdk::PayloadView()> makeLazyFetchClosure( |
| 124 | std::shared_ptr<FetcherOwner> owner, std::shared_ptr<std::mutex> fetch_mutex, LazyFetchContext context) { |
| 125 | // The DSO keepalive lives inside `owner` (FetcherOwner), so it is the single |
| 126 | // source of truth here too — reach it via owner->library_keepalive when |
| 127 | // wrapping each fetched anchor, rather than a parallel capture. |
| 128 | return [owner = std::move(owner), fetch_mutex = std::move(fetch_mutex), |
| 129 | context = std::move(context)]() -> sdk::PayloadView { |
| 130 | PJ_payload_t payload{}; |
| 131 | PJ_error_t err{}; |
| 132 | bool ok = false; |
| 133 | if (owner->fetcher.fetchMessageData != nullptr) { |
| 134 | if (fetch_mutex != nullptr) { |
| 135 | std::lock_guard lock(*fetch_mutex); |
| 136 | ok = owner->fetcher.fetchMessageData(owner->fetcher.ctx, &payload, &err); |
| 137 | } else { |
| 138 | ok = owner->fetcher.fetchMessageData(owner->fetcher.ctx, &payload, &err); |
| 139 | } |
| 140 | } |
| 141 | if (!ok) { |
| 142 | qCWarning(lcIngest) << "[lazy-fetch] failed source=" << QString::fromStdString(context.source_id) |
| 143 | << "dataset=" << context.dataset_id << "topic=" << QString::fromStdString(context.topic_name) |
| 144 | << "object_topic_id=" << context.object_topic_id.id << "timestamp_ns=" << context.timestamp_ns |
| 145 | << "error=" << errorMessage(err); |
| 146 | return {}; |
| 147 | } |
| 148 | if (payload.data == nullptr && payload.size > 0) { |
| 149 | qCWarning(lcIngest) << "[lazy-fetch] null data with nonzero size source=" |
| 150 | << QString::fromStdString(context.source_id) << "dataset=" << context.dataset_id |
| 151 | << "topic=" << QString::fromStdString(context.topic_name) |
| 152 | << "object_topic_id=" << context.object_topic_id.id << "timestamp_ns=" << context.timestamp_ns |
| 153 | << "payload_size=" << payload.size; |
| 154 | if (payload.anchor.release != nullptr) { |
| 155 | payload.anchor.release(payload.anchor.ctx); |
| 156 | } |
| 157 | return {}; |
| 158 | } |
| 159 | if (payload.size == 0) { |
| 160 | qCWarning(lcIngest) << "[lazy-fetch] empty payload source=" << QString::fromStdString(context.source_id) |
| 161 | << "dataset=" << context.dataset_id << "topic=" << QString::fromStdString(context.topic_name) |
| 162 | << "object_topic_id=" << context.object_topic_id.id |
| 163 | << "timestamp_ns=" << context.timestamp_ns; |
| 164 | if (payload.anchor.release != nullptr) { |
| 165 | payload.anchor.release(payload.anchor.ctx); |
| 166 | } |
| 167 | return {}; |
| 168 | } |
| 169 | auto anchor = detail::wrapPayloadAnchor(payload.anchor, owner->library_keepalive); |
| 170 | if (anchor == nullptr) { |
| 171 | // No ownership — must copy because the buffer dies with this call. |
| 172 | auto bytes = std::make_shared<const std::vector<uint8_t>>(copyPayloadBytes(payload)); |
| 173 | return sdk::PayloadView{ |
| 174 | Span<const uint8_t>{bytes->data(), bytes->size()}, |
| 175 | sdk::BufferAnchor{bytes}, |
| 176 | }; |
| 177 | } |
| 178 | return sdk::PayloadView{ |
| 179 | Span<const uint8_t>{payload.data, static_cast<size_t>(payload.size)}, |
| 180 | std::move(anchor), |
no test coverage detected