Replace deprecated Boost.Asio classes

Author: tstenner

src/cancellable_streambuf.h

@@ -1,3 +1,4 @@
+#pragma once
 //
 // cancellable_streambuf.hpp
 // ~~~~~~~~~~~~~~~~~~~~~~~~~~
@@ -12,49 +13,32 @@
 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
 //
 
-#ifndef CANCELLABLE_STREAMBUF_HPP
-#define CANCELLABLE_STREAMBUF_HPP
-
-#if defined(_MSC_VER) && (_MSC_VER >= 1200)
-#pragma once
-#endif // defined(_MSC_VER) && (_MSC_VER >= 1200)
-
+#define BOOST_ASIO_NO_DEPRECATED
 #include "cancellation.h"
-#include <boost/asio/basic_socket.hpp>
-#include <boost/asio/detail/array.hpp>
-#include <boost/asio/detail/config.hpp>
-#include <boost/asio/detail/throw_error.hpp>
+#include <boost/asio/basic_stream_socket.hpp>
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
-#include <boost/asio/stream_socket_service.hpp>
-#include <boost/utility/base_from_member.hpp>
 #include <exception>
-#include <set>
 #include <streambuf>
 
 namespace asio = lslboost::asio;
 using lslboost::system::error_code;
-using asio::basic_socket;
 using asio::io_context;
-using asio::ip::tcp;
 
 namespace lsl {
-typedef tcp Protocol;
+using Protocol = asio::ip::tcp;
+using Socket = asio::basic_stream_socket<Protocol>;
 /// Iostream streambuf for a socket.
 class cancellable_streambuf : public std::streambuf,
-							  private lslboost::base_from_member<io_context>,
-							  public basic_socket<Protocol>,
+							  private asio::io_context,
+							  private Socket,
 							  public lsl::cancellable_obj {
 public:
 	/// Construct a cancellable_streambuf without establishing a connection.
-	cancellable_streambuf()
-		: basic_socket<Protocol>(lslboost::base_from_member<asio::io_context>::member),
-		  cancel_issued_(false), cancel_started_(false) {
-		init_buffers();
-	}
+	cancellable_streambuf() : io_context(), Socket(as_context()) { init_buffers(); }
 
 	/// Destructor flushes buffered data.
-	virtual ~cancellable_streambuf() {
+	virtual ~cancellable_streambuf() override {
 		// no cancel() can fire after this call
 		unregister_from_all();
 		if (pptr() != pbase()) overflow(traits_type::eof());
@@ -69,7 +53,7 @@ class cancellable_streambuf : public std::streambuf,
 		cancel_issued_ = true;
 		std::lock_guard<std::recursive_mutex> lock(cancel_mut_);
 		cancel_started_ = false;
-		this->get_service().get_io_context().post([this]() { close_if_open(); });
+		asio::post(this->as_context(), [this]() { close_if_open(); });
 	}
 
 
@@ -88,15 +72,12 @@ class cancellable_streambuf : public std::streambuf,
 					"Attempt to connect() a cancellable_streambuf after it has been cancelled.");
 
 			init_buffers();
-			this->basic_socket<Protocol>::close(ec_);
-
-			io_handler handler = {this};
-			this->basic_socket<Protocol>::async_connect(endpoint, handler);
-			this->get_service().get_io_context().reset();
+			socket().close(ec_);
+			socket().async_connect(endpoint, [this](const error_code &ec) { this->ec_ = ec; });
+			this->as_context().restart();
 		}
 		ec_ = asio::error::would_block;
-		do
-			this->get_service().get_io_context().run_one();
+		do as_context().run_one();
 		while (!cancel_issued_ && ec_ == asio::error::would_block);
 		return !ec_ ? this : nullptr;
 	}
@@ -108,7 +89,7 @@ class cancellable_streambuf : public std::streambuf,
 	 */
 	cancellable_streambuf *close() {
 		sync();
-		this->basic_socket<Protocol>::close(ec_);
+		socket().close(ec_);
 		if (!ec_) init_buffers();
 		return !ec_ ? this : nullptr;
 	}
@@ -122,36 +103,43 @@ class cancellable_streambuf : public std::streambuf,
 protected:
 	/// Close the socket if it's open.
 	void close_if_open() {
-		if (!cancel_started_ && this->is_open()) {
+		if (!cancel_started_ && socket().is_open()) {
 			cancel_started_ = true;
-			close();
+			socket().close();
 		}
 	}
 
-	/// This function makes sure that a cancellation, if issued, is not being eaten by the
-	/// io_context reset()
+	/// Convenience method to call methods inherited from `Socket`
+	Socket &socket() { return *this; }
+
+	/// Convenience method to call methods inherited from `io_context`
+	asio::io_context &as_context() { return *this; }
+
+	/// Make sure that a cancellation, if issued, is not being eaten by `io_context::reset()`
 	void protected_reset() {
 		std::lock_guard<std::recursive_mutex> lock(cancel_mut_);
 		// if the cancel() comes between completion of a run_one() and this call, close will be
 		// issued right here at the next opportunity
 		if (cancel_issued_) close_if_open();
-		this->get_service().get_io_context().reset();
+		this->as_context().restart();
 		// if the cancel() comes between this call and a completion of run_one(), the posted close
 		// will be processed by the run_one
 	}
 
 	int_type underflow() override {
 		if (gptr() == egptr()) {
-			io_handler handler = {this};
-			this->get_service().async_receive(this->get_implementation(),
-				asio::buffer(asio::buffer(get_buffer_) + putback_max), 0,
-				handler);
+			std::size_t bytes_transferred_;
+			socket().async_receive(asio::buffer(asio::buffer(get_buffer_) + putback_max),
+				[this, &bytes_transferred_](
+					const error_code &ec, std::size_t bytes_transferred = 0) {
+					this->ec_ = ec;
+					bytes_transferred_ = bytes_transferred;
+				});
 
 			ec_ = asio::error::would_block;
 			protected_reset(); // line changed for lsl
-			do
-				this->get_service().get_io_context().run_one();
-			while (ec_ == asio::error::would_block);
+			do as_context().run_one();
+			while (!cancel_issued_ && ec_ == asio::error::would_block);
 			if (ec_) return traits_type::eof();
 
 			setg(&get_buffer_[0], &get_buffer_[0] + putback_max,
@@ -165,18 +153,20 @@ class cancellable_streambuf : public std::streambuf,
 		// Send all data in the output buffer.
 		asio::const_buffer buffer = asio::buffer(pbase(), pptr() - pbase());
 		while (asio::buffer_size(buffer) > 0) {
-			io_handler handler = {this};
-			this->get_service().async_send(
-				this->get_implementation(), asio::buffer(buffer), 0, handler);
+			std::size_t bytes_transferred_;
+			socket().async_send(asio::buffer(buffer),
+				[this, &bytes_transferred_](const error_code &ec, std::size_t bytes_transferred) {
+					this->ec_ = ec;
+					bytes_transferred_ = bytes_transferred;
+				});
 			ec_ = asio::error::would_block;
 			protected_reset(); // line changed for lsl
-			do
-				this->get_service().get_io_context().run_one();
-			while (ec_ == asio::error::would_block);
+			do as_context().run_one();
+			while (!cancel_issued_ && ec_ == asio::error::would_block);
 			if (ec_) return traits_type::eof();
 			buffer = buffer + bytes_transferred_;
 		}
-		setp(&put_buffer_[0], &put_buffer_[0] + put_buffer_.size());
+		setp(&put_buffer_[0], &put_buffer_[0] + sizeof(put_buffer_));
 
 		// If the new character is eof then our work here is done.
 		if (traits_type::eq_int_type(c, traits_type::eof())) return traits_type::not_eof(c);
@@ -194,32 +184,17 @@ class cancellable_streambuf : public std::streambuf,
 		return nullptr;
 	}
 
-	// private:
 	void init_buffers() {
 		setg(&get_buffer_[0], &get_buffer_[0] + putback_max, &get_buffer_[0] + putback_max);
-		setp(&put_buffer_[0], &put_buffer_[0] + put_buffer_.size());
+		setp(&put_buffer_[0], &put_buffer_[0] + sizeof(put_buffer_));
 	}
 
-	struct io_handler;
-	friend struct io_handler;
-	struct io_handler {
-		cancellable_streambuf *this_;
-		void operator()(const error_code & ec, std::size_t bytes_transferred = 0) {
-			this_->ec_ = ec;
-			this_->bytes_transferred_ = bytes_transferred;
-		}
-	};
-
 	enum { putback_max = 8 };
 	enum { buffer_size = 512 };
-	asio::detail::array<char, buffer_size> get_buffer_;
-	asio::detail::array<char, buffer_size> put_buffer_;
+	char get_buffer_[buffer_size], put_buffer_[buffer_size];
 	error_code ec_;
-	std::size_t bytes_transferred_;
-	std::atomic<bool> cancel_issued_;
-	bool cancel_started_;
+	std::atomic<bool> cancel_issued_{false};
+	bool cancel_started_{false};
 	std::recursive_mutex cancel_mut_;
 };
 } // namespace lsl
-
-#endif // CANCELLABLE_STREAMBUF_HPP

src/stream_outlet_impl.cpp

@@ -4,6 +4,7 @@
 #include "send_buffer.h"
 #include "tcp_server.h"
 #include "udp_server.h"
+#include <algorithm>
 #include <memory>
 #include <sstream>