<commit_before>/*******************************************************************************
 * c7a/net/net_dispatcher.hpp
 *
 * Asynchronous callback wrapper around select(), epoll(), or other kernel-level
 * dispatchers.
 *
 * Part of Project c7a.
 *
 * Copyright (C) 2015 <NAME> <<EMAIL>>
 *
 * This file has no license. Only Chuck Norris can compile it.
 ******************************************************************************/

#pragma once
#ifndef C7A_NET_NET_DISPATCHER_HEADER
#define C7A_NET_NET_DISPATCHER_HEADER

#include <c7a/net/net_connection.hpp>
#include <c7a/net/buffer.hpp>
#include <c7a/net/lowlevel/socket.hpp>
#include <c7a/net/lowlevel/select_dispatcher.hpp>
//#include <c7a/net/lowlevel/epoll-dispatcher.hpp> //TODO(tb) can we use a os switch? Do we want that?

#if defined(_LIBCPP_VERSION) || defined(__clang__)
#include <c7a/common/delegate.hpp>
#endif

#include <string>
#include <deque>
#include <queue>
#include <ctime>
#include <chrono>

namespace c7a {
namespace net {

//! \addtogroup net Network Communication
//! \{

/**
 * NetDispatcher is a high level wrapper for asynchronous callback
 * processing.. One can register NetConnection objects for readability and
 * writability checks, buffered reads and writes with completion callbacks, and
 * also timer functions.
 */
class NetDispatcher
{
    static const bool debug = false;

protected:
    //! switch between different low-level dispatchers
    typedef lowlevel::SelectDispatcher<NetConnection&> Dispatcher;
    //typedef lowlevel::EPollDispatcher Dispatcher;

    //! import into class namespace
    typedef lowlevel::Socket Socket;

    //! import into class namespace
    typedef std::chrono::steady_clock steady_clock;

    //! import into class namespace
    typedef std::chrono::milliseconds milliseconds;

#if defined(_LIBCPP_VERSION) || defined(__clang__)
    template <typename Signature>
    using function = common::delegate<Signature>;
#else
    template <typename Signature>
    using function = std::function<Signature>;
#endif

public:
    //! \name Timeout Callbacks
    //! \{

    //! callback signature for timer events
    typedef function<bool ()> TimerCallback;

    //! Register a relative timeout callback
    template <class Rep, class Period>
    void AddRelativeTimeout(const std::chrono::duration<Rep, Period>& timeout,
                            const TimerCallback& cb) {
        timer_pq_.emplace(steady_clock::now() + timeout,
                          std::chrono::duration_cast<milliseconds>(timeout),
                          cb);
    }

    //! \}

    //! \name NetConnection Callbacks
    //! \{

    //! callback signature for socket readable/writable events
    typedef function<bool (NetConnection&)> ConnectionCallback;

    //! Register a buffered read callback and a default exception callback.
    void AddRead(NetConnection& c, const ConnectionCallback& read_cb) {
        return dispatcher_.AddRead(c.GetSocket().fd(), c, read_cb);
    }

    //! Register a buffered write callback and a default exception callback.
    void AddWrite(NetConnection& c, const ConnectionCallback& write_cb) {
        return dispatcher_.AddWrite(c.GetSocket().fd(), c, write_cb);
    }

    //! Register a buffered write callback and a default exception callback.
    void AddReadWrite(
        NetConnection& c,
        const ConnectionCallback& read_cb, const ConnectionCallback& write_cb) {
        return dispatcher_.AddReadWrite(
            c.GetSocket().fd(), c, read_cb, write_cb);
    }

    //! \}

    //! \name Asynchronous Data Reader/Writer Callbacks
    //! \{

    //! callback signature for async read callbacks, they may acquire the buffer
    typedef function<void (NetConnection& c,
                           Buffer&& buffer)> AsyncReadCallback;

    //! asynchronously read n bytes and deliver them to the callback
    virtual void AsyncRead(NetConnection& c, size_t n,
                           AsyncReadCallback done_cb) {
        assert(c.GetSocket().IsValid());

        LOG << "async read on read dispatcher";
        if (n == 0) {
            if (done_cb) done_cb(c, Buffer());<commit_msg>Removing stray virtual.<commit_after>/*******************************************************************************
 * c7a/net/net_dispatcher.hpp
 *
 * Asynchronous callback wrapper around select(), epoll(), or other kernel-level
 * dispatchers.
 *
 * Part of Project c7a.
 *
 * Copyright (C) 2015 <NAME> <<EMAIL>>
 *
 * This file has no license. Only Chuck Norris can compile it.
 ******************************************************************************/

#pragma once
#ifndef C7A_NET_NET_DISPATCHER_HEADER
#define C7A_NET_NET_DISPATCHER_HEADER

#include <c7a/net/net_connection.hpp>
#include <c7a/net/buffer.hpp>
#include <c7a/net/lowlevel/socket.hpp>
#include <c7a/net/lowlevel/select_dispatcher.hpp>
//#include <c7a/net/lowlevel/epoll-dispatcher.hpp> //TODO(tb) can we use a os switch? Do we want that?

#if defined(_LIBCPP_VERSION) || defined(__clang__)
#include <c7a/common/delegate.hpp>
#endif

#include <string>
#include <deque>
#include <queue>
#include <ctime>
#include <chrono>

namespace c7a {
namespace net {

//! \addtogroup net Network Communication
//! \{

/**
 * NetDispatcher is a high level wrapper for asynchronous callback
 * processing.. One can register NetConnection objects for readability and
 * writability checks, buffered reads and writes with completion callbacks, and
 * also timer functions.
 */
class NetDispatcher
{
    static const bool debug = false;

protected:
    //! switch between different low-level dispatchers
    typedef lowlevel::SelectDispatcher<NetConnection&> Dispatcher;
    //typedef lowlevel::EPollDispatcher Dispatcher;

    //! import into class namespace
    typedef lowlevel::Socket Socket;

    //! import into class namespace
    typedef std::chrono::steady_clock steady_clock;

    //! import into class namespace
    typedef std::chrono::milliseconds milliseconds;

#if defined(_LIBCPP_VERSION) || defined(__clang__)
    template <typename Signature>
    using function = common::delegate<Signature>;
#else
    template <typename Signature>
    using function = std::function<Signature>;
#endif

public:
    //! \name Timeout Callbacks
    //! \{

    //! callback signature for timer events
    typedef function<bool ()> TimerCallback;

    //! Register a relative timeout callback
    template <class Rep, class Period>
    void AddRelativeTimeout(const std::chrono::duration<Rep, Period>& timeout,
                            const TimerCallback& cb) {
        timer_pq_.emplace(steady_clock::now() + timeout,
                          std::chrono::duration_cast<milliseconds>(timeout),
                          cb);
    }

    //! \}

    //! \name NetConnection Callbacks
    //! \{

    //! callback signature for socket readable/writable events
    typedef function<bool (NetConnection&)> ConnectionCallback;

    //! Register a buffered read callback and a default exception callback.
    void AddRead(NetConnection& c, const ConnectionCallback& read_cb) {
        return dispatcher_.AddRead(c.GetSocket().fd(), c, read_cb);
    }

    //! Register a buffered write callback and a default exception callback.
    void AddWrite(NetConnection& c, const ConnectionCallback& write_cb) {
        return dispatcher_.AddWrite(c.GetSocket().fd(), c, write_cb);
    }

    //! Register a buffered write callback and a default exception callback.
    void AddReadWrite(
        NetConnection& c,
        const ConnectionCallback& read_cb, const ConnectionCallback& write_cb) {
        return dispatcher_.AddReadWrite(
            c.GetSocket().fd(), c, read_cb, write_cb);
    }

    //! \}

    //! \name Asynchronous Data Reader/Writer Callbacks
    //! \{

    //! callback signature for async read callbacks, they may acquire the buffer
    typedef function<void (NetConnection& c,
                           Buffer&& buffer)> AsyncReadCallback;

    //! asynchronously read n bytes and deliver them to the callback
    void AsyncRead(NetConnection& c, size_t n, AsyncReadCallback done_cb) {
        assert(c.GetSocket().IsValid());

        LOG << "async read on read dispatcher";
        if (n == 0) {
            if (done_cb) done_cb(c, Buffer());
{'fraction_non_alphanumeric': 0.13947018265392525, 'fraction_numerical': 0.0038708116608201283, 'mean_word_length': 3.342436974789916, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5293655', 'n_tokens_mistral': 2293, 'n_tokens_neox': 2181, 'n_words': 678}