kpmpgsmkii/server/src/client.c
2022-01-15 20:01:29 -06:00

332 lines
11 KiB
C

/*
* Kangaroo Punch MultiPlayer Game Server Mark II
* Copyright (C) 2020-2021 Scott Duensing
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*
*/
#include "client.h"
#include "array.h"
#include "network.h"
#include "console.h"
#include "packet.h"
#include "server.h"
#include "rest.h"
static uint8_t clientDequeuePacket(ClientThreadT *client);
static void clientProcessPacket(ClientThreadT *client, PacketDecodeDataT *data);
static uint8_t clientDequeuePacket(ClientThreadT *client) {
uint16_t length = 0;
char *data = NULL;
PacketDecodeDataT decode = { 0 };
ClientRawPacketT *packet = NULL;
// Is there new data to process?
pthread_mutex_lock(&client->packetQueueMutex);
if (arrlenu(client->packetQueue) > 0) {
packet = client->packetQueue[0];
length = packet->length;
data = packet->data;
arrdel(client->packetQueue, 0);
}
pthread_mutex_unlock(&client->packetQueueMutex);
// New data or not, process anything in the queue.
if (packetDecode(client->packetThreadData, &decode, data, length)) {
clientProcessPacket(client, &decode);
if (decode.data) DEL(decode.data);
if (packet) {
DEL(packet->data);
DEL(packet);
}
return 1;
}
// No packet.
return 0;
}
static void clientProcessPacket(ClientThreadT *client, PacketDecodeDataT *data) {
uint64_t i = 0;
uint64_t x = 0;
uint32_t y = 0;
uint64_t length = 0;
char *buffer = NULL;
PacketEncodeDataT encoded = { 0 };
json_object *response = NULL;
RestStringMapT *strings = NULL;
RestIntegerMapT *integers = NULL;
struct timespec timer = { 0 };
double d = 0;
PacketTypeSignUpT *signup = NULL;
PacketTypeSignUpResultT signupResult = { 0 };
switch (data->packetType) {
case PACKET_TYPE_CLIENT_SHUTDOWN:
serverDisconnectClient(client);
break;
case PACKET_TYPE_LOGIN:
consoleMessageQueue("%ld: Channel %d %s %s\n", client->threadIndex, data->channel, ((PacketTypeLoginT *)data->data)->user, ((PacketTypeLoginT *)data->data)->pass);
break;
case PACKET_TYPE_PONG:
// Time round-trip.
clock_gettime(CLOCK_MONOTONIC_RAW, &timer);
d = (timer.tv_sec - client->pingStart.tv_sec) * 1e9;
d = (d + (timer.tv_nsec - client->pingStart.tv_nsec)) * 1e-9;
client->pingStats[client->pingHead] = d;
client->pingHead++;
if (client->pingHead >= PING_STATS_SIZE) client->pingHead = 0;
// ***TODO*** Probably need a mutex here.
if (d > client->pingHigh) client->pingHigh = d;
if (d < client->pingLow) client->pingLow = d;
client->pingAverage = 0;
x = 0;
for (i=0; i<PING_STATS_SIZE; i++) {
if (client->pingStats[i] != 0) {
x++;
client->pingAverage += client->pingStats[i];
}
}
client->pingAverage /= (double)x;
consoleMessageQueue("%ld: Ping: %f Low: %f Average: %f High: %f\n", client->threadIndex, d, client->pingLow, client->pingAverage, client->pingHigh);
break;
case PACKET_TYPE_SIGNUP:
signup = (PacketTypeSignUpT *)data->data;
response = restRequest("USER_CREATE", "sssss",
"first", signup->first,
"last", signup->last,
"user", signup->user,
"pass", signup->pass,
"email", signup->email
);
if (response) {
signupResult.success = (json_object_get_boolean(json_object_object_get(response, "result")) == TRUE) ? 1 : 0;
buffer = (char *)json_object_get_string(json_object_object_get(response, "reason"));
} else {
// Something bad happened.
signupResult.success = 0;
buffer = "Unknown error. Sorry.";
}
memcpy(signupResult.message, buffer, strlen(buffer));
if (response) restRelease(response);
// Build packet.
encoded.control = PACKET_CONTROL_DAT;
encoded.packetType = PACKET_TYPE_SIGNUP_RESULT;
encoded.channel = 0;
encoded.encrypt = 0;
packetEncode(client->packetThreadData, &encoded, (char *)&signupResult, sizeof(PacketTypeSignUpResultT));
// Send it.
packetSend(client->packetThreadData, &encoded);
break;
case PACKET_TYPE_VERSION_BAD:
//***TODO***
break;
case PACKET_TYPE_VERSION_OKAY:
// Fetch string table from REST.
response = restRequest("CONFIG_GET_STRINGS", NULL);
if (!response) {
consoleMessageQueue("%ld: Unable to fetch strings!\n", client->threadIndex);
break;
}
strings = restHelperConfigStringMapGet(response);
if (!strings) {
consoleMessageQueue("%ld: Unable to map strings!\n", client->threadIndex);
break;
}
restRelease(response);
// Send string table to client.
for (i=0; i<(unsigned)shlen(strings); i++) {
// Strings are encoded in a single buffer as: KEY\0DATA\0
x = strlen(strings[i].key);
y = strlen(strings[i].value);
length = x + y + 2;
buffer = (char *)malloc(length);
if (!buffer) {
consoleMessageQueue("%ld: Unable to allocate buffer for string packet!\n", client->threadIndex);
break;
}
memcpy(buffer, strings[i].key, x + 1);
memcpy(&buffer[x + 1], strings[i].value, y + 1);
//consoleMessageQueue("[%s]=[%s]\n", strings[i].key, strings[i].value);
// Build packet.
encoded.control = PACKET_CONTROL_DAT;
encoded.packetType = PACKET_TYPE_STRING;
encoded.channel = 0;
encoded.encrypt = 0;
packetEncode(client->packetThreadData, &encoded, buffer, length);
// Send it.
packetSend(client->packetThreadData, &encoded);
DEL(buffer);
}
restHelperConfigStringMapRelease(strings);
// Fetch number table from REST.
response = restRequest("CONFIG_GET_NUMBERS", NULL);
if (!response) {
consoleMessageQueue("%ld: Unable to fetch numbers!\n", client->threadIndex);
break;
}
integers = restHelperConfigIntegerMapGet(response);
if (!integers) {
consoleMessageQueue("%ld: Unable to map numbers!\n", client->threadIndex);
break;
}
restRelease(response);
// Send number table to client.
for (i=0; i<(unsigned)shlen(integers); i++) {
// Integers are encoded in a single buffer as: 1234DATA\0
// Integers are 64 bit until sent to the client when they are truncated to 32.
x = strlen(integers[i].key);
y = integers[i].value; // 64->32
length = x + 5;
buffer = (char *)malloc(length);
if (!buffer) {
consoleMessageQueue("%ld: Unable to allocate buffer for number packet!\n\r", client->threadIndex);
break;
}
memcpy(buffer, &y, 4);
memcpy(&buffer[4], integers[i].key, x + 1);
// Build packet.
encoded.control = PACKET_CONTROL_DAT;
encoded.packetType = PACKET_TYPE_NUMBER;
encoded.channel = 0;
encoded.encrypt = 0;
packetEncode(client->packetThreadData, &encoded, buffer, length);
// Send it.
packetSend(client->packetThreadData, &encoded);
DEL(buffer);
//logWrite("[%s] = [%d]\r\n", integers[i].key, integers[i].value);
}
restHelperConfigIntegerMapRelease(integers);
// Build PROCEED packet.
encoded.control = PACKET_CONTROL_DAT;
encoded.packetType = PACKET_TYPE_PROCEED;
encoded.channel = 0;
encoded.encrypt = 0;
packetEncode(client->packetThreadData, &encoded, NULL, 0);
// Send it.
packetSend(client->packetThreadData, &encoded);
break;
default:
consoleMessageQueue("%ld: Channel %d Unknown Packet %d\n", client->threadIndex, data->channel, data->packetType);
break;
}
}
void clientQueuePacket(ClientThreadT *client, uint8_t *data, uint32_t length) {
ClientRawPacketT *packet = NULL;
packet = (ClientRawPacketT *)malloc(sizeof(ClientRawPacketT));
if (packet) {
packet->data = (char *)malloc(length);
if (packet->data) {
memcpy(packet->data, data, length);
packet->length = length;
}
//consoleMessageQueue("%ld: Channel %d Bytes in %d\n", client->threadIndex, length);
/*
logWrite("Bytes %d\n\r", length);
for (size_t x=0; x<length; x++) {
logWrite("[%x] '%c'\n\r", data[x], data[x]);
}
*/
pthread_mutex_lock(&client->packetQueueMutex);
arrput(client->packetQueue, packet);
pthread_mutex_unlock(&client->packetQueueMutex);
}
}
void *clientThread(void *data) {
ENetPeer *peer = (ENetPeer *)data;
ClientThreadT *client = (ClientThreadT *)peer->data;
PacketEncodeDataT encoded = { 0 };
struct timespec remaining = { 0, 0 };
struct timespec sleepTime = { 0, 1000000000/100 }; // 1/100th second.
PacketTypeVersionT version = { 0 };
uint8_t versionSent = 0;
time_t ticks = { 0 };
time_t lastTicks = { 0 };
int8_t pingTimeout = 0;
// Process packets until we're done.
while (client->running) {
if (!clientDequeuePacket(client)) {
// Start communications with client as soon as encryption channel is ready.
if (!versionSent) {
if (packetEncryptionReady()) {
// Send required protocol version.
version.version = PACKET_PROTOCOL_VERSION;
encoded.control = PACKET_CONTROL_DAT;
encoded.packetType = PACKET_TYPE_VERSION;
encoded.channel = 0;
encoded.encrypt = 0;
packetEncode(client->packetThreadData, &encoded, (char *)&version, sizeof(PacketTypeVersionT));
packetSend(client->packetThreadData, &encoded);
versionSent = 1;
}
}
// Ping the client every 5 seconds.
ticks = time(NULL);
if (ticks != lastTicks) {
lastTicks = ticks;
pingTimeout++;
if (pingTimeout >= 5) {
pingTimeout = 0;
encoded.control = PACKET_CONTROL_DAT;
encoded.packetType = PACKET_TYPE_PING;
encoded.channel = 0;
encoded.encrypt = 0;
packetEncode(client->packetThreadData, &encoded, (char *)&version, sizeof(PacketTypeVersionT));
packetSend(client->packetThreadData, &encoded);
clock_gettime(CLOCK_MONOTONIC_RAW, &client->pingStart);
}
}
// Don't eat all the CPU.
nanosleep(&remaining, &sleepTime);
}
}
// ***TODO*** Write ping stats to database.
// Clean up client data on the way out.
while (arrlen(client->packetQueue) > 0) {
if (client->packetQueue[0]->data) DEL(client->packetQueue[0]->data);
arrdel(client->packetQueue, 0);
}
arrfree(client->packetQueue);
pthread_mutex_destroy(&client->packetQueueMutex);
packetThreadDataDestroy(&client->packetThreadData);
DEL(client);
pthread_exit(NULL);
}