src/lib/dcpomatic_socket.cc

   1 /*
   2     Copyright (C) 2012-2020 Carl Hetherington <cth@carlh.net>
   3
   4     This file is part of DCP-o-matic.
   5
   6     DCP-o-matic is free software; you can redistribute it and/or modify
   7     it under the terms of the GNU General Public License as published by
   8     the Free Software Foundation; either version 2 of the License, or
   9     (at your option) any later version.
  10
  11     DCP-o-matic is distributed in the hope that it will be useful,
  12     but WITHOUT ANY WARRANTY; without even the implied warranty of
  13     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14     GNU General Public License for more details.
  15
  16     You should have received a copy of the GNU General Public License
  17     along with DCP-o-matic.  If not, see <http://www.gnu.org/licenses/>.
  18
  19 */
  20
  21
  22 #include "compose.hpp"
  23 #include "dcpomatic_assert.h"
  24 #include "dcpomatic_log.h"
  25 #include "dcpomatic_socket.h"
  26 #include "exceptions.h"
  27 #include <boost/bind/bind.hpp>
  28 #include <boost/lambda/lambda.hpp>
  29 #include <iostream>
  30
  31 #include "i18n.h"
  32
  33
  34 using std::shared_ptr;
  35 using std::weak_ptr;
  36
  37
  38 /** @param timeout Timeout in seconds */
  39 Socket::Socket (int timeout)
  40         : _deadline (_io_service)
  41         , _socket (_io_service)
  42         , _timeout (timeout)
  43 {
  44         _deadline.expires_at (boost::posix_time::pos_infin);
  45         check ();
  46 }
  47
  48 void
  49 Socket::check ()
  50 {
  51         if (_deadline.expires_at() <= boost::asio::deadline_timer::traits_type::now ()) {
  52                 _socket.close ();
  53                 _deadline.expires_at (boost::posix_time::pos_infin);
  54         }
  55
  56         _deadline.async_wait (boost::bind (&Socket::check, this));
  57 }
  58
  59
  60 /** Blocking connect.
  61  *  @param endpoint End-point to connect to.
  62  */
  63 void
  64 Socket::connect (boost::asio::ip::tcp::endpoint endpoint)
  65 {
  66         _deadline.expires_from_now (boost::posix_time::seconds (_timeout));
  67         boost::system::error_code ec = boost::asio::error::would_block;
  68         _socket.async_connect (endpoint, boost::lambda::var(ec) = boost::lambda::_1);
  69         do {
  70                 _io_service.run_one();
  71         } while (ec == boost::asio::error::would_block);
  72
  73         if (ec) {
  74                 throw NetworkError (String::compose (_("error during async_connect (%1)"), ec.value ()));
  75         }
  76
  77         if (!_socket.is_open ()) {
  78                 throw NetworkError (_("connect timed out"));
  79         }
  80
  81         if (_send_buffer_size) {
  82                 boost::asio::socket_base::send_buffer_size old_size;
  83                 _socket.get_option(old_size);
  84
  85                 boost::asio::socket_base::send_buffer_size new_size(*_send_buffer_size);
  86                 _socket.set_option(new_size);
  87
  88                 LOG_GENERAL("Changed socket send buffer size from %1 to %2", old_size.value(), *_send_buffer_size);
  89         }
  90 }
  91
  92
  93 /** Blocking write.
  94  *  @param data Buffer to write.
  95  *  @param size Number of bytes to write.
  96  */
  97 void
  98 Socket::write (uint8_t const * data, int size)
  99 {
 100         _deadline.expires_from_now (boost::posix_time::seconds (_timeout));
 101         boost::system::error_code ec = boost::asio::error::would_block;
 102
 103         boost::asio::async_write (_socket, boost::asio::buffer (data, size), boost::lambda::var(ec) = boost::lambda::_1);
 104
 105         do {
 106                 _io_service.run_one ();
 107         } while (ec == boost::asio::error::would_block);
 108
 109         if (ec) {
 110                 throw NetworkError (String::compose (_("error during async_write (%1)"), ec.value ()));
 111         }
 112
 113         if (_write_digester) {
 114                 _write_digester->add (data, static_cast<size_t>(size));
 115         }
 116 }
 117
 118
 119 void
 120 Socket::write (uint32_t v)
 121 {
 122         v = htonl (v);
 123         write (reinterpret_cast<uint8_t*> (&v), 4);
 124 }
 125
 126
 127 /** Blocking read.
 128  *  @param data Buffer to read to.
 129  *  @param size Number of bytes to read.
 130  */
 131 void
 132 Socket::read (uint8_t* data, int size)
 133 {
 134         _deadline.expires_from_now (boost::posix_time::seconds (_timeout));
 135         boost::system::error_code ec = boost::asio::error::would_block;
 136
 137         boost::asio::async_read (_socket, boost::asio::buffer (data, size), boost::lambda::var(ec) = boost::lambda::_1);
 138
 139         do {
 140                 _io_service.run_one ();
 141         } while (ec == boost::asio::error::would_block);
 142
 143         if (ec) {
 144                 throw NetworkError (String::compose (_("error during async_read (%1)"), ec.value ()));
 145         }
 146
 147         if (_read_digester) {
 148                 _read_digester->add (data, static_cast<size_t>(size));
 149         }
 150 }
 151
 152
 153 uint32_t
 154 Socket::read_uint32 ()
 155 {
 156         uint32_t v;
 157         read (reinterpret_cast<uint8_t *> (&v), 4);
 158         return ntohl (v);
 159 }
 160
 161
 162 void
 163 Socket::start_read_digest ()
 164 {
 165         DCPOMATIC_ASSERT (!_read_digester);
 166         _read_digester.reset (new Digester());
 167 }
 168
 169
 170 void
 171 Socket::start_write_digest ()
 172 {
 173         DCPOMATIC_ASSERT (!_write_digester);
 174         _write_digester.reset (new Digester());
 175 }
 176
 177
 178 Socket::ReadDigestScope::ReadDigestScope (shared_ptr<Socket> socket)
 179         : _socket (socket)
 180 {
 181         socket->start_read_digest ();
 182 }
 183
 184
 185 bool
 186 Socket::ReadDigestScope::check ()
 187 {
 188         auto sp = _socket.lock ();
 189         if (!sp) {
 190                 return false;
 191         }
 192
 193         return sp->check_read_digest ();
 194 }
 195
 196
 197 Socket::WriteDigestScope::WriteDigestScope (shared_ptr<Socket> socket)
 198         : _socket (socket)
 199 {
 200         socket->start_write_digest ();
 201 }
 202
 203
 204 Socket::WriteDigestScope::~WriteDigestScope ()
 205 {
 206         auto sp = _socket.lock ();
 207         if (sp) {
 208                 try {
 209                         sp->finish_write_digest ();
 210                 } catch (...) {
 211                         /* If we can't write our digest, something bad has happened
 212                          * so let's just let it happen.
 213                          */
 214                 }
 215         }
 216 }
 217
 218
 219 bool
 220 Socket::check_read_digest ()
 221 {
 222         DCPOMATIC_ASSERT (_read_digester);
 223         int const size = _read_digester->size ();
 224
 225         uint8_t ref[size];
 226         _read_digester->get (ref);
 227
 228         /* Make sure _read_digester is gone before we call read() so that the digest
 229          * isn't itself digested.
 230          */
 231         _read_digester.reset ();
 232
 233         uint8_t actual[size];
 234         read (actual, size);
 235
 236         return memcmp(ref, actual, size) == 0;
 237 }
 238
 239
 240 void
 241 Socket::finish_write_digest ()
 242 {
 243         DCPOMATIC_ASSERT (_write_digester);
 244         int const size = _write_digester->size();
 245
 246         uint8_t buffer[size];
 247         _write_digester->get (buffer);
 248
 249         /* Make sure _write_digester is gone before we call write() so that the digest
 250          * isn't itself digested.
 251          */
 252         _write_digester.reset ();
 253
 254         write (buffer, size);
 255 }
 256
 257
 258 void
 259 Socket::set_send_buffer_size (int size)
 260 {
 261         _send_buffer_size = size;
 262 }
 263