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
|
// message.h
#pragma once
#include "../util/sock.h"
class Message;
class MessagingPort;
typedef WrappingInt MSGID;
const int DBPort = 27017;
class Listener {
public:
Listener(int p) : port(p) { }
void listen(); // never returns (start a thread)
/* spawn a thread, etc., then return */
virtual void accepted(MessagingPort *mp) = 0;
private:
int port;
};
class AbstractMessagingPort {
public:
virtual void reply(Message& received, Message& response) = 0;
};
class MessagingPort : public AbstractMessagingPort {
public:
MessagingPort(int sock, SockAddr& farEnd);
MessagingPort();
~MessagingPort();
void shutdown();
bool connect(SockAddr& farEnd);
/* it's assumed if you reuse a message object, that it doesn't cross MessagingPort's.
also, the Message data will go out of scope on the subsequent recv call.
*/
bool recv(Message& m);
void reply(Message& received, Message& response);
bool call(SockAddr& to, Message& toSend, Message& response);
void say(SockAddr& to, Message& toSend, int responseTo = -1);
private:
int sock;
SockAddr farEnd;
};
#pragma pack(push)
#pragma pack(1)
enum Operations {
opReply = 1, /* reply. responseTo is set. */
dbMsg = 1000, /* generic msg command followed by a string */
dbUpdate = 2001, /* update object */
dbInsert = 2002,
// dbGetByOID = 2003,
dbQuery = 2004,
dbGetMore = 2005,
dbDelete = 2006,
dbKillCursors = 2007
};
struct MsgData {
int len; /* len of the msg, including this field */
MSGID id; /* request/reply id's match... */
int responseTo; /* id of the message we are responding to */
int operation;
char _data[4];
int dataLen();
};
const int MsgDataHeaderSize = sizeof(MsgData) - 4;
inline int MsgData::dataLen() { return len - MsgDataHeaderSize; }
#pragma pack(pop)
class Message {
public:
Message() { data = 0; freeIt = false; }
Message( void * _data , bool _freeIt ){ data = (MsgData*)_data; freeIt = _freeIt; };
~Message() { reset(); }
SockAddr from;
MsgData *data;
Message& operator=(Message& r) {
assert( data == 0 );
data = r.data;
assert( r.freeIt );
r.freeIt = false;
r.data = 0;
freeIt = true;
return *this;
}
void reset() {
if( freeIt && data )
free(data);
data = 0; freeIt = false;
}
void setData(MsgData *d, bool _freeIt) {
assert( data == 0 );
freeIt = _freeIt;
data = d;
}
void setData(int operation, const char *msgtxt) {
setData(operation, msgtxt, strlen(msgtxt)+1);
}
void setData(int operation, const char *msgdata, int len) {
assert(data == 0);
int dataLen = len + sizeof(MsgData) - 4;
MsgData *d = (MsgData *) malloc(dataLen);
memcpy(d->_data, msgdata, len);
d->len = dataLen;
d->operation = operation;
freeIt= true;
data = d;
}
private:
bool freeIt;
};
|
