dht_proxy_server.h

/*
 *  Copyright (C) 2017-2018 Savoir-faire Linux Inc.
 *  Author: Sébastien Blin <sebastien.blin@savoirfairelinux.com>
 *          Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 3 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program. If not, see <https://www.gnu.org/licenses/>.
 */

#if OPENDHT_PROXY_SERVER

#pragma once

#include "callbacks.h"
#include "def.h"
#include "infohash.h"
#include "proxy.h"
#include "scheduler.h"
#include "sockaddr.h"
#include "value.h"

#include <thread>
#include <memory>
#include <mutex>
#include <restbed>

namespace Json {
    class Value;
}

namespace dht {

class DhtRunner;

class OPENDHT_PUBLIC DhtProxyServer
{
public:
    DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port = 8000, const std::string& pushServer = "");
    virtual ~DhtProxyServer();

    DhtProxyServer(const DhtProxyServer& other) = delete;
    DhtProxyServer(DhtProxyServer&& other) = delete;
    DhtProxyServer& operator=(const DhtProxyServer& other) = delete;
    DhtProxyServer& operator=(DhtProxyServer&& other) = delete;

    struct ServerStats {
        size_t listenCount;
        size_t putCount;
        size_t pushListenersCount;
        double requestRate;

        std::string toString() const {
            std::ostringstream ss;
            ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
            ss << "Requests: " << requestRate << " per second." << std::endl;
            return ss.str();
        }
    };
    ServerStats getStats() const;

    std::shared_ptr<DhtRunner> getNode() const { return dht_; }

    void stop();

private:
    void getNodeInfo(const std::shared_ptr<restbed::Session>& session) const;

    void get(const std::shared_ptr<restbed::Session>& session) const;

    void listen(const std::shared_ptr<restbed::Session>& session);

    void put(const std::shared_ptr<restbed::Session>& session);

    void cancelPut(const InfoHash& key, Value::Id vid);

#if OPENDHT_PROXY_SERVER_IDENTITY

    void putSigned(const std::shared_ptr<restbed::Session>& session) const;

    void putEncrypted(const std::shared_ptr<restbed::Session>& session) const;
#endif // OPENDHT_PROXY_SERVER_IDENTITY

    void getFiltered(const std::shared_ptr<restbed::Session>& session) const;

    void handleOptionsMethod(const std::shared_ptr<restbed::Session>& session) const;

    void removeClosedListeners(bool testSession = true);

#if OPENDHT_PUSH_NOTIFICATIONS

    void subscribe(const std::shared_ptr<restbed::Session>& session);
    void unsubscribe(const std::shared_ptr<restbed::Session>& session);
    void sendPushNotification(const std::string& key, const Json::Value& json, bool isAndroid) const;

    void cancelPushListen(const std::string& pushToken, const InfoHash& key, const std::string& clientId);


#endif //OPENDHT_PUSH_NOTIFICATIONS

    using clock = std::chrono::steady_clock;
    using time_point = clock::time_point;

    std::thread server_thread {};
    std::unique_ptr<restbed::Service> service_;
    std::shared_ptr<DhtRunner> dht_;

    std::mutex schedulerLock_;
    std::condition_variable schedulerCv_;
    Scheduler scheduler_;
    std::thread schedulerThread_;

    Sp<Scheduler::Job> printStatsJob_;
    mutable std::mutex statsMutex_;
    mutable NodeInfo nodeInfo_ {};

    // Handle client quit for listen.
    // NOTE: can be simplified when we will supports restbed 5.0
    std::thread listenThread_;
    struct SessionToHashToken {
        std::shared_ptr<restbed::Session> session;
        InfoHash hash;
        std::future<size_t> token;
    };
    std::vector<SessionToHashToken> currentListeners_;
    std::mutex lockListener_;
    std::atomic_bool stopListeners {false};

    struct PermanentPut;
    struct SearchPuts;
    std::map<InfoHash, SearchPuts> puts_;

    mutable std::atomic<size_t> requestNum_ {0};
    mutable std::atomic<time_point> lastStatsReset_ {time_point::min()};

    const std::string pushServer_;

#if OPENDHT_PUSH_NOTIFICATIONS
    struct Listener;
    struct PushListener;
    std::mutex lockPushListeners_;
    std::map<std::string, PushListener> pushListeners_;
    proxy::ListenToken tokenPushNotif_ {0};
#endif //OPENDHT_PUSH_NOTIFICATIONS
};

}

#endif //OPENDHT_PROXY_SERVER