//  Copyright (c) 2005-2009 Hartmut Kaiser
//  Copyright (c) 2005-2007 Andre Merzky (andre@merzky.net)
// 
//  Distributed under the Boost Software License, Version 1.0. (See accompanying 
//  file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

#include <boost/asio.hpp>

#include <map>
#include <vector>

#include <saga/saga/exception.hpp>
#include <saga/saga/url.hpp>
#include <saga/impl/config.hpp>

#include <saga/saga/adaptors/task.hpp>
#include <saga/saga/adaptors/metric.hpp>
#include <saga/saga/adaptors/utils/ini/ini.hpp>
#include <saga/saga/packages/stream/adaptors/stream.hpp>

#include "stream_adaptor.hpp"
#include "stream_server.hpp"
#include "stream_cpi_data.hpp"

///////////////////////////////////////////////////////////////////////////////
namespace stream
{

///////////////////////////////////////////////////////////////////////////////
namespace detail
{
    std::string get_default_url(saga::ini::ini const& adap_ini)
    {
        // try to find default url in configuration file
        if (adap_ini.has_section("preferences")) 
        {
            saga::ini::ini prefs = adap_ini.get_section ("preferences");
            if (prefs.has_entry("url"))
            {
                return prefs.get_entry("url");
            }
        }

    // throw error 
        SAGA_OSSTREAM strm;
        strm << "stream_server::init: "
                "no default stream server URL specified in configuration.";
        SAGA_ADAPTOR_THROW_NO_CONTEXT(SAGA_OSSTREAM_GETSTRING(strm), 
            saga::BadParameter);
        return std::string();
    }
}

///////////////////////////////////////////////////////////////////////////////
stream_server::stream_server (proxy* p, cpi_info const& info, 
        saga::ini::ini const& glob_ini, saga::ini::ini const& adap_ini, 
        TR1::shared_ptr<saga::adaptor> adaptor)
  : base_cpi (p, info, adaptor, cpi::Noflags), 
    acceptor_(get_adaptor()->io_service_)
{
    instance_data_type data(this);
    if (data->location_.get_url().empty()) 
    {
        // choose implementation defined value
        data->location_ = detail::get_default_url(adap_ini);
    }
    
    // we implement only 'any' and 'tcp' schemes
    saga::url service_url(data->location_);
    std::string scheme(service_url.get_scheme());
    if (!scheme.empty() && scheme != "tcp" && scheme != "any")
    {
        SAGA_ADAPTOR_THROW("Can't use schemes others from 'any' or 'tcp' " 
            "for stream management.", saga::adaptors::AdaptorDeclined);
    }
}

void stream_server::initialize_service()
{
    if (acceptor_.is_open())
        return;     // initialize only once

    try {
        saga::url url;
        {
            instance_data_type data(this);
            url = data->location_;
        }

    // initialize acceptor
        mutex_type::scoped_lock lock(mtx_);
        int port = url.get_port();
        if (-1 == port) 
            port = SAGA_DEFAULT_STREAM_DEFAULT_PORT;

        using boost::asio::ip::tcp;

        tcp::endpoint endpoint(tcp::v4(), port);
        acceptor_.open(endpoint.protocol());
        acceptor_.set_option(tcp::acceptor::reuse_address(true));
        acceptor_.bind(endpoint);
        acceptor_.listen();
    }
    catch (boost::system::system_error const& e) {
        SAGA_OSSTREAM errstrm;
        errstrm << "stream_server::initialize_service: could not initialize "
                   "service: " << e.what();
        SAGA_ADAPTOR_THROW(SAGA_OSSTREAM_GETSTRING(errstrm), 
            saga::NoSuccess);
    } 
}

///////////////////////////////////////////////////////////////////////////////
stream_server::~stream_server (void)
{
    try {
        if (acceptor_.is_open()) 
            acceptor_.close();
    }
    catch (boost::system::system_error const& e) {
        SAGA_OSSTREAM errstrm;
        errstrm << "stream_server::~stream_server: could not close service: "
                << e.what();
        SAGA_LOG_CRITICAL(SAGA_OSSTREAM_GETSTRING(errstrm).c_str());
    } 
}

///////////////////////////////////////////////////////////////////////////////
// SAGA API functions
void stream_server::sync_get_url (saga::url& url)
{
    instance_data_type data(this);
    url = data->location_;
}

void stream_server::sync_serve(saga::stream::stream& result, double timeout)
{
    saga::url location;
    {
        instance_data_type data(this);
        location = data->location_; 
    }

    // make sure acceptor is initialized
    initialize_service();

    {
        mutex_type::scoped_lock lock(mtx_);

        // create the acceptor
        using boost::asio::ip::tcp;
        TR1::shared_ptr<tcp::socket> sock(new tcp::socket(get_adaptor()->io_service_));

        // try to accept a connection
        boost::system::error_code ec;
        acceptor_.accept(*sock, ec);
        if (ec) 
        {
            SAGA_OSSTREAM errstrm;
            errstrm << "stream_server::sync_serve: could not establish connection: "
                    << ec.message();
            SAGA_ADAPTOR_THROW(SAGA_OSSTREAM_GETSTRING(errstrm), 
                saga::BadParameter);
        }

        // create the SAGA stream object
        saga::adaptors::stream stream(
            new stream::stream_cpi_data(location, sock), 
            proxy_->get_session());

        result = stream;
    }

    // fire corresponding metric
    saga::monitorable monitor (this->proxy_);
    saga::adaptors::metric m (
        monitor.get_metric(saga::stream::metrics::server_clientconnect));
    m.set_attribute(saga::attributes::metric_value, "1");
    m.fire();
}

void stream_server::sync_close(saga::impl::void_t& ret, double timeout)
{
    // verify stream_server is operational
    if (!acceptor_.is_open())
    {
        SAGA_ADAPTOR_THROW(
            "stream_server::sync_close: stream has been closed already.",
            saga::IncorrectState);
    }

    // positive timeout is not supported
    if (timeout > 0.0)
    {
        SAGA_ADAPTOR_THROW(
            "stream_server::sync_close: positive timeout parameter values "
            "are not supported.", saga::BadParameter);
    }

    // close the service
    try {
        if (acceptor_.is_open()) 
            acceptor_.close();
    }
    catch (boost::system::system_error const& e) {
        SAGA_OSSTREAM errstrm;
        errstrm << "stream_server::sync_close: could not close service: "
                << e.what();
        SAGA_ADAPTOR_THROW(SAGA_OSSTREAM_GETSTRING(errstrm),
            saga::NoSuccess);
    } 
}

void stream_server::sync_connect (saga::stream::stream& ret, double timeout)
{
    if (timeout != -1.0)
    {
        SAGA_OSSTREAM errstrm;
        errstrm << "stream_server::sync_connect: timeout (other than -1.0) is "
                   "not supported, provided value: "
                << timeout;
        SAGA_ADAPTOR_THROW(SAGA_OSSTREAM_GETSTRING(errstrm),
            saga::NotImplemented);
    }

    saga::url url;
    {
        instance_data_type data(this);
        url = data->location_;
    }

    // create a new stream connected to this server
    saga::stream::stream strm (this->get_session(), url);
    strm.connect();

    ret = strm;
}

#if !defined(SAGA_DEFAULT_STREAM_ADAPTOR_NO_ASYNCS)
///////////////////////////////////////////////////////////////////////////////
// This adaptor implements the async functions based on its own synchronous
// functions for testing purposes only. Since there is no principal need
// to do so, we allow these to be 'switched off'.

saga::task stream_server::async_get_url () 
{
    return (saga::adaptors::task("stream_server::async_get_url",
        shared_from_this(), &stream_server::sync_get_url)); 
}

saga::task stream_server::async_serve(double timeout)
{
    return (saga::adaptors::task("stream_server::async_serve",
        shared_from_this(), &stream_server::sync_serve, 
        timeout)); 
}

saga::task stream_server::async_close(double timeout)
{
    return (saga::adaptors::task("stream_server::async_close",
        shared_from_this(), &stream_server::sync_close, timeout)); 
}

saga::task stream_server::async_connect(double timeout)
{
    return (saga::adaptors::task("stream_server::async_connect",
        shared_from_this(), &stream_server::sync_connect, timeout)); 
}

#endif

}   // namespace stream