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