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