Increase Communicator buffer size to 64k
[libs/net.git] / 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(65536),
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