<commit_before>/*
 * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */

/* This must be included before anything else */
#if HAVE_CONFIG_H
#  include <config.h>
#endif


#include <yajr/rpc/internal/json_stream_wrappers.hpp>
#include <yajr/rpc/message_factory.inl.hpp>
#include <yajr/internal/comms.hpp>

#include <opflex/logging/internal/logging.hpp>

#include <rapidjson/error/en.h>

namespace yajr {
    namespace comms {
        namespace internal {

#ifdef COMMS_DEBUG_OBJECT_COUNT
::boost::atomic<size_t> CommunicationPeer::counter(0);
#endif

void CommunicationPeer::startKeepAlive(
        uint64_t begin,
        uint64_t repeat,
        uint64_t interval) {

    LOG(DEBUG) << this
        << " interval=" << interval
        << " begin=" << begin
        << " repeat=" << repeat
    ;

    sendEchoReq();
    bumpLastHeard();

    keepAliveInterval_ = interval;
    uv_timer_start(&keepAliveTimer_, on_timeout, begin, repeat);
}

void CommunicationPeer::stopKeepAlive() {
    LOG(DEBUG) << this;
 // assert(keepAliveInterval_ && uv_is_active((uv_handle_t *)&keepAliveTimer_));

    uv_timer_stop(&keepAliveTimer_);
    keepAliveInterval_ = 0;
}

void CommunicationPeer::on_timeout(uv_timer_t * timer) {
    LOG(DEBUG);

    get(timer)->timeout();

}

void CommunicationPeer::bumpLastHeard() const {
    LOG(DEBUG) << this << " " << lastHeard_ << " -> " << now();
    lastHeard_ = now();
}

void CommunicationPeer::onConnect() {
    connected_ = 1;
    keepAliveTimer_.data = this;
    LOG(DEBUG) << this << " up() for a timer init";
    up();
    uv_timer_init(getUvLoop(), &keepAliveTimer_);
    uv_unref((uv_handle_t*) &keepAliveTimer_);

    connectionHandler_(this, data_, ::yajr::StateChange::CONNECT, 0);

    /* some transports, like for example SSL/TLS, need to start talking
     * before there's anything to say */
    (void) write();
}

void CommunicationPeer::onDisconnect(bool now) {
    LOG(DEBUG)
        << this
        << " connected_ = "
        << static_cast< bool >(connected_)
        << " now = "
        << now
    ;

    if (connected_ || now) {
        LOG(DEBUG)
            << this
            << " issuing close for tcp handle"
        ;
        if (!uv_is_closing((uv_handle_t*)&handle_)) {
            uv_close((uv_handle_t*)&handle_, on_close);
         // uv_close((uv_handle_t*)&handle_, connected_? on_close : NULL);
        }
    }

    if (connected_) {

        connected_ = 0;

        if (getKeepAliveInterval()) {
            stopKeepAlive();
        }

        if (!uv_is_closing((uv_handle_t*)&keepAliveTimer_)) {
            uv_close((uv_handle_t*)&keepAliveTimer_, on_close);
        }
     // uv_close((uv_handle_t*)&handle_, on_close);

        /* FIXME: might be called too many times? */
        connectionHandler_(this, data_, ::yajr::StateChange::DISCONNECT, 0);

    }

    unlink();

    if (destroying_) {
        return;
    }

    if (!passive_) {
        LOG(DEBUG) << this << " active => retry queue";
        /* we should attempt to reconnect later */
        insert(internal::Peer::LoopData::RETRY_TO_CONNECT);
    } else {
        LOG(DEBUG) << this << " passive => eventually drop";
        /* whoever it was, hopefully will reconnect again */
        insert(internal::Peer::LoopData::PENDING_DELETE);
    }
#ifdef OLD_VERSION
    if (!connected_) {
        return;
    }

    connected_ = 0;

    if (getKeepAliveInterval()) {
        stopKeepAlive();
    }<commit_msg>start afresh each connection, wipe all queues<commit_after>/*
 * Copyright (c) 2014 Cisco Systems, Inc. and others.  All rights reserved.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
 * and is available at http://www.eclipse.org/legal/epl-v10.html
 */

/* This must be included before anything else */
#if HAVE_CONFIG_H
#  include <config.h>
#endif


#include <yajr/rpc/internal/json_stream_wrappers.hpp>
#include <yajr/rpc/message_factory.inl.hpp>
#include <yajr/internal/comms.hpp>

#include <opflex/logging/internal/logging.hpp>

#include <rapidjson/error/en.h>

namespace yajr {
    namespace comms {
        namespace internal {

#ifdef COMMS_DEBUG_OBJECT_COUNT
::boost::atomic<size_t> CommunicationPeer::counter(0);
#endif

void CommunicationPeer::startKeepAlive(
        uint64_t begin,
        uint64_t repeat,
        uint64_t interval) {

    LOG(DEBUG) << this
        << " interval=" << interval
        << " begin=" << begin
        << " repeat=" << repeat
    ;

    sendEchoReq();
    bumpLastHeard();

    keepAliveInterval_ = interval;
    uv_timer_start(&keepAliveTimer_, on_timeout, begin, repeat);
}

void CommunicationPeer::stopKeepAlive() {
    LOG(DEBUG) << this;
 // assert(keepAliveInterval_ && uv_is_active((uv_handle_t *)&keepAliveTimer_));

    uv_timer_stop(&keepAliveTimer_);
    keepAliveInterval_ = 0;
}

void CommunicationPeer::on_timeout(uv_timer_t * timer) {
    LOG(DEBUG);

    get(timer)->timeout();

}

void CommunicationPeer::bumpLastHeard() const {
    LOG(DEBUG) << this << " " << lastHeard_ << " -> " << now();
    lastHeard_ = now();
}

void CommunicationPeer::onConnect() {
    connected_ = 1;
    keepAliveTimer_.data = this;
    LOG(DEBUG) << this << " up() for a timer init";
    up();
    uv_timer_init(getUvLoop(), &keepAliveTimer_);
    uv_unref((uv_handle_t*) &keepAliveTimer_);

    connectionHandler_(this, data_, ::yajr::StateChange::CONNECT, 0);

    /* some transports, like for example SSL/TLS, need to start talking
     * before there's anything to say */
    (void) write();
}

void CommunicationPeer::onDisconnect(bool now) {
    LOG(DEBUG)
        << this
        << " connected_ = "
        << static_cast< bool >(connected_)
        << " now = "
        << now
    ;

    if (connected_ || now) {
        LOG(DEBUG)
            << this
            << " issuing close for tcp handle"
        ;
        if (!uv_is_closing((uv_handle_t*)&handle_)) {
            uv_close((uv_handle_t*)&handle_, on_close);
         // uv_close((uv_handle_t*)&handle_, connected_? on_close : NULL);
        }
    }

    if (connected_) {

        /* wipe ssIn_ out */
        // std::stringstream().swap(ssIn_); // C++11 only
        ssIn_.~basic_stringstream();
        new ((void *) &ssIn_) std::stringstream();

        /* wipe deque out and reset pendingBytes_ */
        s_.deque_.clear();
        pendingBytes_ = 0;

        connected_ = 0;

        if (getKeepAliveInterval()) {
            stopKeepAlive();
        }

        if (!uv_is_closing((uv_handle_t*)&keepAliveTimer_)) {
            uv_close((uv_handle_t*)&keepAliveTimer_, on_close);
        }
     // uv_close((uv_handle_t*)&handle_, on_close);

        /* FIXME: might be called too many times? */
        connectionHandler_(this, data_, ::yajr::StateChange::DISCONNECT, 0);

    }

    unlink();

    if (destroying_) {
        LOG(DEBUG)
            << this
            << " already destroying"
        ;
        return;
    }

    if (!passive_) {
        LOG(DEBUG) << this << " active => retry queue";
        /* we should attempt to reconnect later */
        insert(internal::Peer::LoopData::RETRY_TO_CONNECT);
    } else {
        LOG(DEBUG) << this << " passive => eventually drop";
        /* whoever it was, hopefully will reconnect again */
        insert(internal::Peer::LoopData::PENDING_DELETE);
    }
#ifdef OLD_VERSION
    if (!connected_) {
        return;
    }

    connected_ = 0;

    if (getKeepAliveInterval()) {
        stopKeepAlive();
    }
{'fraction_non_alphanumeric': 0.1349165596919127, 'fraction_numerical': 0.005776636713735558, 'mean_word_length': 2.765587240212663, 'pattern_counts': {'":': 0, '<': 127, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '7134090', 'n_tokens_mistral': 2540, 'n_tokens_neox': 2433, 'n_words': 538}