/usr/include/pion/tcp/stream.hpp is in libpion-dev 5.0.7+dfsg-4.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 | // ---------------------------------------------------------------------
// pion: a Boost C++ framework for building lightweight HTTP interfaces
// ---------------------------------------------------------------------
// Copyright (C) 2007-2014 Splunk Inc. (https://github.com/splunk/pion)
//
// Distributed under the Boost Software License, Version 1.0.
// See http://www.boost.org/LICENSE_1_0.txt
//
#ifndef __PION_TCP_STREAM_HEADER__
#define __PION_TCP_STREAM_HEADER__
#include <cstring>
#include <istream>
#include <streambuf>
#include <boost/bind.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>
#include <pion/config.hpp>
#include <pion/tcp/connection.hpp>
namespace pion { // begin namespace pion
namespace tcp { // begin namespace tcp
///
/// stream_buffer: std::basic_streambuf wrapper for TCP network connections.
/// Based in part on section 13.13.3 of "The Standard C++ Library"
/// by Nicolai M. Josuttis, published in 1999 by Addison-Wesley
///
class stream_buffer
: public std::basic_streambuf<char, std::char_traits<char> >
{
public:
// data type definitions required for iostream compatability
typedef char char_type;
typedef std::char_traits<char>::int_type int_type;
typedef std::char_traits<char>::off_type off_type;
typedef std::char_traits<char>::pos_type pos_type;
typedef std::char_traits<char> traits_type;
// some integer constants used within stream_buffer
enum {
PUT_BACK_MAX = 10, //< number of bytes that can be put back into the read buffer
WRITE_BUFFER_SIZE = 8192 //< size of the write buffer
};
/**
* constructs a TCP stream buffer object for an existing TCP connection
*
* @param conn_ptr pointer to the TCP connection to use for reading & writing
*/
explicit stream_buffer(const tcp::connection_ptr& conn_ptr)
: m_conn_ptr(conn_ptr), m_read_buf(m_conn_ptr->get_read_buffer().c_array())
{
setup_buffers();
}
/**
* constructs a TCP stream buffer object for a new TCP connection
*
* @param io_service asio service associated with the connection
* @param ssl_flag if true then the connection will be encrypted using SSL
*/
explicit stream_buffer(boost::asio::io_service& io_service,
const bool ssl_flag = false)
: m_conn_ptr(new connection(io_service, ssl_flag)),
m_read_buf(m_conn_ptr->get_read_buffer().c_array())
{
setup_buffers();
}
/**
* constructs a TCP stream buffer object for a new SSL/TCP connection
*
* @param io_service asio service associated with the connection
* @param ssl_context asio ssl context associated with the connection
*/
stream_buffer(boost::asio::io_service& io_service,
connection::ssl_context_type& ssl_context)
: m_conn_ptr(new connection(io_service, ssl_context)),
m_read_buf(m_conn_ptr->get_read_buffer().c_array())
{
setup_buffers();
}
/// virtual destructor flushes the write buffer
virtual ~stream_buffer() { sync(); }
/// returns a reference to the current TCP connection
connection& get_connection(void) { return *m_conn_ptr; }
/// returns a const reference to the current TCP connection
const connection& get_connection(void) const { return *m_conn_ptr; }
protected:
/// sets up the read and write buffers for input and output
inline void setup_buffers(void) {
// use the TCP connection's read buffer and allow for bytes to be put back
setg(m_read_buf+PUT_BACK_MAX, m_read_buf+PUT_BACK_MAX, m_read_buf+PUT_BACK_MAX);
// set write buffer size-1 so that we have an extra char avail for overflow
setp(m_write_buf, m_write_buf+(WRITE_BUFFER_SIZE-1));
}
/**
* writes data in the output buffer to the TCP connection
*
* @return int_type the number of bytes sent, or eof() if there was an error
*/
inline int_type flush_output(void) {
const std::streamsize bytes_to_send = std::streamsize(pptr() - pbase());
int_type bytes_sent = 0;
if (bytes_to_send > 0) {
boost::mutex::scoped_lock async_lock(m_async_mutex);
m_bytes_transferred = 0;
m_conn_ptr->async_write(boost::asio::buffer(pbase(), bytes_to_send),
boost::bind(&stream_buffer::operation_finished, this,
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
m_async_done.wait(async_lock);
bytes_sent = m_bytes_transferred;
pbump(-bytes_sent);
if (m_async_error)
bytes_sent = traits_type::eof();
}
return bytes_sent;
}
/**
* this function is called when the read buffer has no more characters available
*
* @return int_type the next character available for reading, or eof() if there was an error
*/
virtual int_type underflow(void) {
// first check if we still have bytes available in the read buffer
if (gptr() < egptr())
return traits_type::to_int_type(*gptr());
// calculate the number of bytes we will allow to be put back
std::streamsize put_back_num = std::streamsize(gptr() - eback());
if (put_back_num > PUT_BACK_MAX)
put_back_num = PUT_BACK_MAX;
// copy the last bytes read to the beginning of the buffer (for put back)
if (put_back_num > 0)
memmove(m_read_buf+(PUT_BACK_MAX-put_back_num), gptr()-put_back_num, put_back_num);
// read data from the TCP connection
// note that this has to be an ansynchronous call; otherwise, it cannot
// be cancelled by other threads and will block forever (such as during shutdown)
boost::mutex::scoped_lock async_lock(m_async_mutex);
m_bytes_transferred = 0;
m_conn_ptr->async_read_some(boost::asio::buffer(m_read_buf+PUT_BACK_MAX,
connection::READ_BUFFER_SIZE-PUT_BACK_MAX),
boost::bind(&stream_buffer::operation_finished, this,
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
m_async_done.wait(async_lock);
if (m_async_error)
return traits_type::eof();
// reset buffer pointers now that data is available
setg(m_read_buf+(PUT_BACK_MAX-put_back_num), //< beginning of putback bytes
m_read_buf+PUT_BACK_MAX, //< read position
m_read_buf+PUT_BACK_MAX+m_bytes_transferred); //< end of buffer
// return next character available
return traits_type::to_int_type(*gptr());
}
/**
* this function is called when the write buffer for the stream is full
*
* @param c character that has not been written yet, or eof() if we are flushing
* @return int_type the last character written, or eof() if there was an error
*/
virtual int_type overflow(int_type c) {
if (! traits_type::eq_int_type(c, traits_type::eof())) {
// character is not eof -> add it to the end of the write buffer
// we can push this to the back of the write buffer because we set
// the size of the write buffer to 1 less than the actual size using setp()
*pptr() = c;
pbump(1);
}
// flush data in the write buffer by sending it to the TCP connection
return ((flush_output() == traits_type::eof())
? traits_type::eof() : traits_type::not_eof(c));
}
/**
* writes a sequence of characters
*
* @param s pointer to a sequence of characters
* @param n number of characters in the sequence to write
*
* @return std::streamsize number of character written
*/
virtual std::streamsize xsputn(const char_type *s, std::streamsize n) {
const std::streamsize bytes_available = std::streamsize(epptr() - pptr());
std::streamsize bytes_sent = 0;
if (bytes_available >= n) {
// there is enough room in the buffer -> just put it in there
memcpy(pptr(), s, n);
pbump(n);
bytes_sent = n;
} else {
// there is not enough room left in the buffer
if (bytes_available > 0) {
// fill up the buffer
memcpy(pptr(), s, bytes_available);
pbump(bytes_available);
}
// flush data in the write buffer by sending it to the TCP connection
if (flush_output() == traits_type::eof())
return 0;
if ((n-bytes_available) >= (WRITE_BUFFER_SIZE-1)) {
// the remaining data to send is larger than the buffer available
// send it all now rather than buffering
boost::mutex::scoped_lock async_lock(m_async_mutex);
m_bytes_transferred = 0;
m_conn_ptr->async_write(boost::asio::buffer(s+bytes_available,
n-bytes_available),
boost::bind(&stream_buffer::operation_finished, this,
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
m_async_done.wait(async_lock);
bytes_sent = bytes_available + m_bytes_transferred;
} else {
// the buffer is larger than the remaining data
// put remaining data to the beginning of the output buffer
memcpy(pbase(), s+bytes_available, n-bytes_available);
pbump(n-bytes_available);
bytes_sent = n;
}
}
return bytes_sent;
}
/**
* reads a sequence of characters
*
* @param s pointer to where the sequence of characters will be stored
* @param n number of characters in the sequence to read
*
* @return std::streamsize number of character read
*/
virtual std::streamsize xsgetn(char_type *s, std::streamsize n) {
std::streamsize bytes_remaining = n;
while (bytes_remaining > 0) {
const std::streamsize bytes_available = std::streamsize(egptr() - gptr());
const std::streamsize bytes_next_read = ((bytes_available >= bytes_remaining)
? bytes_remaining : bytes_available);
// copy available input data from buffer
if (bytes_next_read > 0) {
memcpy(s, gptr(), bytes_next_read);
gbump(bytes_next_read);
bytes_remaining -= bytes_next_read;
s += bytes_next_read;
}
if (bytes_remaining > 0) {
// call underflow() to read more data
if (traits_type::eq_int_type(underflow(), traits_type::eof()))
break;
}
}
return(n-bytes_remaining);
}
/**
* synchronize buffers with the TCP connection
*
* @return 0 if successful, -1 if there was an error
*/
virtual int_type sync(void) {
return ((flush_output() == traits_type::eof()) ? -1 : 0);
}
private:
/// function called after an asynchronous operation has completed
inline void operation_finished(const boost::system::error_code& error_code,
std::size_t bytes_transferred)
{
boost::mutex::scoped_lock async_lock(m_async_mutex);
m_async_error = error_code;
m_bytes_transferred = bytes_transferred;
m_async_done.notify_one();
}
/// pointer to the underlying TCP connection used for reading & writing
tcp::connection_ptr m_conn_ptr;
/// condition signaled whenever an asynchronous operation has completed
boost::mutex m_async_mutex;
/// condition signaled whenever an asynchronous operation has completed
boost::condition m_async_done;
/// used to keep track of the result from the last asynchronous operation
boost::system::error_code m_async_error;
/// the number of bytes transferred by the last asynchronous operation
std::size_t m_bytes_transferred;
/// pointer to the start of the TCP connection's read buffer
char_type * m_read_buf;
/// buffer used to write output
char_type m_write_buf[WRITE_BUFFER_SIZE];
};
///
/// stream: std::basic_iostream wrapper for TCP network connections
///
class stream
: public std::basic_iostream<char, std::char_traits<char> >
{
public:
// data type definitions required for iostream compatability
typedef char char_type;
typedef std::char_traits<char>::int_type int_type;
typedef std::char_traits<char>::off_type off_type;
typedef std::char_traits<char>::pos_type pos_type;
typedef std::char_traits<char> traits_type;
/**
* constructs a TCP stream object for an existing TCP connection
*
* @param conn_ptr pointer to the TCP connection to use for reading & writing
*/
explicit stream(const tcp::connection_ptr& conn_ptr)
: std::basic_iostream<char, std::char_traits<char> >(NULL), m_tcp_buf(conn_ptr)
{
// initialize basic_iostream with pointer to the stream buffer
std::basic_ios<char,std::char_traits<char> >::init(&m_tcp_buf);
}
/**
* constructs a TCP stream object for a new TCP connection
*
* @param io_service asio service associated with the connection
* @param ssl_flag if true then the connection will be encrypted using SSL
*/
explicit stream(boost::asio::io_service& io_service,
const bool ssl_flag = false)
: std::basic_iostream<char, std::char_traits<char> >(NULL), m_tcp_buf(io_service, ssl_flag)
{
// initialize basic_iostream with pointer to the stream buffer
std::basic_ios<char,std::char_traits<char> >::init(&m_tcp_buf);
}
/**
* constructs a TCP stream object for a new SSL/TCP connection
*
* @param io_service asio service associated with the connection
* @param ssl_context asio ssl context associated with the connection
*/
stream(boost::asio::io_service& io_service,
connection::ssl_context_type& ssl_context)
: std::basic_iostream<char, std::char_traits<char> >(NULL), m_tcp_buf(io_service, ssl_context)
{
// initialize basic_iostream with pointer to the stream buffer
std::basic_ios<char,std::char_traits<char> >::init(&m_tcp_buf);
}
/**
* accepts a new tcp connection and performs SSL handshake if necessary
*
* @param tcp_acceptor object used to accept new connections
* @return boost::system::error_code contains error code if the connection fails
*
* @see boost::asio::basic_socket_acceptor::accept()
*/
inline boost::system::error_code accept(boost::asio::ip::tcp::acceptor& tcp_acceptor)
{
boost::system::error_code ec = m_tcp_buf.get_connection().accept(tcp_acceptor);
if (! ec && get_ssl_flag()) ec = m_tcp_buf.get_connection().handshake_server();
return ec;
}
/**
* connects to a remote endpoint and performs SSL handshake if necessary
*
* @param tcp_endpoint remote endpoint to connect to
* @return boost::system::error_code contains error code if the connection fails
*
* @see boost::asio::basic_socket_acceptor::connect()
*/
inline boost::system::error_code connect(boost::asio::ip::tcp::endpoint& tcp_endpoint)
{
boost::system::error_code ec = m_tcp_buf.get_connection().connect(tcp_endpoint);
if (! ec && get_ssl_flag()) ec = m_tcp_buf.get_connection().handshake_client();
return ec;
}
/**
* connects to a (IPv4) remote endpoint and performs SSL handshake if necessary
*
* @param remote_addr remote IP address (v4) to connect to
* @param remote_port remote port number to connect to
* @return boost::system::error_code contains error code if the connection fails
*
* @see boost::asio::basic_socket_acceptor::connect()
*/
inline boost::system::error_code connect(const boost::asio::ip::address& remote_addr,
const unsigned int remote_port)
{
boost::asio::ip::tcp::endpoint tcp_endpoint(remote_addr, remote_port);
boost::system::error_code ec = m_tcp_buf.get_connection().connect(tcp_endpoint);
if (! ec && get_ssl_flag()) ec = m_tcp_buf.get_connection().handshake_client();
return ec;
}
/// closes the tcp connection
inline void close(void) { m_tcp_buf.get_connection().close(); }
/*
Use close instead; basic_socket::cancel is deprecated for Windows XP.
/// cancels any asynchronous operations pending on the tcp connection
inline void cancel(void) { m_tcp_buf.get_connection().cancel(); }
*/
/// returns true if the connection is currently open
inline bool is_open(void) const { return m_tcp_buf.get_connection().is_open(); }
/// returns true if the connection is encrypted using SSL
inline bool get_ssl_flag(void) const { return m_tcp_buf.get_connection().get_ssl_flag(); }
/// returns the client's IP address
inline boost::asio::ip::address get_remote_ip(void) const {
return m_tcp_buf.get_connection().get_remote_ip();
}
/// returns a pointer to the stream buffer in use
stream_buffer *rdbuf(void) { return &m_tcp_buf; }
private:
/// the underlying TCP stream buffer used for reading & writing
stream_buffer m_tcp_buf;
};
} // end namespace tcp
} // end namespace pion
#endif
|