source/net/communicator.cpp

   1 #include <cstring>
   2 #include "communicator.h"
   3 #include "streamsocket.h"
   4
   5 using namespace std;
   6
   7 namespace {
   8
   9 using namespace Msp::Net;
  10
  11 struct Handshake
  12 {
  13         Msp::UInt64 hash;
  14 };
  15
  16
  17 class HandshakeProtocol: public Protocol
  18 {
  19 public:
  20         HandshakeProtocol();
  21 };
  22
  23 HandshakeProtocol::HandshakeProtocol():
  24         Protocol(0x7F00)
  25 {
  26         add<Handshake>()(&Handshake::hash);
  27 }
  28
  29
  30 class HandshakeReceiver: public PacketReceiver<Handshake>
  31 {
  32 private:
  33         Msp::UInt64 hash;
  34
  35 public:
  36         HandshakeReceiver();
  37         Msp::UInt64 get_hash() const { return hash; }
  38         virtual void receive(const Handshake &);
  39 };
  40
  41 HandshakeReceiver::HandshakeReceiver():
  42         hash(0)
  43 { }
  44
  45 void HandshakeReceiver::receive(const Handshake &shake)
  46 {
  47         hash = shake.hash;
  48 }
  49
  50 }
  51
  52
  53 namespace Msp {
  54 namespace Net {
  55
  56 Communicator::Communicator(StreamSocket &s, const Protocol &p, ReceiverBase &r):
  57         socket(s),
  58         protocol(p),
  59         receiver(r),
  60         handshake_status(0),
  61         buf_size(1024),
  62         in_buf(new char[buf_size]),
  63         in_begin(in_buf),
  64         in_end(in_buf),
  65         out_buf(new char[buf_size]),
  66         good(true)
  67 {
  68         socket.signal_data_available.connect(sigc::mem_fun(this, &Communicator::data_available));
  69 }
  70
  71 Communicator::~Communicator()
  72 {
  73         delete[] in_buf;
  74         delete[] out_buf;
  75 }
  76
  77 void Communicator::initiate_handshake()
  78 {
  79         if(handshake_status!=0)
  80                 throw sequence_error("handshaking already done");
  81
  82         send_handshake();
  83         handshake_status = 1;
  84 }
  85
  86 void Communicator::send_data(unsigned size)
  87 {
  88         if(!good)
  89                 throw sequence_error("connection aborted");
  90         if(handshake_status!=2)
  91                 throw sequence_error("handshake incomplete");
  92
  93         try
  94         {
  95                 socket.write(out_buf, size);
  96         }
  97         catch(const std::exception &e)
  98         {
  99                 good = false;
 100                 if(signal_error.empty())
 101                         throw;
 102                 signal_error.emit(e);
 103         }
 104 }
 105
 106 void Communicator::data_available()
 107 {
 108         if(!good)
 109                 return;
 110
 111         try
 112         {
 113                 in_end += socket.read(in_end, in_buf+buf_size-in_end);
 114
 115                 bool more = true;
 116                 while(more)
 117                 {
 118                         if(handshake_status==2)
 119                         {
 120                                 more = receive_packet(protocol, receiver);
 121                         }
 122                         else
 123                         {
 124                                 HandshakeProtocol hsproto;
 125                                 HandshakeReceiver hsrecv;
 126                                 if((more = receive_packet(hsproto, hsrecv)))
 127                                 {
 128                                         if(handshake_status==0)
 129                                                 send_handshake();
 130
 131                                         if(hsrecv.get_hash()==protocol.get_hash())
 132                                         {
 133                                                 handshake_status = 2;
 134                                                 signal_handshake_done.emit();
 135                                         }
 136                                         else
 137                                                 throw incompatible_protocol("hash mismatch");
 138                                 }
 139                         }
 140                 }
 141         }
 142         catch(const exception &e)
 143         {
 144                 good = false;
 145                 if(signal_error.empty())
 146                         throw;
 147                 signal_error.emit(e);
 148         }
 149 }
 150
 151 bool Communicator::receive_packet(const Protocol &proto, ReceiverBase &recv)
 152 {
 153         int psz = proto.get_packet_size(in_begin, in_end-in_begin);
 154         if(psz && psz<=in_end-in_begin)
 155         {
 156                 char *pkt = in_begin;
 157                 in_begin += psz;
 158                 proto.dispatch(recv, pkt, psz);
 159                 return true;
 160         }
 161         else
 162         {
 163                 if(in_end==in_buf+buf_size)
 164                 {
 165                         unsigned used = in_end-in_begin;
 166                         memmove(in_buf, in_begin, used);
 167                         in_begin = in_buf;
 168                         in_end = in_begin+used;
 169                 }
 170                 return false;
 171         }
 172 }
 173
 174 void Communicator::send_handshake()
 175 {
 176         Handshake shake;
 177         shake.hash = protocol.get_hash();
 178
 179         HandshakeProtocol hsproto;
 180         unsigned size = hsproto.serialize(shake, out_buf, buf_size);
 181         socket.write(out_buf, size);
 182 }
 183
 184 } // namespace Net
 185 } // namespace Msp