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 }
  89
  90
  91 /** Blocking write.
  92  *  @param data Buffer to write.
  93  *  @param size Number of bytes to write.
  94  */
  95 void
  96 Socket::write (uint8_t const * data, int size)
  97 {
  98         _deadline.expires_from_now (boost::posix_time::seconds (_timeout));
  99         boost::system::error_code ec = boost::asio::error::would_block;
 100
 101         boost::asio::async_write (_socket, boost::asio::buffer (data, size), boost::lambda::var(ec) = boost::lambda::_1);
 102
 103         do {
 104                 _io_service.run_one ();
 105         } while (ec == boost::asio::error::would_block);
 106
 107         if (ec) {
 108                 throw NetworkError (String::compose (_("error during async_write (%1)"), ec.value ()));
 109         }
 110
 111         if (_write_digester) {
 112                 _write_digester->add (data, static_cast<size_t>(size));
 113         }
 114 }
 115
 116
 117 void
 118 Socket::write (uint32_t v)
 119 {
 120         v = htonl (v);
 121         write (reinterpret_cast<uint8_t*> (&v), 4);
 122 }
 123
 124
 125 /** Blocking read.
 126  *  @param data Buffer to read to.
 127  *  @param size Number of bytes to read.
 128  */
 129 void
 130 Socket::read (uint8_t* data, int size)
 131 {
 132         _deadline.expires_from_now (boost::posix_time::seconds (_timeout));
 133         boost::system::error_code ec = boost::asio::error::would_block;
 134
 135         boost::asio::async_read (_socket, boost::asio::buffer (data, size), boost::lambda::var(ec) = boost::lambda::_1);
 136
 137         do {
 138                 _io_service.run_one ();
 139         } while (ec == boost::asio::error::would_block);
 140
 141         if (ec) {
 142                 throw NetworkError (String::compose (_("error during async_read (%1)"), ec.value ()));
 143         }
 144
 145         if (_read_digester) {
 146                 _read_digester->add (data, static_cast<size_t>(size));
 147         }
 148 }
 149
 150
 151 uint32_t
 152 Socket::read_uint32 ()
 153 {
 154         uint32_t v;
 155         read (reinterpret_cast<uint8_t *> (&v), 4);
 156         return ntohl (v);
 157 }
 158
 159
 160 void
 161 Socket::start_read_digest ()
 162 {
 163         DCPOMATIC_ASSERT (!_read_digester);
 164         _read_digester.reset (new Digester());
 165 }
 166
 167
 168 void
 169 Socket::start_write_digest ()
 170 {
 171         DCPOMATIC_ASSERT (!_write_digester);
 172         _write_digester.reset (new Digester());
 173 }
 174
 175
 176 Socket::ReadDigestScope::ReadDigestScope (shared_ptr<Socket> socket)
 177         : _socket (socket)
 178 {
 179         socket->start_read_digest ();
 180 }
 181
 182
 183 bool
 184 Socket::ReadDigestScope::check ()
 185 {
 186         auto sp = _socket.lock ();
 187         if (!sp) {
 188                 return false;
 189         }
 190
 191         return sp->check_read_digest ();
 192 }
 193
 194
 195 Socket::WriteDigestScope::WriteDigestScope (shared_ptr<Socket> socket)
 196         : _socket (socket)
 197 {
 198         socket->start_write_digest ();
 199 }
 200
 201
 202 Socket::WriteDigestScope::~WriteDigestScope ()
 203 {
 204         auto sp = _socket.lock ();
 205         if (sp) {
 206                 try {
 207                         sp->finish_write_digest ();
 208                 } catch (...) {
 209                         /* If we can't write our digest, something bad has happened
 210                          * so let's just let it happen.
 211                          */
 212                 }
 213         }
 214 }
 215
 216
 217 bool
 218 Socket::check_read_digest ()
 219 {
 220         DCPOMATIC_ASSERT (_read_digester);
 221         int const size = _read_digester->size ();
 222
 223         uint8_t ref[size];
 224         _read_digester->get (ref);
 225
 226         /* Make sure _read_digester is gone before we call read() so that the digest
 227          * isn't itself digested.
 228          */
 229         _read_digester.reset ();
 230
 231         uint8_t actual[size];
 232         read (actual, size);
 233
 234         return memcmp(ref, actual, size) == 0;
 235 }
 236
 237
 238 void
 239 Socket::finish_write_digest ()
 240 {
 241         DCPOMATIC_ASSERT (_write_digester);
 242         int const size = _write_digester->size();
 243
 244         uint8_t buffer[size];
 245         _write_digester->get (buffer);
 246
 247         /* Make sure _write_digester is gone before we call write() so that the digest
 248          * isn't itself digested.
 249          */
 250         _write_digester.reset ();
 251
 252         write (buffer, size);
 253 }
 254
 255
 256 void
 257 Socket::set_send_buffer_size (int size)
 258 {
 259         _send_buffer_size = size;
 260 }
 261