blob: a589f685f8ce6ab1fdc6499257ab2fe1ca57cde8 (
plain) (
blame)
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
|
// Copyright Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#include "messagetypepolicy.h"
#include <vespa/documentapi/messagebus/documentprotocol.h>
#include <vespa/messagebus/message.h>
#include <vespa/vespalib/stllike/hash_map.hpp>
#include <vespa/config/helper/configfetcher.hpp>
#include <vespa/config/subscription/configuri.h>
using vespa::config::content::MessagetyperouteselectorpolicyConfig;
namespace documentapi {
namespace policy {
using MessageTypeMapT = vespalib::hash_map<int, mbus::Route>;
class MessageTypeMap : public MessageTypeMapT {
public:
using MessageTypeMapT::MessageTypeMapT;
};
}
MessageTypePolicy::MessageTypePolicy(const config::ConfigUri & configUri) :
mbus::IRoutingPolicy(),
config::IFetcherCallback<MessagetyperouteselectorpolicyConfig>(),
_map(),
_defaultRoute(),
_fetcher(std::make_unique<config::ConfigFetcher>(configUri.getContext()))
{
_fetcher->subscribe<MessagetyperouteselectorpolicyConfig>(configUri.getConfigId(), this);
_fetcher->start();
}
MessageTypePolicy::~MessageTypePolicy() {}
void
MessageTypePolicy::configure(std::unique_ptr<MessagetyperouteselectorpolicyConfig> cfg)
{
auto map = std::make_unique<policy::MessageTypeMap>();
for (size_t i(0), m(cfg->route.size()); i < m; i++) {
const MessagetyperouteselectorpolicyConfig::Route & r = cfg->route[i];
(*map)[r.messagetype] = mbus::Route::parse(r.name);
}
_map.set(map.release());
_defaultRoute.set(new mbus::Route(mbus::Route::parse(cfg->defaultroute)));
_map.latch();
_defaultRoute.latch();
}
void
MessageTypePolicy::select(mbus::RoutingContext & context)
{
int messageType = context.getMessage().getType();
std::shared_ptr<policy::MessageTypeMap> map = _map.get();
policy::MessageTypeMap::const_iterator found = map->find(messageType);
if (found != map->end()) {
context.addChild(found->second);
} else {
context.addChild(*_defaultRoute.get());
}
}
void
MessageTypePolicy::merge(mbus::RoutingContext &context)
{
DocumentProtocol::merge(context);
}
} // namespace documentapi
|
