1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
|
#include "MessageTcp.h"
#include <assert.h>
#include "itextstream.h"
#include "clsocket/ActiveSocket.h"
namespace gameconn
{
MessageTcp::~MessageTcp() {}
MessageTcp::MessageTcp() {}
void MessageTcp::init(std::unique_ptr<CActiveSocket> &&connection) {
tcp = std::move(connection);
inputBuffer.clear();
outputBuffer.clear();
inputPos = 0;
outputPos = 0;
}
bool MessageTcp::isAlive() const {
return tcp.get() && tcp->IsSocketValid(); //TODO: IsAlive?
}
bool MessageTcp::readMessage(std::vector<char> &message) {
message.clear();
think();
const char *buffer = inputBuffer.data() + inputPos;
auto remains = inputBuffer.size() - inputPos;
auto pull = [&](void* ptr, std::size_t size) -> void {
assert(size <= remains);
memcpy(ptr, buffer, size);
buffer += size;
remains -= size;
};
if (remains < 12)
return false;
int len = -1;
char magic[5] = {0};
//note: little-endianness is assumed
pull(magic, 4);
if (strcmp(magic, "TDM[") != 0)
goto zomg;
pull(&len, 4);
if (len < 0)
goto zomg;
pull(magic, 4);
if (strcmp(magic, "] ") != 0)
goto zomg;
if (remains < static_cast<std::size_t>(len) + 12)
return false;
message.reserve(len + 1);
message.resize(len);
pull(message.data(), len);
message.data()[len] = '\0';
pull(magic, 4);
if (strcmp(magic, " (") != 0)
goto zomg;
int len2;
pull(&len2, 4);
if (len2 != len)
goto zomg;
pull(magic, 4);
if (strcmp(magic, ")TDM") != 0)
goto zomg;
inputPos = buffer - inputBuffer.data();
return true;
zomg:
rError() << "ERROR: MessageTCP: wrong message format\n";
message.clear();
init({});
return false;
}
void MessageTcp::writeMessage(const char *message, int len) {
auto where = outputBuffer.size();
outputBuffer.resize(where + len + 24);
auto push = [&](const void *ptr, std::size_t size) -> void {
memcpy(&outputBuffer[where], ptr, size);
where += size;
};
//note: little-endianness is assumed
push("TDM[", 4);
push(&len, 4);
push("] ", 4);
push(message, len);
push(" (", 4);
push(&len, 4);
push(")TDM", 4);
assert(where == outputBuffer.size());
think();
}
void MessageTcp::think() {
if (!tcp)
return;
static const int BUFFER_SIZE = 1024;
//if data in buffer is too far from start, then it is moved to the beginning
auto compactBuffer = [](std::vector<char>& vec, std::size_t& pos) -> void {
auto remains = vec.size() - pos;
if (pos > remains + BUFFER_SIZE) {
memcpy(vec.data(), vec.data() + pos, remains);
vec.resize(remains);
pos = 0;
}
};
compactBuffer(inputBuffer, inputPos);
//fetch incoming data from socket
static_assert(BUFFER_SIZE >= 256); //ZeroMQ identity fits
char buffer[BUFFER_SIZE];
for (int iter = 0; ; iter++) {
int read = tcp->Receive(BUFFER_SIZE, (uint8*)buffer);
if (read == -1 && tcp->GetSocketError() == CSimpleSocket::SocketEwouldblock)
break; //no more data
if (read == -1)
goto onerror; //socket error
if (read == 0)
goto onerror; //connection closed on other side
inputBuffer.resize(inputBuffer.size() + read);
memcpy(inputBuffer.data() + inputBuffer.size() - read, buffer, read);
}
//push outcoming data to socket
while (outputPos < outputBuffer.size()) {
auto remains = outputBuffer.size() - outputPos;
int written = tcp->Send((uint8*)&outputBuffer[outputPos], remains);
if (written == -1 && tcp->GetSocketError() == CSimpleSocket::SocketEwouldblock)
break; //no more data
if (written == -1)
goto onerror; //socket error
if (written == 0)
goto onerror; //connection closed on other side
outputPos += written;
}
compactBuffer(outputBuffer, outputPos);
return;
onerror:
rError() << "Automation lost connection\n";
tcp.reset();
}
}
|
