1 | #include "MessageProcessor.h"
|
---|
2 |
|
---|
3 | #include <iostream>
|
---|
4 | #include <fstream>
|
---|
5 |
|
---|
6 | #include "Compiler.h"
|
---|
7 |
|
---|
8 | #if defined WINDOWS
|
---|
9 | #include <ws2tcpip.h>
|
---|
10 | #endif
|
---|
11 |
|
---|
12 | #include "Common.h"
|
---|
13 |
|
---|
14 | MessageProcessor::MessageProcessor() {
|
---|
15 | this->sock = 0;
|
---|
16 | this->outputLog = NULL;
|
---|
17 | lastUsedId = 0;
|
---|
18 | }
|
---|
19 |
|
---|
20 | MessageProcessor::MessageProcessor(int sock, ofstream* outputLog) {
|
---|
21 | this->sock = sock;
|
---|
22 | this->outputLog = outputLog;
|
---|
23 | lastUsedId = 0;
|
---|
24 | }
|
---|
25 |
|
---|
26 | MessageProcessor::~MessageProcessor() {
|
---|
27 | }
|
---|
28 |
|
---|
29 | int MessageProcessor::sendMessage(NETWORK_MSG *msg, struct sockaddr_in *dest) {
|
---|
30 | cout << "Sending message of type " << msg->type << endl;
|
---|
31 |
|
---|
32 | msg->id = ++lastUsedId;
|
---|
33 | MessageContainer message(*msg, *dest);
|
---|
34 |
|
---|
35 | if (outputLog)
|
---|
36 | (*outputLog) << "Sending message (id " << msg->id << ") of type " << MessageContainer::getMsgTypeString(msg->type) << endl;
|
---|
37 |
|
---|
38 | sentMessages[msg->id][dest->sin_addr.s_addr] = message;
|
---|
39 |
|
---|
40 | int ret = sendto(sock, (char*)msg, sizeof(NETWORK_MSG), 0, (struct sockaddr *)dest, sizeof(struct sockaddr_in));
|
---|
41 |
|
---|
42 | if (ret < 0)
|
---|
43 | error("sendMessage");
|
---|
44 |
|
---|
45 | return ret;
|
---|
46 | }
|
---|
47 |
|
---|
48 | int MessageProcessor::receiveMessage(NETWORK_MSG *msg, struct sockaddr_in *source) {
|
---|
49 | socklen_t socklen = sizeof(struct sockaddr_in);
|
---|
50 |
|
---|
51 | // assume we don't care about the value of socklen
|
---|
52 | cout << "Waiting for message from server" << endl;
|
---|
53 | int ret = recvfrom(sock, (char*)msg, sizeof(NETWORK_MSG), 0, (struct sockaddr *)source, &socklen);
|
---|
54 | cout << "Returned from wait" << endl;
|
---|
55 |
|
---|
56 | if (ret == -1)
|
---|
57 | return ret;
|
---|
58 |
|
---|
59 | // add id to the NETWORK_MSG struct
|
---|
60 | if (msg->type == MSG_TYPE_ACK) {
|
---|
61 | if (!sentMessages[msg->id][source->sin_addr.s_addr].getAcked()) {
|
---|
62 | sentMessages[msg->id][source->sin_addr.s_addr].setAcked(true);
|
---|
63 | sentMessages[msg->id][source->sin_addr.s_addr].setTimeAcked(getCurrentMillis());
|
---|
64 | if (outputLog)
|
---|
65 | (*outputLog) << "Received ack for message id " << msg->id << endl;
|
---|
66 | }
|
---|
67 |
|
---|
68 | return -1; // don't do any further processing
|
---|
69 | }else {
|
---|
70 | bool isDuplicate = false;
|
---|
71 | map<unsigned int, unsigned long long>& ackedPlayerMessages = ackedMessages[source->sin_addr.s_addr];
|
---|
72 |
|
---|
73 | if (ackedPlayerMessages.find(msg->id) != ackedPlayerMessages.end()) {
|
---|
74 | isDuplicate = true;
|
---|
75 | cout << "Got duplicate of type " << msg->type << endl;
|
---|
76 | if (outputLog)
|
---|
77 | (*outputLog) << "Received duplicate (id " << msg->id << ") of type " << MessageContainer::getMsgTypeString(msg->type) << endl;
|
---|
78 | }else {
|
---|
79 | cout << "Got message of type " << msg->type << endl;
|
---|
80 | if (outputLog)
|
---|
81 | (*outputLog) << "Received message (id " << msg->id << ") of type " << MessageContainer::getMsgTypeString(msg->type) << endl;
|
---|
82 | }
|
---|
83 |
|
---|
84 | ackedPlayerMessages[msg->id] = getCurrentMillis();
|
---|
85 |
|
---|
86 | NETWORK_MSG ack;
|
---|
87 | ack.id = msg->id;
|
---|
88 | ack.type = MSG_TYPE_ACK;
|
---|
89 |
|
---|
90 | sendto(sock, (char*)&ack, sizeof(NETWORK_MSG), 0, (struct sockaddr *)source, sizeof(struct sockaddr_in));
|
---|
91 |
|
---|
92 | if (isDuplicate)
|
---|
93 | return -1;
|
---|
94 | }
|
---|
95 |
|
---|
96 | return ret;
|
---|
97 | }
|
---|
98 |
|
---|
99 | void MessageProcessor::broadcastMessage(NETWORK_MSG &msg, map<unsigned int, Player*>& players) {
|
---|
100 | map<unsigned int, Player*>::iterator it;
|
---|
101 | for (it = players.begin(); it != players.end(); it++) {
|
---|
102 | this->sendMessage(&msg, &(it->second->addr));
|
---|
103 | }
|
---|
104 | }
|
---|
105 |
|
---|
106 | void MessageProcessor::resendUnackedMessages() {
|
---|
107 | map<unsigned int, map<unsigned long, MessageContainer> >::iterator it;
|
---|
108 | map<unsigned long, MessageContainer>::iterator it2;
|
---|
109 | map<unsigned long, MessageContainer> sentMsg;
|
---|
110 |
|
---|
111 | for (it = sentMessages.begin(); it != sentMessages.end(); it++) {
|
---|
112 | sentMsg = it->second;
|
---|
113 | for (it2 = sentMsg.begin(); it2 != sentMsg.end(); it2++) {
|
---|
114 | if (!(it2->second.getAcked())) {
|
---|
115 | sendto(sock, (const char*)it2->second.getMessage(), sizeof(NETWORK_MSG), 0, (struct sockaddr *)&it2->first, sizeof(struct sockaddr_in));
|
---|
116 | }
|
---|
117 | }
|
---|
118 | }
|
---|
119 | }
|
---|
120 |
|
---|
121 | void MessageProcessor::cleanAckedMessages() {
|
---|
122 | map<unsigned int, map<unsigned long, MessageContainer> >::iterator it = sentMessages.begin();
|
---|
123 | map<unsigned long, MessageContainer>::iterator it2;
|
---|
124 |
|
---|
125 | while (it != sentMessages.end()) {
|
---|
126 | it2 = it->second.begin();
|
---|
127 | while (it2 != it->second.end()) {
|
---|
128 | if (it2->second.getAcked()) {
|
---|
129 | if ((getCurrentMillis() - it2->second.getTimeAcked()) > 1000) {
|
---|
130 | if (outputLog)
|
---|
131 | (*outputLog) << "Removing id " << it2->second.getMessage()->id << " from the acked record" << endl;
|
---|
132 | it->second.erase(it2++);
|
---|
133 | }else
|
---|
134 | it2++;
|
---|
135 | }else
|
---|
136 | it2++;
|
---|
137 | }
|
---|
138 |
|
---|
139 | if (it->second.size() == 0)
|
---|
140 | sentMessages.erase(it++);
|
---|
141 | else
|
---|
142 | it++;
|
---|
143 | }
|
---|
144 |
|
---|
145 | map<unsigned long, map<unsigned int, unsigned long long> >::iterator it3 = ackedMessages.begin();
|
---|
146 | map<unsigned int, unsigned long long>::iterator it4;
|
---|
147 |
|
---|
148 | // somehow want to delete the inner map once that player logs out
|
---|
149 | while (it3 != ackedMessages.end()) {
|
---|
150 | it4 = it3->second.begin();
|
---|
151 | while (it4 != it3->second.end()) {
|
---|
152 | if ((getCurrentMillis() - it4->second) > 500)
|
---|
153 | it3->second.erase(it4++);
|
---|
154 | else
|
---|
155 | it4++;
|
---|
156 | }
|
---|
157 | it3++;
|
---|
158 | }
|
---|
159 | }
|
---|
160 |
|
---|
161 | map<unsigned int, map<unsigned long, MessageContainer> >& MessageProcessor::getSentMessages() {
|
---|
162 | return this->sentMessages;
|
---|
163 | }
|
---|
164 |
|
---|
165 | map<unsigned long, map<unsigned int, unsigned long long> >& MessageProcessor::getAckedMessages() {
|
---|
166 | return this->ackedMessages;
|
---|
167 | }
|
---|