compactor_registry.h

Go to the documentation of this file.

// SPDX-License-Identifier: Apache-2.0
#pragma once
 
#include <entropic/core/compaction.h>
#include <entropic/entropic.h>
 
#include <functional>
#include <shared_mutex>
#include <string>
#include <unordered_map>
 
namespace entropic {
 
using CompactorFn = std::function<CompactionResult(
    const std::vector<Message>& messages,
    const CompactionConfig& config,
    const std::string& identity)>;
 
struct CompactorEntry {
    entropic_compactor_fn c_callback = nullptr; 
    void* user_data = nullptr;                  
    CompactorFn cpp_fn;                         
};
 
class CompactorRegistry {
public:
    explicit CompactorRegistry(CompactionManager& default_manager);
 
    entropic_error_t register_compactor(
        const std::string& identity,
        entropic_compactor_fn compactor,
        void* user_data);
 
    entropic_error_t deregister_compactor(
        const std::string& identity);
 
    CompactionResult compact(
        const std::string& identity,
        const std::vector<Message>& messages,
        const CompactionConfig& config);
 
    bool has_custom_compactor(const std::string& identity) const;
 
    CompactionManager& default_manager() { return default_manager_; }
 
private:
    CompactionResult run_default(
        const std::string& identity,
        const std::vector<Message>& messages,
        const CompactionConfig& config);
 
    CompactionResult run_custom(
        const CompactorFn& selected,
        const std::string& source,
        const std::string& identity,
        const std::vector<Message>& messages,
        const CompactionConfig& config);
 
    static CompactorFn wrap_c_compactor(
        entropic_compactor_fn compactor,
        void* user_data);
 
    CompactionManager& default_manager_;                        
    CompactorFn default_compactor_;                             
    std::unordered_map<std::string, CompactorEntry> compactors_; 
    mutable std::shared_mutex mutex_;                            
};
 
} // namespace entropic