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                                 more = receive_packet(protocol, receiver);
 120                         else
 121                         {
 122                                 HandshakeProtocol hsproto;
 123                                 HandshakeReceiver hsrecv;
 124                                 if((more = receive_packet(hsproto, hsrecv)))
 125                                 {
 126                                         if(handshake_status==0)
 127                                                 send_handshake();
 128
 129                                         if(hsrecv.get_hash()==protocol.get_hash())
 130                                         {
 131                                                 handshake_status = 2;
 132                                                 signal_handshake_done.emit();
 133                                         }
 134                                         else
 135                                                 throw incompatible_protocol("hash mismatch");
 136                                 }
 137                         }
 138                 }
 139         }
 140         catch(const exception &e)
 141         {
 142                 good = false;
 143                 if(signal_error.empty())
 144                         throw;
 145                 signal_error.emit(e);
 146         }
 147 }
 148
 149 bool Communicator::receive_packet(const Protocol &proto, ReceiverBase &recv)
 150 {
 151         int psz = proto.get_packet_size(in_begin, in_end-in_begin);
 152         if(psz && psz<=in_end-in_begin)
 153         {
 154                 char *pkt = in_begin;
 155                 in_begin += psz;
 156                 proto.dispatch(recv, pkt, psz);
 157                 return true;
 158         }
 159         else
 160         {
 161                 if(in_end==in_buf+buf_size)
 162                 {
 163                         unsigned used = in_end-in_begin;
 164                         memmove(in_buf, in_begin, used);
 165                         in_begin = in_buf;
 166                         in_end = in_begin+used;
 167                 }
 168                 return false;
 169         }
 170 }
 171
 172 void Communicator::send_handshake()
 173 {
 174         Handshake shake;
 175         shake.hash = protocol.get_hash();
 176
 177         HandshakeProtocol hsproto;
 178         unsigned size = hsproto.serialize(shake, out_buf, buf_size);
 179         socket.write(out_buf, size);
 180 }
 181
 182 } // namespace Net
 183 } // namespace Msp