#pragma once

#include <jlib/dp.h>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>

namespace grpc {
class Server;
}

class RpcServer : public jlib::dp::singleton<RpcServer>
{
protected:
    RpcServer();

public:
    void startup();
    void shutdown();

private:
    class AlarmServerImpl;
    std::shared_ptr<grpc::Server> server_ = {};

    bool running_ = false;
    bool shutdown_ok_ = false;

    std::thread thread_ = {};
    std::mutex mutex_ = {};
    std::condition_variable cv_ = {};
};

#include "rpc_server.h"
#include <grpcpp/grpcpp.h>
#include "alarm_server.grpc.pb.h"
#include "../core/machine.h"
#include <jlib/log2.h>

class RpcServer::AlarmServerImpl final : public ::alarm_server::AlarmServer::Service
{
protected:
virtual ::grpc::Status IsNeedSync(::grpc::ServerContext* context,
            const ::alarm_server::IsNeedSyncRequest* request,
            ::alarm_server::IsNeedSyncReply* reply) override {
    reply->set_changedmachinecount(core::machine_mgr::get_instance()    ->getUpdatedMachineCount());
    return ::grpc::Status::OK;
}

virtual ::grpc::Status Sync(::grpc::ServerContext* context,
            const ::alarm_server::GetMachineRequest* request,
            ::grpc::ServerWriter<::alarm_server::MachineReply>* writer) override {
        static auto machine2reply = [](const core::machine_ptr& machine) {
        static auto machine2alarm_msg = [](const core::machine_ptr& machine, alarm_server::AlarmMessage* alarmMsg) {
        if (machine->has_alarm()) {
            alarmMsg->set_valid(true);
            alarmMsg->set_zone(machine->get_prev_alarm_exception_zone());
            alarmMsg->set_event(machine->get_prev_alarm_exception_event());
            alarmMsg->set_timestamp(std::chrono::system_clock::to_time_t(machine->get_prev_alarm_time()));

            core::machine_mgr::get_instance()->clear_machine_alarm(machine);
            } else {
                alarmMsg->set_valid(false);
            }
        };

        alarm_server::MachineReply reply;
        reply.set_account(machine->get_account());
        reply.set_online(machine->is_online());
        reply.set_type(machine->get_type());
        reply.set_status(machine->get_status());
        reply.set_password(machine->get_psw());
        machine2alarm_msg(machine, reply.mutable_alarmmessage());
        return reply;
    };

    if (request->account().empty()) {
        for (auto machine : core::machine_mgr::get_instance()->getUpdatedMachines()) {
        if (!writer->Write(machine2reply(machine))) {
            JLOG_ERRO("AlarmServerImpl::Sync failed");
            break;
        }
        }
    } else {
        auto machine = core::machine_mgr::get_instance()->get(request->account());
        if (machine) {
            if (!writer->Write(machine2reply(machine))) {
                JLOG_ERRO("AlarmServerImpl::Sync failed");
            }
        }
    }

    return ::grpc::Status::OK;
}

public:

};



RpcServer::RpcServer()
{

}

void RpcServer::startup()
{
    running_ = true;

    thread_ = std::thread([this]() {
        AUTO_LOG_FUNCTION;
        try {
            AlarmServerImpl service;
            std::string server_address("0.0.0.0:50051");
            ::grpc::ServerBuilder builder;
            builder.AddListeningPort(server_address, ::grpc::InsecureServerCredentials());
            builder.RegisterService(&service);
            std::shared_ptr<grpc::Server> server = builder.BuildAndStart();
            server_ = server;

            JLOG_INFO("RpcServer::AlarmServerImpl started on {}", server_address);

            server->Wait();

            {
                std::unique_lock<std::mutex> ul(mutex_);
                cv_.wait(ul, [this]() {return shutdown_ok_; });
            }
        } catch (...) {
            return;
        }
    });
}

void RpcServer::shutdown()
{
    AUTO_LOG_FUNCTION;
    running_ = false;

    {
        std::lock_guard<std::mutex> lg(mutex_);
        server_->Shutdown();
        server_ = nullptr;
        shutdown_ok_ = true;

        JLOG_INFO("server_ done shutdown");
    }

    cv_.notify_one();
    thread_.join();
    JLOG_INFO("thread_ done joined, RpcServer::shutdown OK");
}

#include <iostream>
#include <memory>
#include <string>
#include <chrono>
#include <ctime>
#include <thread>
#include <grpcpp/grpcpp.h>
#include "../wechat_server/rpc/alarm_server.grpc.pb.h"
#include "../wechat_server/core/machine.h"

using grpc::Channel;
using grpc::ClientContext;
using grpc::Status;
using namespace alarm_server;

class AlarmClient {
public:
    AlarmClient(std::shared_ptr<Channel> channel)
        : stub_(AlarmServer::NewStub(channel)) {}

    size_t IsNeedSync() {
        IsNeedSyncRequest request;
        IsNeedSyncReply reply;
        ClientContext context;

        Status status = stub_->IsNeedSync(&context, request, &reply);

        if(status.ok()){
            return reply.changedmachinecount();
        }else{
            std::cout << "IsNeedSync: " << status.error_code() << ": " << status.error_message() << std::endl;
            return 0;
        }
    }

    void sync(){
        GetMachineRequest request;
        ClientContext context;
        MachineReply reply;
        auto reader = stub_->Sync(&context, request);
        int i = 0;
        //std::cout << "--------------------sync started" << std::endl;
        while(reader->Read(&reply)){
            std::cout << "#" << ++i << " "
                << reply.account() << " "
                << (reply.online() ? "online " : "offline ")
                << core::machine_info::get_type_string(core::machine_info::integer2type(reply.type())) << " "
                << core::machine_info::get_status_string(core::machine_info::integer2status(reply.status())) << " "
                << reply.password() << " ";

            const auto& alarmMsg = reply.alarmmessage();
            if(alarmMsg.valid()){
                std::cout << "alarm message:" << " "
                    << alarmMsg.zone()  << " "
                    << alarmMsg.event()  << " ";
                auto t = static_cast<time_t>(alarmMsg.timestamp());
                std::tm * ptm = std::localtime(&t);
                char buffer[32];
                // Format: 2009-12-31 20:20:00
                std::strftime(buffer, 32, "%Y-%m-%d %H:%M:%S", ptm);
                std::cout << buffer << std::endl;
            }else{
                std::cout << "no alarm message" << std::endl;
            }
        }

        std::cout << std::endl;
        //std::cout << "sync finished-------------------" << std::endl << std::endl;
    }

    void sync(const std::string& account){
        GetMachineRequest request;
        request.set_account(account);
        ClientContext context;
        MachineReply reply;
        auto reader = stub_->Sync(&context, request);
        int i = 0;
        //std::cout << "--------------------sync started" << std::endl;
        while(reader->Read(&reply)){
            std::cout << "#" << ++i << " "
                << reply.account() << " "
                << (reply.online() ? "online " : "offline ")
                << core::machine_info::get_type_string(core::machine_info::integer2type(reply.type())) << " "
                << core::machine_info::get_status_string(core::machine_info::integer2status(reply.status())) << " "
                << reply.password() << " ";

            const auto& alarmMsg = reply.alarmmessage();
            if(alarmMsg.valid()){
                std::cout << "alarm message:" << " "
                    << alarmMsg.zone()  << " "
                    << alarmMsg.event()  << " ";
                auto t = static_cast<time_t>(alarmMsg.timestamp());
                std::tm * ptm = std::localtime(&t);
                char buffer[32];
                // Format: 2009-12-31 20:20:00
                std::strftime(buffer, 32, "%Y-%m-%d %H:%M:%S", ptm);
                std::cout << buffer << std::endl;
            }else{
                std::cout << "no alarm message" << std::endl;
            }
        }

        std::cout << std::endl;
    }

private:
    std::unique_ptr<AlarmServer::Stub> stub_;
};

int main()
{
    AlarmClient client(grpc::CreateChannel("localhost:50051", grpc::InsecureChannelCredentials()));

    std::cout << "Testing sync with account..." << std::endl;

    std::cout << "sync 70801150121796" << std::endl;
    client.sync("70801150121796");

    std::cout << "sync 70801151915015" << std::endl;
    client.sync("70801151915015");

    std::cout << "sync 70801163150828" << std::endl;
    client.sync("70801163150828");

    std::cout << "Testing sync without account..." << std::endl;
    while(true){
        auto cnt = client.IsNeedSync();
        if(cnt > 0){
            //std::cout << std::endl << cnt << " machines updated, syncing..."<< std::endl;
            client.sync();
        }else{
            //std::cout << "no machines updated"<< std::endl;
        }

        std::this_thread::sleep_for(std::chrono::seconds(1));
    }

    return 0;
}