forked from Restream/reindexer
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
d629021
commit 7bd3b34
Showing
13 changed files
with
226 additions
and
35 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| #include "event_subscriber_config.h" | ||
| #include "core/cjson/jsonbuilder.h" | ||
|
|
||
| namespace reindexer { | ||
|
|
||
| using namespace std::string_view_literals; | ||
|
|
||
| Error EventSubscriberConfig::FromJSON(span<char> json) { | ||
| try { | ||
| FromJSON(gason::JsonParser().Parse(json)); | ||
| } catch (const gason::Exception &ex) { | ||
| return Error(errParseJson, "UpdatesFilter: %s", ex.what()); | ||
| } catch (const Error &err) { | ||
| return err; | ||
| } | ||
| return {}; | ||
| } | ||
|
|
||
| void EventSubscriberConfig::FromJSON(const gason::JsonNode &root) { | ||
| formatVersion_ = root["version"sv].As<int>(-1); | ||
| if (formatVersion_ < kMinSubscribersConfigFormatVersion) { | ||
| throw Error(errParams, "Min supported subscribers config format version is %d, but %d version was found in JSON", | ||
| kMinSubscribersConfigFormatVersion, formatVersion_) | ||
| } | ||
| streams_.clear(); | ||
| streams_.resize(kMaxStreamsPerSub); | ||
|
|
||
| withDBName_ = root["with_db_name"sv].As<bool>(false); | ||
| for (const auto &stream : root["streams"sv]) { | ||
| const int id = stream["id"].As<int>(-1); | ||
| if (id < 0 || id >= streams_.size()) { | ||
| throw Error(errParams, "Stream ID %d is out of range", id); | ||
| } | ||
| if (streams_[id].has_value()) { | ||
| throw Error(errParams, "Stream ID %d is duplicated", id); | ||
| } | ||
|
|
||
| auto &s = streams_[id].emplace(); | ||
| s.withConfigNamespace = stream["with_config_namespace"sv].As<bool>(false); | ||
| for (const auto &ns : stream["namespaces"sv]) { | ||
| auto name = ns["name"sv].As<std::string_view>(); | ||
| for (const auto &f : ns["filters"sv]) { | ||
| UpdatesFilters::Filter filter; | ||
| filter.FromJSON(f); | ||
| s.filters.AddFilter(name, std::move(filter)); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| void EventSubscriberConfig::GetJSON(WrSerializer &ser) const { | ||
| JsonBuilder builder(ser); | ||
| { | ||
| builder.Put("version"sv, formatVersion_); | ||
| builder.Put("with_config_namespace"sv, withDBName_); | ||
| auto streamArr = builder.Array("streams"sv); | ||
| for (size_t i = 0; i < streams_.size(); ++i) { | ||
| auto streamObj = streamArr.Object(); | ||
| if (streams_[i].has_value()) { | ||
| streamObj.Put("with_config_namespace"sv, streams_[i]->withConfigNamespace); | ||
| auto nsArr = streamObj.Array("namespaces"sv); | ||
| for (const auto &nsFilters : streams_[i]->filters) { | ||
| auto obj = nsArr.Object(); | ||
| obj.Put("name"sv, nsFilters.first); | ||
| auto arrFilters = obj.Array("filters"sv); | ||
| for (const auto &filter : nsFilters.second) { | ||
| auto filtersObj = arrFilters.Object(); | ||
| filter.GetJSON(filtersObj); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } // namespace reindexer |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,79 @@ | ||
| #pragma once | ||
|
|
||
| #include <optional> | ||
| #include "core/type_consts.h" | ||
| #include "estl/fast_hash_map.h" | ||
| #include "estl/span.h" | ||
| #include "tools/errors.h" | ||
| #include "tools/stringstools.h" | ||
|
|
||
| namespace gason { | ||
| struct JsonNode; | ||
| } | ||
|
|
||
| namespace reindexer { | ||
|
|
||
| class JsonBuilder; | ||
|
|
||
| /// Object of this class contains filters set. Filters are separated by namespace and concatenated with disjunction | ||
| class UpdatesFilters { | ||
| public: | ||
| class Filter { | ||
| public: | ||
| // TODO: Any additional condition check should be added here | ||
| bool Check() const { return true; } | ||
| void FromJSON(const gason::JsonNode &) {} | ||
| void GetJSON(JsonBuilder &) const {} | ||
|
|
||
| bool operator==(const Filter &) const { return true; } | ||
| }; | ||
|
|
||
| /// Merge two filters sets | ||
| /// If one of the filters set is empty, result filters set will also be empty | ||
| /// If one of the filters set contains some conditions for specific namespace, | ||
| /// then result filters set will also contain this conditions | ||
| /// @param rhs - Another filters set | ||
| void Merge(const UpdatesFilters &rhs); | ||
| /// Add new filter for specified namespace. Doesn't merge filters, just concatenates it into disjunction sequence | ||
| /// @param ns - Namespace | ||
| /// @param filter - Filter to add | ||
| void AddFilter(std::string_view ns, Filter filter); | ||
| /// Check if filters set allows this namespace | ||
| /// @param ns - Namespace | ||
| /// @return 'true' if filter's conditions are satisfied | ||
| bool Check(std::string_view ns) const; | ||
|
|
||
| Error FromJSON(span<char> json); | ||
| void FromJSON(const gason::JsonNode &root); | ||
| void GetJSON(WrSerializer &ser) const; | ||
|
|
||
| bool operator==(const UpdatesFilters &rhs) const; | ||
|
|
||
| private: | ||
| using FiltersList = h_vector<Filter, 4>; | ||
|
|
||
| fast_hash_map<std::string, FiltersList, nocase_hash_str, nocase_equal_str, nocase_less_str> filters_; | ||
| }; | ||
|
|
||
| class EventSubscriberConfig { | ||
| public: | ||
| struct StreamConfig { | ||
| UpdatesFilters filters; | ||
| bool withConfigNamespace = false; | ||
| }; | ||
| using StreamsContainerT = std::vector<std::optional<StreamConfig>>; | ||
|
|
||
| Error FromJSON(span<char> json); | ||
| void FromJSON(const gason::JsonNode &root); | ||
| void GetJSON(WrSerializer &ser) const; | ||
|
|
||
| const StreamsContainerT &Streams() const noexcept { return streams_; } | ||
| bool WithDBName() const noexcept { return withDBName_; } | ||
|
|
||
| private: | ||
| int formatVersion_ = kSubscribersConfigFormatVersion; | ||
| bool withDBName_ = false; | ||
| StreamsContainerT streams_; | ||
| }; | ||
|
|
||
| } // namespace reindexer |
Oops, something went wrong.