FlashMQ/mainapp.h at master · halfgaar/FlashMQ · 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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
/*
This file is part of FlashMQ (https://www.flashmq.org)
Copyright (C) 2021-2023 Wiebe Cazemier

FlashMQ is free software: you can redistribute it and/or modify
it under the terms of The Open Software License 3.0 (OSL-3.0).

See LICENSE for license details.
*/

#ifndef MAINAPP_H
#define MAINAPP_H

#include <iostream>
#include <sys/socket.h>
#include <stdexcept>
#include <netinet/in.h>
#include <fcntl.h>
#include <thread>
#include <vector>
#include <functional>
#include <forward_list>
#include <list>
#include <sys/resource.h>

#include "threaddata.h"
#include "subscriptionstore.h"
#include "configfileparser.h"
#include "scopedsocket.h"
#include "oneinstancelock.h"
#include "bridgeinfodb.h"
#include "backgroundworker.h"
#include "driftcounter.h"
#include "globals.h"

class MainApp
{
#ifdef TESTING
    friend class MainAppInThread;
    friend class MainAppAsFork;
#endif

    std::weak_ptr<MainApp> mSelf;
    int num_threads = 0;

    bool started = false;
    bool running = true;
    std::vector<ThreadDataOwner> threads;
    std::shared_ptr<SubscriptionStore> subscriptionStore;
    std::unique_ptr<ConfigFileParser> confFileParser;
    int epollFdAccept = -1;
    int taskEventFd = -1;
    bool doConfigReload = false;
    bool doLogFileReOpen = false;
    bool doQuitAction = false;
    bool errorExit = false;
    bool doMemoryTrim = false;
    size_t threadsPendingInit = 0;
    QueuedTasks timed_tasks;
    MutexOwned<std::vector<std::function<void()>>> taskQueue;

    uint overloadLogCounter = 0;
    DriftCounter drift;
    std::chrono::milliseconds medianThreadDrift = std::chrono::milliseconds(0);

    Settings settings;

    std::list<std::shared_ptr<Listener>> listeners;
    std::unordered_map<int, ScopedSocket> activeListenSockets;

    std::unordered_map<std::string, BridgeConfig> bridgeConfigs;
    std::mutex quitMutex;
    std::string fuzzFilePath;
    OneInstanceLock oneInstanceLock;
    const pthread_t mInitializedByThread = pthread_self();

    Logger *logger = Logger::getInstance();

    BackgroundWorker bgWorker;

    void setSelf(const std::shared_ptr<MainApp> &self);
    bool getFuzzMode() const;
    void setlimits();
    void loadConfig(bool reload);
    void reloadConfig();
    void reopenLogfile();
    void reloadTimers(const Settings *old_settings);
    static void doHelp(const char *arg);
    static void showLicense();
    std::list<ScopedSocket> createListenSocket(const std::shared_ptr<Listener> &listener, bool do_listen);
    void wakeUpThread();
    void addImmediateTask(std::function<void ()> f);
    void queueKeepAliveCheckAtAllThreads();
    void queuePasswordFileReloadAllThreads();
    void queuepluginPeriodicEventAllThreads();
    void setFuzzFile(const std::string &fuzzFilePath);
    void queuePublishStatsOnDollarTopic();
    void saveStateInThread();
    void queueSendQueuedWills();
    void waitForWillsQueued();
    void queueRetainedMessageExpiration();
    void sendBridgesToThreads();
    void queueBridgeReconnectAllThreads();
    void queueInternalHeartbeat();
    void performAllImmediateTasks();

    MainApp(const std::string &configFilePath);
public:
    MainApp(const MainApp &rhs) = delete;
    MainApp(MainApp &&rhs) = delete;
    ~MainApp();
    static std::shared_ptr<MainApp> initMainApp(int argc, char *argv[]);
    void start();
    void queueQuit();
    void quit();
    bool getStarted() const {return started;}
    static void testConfig();

    void queueConfigReload();
    void queueReopenLogFile();
    void queueCleanup();
    void queuePurgeSubscriptionTree();
    void queueMemoryTrim();
    void memoryTrim();
    void queueThreadInitDecrement();
};

#endif // MAINAPP_H