#ifndef IPC_SERVER_H
#define IPC_SERVER_H

#include <memory> //std::enable_shared_from_this

#include "ipc_session_holder.h"
#include "ipc_session.h"

namespace ipc {

template<typename CustomSession>
class Server : public SessionHolder {
  public:
    Server(asio::io_service& io_service, std::string address, std::string port) : acceptor_(io_service), socket_(io_service) {
      if (address.empty()) {
        address_ = "*";
      } else {
        address_ = std::move(address);
      }
      port_ = std::move(port);
      stopped_ = true;
    }

    ~Server() {
      stop();
    }

  /**
    * @brief Is server running.
    * Checks if the server is running. It might have been stopped with stop() or because of bind error.
    */
    inline bool isRunning() const {
      return !stopped_;
    }

  /**
    * @brief Starts the server.
    * Start accepting clients, by starting an acceptor and binding to an endpoint. 
    */
    void start() {
      if (stopped_ == true) {
        asio::ip::tcp::endpoint endpoint((address_ == "*" ? asio::ip::address_v4::from_string("0.0.0.0") : asio::ip::address_v4::from_string(address_)), 
                                        static_cast<unsigned short>(strtoul(port_.c_str(), nullptr, 0)));
        acceptor_.open(endpoint.protocol());
        acceptor_.set_option(asio::socket_base::reuse_address(true));
        std::error_code ec;
        acceptor_.bind(endpoint, ec);
        if (ec) {
          LOG_ERROR("IPC::Server - host:port = %s:%s - Error = %s - Try closing the server already in use before creating a new one!", address_.c_str(), port_.c_str(), ec.message().c_str());
          acceptor_.cancel();
          acceptor_.close();
        } else {
          acceptor_.listen();
          stopped_ = false;
          accept();
        }
      }
    }

  /**
    * @brief Stops the server.
    * Closes all sessions and cancels acceptor.
    */
    void stop() {
      std::lock_guard<std::mutex> lock(sessions_mutex_);
      if (stopped_ == false) {
        stopped_ = true;
        for (auto &s: active_sessions_) {
          if (s) {
            s->close();
          }
        }
        acceptor_.cancel();
        acceptor_.close();
        active_sessions_.clear();
      }
    }
 
   /**
    * @brief Broadcasts a message to all clients that are connected, subscribed and has the correct filter.
    * @param buff Buffer to broadcast
    */
    void broadcast(const Buffer &buff) const {
      if (buff.type() == MsgType::Error) {
        return;
      }
      std::lock_guard<std::mutex> lock(sessions_mutex_);
      for (auto &s: active_sessions_) {
        // TODO(harijs): This check might not be required if active_sessions are never nullptr and are always connected.
        // Shouldn't be because of the mutex'd GC.
        if (s && s->connected() == true) {
          if (s->subscribed() == true && s->subscribedToMessage(buff.type()) == true) {
            s->write(buff);
          }
        }
      }
    }

   /**
    * @brief Returns the number of active sessions.
    * They might be disconnected and not removed if broadcast() is never called, because it performs garbage collection there!
    * So connected() should still be checked for all of the sessions before use.
    */
    inline size_t sessionCount() const {
      std::lock_guard<std::mutex> lock(sessions_mutex_);
      return active_sessions_.size(); 
    }

  private:
   /**
    * @brief Accept new client connection.
    * Creates an async accept loop.
    */
    void accept() {
      if (stopped_ == true) { return; }
      acceptor_.async_accept(socket_,
          [this](const std::error_code &ec) {
            if (!ec) {
              auto session = std::make_shared<CustomSession>(std::move(socket_), this, false);
              session->start();
              std::lock_guard<std::mutex> lock(sessions_mutex_);
              active_sessions_.push_back(std::move(session));
            } else {
              LOG_WARNING("IPC::Server::accept() - host:port = %s:%s - Error = %s - Trying again!", address_.c_str(), port_.c_str(), ec.message().c_str());
            }

            accept();
          });
    }

   /**
    * @brief Perform garbage collection on session disconnect.
    */
    void sessionDisconnected() override final {
      std::lock_guard<std::mutex> lock(sessions_mutex_);
      auto session = std::begin(active_sessions_);
      while (session != std::end(active_sessions_)) {
        bool deleted = false;
        if (*session) {
          if ((*session)->connected() == false) {
            session = active_sessions_.erase(session);
            deleted = true;
          }
        }
        if (deleted == false) {
          session++;
        }
      }
    }

   /**
    * @brief Broadcast via the server session
    * @param buff Buffer to broadcast
    */
    inline void sessionBroadcast(const Buffer &buff) const override final {
      broadcast(buff);
    }

    std::string address_;
    std::string port_;
    asio::ip::tcp::acceptor acceptor_;
    asio::ip::tcp::socket   socket_;
    std::vector<std::shared_ptr<CustomSession> > active_sessions_;

    std::atomic<bool> stopped_;

    mutable std::mutex sessions_mutex_;
};

}

#endif //IPC_SERVER_H