<commit_before>//  ===== ---- ofp/sys/buffered.h --------------------------*- C++ -*- =====  //
//
//  Copyright (c) 2013 <NAME>
//
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
//  
//  ===== ------------------------------------------------------------ =====  //
/// \file
/// \brief Defines the sys::Buffered class for use by ASIO streams.
//  ===== ------------------------------------------------------------ =====  //

#ifndef OFP_SYS_BUFFERED_H_
#define OFP_SYS_BUFFERED_H_

#include "ofp/sys/asio_utils.h"
#include "ofp/bytelist.h"

namespace ofp {
namespace sys {

OFP_BEGIN_IGNORE_PADDING

template <class StreamType>
class Buffered : private StreamType {
  using inherited = StreamType;

public:
  using next_layer_type = inherited;
  using lowest_layer_type = typename inherited::lowest_layer_type;

  using inherited::inherited;

  Buffered(tcp::socket sock, asio::ssl::context &context)
    : inherited{sock.get_io_service(), context} {
      lowest_layer() = std::move(sock);
  }

  using inherited::get_io_service;
  using inherited::async_read_some;
  // using inherited::async_write_some;
  using inherited::lowest_layer;
  using inherited::async_handshake;
  using inherited::async_shutdown;

  // using inherited::async_connect;
  // using inherited::remote_endpoint;
  // using inherited::is_open;
  // using inherited::set_option;
  // using inherited::close;

  ~Buffered() {
    // Shutdown the underlying socket before destroying buffers.
    shutdownLowestLayer();
  }

  const next_layer_type &next_layer() const { return *this; }
  next_layer_type &next_layer() { return *this; }

  bool is_open() const { return lowest_layer().is_open(); }

  void buf_write(const void *data, size_t length) {
    buffer_[bufferIdx_].add(data, length);
  }

  template <class CompletionHandler>
  void buf_flush(CompletionHandler &&handler);

  void shutdownLowestLayer();

private:
  // Use a two buffer strategy for async-writes. We queue up data in one
  // growable buffer while we're in the process of flushing the other buffer.
  ByteList buffer_[2];
  int bufferIdx_ = 0;
  bool isFlushing_ = false;
  bool isShutdown_ = false;
};

OFP_END_IGNORE_PADDING

template <class StreamType>
template <class CompletionHandler>
void Buffered<StreamType>::buf_flush(CompletionHandler &&handler) {
  const ByteList &outgoing = buffer_[bufferIdx_];
  if (isFlushing_ || outgoing.size() == 0)
    return;

  bufferIdx_ = !bufferIdx_;
  isFlushing_ = true;

  log::trace("write", outgoing.data(), outgoing.size());

  async_write(next_layer(), asio::buffer(outgoing.data(), outgoing.size()),
              [this, handler](const asio::error_code &err, size_t bytes_transferred) {

    if (!err) {
      assert(bytes_transferred == buffer_[!bufferIdx_].size());

      log::debug("Write:", bytes_transferred);
      isFlushing_ = false;
      buffer_[!bufferIdx_].clear();
      if (buffer_[bufferIdx_].size() > 0) {
        // Start another async write for the other output buffer.
        buf_flush(handler);
      } else {
        // Call completion handler.
        handler(err);
      }

    } else {
      log::debug("Write error ", err);
      handler(err);
    }
  });
}

template <class StreamType>
void Buffered<StreamType>::shutdownLowestLayer() {
  if (!isShutdown_ && is_open()) {
    std::error_code err;
    lowest_layer().shutdown(tcp::socket::shutdown_both, err);
    if (err)
      log::info("Buffered::shutdownLowestLayer error:", err.message());
    isShutdown_ = true;
  }
}

}  // namespace sys
}  // namespace ofp

#endif  // OFP_SYS_BUFFERED_H_<commit_msg>Remove isShutdown_ ivar.<commit_after>//  ===== ---- ofp/sys/buffered.h --------------------------*- C++ -*- =====  //
//
//  Copyright (c) 2013 <NAME>
//
//  Licensed under the Apache License, Version 2.0 (the "License");
//  you may not use this file except in compliance with the License.
//  You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
//  Unless required by applicable law or agreed to in writing, software
//  distributed under the License is distributed on an "AS IS" BASIS,
//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//  See the License for the specific language governing permissions and
//  limitations under the License.
//  
//  ===== ------------------------------------------------------------ =====  //
/// \file
/// \brief Defines the sys::Buffered class for use by ASIO streams.
//  ===== ------------------------------------------------------------ =====  //

#ifndef OFP_SYS_BUFFERED_H_
#define OFP_SYS_BUFFERED_H_

#include "ofp/sys/asio_utils.h"
#include "ofp/bytelist.h"

namespace ofp {
namespace sys {

OFP_BEGIN_IGNORE_PADDING

template <class StreamType>
class Buffered : private StreamType {
  using inherited = StreamType;

public:
  using next_layer_type = inherited;
  using lowest_layer_type = typename inherited::lowest_layer_type;

  using inherited::inherited;

  Buffered(tcp::socket sock, asio::ssl::context &context)
    : inherited{sock.get_io_service(), context} {
      lowest_layer() = std::move(sock);
  }

  using inherited::get_io_service;
  using inherited::async_read_some;
  // using inherited::async_write_some;
  using inherited::lowest_layer;
  using inherited::async_handshake;
  using inherited::async_shutdown;

  // using inherited::async_connect;
  // using inherited::remote_endpoint;
  // using inherited::is_open;
  // using inherited::set_option;
  // using inherited::close;

  const next_layer_type &next_layer() const { return *this; }
  next_layer_type &next_layer() { return *this; }

  bool is_open() const { return lowest_layer().is_open(); }

  void buf_write(const void *data, size_t length) {
    buffer_[bufferIdx_].add(data, length);
  }

  template <class CompletionHandler>
  void buf_flush(CompletionHandler &&handler);

  void shutdownLowestLayer();

private:
  // Use a two buffer strategy for async-writes. We queue up data in one
  // growable buffer while we're in the process of flushing the other buffer.
  ByteList buffer_[2];
  int bufferIdx_ = 0;
  bool isFlushing_ = false;
};

OFP_END_IGNORE_PADDING

template <class StreamType>
template <class CompletionHandler>
void Buffered<StreamType>::buf_flush(CompletionHandler &&handler) {
  const ByteList &outgoing = buffer_[bufferIdx_];
  if (isFlushing_ || outgoing.size() == 0)
    return;

  bufferIdx_ = !bufferIdx_;
  isFlushing_ = true;

  log::trace("write", outgoing.data(), outgoing.size());

  async_write(next_layer(), asio::buffer(outgoing.data(), outgoing.size()),
              [this, handler](const asio::error_code &err, size_t bytes_transferred) {

    if (!err) {
      assert(bytes_transferred == buffer_[!bufferIdx_].size());

      isFlushing_ = false;
      buffer_[!bufferIdx_].clear();
      if (buffer_[bufferIdx_].size() > 0) {
        // Start another async write for the other output buffer.
        buf_flush(handler);
      } else {
        // Call completion handler.
        handler(err);
      }

    } else {
      log::debug("Buffered::buf_flush error ", err);
      handler(err);
    }
  });
}

template <class StreamType>
void Buffered<StreamType>::shutdownLowestLayer() {
  if (is_open()) {
    std::error_code err;
    lowest_layer().shutdown(tcp::socket::shutdown_both, err);
    if (err)
      log::debug("Buffered::shutdownLowestLayer error:", err.message());
  }
}

}  // namespace sys
}  // namespace ofp

#endif  // OFP_SYS_BUFFERED_H_
{'fraction_non_alphanumeric': 0.1630407601900475, 'fraction_numerical': 0.003000750187546887, 'mean_word_length': 3.1125964010282776, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 21, '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': '3595897', 'n_tokens_mistral': 2458, 'n_tokens_neox': 2386, 'n_words': 640}