counters/CountersServer.cpp at master · smyte/counters · GitHub

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
#include <memory>
#include <string>

#include "counters/CountersDecrementKafkaStoreConsumer.h"
#include "counters/CountersHandler.h"
#include "counters/CountersIncrementKafkaConsumer.h"
#include "pipeline/RedisPipelineBootstrap.h"
#include "platform/gcloud/GoogleCloudStorage.h"

namespace counters {

static pipeline::RedisPipelineBootstrap::Config config{
  redisHandlerFactory : [](pipeline::RedisPipelineBootstrap* bootstrap) -> std::shared_ptr<pipeline::RedisHandler> {
    return std::make_shared<CountersHandler>(bootstrap->getDatabaseManager(), bootstrap->getKafkaConsumerHelper());
  },

  kafkaConsumerFactoryMap :
      {{
           CountersIncrementKafkaConsumer::name(),
           [](const std::string& brokerList, const pipeline::KafkaConsumerConfig& consumerConfig,
              const std::string& offsetKey,
              pipeline::RedisPipelineBootstrap* bootstrap) -> std::shared_ptr<infra::kafka::AbstractConsumer> {
             return std::make_shared<CountersIncrementKafkaConsumer>(
                 brokerList, consumerConfig.topic, consumerConfig.partition, consumerConfig.groupId, offsetKey,
                 consumerConfig.lowLatency, bootstrap->getKafkaConsumerHelper());
           },
       },
       {
           CountersDecrementKafkaStoreConsumer::name(),
           [](const std::string& brokerList, const pipeline::KafkaConsumerConfig& consumerConfig,
              const std::string& offsetKey,
              pipeline::RedisPipelineBootstrap* bootstrap) -> std::shared_ptr<infra::kafka::AbstractConsumer> {
             return std::make_shared<CountersDecrementKafkaStoreConsumer>(
                 brokerList, consumerConfig.objectStoreBucketName, consumerConfig.objectStoreObjectNamePrefix,
                 consumerConfig.topic, consumerConfig.partition, consumerConfig.groupId, offsetKey,
                 consumerConfig.offsetKeySuffix, bootstrap->getKafkaConsumerHelper(),
                 std::make_shared<platform::gcloud::GoogleCloudStorage>());
           },
       }},

  databaseManagerFactory : nullptr,

  scheduledTaskQueueFactory : nullptr,

  rocksDbCfConfiguratorMap : {
      {
          pipeline::DatabaseManager::defaultColumnFamilyName(), CountersHandler::optimizeColumnFamily,
      },
  },

  rocksDbConfigurator : nullptr,

  singletonRedisHandler : false,  // in order to support transactions
};

static auto redisPipelineBootstrap = pipeline::RedisPipelineBootstrap::create(config);

}  // namespace counters