📄 reactive_socket_service.hpp
字号:
if (impl.flags_ & implementation_type::enable_connection_aborted) *target = 1; else *target = 0; option.resize(impl.protocol_, sizeof(int)); ec = asio::error_code(); } return ec; } else { size_t size = option.size(impl.protocol_); socket_ops::getsockopt(impl.socket_, option.level(impl.protocol_), option.name(impl.protocol_), option.data(impl.protocol_), &size, ec); if (!ec) option.resize(impl.protocol_, size); return ec; } } // Perform an IO control command on the socket. template <typename IO_Control_Command> asio::error_code io_control(implementation_type& impl, IO_Control_Command& command, asio::error_code& ec) { if (!is_open(impl)) { ec = asio::error::bad_descriptor; return ec; } if (command.name() == static_cast<int>(FIONBIO)) { if (command.get()) impl.flags_ |= implementation_type::user_set_non_blocking; else impl.flags_ &= ~implementation_type::user_set_non_blocking; ec = asio::error_code(); } else { socket_ops::ioctl(impl.socket_, command.name(), static_cast<ioctl_arg_type*>(command.data()), ec); } return ec; } // Get the local endpoint. endpoint_type local_endpoint(const implementation_type& impl, asio::error_code& ec) const { if (!is_open(impl)) { ec = asio::error::bad_descriptor; return endpoint_type(); } endpoint_type endpoint; std::size_t addr_len = endpoint.capacity(); if (socket_ops::getsockname(impl.socket_, endpoint.data(), &addr_len, ec)) return endpoint_type(); endpoint.resize(addr_len); return endpoint; } // Get the remote endpoint. endpoint_type remote_endpoint(const implementation_type& impl, asio::error_code& ec) const { if (!is_open(impl)) { ec = asio::error::bad_descriptor; return endpoint_type(); } endpoint_type endpoint; std::size_t addr_len = endpoint.capacity(); if (socket_ops::getpeername(impl.socket_, endpoint.data(), &addr_len, ec)) return endpoint_type(); endpoint.resize(addr_len); return endpoint; } /// Disable sends or receives on the socket. asio::error_code shutdown(implementation_type& impl, socket_base::shutdown_type what, asio::error_code& ec) { if (!is_open(impl)) { ec = asio::error::bad_descriptor; return ec; } socket_ops::shutdown(impl.socket_, what, ec); return ec; } // Send the given data to the peer. template <typename ConstBufferSequence> size_t send(implementation_type& impl, const ConstBufferSequence& buffers, socket_base::message_flags flags, asio::error_code& ec) { if (!is_open(impl)) { ec = asio::error::bad_descriptor; return 0; } // Copy buffers into array. socket_ops::buf bufs[max_buffers]; typename ConstBufferSequence::const_iterator iter = buffers.begin(); typename ConstBufferSequence::const_iterator end = buffers.end(); size_t i = 0; size_t total_buffer_size = 0; for (; iter != end && i < max_buffers; ++iter, ++i) { asio::const_buffer buffer(*iter); socket_ops::init_buf(bufs[i], asio::buffer_cast<const void*>(buffer), asio::buffer_size(buffer)); total_buffer_size += asio::buffer_size(buffer); } // A request to receive 0 bytes on a stream socket is a no-op. if (impl.protocol_.type() == SOCK_STREAM && total_buffer_size == 0) { ec = asio::error_code(); return 0; } // Make socket non-blocking if user wants non-blocking. if (impl.flags_ & implementation_type::user_set_non_blocking) { if (!(impl.flags_ & implementation_type::internal_non_blocking)) { ioctl_arg_type non_blocking = 1; if (socket_ops::ioctl(impl.socket_, FIONBIO, &non_blocking, ec)) return 0; impl.flags_ |= implementation_type::internal_non_blocking; } } // Send the data. for (;;) { // Try to complete the operation without blocking. int bytes_sent = socket_ops::send(impl.socket_, bufs, i, flags, ec); // Check if operation succeeded. if (bytes_sent >= 0) return bytes_sent; // Operation failed. if ((impl.flags_ & implementation_type::user_set_non_blocking) || (ec != asio::error::would_block && ec != asio::error::try_again)) return 0; // Wait for socket to become ready. if (socket_ops::poll_write(impl.socket_, ec) < 0) return 0; } } template <typename ConstBufferSequence, typename Handler> class send_handler { public: send_handler(socket_type socket, asio::io_service& io_service, const ConstBufferSequence& buffers, socket_base::message_flags flags, Handler handler) : socket_(socket), io_service_(io_service), work_(io_service), buffers_(buffers), flags_(flags), handler_(handler) { } bool operator()(const asio::error_code& result) { // Check whether the operation was successful. if (result) { io_service_.post(bind_handler(handler_, result, 0)); return true; } // Copy buffers into array. socket_ops::buf bufs[max_buffers]; typename ConstBufferSequence::const_iterator iter = buffers_.begin(); typename ConstBufferSequence::const_iterator end = buffers_.end(); size_t i = 0; for (; iter != end && i < max_buffers; ++iter, ++i) { asio::const_buffer buffer(*iter); socket_ops::init_buf(bufs[i], asio::buffer_cast<const void*>(buffer), asio::buffer_size(buffer)); } // Send the data. asio::error_code ec; int bytes = socket_ops::send(socket_, bufs, i, flags_, ec); // Check if we need to run the operation again. if (ec == asio::error::would_block || ec == asio::error::try_again) return false; io_service_.post(bind_handler(handler_, ec, bytes < 0 ? 0 : bytes)); return true; } private: socket_type socket_; asio::io_service& io_service_; asio::io_service::work work_; ConstBufferSequence buffers_; socket_base::message_flags flags_; Handler handler_; }; // Start an asynchronous send. The data being sent must be valid for the // lifetime of the asynchronous operation. template <typename ConstBufferSequence, typename Handler> void async_send(implementation_type& impl, const ConstBufferSequence& buffers, socket_base::message_flags flags, Handler handler) { if (!is_open(impl)) { this->get_io_service().post(bind_handler(handler, asio::error::bad_descriptor, 0)); } else { if (impl.protocol_.type() == SOCK_STREAM) { // Determine total size of buffers. typename ConstBufferSequence::const_iterator iter = buffers.begin(); typename ConstBufferSequence::const_iterator end = buffers.end(); size_t i = 0; size_t total_buffer_size = 0; for (; iter != end && i < max_buffers; ++iter, ++i) { asio::const_buffer buffer(*iter); total_buffer_size += asio::buffer_size(buffer); } // A request to receive 0 bytes on a stream socket is a no-op. if (total_buffer_size == 0) { this->get_io_service().post(bind_handler(handler, asio::error_code(), 0)); return; } } // Make socket non-blocking. if (!(impl.flags_ & implementation_type::internal_non_blocking)) { ioctl_arg_type non_blocking = 1; asio::error_code ec; if (socket_ops::ioctl(impl.socket_, FIONBIO, &non_blocking, ec)) { this->get_io_service().post(bind_handler(handler, ec, 0)); return; } impl.flags_ |= implementation_type::internal_non_blocking; } reactor_.start_write_op(impl.socket_, send_handler<ConstBufferSequence, Handler>( impl.socket_, this->get_io_service(), buffers, flags, handler)); } } // Send a datagram to the specified endpoint. Returns the number of bytes // sent. template <typename ConstBufferSequence> size_t send_to(implementation_type& impl, const ConstBufferSequence& buffers, const endpoint_type& destination, socket_base::message_flags flags, asio::error_code& ec) { if (!is_open(impl)) { ec = asio::error::bad_descriptor; return 0; } // Copy buffers into array. socket_ops::buf bufs[max_buffers]; typename ConstBufferSequence::const_iterator iter = buffers.begin(); typename ConstBufferSequence::const_iterator end = buffers.end(); size_t i = 0; for (; iter != end && i < max_buffers; ++iter, ++i) { asio::const_buffer buffer(*iter); socket_ops::init_buf(bufs[i], asio::buffer_cast<const void*>(buffer), asio::buffer_size(buffer)); } // Make socket non-blocking if user wants non-blocking. if (impl.flags_ & implementation_type::user_set_non_blocking) { if (!(impl.flags_ & implementation_type::internal_non_blocking)) { ioctl_arg_type non_blocking = 1; if (socket_ops::ioctl(impl.socket_, FIONBIO, &non_blocking, ec)) return 0; impl.flags_ |= implementation_type::internal_non_blocking; } } // Send the data. for (;;) { // Try to complete the operation without blocking. int bytes_sent = socket_ops::sendto(impl.socket_, bufs, i, flags, destination.data(), destination.size(), ec); // Check if operation succeeded. if (bytes_sent >= 0) return bytes_sent; // Operation failed. if ((impl.flags_ & implementation_type::user_set_non_blocking) || (ec != asio::error::would_block && ec != asio::error::try_again)) return 0; // Wait for socket to become ready. if (socket_ops::poll_write(impl.socket_, ec) < 0) return 0; } } template <typename ConstBufferSequence, typename Handler> class send_to_handler { public: send_to_handler(socket_type socket, asio::io_service& io_service, const ConstBufferSequence& buffers, const endpoint_type& endpoint, socket_base::message_flags flags, Handler handler) : socket_(socket), io_service_(io_service), work_(io_service), buffers_(buffers), destination_(endpoint), flags_(flags), handler_(handler) { } bool operator()(const asio::error_code& result) { // Check whether the operation was successful. if (result) { io_service_.post(bind_handler(handler_, result, 0)); return true; } // Copy buffers into array. socket_ops::buf bufs[max_buffers]; typename ConstBufferSequence::const_iterator iter = buffers_.begin(); typename ConstBufferSequence::const_iterator end = buffers_.end(); size_t i = 0; for (; iter != end && i < max_buffers; ++iter, ++i) { asio::const_buffer buffer(*iter); socket_ops::init_buf(bufs[i], asio::buffer_cast<const void*>(buffer), asio::buffer_size(buffer)); } // Send the data. asio::error_code ec; int bytes = socket_ops::sendto(socket_, bufs, i, flags_, destination_.data(), destination_.size(), ec); // Check if we need to run the operation again. if (ec == asio::error::would_block || ec == asio::error::try_again) return false; io_service_.post(bind_handler(handler_, ec, bytes < 0 ? 0 : bytes)); return true; } private: socket_type socket_;
⌨️ 快捷键说明
复制代码
Ctrl + C
搜索代码
Ctrl + F
全屏模式
F11
切换主题
Ctrl + Shift + D
显示快捷键
?
增大字号
Ctrl + =
减小字号
Ctrl + -