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