| 210 | } |
| 211 | |
| 212 | bool BinManager::offer(const std::string &group, std::shared_ptr<core::FlowFile> flow) { |
| 213 | std::lock_guard < std::mutex > lock(mutex_); |
| 214 | if (flow->getSize() > maxSize_) { |
| 215 | // could not be added to a bin -- too large by itself, so create a separate bin for just this guy. |
| 216 | std::unique_ptr<Bin> bin = std::unique_ptr < Bin > (new Bin(0, ULLONG_MAX, 1, INT_MAX, "", group)); |
| 217 | if (!bin->offer(flow)) |
| 218 | return false; |
| 219 | readyBin_.push_back(std::move(bin)); |
| 220 | logger_->log_debug("BinManager move bin %s to ready bins for group %s", readyBin_.back()->getUUIDStr(), group); |
| 221 | return true; |
| 222 | } |
| 223 | auto search = groupBinMap_.find(group); |
| 224 | if (search != groupBinMap_.end()) { |
| 225 | std::unique_ptr < std::deque<std::unique_ptr<Bin>>>&queue = search->second; |
| 226 | if (!queue->empty()) { |
| 227 | std::unique_ptr<Bin> &tail = queue->back(); |
| 228 | if (!tail->offer(flow)) { |
| 229 | // last bin can not offer the flow |
| 230 | std::unique_ptr<Bin> bin = std::unique_ptr < Bin > (new Bin(minSize_, maxSize_, minEntries_, maxEntries_, fileCount_, group)); |
| 231 | if (!bin->offer(flow)) |
| 232 | return false; |
| 233 | queue->push_back(std::move(bin)); |
| 234 | logger_->log_debug("BinManager add bin %s to group %s", queue->back()->getUUIDStr(), group); |
| 235 | binCount_++; |
| 236 | } |
| 237 | } else { |
| 238 | std::unique_ptr<Bin> bin = std::unique_ptr < Bin > (new Bin(minSize_, maxSize_, minEntries_, maxEntries_, fileCount_, group)); |
| 239 | if (!bin->offer(flow)) |
| 240 | return false; |
| 241 | queue->push_back(std::move(bin)); |
| 242 | binCount_++; |
| 243 | logger_->log_debug("BinManager add bin %s to group %s", queue->back()->getUUIDStr(), group); |
| 244 | } |
| 245 | } else { |
| 246 | std::unique_ptr<std::deque<std::unique_ptr<Bin>>> queue = std::unique_ptr<std::deque<std::unique_ptr<Bin>>> (new std::deque<std::unique_ptr<Bin>>()); |
| 247 | std::unique_ptr<Bin> bin = std::unique_ptr < Bin > (new Bin(minSize_, maxSize_, minEntries_, maxEntries_, fileCount_, group)); |
| 248 | if (!bin->offer(flow)) |
| 249 | return false; |
| 250 | queue->push_back(std::move(bin)); |
| 251 | logger_->log_debug("BinManager add bin %s to group %s", queue->back()->getUUIDStr(), group); |
| 252 | groupBinMap_.insert(std::make_pair(group, std::move(queue))); |
| 253 | binCount_++; |
| 254 | } |
| 255 | |
| 256 | return true; |
| 257 | } |
| 258 | |
| 259 | void BinFiles::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) { |
| 260 | // Rollback is not viable for this processor!! |