Boost.Bind -> c++11 lambda

Author: tstenner

src/cancellable_streambuf.h

@@ -27,7 +27,6 @@
 #include <boost/asio/io_context.hpp>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/stream_socket_service.hpp>
-#include <boost/bind/bind.hpp>
 #include <boost/utility/base_from_member.hpp>
 #include <exception>
 #include <set>
@@ -68,8 +67,7 @@ class cancellable_streambuf : public std::streambuf,
 		cancel_issued_ = true;
 		lslboost::lock_guard<lslboost::recursive_mutex> lock(cancel_mut_);
 		cancel_started_ = false;
-		this->get_service().get_io_context().post(
-			lslboost::bind(&cancellable_streambuf::close_if_open, this));
+		this->get_service().get_io_context().post([this]() { close_if_open(); });
 	}
 
 

src/data_receiver.cpp

@@ -2,7 +2,6 @@
 #include "cancellable_streambuf.h"
 #include "sample.h"
 #include "socket_utils.h"
-#include <boost/bind.hpp>
 #include <iostream>
 #include <memory>
 

src/info_receiver.cpp

@@ -1,7 +1,6 @@
 #include "info_receiver.h"
 #include "cancellable_streambuf.h"
 #include "inlet_connection.h"
-#include <boost/bind.hpp>
 #include <iostream>
 #include <memory>
 

src/resolve_attempt_udp.cpp

@@ -4,13 +4,10 @@
 #include "socket_utils.h"
 #include <boost/asio/ip/multicast.hpp>
 #include <boost/asio/placeholders.hpp>
-#include <boost/bind.hpp>
-
-
-// === implementation of the resolver_burst_udp class ===
 
 using namespace lsl;
 using namespace lslboost::asio;
+using err_t = const lslboost::system::error_code &;
 
 resolve_attempt_udp::resolve_attempt_udp(io_context &io, const udp &protocol,
 	const std::vector<udp::endpoint> &targets, const std::string &query, result_container &results,
@@ -73,22 +70,25 @@ void resolve_attempt_udp::begin() {
 	// also initiate the cancel event, if desired
 	if (cancel_after_ != FOREVER) {
 		cancel_timer_.expires_after(timeout_sec(cancel_after_));
-		cancel_timer_.async_wait(lslboost::bind(
-			&resolve_attempt_udp::handle_timeout, shared_from_this(), placeholders::error));
+		auto keepalive(shared_from_this());
+		cancel_timer_.async_wait([keepalive, this](err_t err) {
+			if (!err) do_cancel();
+		});
 	}
 }
 
 void resolve_attempt_udp::cancel() {
-	post(io_, lslboost::bind(&resolve_attempt_udp::do_cancel, shared_from_this()));
+	auto keepalive(shared_from_this());
+	post(io_, [keepalive]() { keepalive->do_cancel(); });
 }
 
 
 // === receive loop ===
 
 void resolve_attempt_udp::receive_next_result() {
+	auto keepalive(shared_from_this());
 	recv_socket_.async_receive_from(buffer(resultbuf_), remote_endpoint_,
-		lslboost::bind(&resolve_attempt_udp::handle_receive_outcome, shared_from_this(),
-			placeholders::error, placeholders::bytes_transferred));
+		[keepalive, this](err_t err, size_t len) { handle_receive_outcome(err, len); });
 }
 
 void resolve_attempt_udp::handle_receive_outcome(error_code err, std::size_t len) {
@@ -142,9 +142,9 @@ void resolve_attempt_udp::handle_receive_outcome(error_code err, std::size_t len
 
 // === send loop ===
 
-void resolve_attempt_udp::send_next_query(endpoint_list::const_iterator i) {
-	if (i != targets_.end() && !cancelled_) {
-		udp::endpoint ep(*i);
+void resolve_attempt_udp::send_next_query(endpoint_list::const_iterator next) {
+	if (next != targets_.end() && !cancelled_) {
+		udp::endpoint ep(*next++);
 		// endpoint matches our active protocol?
 		if (ep.protocol() == recv_socket_.local_endpoint().protocol()) {
 			// select socket to use
@@ -153,29 +153,19 @@ void resolve_attempt_udp::send_next_query(endpoint_list::const_iterator i) {
 					? broadcast_socket_
 					: (ep.address().is_multicast() ? multicast_socket_ : unicast_socket_);
 			// and send the query over it
-			sock.async_send_to(lslboost::asio::buffer(query_msg_), ep,
-				lslboost::bind(&resolve_attempt_udp::handle_send_outcome, shared_from_this(), ++i,
-					placeholders::error));
+			auto keepalive(shared_from_this());
+			sock.async_send_to(
+				lslboost::asio::buffer(query_msg_), ep, [keepalive, this, next](err_t err, size_t) {
+					if (!cancelled_ && err != error::operation_aborted &&
+						err != error::not_connected && err != error::not_socket)
+						send_next_query(next);
+				});
 		} else
 			// otherwise just go directly to the next query
-			send_next_query(++i);
+			send_next_query(next);
 	}
 }
 
-/// Handler that gets called when a send has completed
-void resolve_attempt_udp::handle_send_outcome(endpoint_list::const_iterator i, error_code err) {
-	if (!cancelled_ && err != error::operation_aborted && err != error::not_connected &&
-		err != error::not_socket)
-		send_next_query(i);
-}
-
-
-// === cancellation logic ===
-
-void resolve_attempt_udp::handle_timeout(error_code err) {
-	if (!err) do_cancel();
-}
-
 void resolve_attempt_udp::do_cancel() {
 	try {
 		cancelled_ = true;

src/resolve_attempt_udp.h

@@ -86,14 +86,8 @@ class resolve_attempt_udp : public cancellable_obj,
 	/// Handler that gets called when a receive has completed.
 	void handle_receive_outcome(error_code err, std::size_t len);
 
-	/// Handler that gets called when a send has completed.
-	void handle_send_outcome(endpoint_list::const_iterator i, error_code err);
-
 	// === cancellation ===
 
-	/// Handler that gets called when the give up timeout has expired.
-	void handle_timeout(error_code err);
-
 	/// Cancel the outstanding operations.
 	void do_cancel();
 

src/resolver_impl.cpp

@@ -4,7 +4,6 @@
 #include "socket_utils.h"
 #include <boost/asio/ip/udp.hpp>
 #include <boost/asio/placeholders.hpp>
-#include <boost/bind.hpp>
 #include <boost/thread/thread_only.hpp>
 #include <memory>
 
@@ -137,7 +136,8 @@ void resolver_impl::resolve_continuous(const std::string &query, double forget_a
 	// start a wave of resolve packets
 	next_resolve_wave();
 	// spawn a thread that runs the IO operations
-	background_io_ = std::make_shared<lslboost::thread>(lslboost::bind(&io_context::run, io_));
+	auto io_keepalive(io_);
+	background_io_ = std::make_shared<lslboost::thread>([io_keepalive]() { io_keepalive->run(); });
 }
 
 std::vector<stream_info_impl> resolver_impl::results(uint32_t max_results) {

src/tcp_server.cpp

@@ -23,6 +23,7 @@
 #include "portable_archive/portable_oarchive.hpp"
 
 using namespace lslboost::asio;
+using err_t = const lslboost::system::error_code &;
 
 namespace lsl {
 /**
@@ -77,16 +78,9 @@ class client_session : public std::enable_shared_from_this<client_session> {
 	/// Handler that gets called after finishing reading of the query line.
 	void handle_read_query_outcome(error_code err);
 
-	/// Handler that gets called after finishing the sending of a reply (nothing to do here).
-	void handle_send_outcome(error_code) {}
-
 	/// Helper function to send a status message to the connected party.
 	void send_status_message(const std::string &msg);
 
-	/// Handler that gets called after finishing the sending of a message, holding a reference
-	/// to the message.
-	void handle_status_outcome(string_p, error_code) {}
-
 	/// Handler that gets called after finishing the reading of feedparameters.
 	void handle_read_feedparams(
 		int request_protocol_version, std::string request_uid, error_code err);
@@ -188,7 +182,8 @@ void tcp_server::end_serving() {
 	shutdown_ = true;
 	// issue closure of the server socket; this will result in a cancellation of the associated IO
 	// operations
-	post(*io_, lslboost::bind(&tcp::acceptor::close, acceptor_));
+	auto keepalive(acceptor_);
+	post(*io_, [keepalive]() { keepalive->close(); });
 	// issue closure of all active client session sockets; cancels the related outstanding IO jobs
 	close_inflight_sockets();
 	// also notify any transfer threads that are blocked waiting for a sample by sending them one (=
@@ -204,9 +199,9 @@ void tcp_server::accept_next_connection() {
 		std::shared_ptr<client_session> newsession{
 			std::make_shared<client_session>(shared_from_this())};
 		// accept a connection on the session's socket
-		acceptor_->async_accept(
-			*newsession->socket(), lslboost::bind(&tcp_server::handle_accept_outcome,
-									   shared_from_this(), newsession, placeholders::error));
+		auto keepalive(shared_from_this());
+		acceptor_->async_accept(*newsession->socket(),
+			[keepalive, newsession, this](err_t err) { handle_accept_outcome(newsession, err); });
 	} catch (std::exception &e) {
 		LOG_F(ERROR, "Error during tcp_server::accept_next_connection: %s", e.what());
 	}
@@ -234,22 +229,22 @@ void tcp_server::unregister_inflight_socket(const tcp_socket_p &sock) {
 	inflight_.erase(sock);
 }
 
-template <class SocketPtr, class Protocol> void shutdown_and_close(SocketPtr sock) {
-	try {
-		if (sock->is_open()) {
-			try {
-				// (in some cases shutdown may fail)
-				sock->shutdown(Protocol::socket::shutdown_both);
-			} catch (...) {}
-			sock->close();
-		}
-	} catch (std::exception &e) { LOG_F(WARNING, "Error during shutdown_and_close: %s", e.what()); }
-}
-
 void tcp_server::close_inflight_sockets() {
 	lslboost::lock_guard<lslboost::recursive_mutex> lock(inflight_mut_);
-	for (const auto &i : inflight_)
-		post(*io_, lslboost::bind(&shutdown_and_close<tcp_socket_p, tcp>, i));
+	for (auto sock : inflight_)
+		post(*io_, [sock]() {
+			try {
+				if (sock->is_open()) {
+					try {
+						// (in some cases shutdown may fail)
+						sock->shutdown(sock->shutdown_both);
+					} catch (...) {}
+					sock->close();
+				}
+			} catch (std::exception &e) {
+				LOG_F(WARNING, "Error during shutdown_and_close: %s", e.what());
+			}
+		});
 }
 
 
@@ -272,9 +267,9 @@ void client_session::begin_processing() {
 		serv_->register_inflight_socket(sock_);
 		registered_ = true;
 		// read the request line
+		auto keepalive(shared_from_this());
 		async_read_until(*sock_, requestbuf_, "\r\n",
-			lslboost::bind(&client_session::handle_read_command_outcome, shared_from_this(),
-				placeholders::error));
+			[keepalive, this](err_t err, size_t) { handle_read_command_outcome(err); });
 	} catch (std::exception &e) {
 		LOG_F(ERROR, "Error during client_session::begin_processing: %s", e.what());
 	}
@@ -287,29 +282,29 @@ void client_session::handle_read_command_outcome(error_code err) {
 			std::string method;
 			getline(requeststream_, method);
 			method = trim(method);
+			auto keepalive(shared_from_this());
 			if (method == "LSL:shortinfo")
 				// shortinfo request: read the content query string
 				async_read_until(*sock_, requestbuf_, "\r\n",
-					lslboost::bind(&client_session::handle_read_query_outcome, shared_from_this(),
-						placeholders::error));
+					[keepalive, this](err_t err, std::size_t) { handle_read_query_outcome(err); });
 			if (method == "LSL:fullinfo")
 				// fullinfo request: reply right away
 				async_write(*sock_, lslboost::asio::buffer(serv_->fullinfo_msg_),
-					lslboost::bind(&client_session::handle_send_outcome, shared_from_this(),
-						placeholders::error));
+					[keepalive, this](err_t, std::size_t) { });
 			if (method == "LSL:streamfeed")
 				// streamfeed request (1.00): read feed parameters
-				async_read_until(*sock_, requestbuf_, "\r\n",
-					lslboost::bind(&client_session::handle_read_feedparams, shared_from_this(), 100,
-						"", placeholders::error));
+				async_read_until(
+					*sock_, requestbuf_, "\r\n", [keepalive, this](err_t err, std::size_t) {
+						handle_read_feedparams(100, "", err);
+					});
 			if (method.compare(0, 15, "LSL:streamfeed/") == 0) {
 				// streamfeed request with version: read feed parameters
 				std::vector<std::string> parts = splitandtrim(method, ' ', true);
 				int request_protocol_version = std::stoi(parts[0].substr(15));
 				std::string request_uid = (parts.size() > 1) ? parts[1] : "";
-				async_read_until(*sock_, requestbuf_, "\r\n\r\n",
-					lslboost::bind(&client_session::handle_read_feedparams, shared_from_this(),
-						request_protocol_version, request_uid, placeholders::error));
+				async_read_until(*sock_, requestbuf_, "\r\n\r\n", [=](err_t err, std::size_t) {
+					keepalive->handle_read_feedparams(request_protocol_version, request_uid, err);
+				});
 			}
 		}
 	} catch (std::exception &e) {
@@ -324,12 +319,14 @@ void client_session::handle_read_query_outcome(error_code err) {
 			std::string query;
 			getline(requeststream_, query);
 			query = trim(query);
-			if (serv_->info_->matches_query(query))
+			if (serv_->info_->matches_query(query)) {
 				// matches: reply (otherwise just close the stream)
+				auto keepalive(shared_from_this());
 				async_write(*sock_, lslboost::asio::buffer(serv_->shortinfo_msg_),
-					lslboost::bind(&client_session::handle_send_outcome, shared_from_this(),
-						placeholders::error));
-			else
+					[keepalive](err_t, std::size_t) {
+						/* keep the client_session alive until the shortinfo is sent completely*/
+					});
+			} else
 				LOG_F(INFO, "%p got a shortinfo query response for the wrong query", this);
 		}
 	} catch (std::exception &e) {
@@ -338,10 +335,11 @@ void client_session::handle_read_query_outcome(error_code err) {
 }
 
 void client_session::send_status_message(const std::string &str) {
-	string_p msg(std::make_shared<std::string>(str));
+	auto msg(std::make_shared<std::string>(str));
+	auto keepalive(shared_from_this());
 	async_write(*sock_, lslboost::asio::buffer(*msg),
-		lslboost::bind(
-			&client_session::handle_status_outcome, shared_from_this(), msg, placeholders::error));
+		[msg, keepalive](
+			err_t, std::size_t) { /* keep objects alive until the message is sent */ });
 }
 
 void client_session::handle_read_feedparams(
@@ -489,9 +487,10 @@ void client_session::handle_read_feedparams(
 			else
 				*outarch_ << *temp;
 			// send off the newly created feedheader
-			async_write(*sock_, feedbuf_.data(),
-				lslboost::bind(&client_session::handle_send_feedheader_outcome, shared_from_this(),
-					placeholders::error, placeholders::bytes_transferred));
+			auto keepalive(shared_from_this());
+			async_write(*sock_, feedbuf_.data(), [keepalive, this](err_t err, size_t len) {
+				handle_send_feedheader_outcome(err, len);
+			});
 			DLOG_F(4, "%p sent test pattern samples", this);
 		}
 	} catch (std::exception &e) {
@@ -545,10 +544,10 @@ void client_session::transfer_samples_thread(std::shared_ptr<client_session>) {
 					// send off the chunk that we aggregated so far
 					lslboost::unique_lock<lslboost::mutex> lock(completion_mut_);
 					transfer_completed_ = false;
-					async_write(*sock_, feedbuf_.data(),
-						lslboost::bind(&client_session::handle_chunk_transfer_outcome,
-							shared_from_this(), placeholders::error,
-							placeholders::bytes_transferred));
+					auto keepalive(shared_from_this());
+					async_write(*sock_, feedbuf_.data(), [keepalive, this](err_t err, size_t len) {
+						handle_chunk_transfer_outcome(err, len);
+					});
 					// wait for the completion condition
 					completion_cond_.wait(lock, [this]() { return transfer_completed_; });
 					// handle transfer outcome