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 #include "dcpomatic_socket.h"
  22 #include "compose.hpp"
  23 #include "exceptions.h"
  24 #include "dcpomatic_assert.h"
  25 #include <boost/bind.hpp>
  26 #include <boost/lambda/lambda.hpp>
  27 #include <iostream>
  28
  29 #include "i18n.h"
  30
  31 using boost::shared_ptr;
  32 using boost::weak_ptr;
  33
  34 /** @param timeout Timeout in seconds */
  35 Socket::Socket (int timeout)
  36         : _deadline (_io_service)
  37         , _socket (_io_service)
  38         , _timeout (timeout)
  39 {
  40         _deadline.expires_at (boost::posix_time::pos_infin);
  41         check ();
  42 }
  43
  44 void
  45 Socket::check ()
  46 {
  47         if (_deadline.expires_at() <= boost::asio::deadline_timer::traits_type::now ()) {
  48                 _socket.close ();
  49                 _deadline.expires_at (boost::posix_time::pos_infin);
  50         }
  51
  52         _deadline.async_wait (boost::bind (&Socket::check, this));
  53 }
  54
  55 /** Blocking connect.
  56  *  @param endpoint End-point to connect to.
  57  */
  58 void
  59 Socket::connect (boost::asio::ip::tcp::endpoint endpoint)
  60 {
  61         _deadline.expires_from_now (boost::posix_time::seconds (_timeout));
  62         boost::system::error_code ec = boost::asio::error::would_block;
  63         _socket.async_connect (endpoint, boost::lambda::var(ec) = boost::lambda::_1);
  64         do {
  65                 _io_service.run_one();
  66         } while (ec == boost::asio::error::would_block);
  67
  68         if (ec) {
  69                 throw NetworkError (String::compose (_("error during async_connect (%1)"), ec.value ()));
  70         }
  71
  72         if (!_socket.is_open ()) {
  73                 throw NetworkError (_("connect timed out"));
  74         }
  75 }
  76
  77 /** Blocking write.
  78  *  @param data Buffer to write.
  79  *  @param size Number of bytes to write.
  80  */
  81 void
  82 Socket::write (uint8_t const * data, int size)
  83 {
  84         _deadline.expires_from_now (boost::posix_time::seconds (_timeout));
  85         boost::system::error_code ec = boost::asio::error::would_block;
  86
  87         boost::asio::async_write (_socket, boost::asio::buffer (data, size), boost::lambda::var(ec) = boost::lambda::_1);
  88
  89         do {
  90                 _io_service.run_one ();
  91         } while (ec == boost::asio::error::would_block);
  92
  93         if (ec) {
  94                 throw NetworkError (String::compose (_("error during async_write (%1)"), ec.value ()));
  95         }
  96
  97         if (_write_digester) {
  98                 _write_digester->add (data, static_cast<size_t>(size));
  99         }
 100 }
 101
 102 void
 103 Socket::write (uint32_t v)
 104 {
 105         v = htonl (v);
 106         write (reinterpret_cast<uint8_t*> (&v), 4);
 107 }
 108
 109 /** Blocking read.
 110  *  @param data Buffer to read to.
 111  *  @param size Number of bytes to read.
 112  */
 113 void
 114 Socket::read (uint8_t* data, int size)
 115 {
 116         _deadline.expires_from_now (boost::posix_time::seconds (_timeout));
 117         boost::system::error_code ec = boost::asio::error::would_block;
 118
 119         boost::asio::async_read (_socket, boost::asio::buffer (data, size), boost::lambda::var(ec) = boost::lambda::_1);
 120
 121         do {
 122                 _io_service.run_one ();
 123         } while (ec == boost::asio::error::would_block);
 124
 125         if (ec) {
 126                 throw NetworkError (String::compose (_("error during async_read (%1)"), ec.value ()));
 127         }
 128
 129         if (_read_digester) {
 130                 _read_digester->add (data, static_cast<size_t>(size));
 131         }
 132 }
 133
 134 uint32_t
 135 Socket::read_uint32 ()
 136 {
 137         uint32_t v;
 138         read (reinterpret_cast<uint8_t *> (&v), 4);
 139         return ntohl (v);
 140 }
 141
 142
 143 void
 144 Socket::start_read_digest ()
 145 {
 146         DCPOMATIC_ASSERT (!_read_digester);
 147         _read_digester.reset (new Digester());
 148 }
 149
 150 void
 151 Socket::start_write_digest ()
 152 {
 153         DCPOMATIC_ASSERT (!_write_digester);
 154         _write_digester.reset (new Digester());
 155 }
 156
 157
 158 Socket::ReadDigestScope::ReadDigestScope (shared_ptr<Socket> socket)
 159         : _socket (socket)
 160 {
 161         socket->start_read_digest ();
 162 }
 163
 164
 165 bool
 166 Socket::ReadDigestScope::check ()
 167 {
 168         shared_ptr<Socket> sp = _socket.lock ();
 169         if (!sp) {
 170                 return false;
 171         }
 172
 173         return sp->check_read_digest ();
 174 }
 175
 176
 177 Socket::WriteDigestScope::WriteDigestScope (shared_ptr<Socket> socket)
 178         : _socket (socket)
 179 {
 180         socket->start_write_digest ();
 181 }
 182
 183
 184 Socket::WriteDigestScope::~WriteDigestScope ()
 185 {
 186         shared_ptr<Socket> sp = _socket.lock ();
 187         if (sp) {
 188                 try {
 189                         sp->finish_write_digest ();
 190                 } catch (...) {
 191                         /* If we can't write our digest, something bad has happened
 192                          * so let's just let it happen.
 193                          */
 194                 }
 195         }
 196 }
 197
 198
 199 bool
 200 Socket::check_read_digest ()
 201 {
 202         DCPOMATIC_ASSERT (_read_digester);
 203         int const size = _read_digester->size ();
 204
 205         uint8_t ref[size];
 206         _read_digester->get (ref);
 207
 208         /* Make sure _read_digester is gone before we call read() so that the digest
 209          * isn't itself digested.
 210          */
 211         _read_digester.reset ();
 212
 213         uint8_t actual[size];
 214         read (actual, size);
 215
 216         return memcmp(ref, actual, size) == 0;
 217 }
 218
 219 void
 220 Socket::finish_write_digest ()
 221 {
 222         DCPOMATIC_ASSERT (_write_digester);
 223         int const size = _write_digester->size();
 224
 225         uint8_t buffer[size];
 226         _write_digester->get (buffer);
 227
 228         /* Make sure _write_digester is gone before we call write() so that the digest
 229          * isn't itself digested.
 230          */
 231         _write_digester.reset ();
 232
 233         write (buffer, size);
 234 }
 235