/*
* Open Pixel Control and HTTP server for Fadecandy.
*
* This is a TCP server which accepts either Open Pixel Control, HTTP, or WebSockets
* connections on a single port. We use a fork of libwebsockets which supports serving
* external non-HTTP protocols via a low-level receive callback.
*
* Copyright (c) 2013 Micah Elizabeth Scott
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to
* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
* the Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
* IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include "netserver.h"
#include "version.h"
#include "libwebsockets.h"
#include "rapidjson/stringbuffer.h"
#include "rapidjson/writer.h"
#include <iostream>
#include <algorithm>
NetServer::NetServer(OPC::callback_t opcCallback, jsonCallback_t jsonCallback,
void *context, bool verbose)
: mOpcCallback(opcCallback), mJsonCallback(jsonCallback),
mUserContext(context), mThread(0), mVerbose(verbose)
{}
bool NetServer::start(const char *host, int port)
{
const int llNormal = LLL_ERR | LLL_WARN;
const int llVerbose = llNormal | LLL_NOTICE;
static struct libwebsocket_protocols protocols[] = {
// Only one protocol for now. Handles HTTP as well as our default WebSockets protocol.
{
"fcserver", // Name
lwsCallback, // Callback
sizeof(Client), // Protocol-specific data size
sizeof(OPC::Message), // Max frame size / rx buffer
},
{ NULL, NULL, 0, 0 } // terminator
};
struct lws_context_creation_info info;
memset(&info, 0, sizeof info);
info.gid = -1;
info.uid = -1;
info.host = host;
info.port = port;
info.protocols = protocols;
info.user = this;
// Quieter during create_context, since it's kind of chatty.
lws_set_log_level(llNormal, NULL);
struct libwebsocket_context *context = libwebsocket_create_context(&info);
if (!context) {
lwsl_err("libwebsocket init failed\n");
return false;
}
// Maybe set up a more verbose log level now.
if (mVerbose) {
lws_set_log_level(llVerbose, NULL);
}
lwsl_notice("Server listening on %s:%d\n", host ? host : "*", port);
// Note that we pass ownership of all libwebsockets state to this new thread.
// We shouldn't access it on the other threads afterwards.
mThread = new tthread::thread(threadFunc, context);
return true;
}
void NetServer::threadFunc(void *arg)
{
struct libwebsocket_context *context = (libwebsocket_context*) arg;
NetServer *self = (NetServer*) libwebsocket_context_user(context);
/*
* Mostly we're just handling incoming events from libwebsocket's poll(),
* but we do have some non-latency-critical broadcast events to flush out
* periodically.
*
* These would be faster if we could wake up libwebsocket's poll() from another
* thread, but there isn't an easy way to do that now, and we don't really care
* that much. During normal operation we'll be receiving lots of data over the
* network anyway.
*/
while (libwebsocket_service(context, 100) >= 0) {
self->flushBroadcastList();
}
libwebsocket_context_destroy(context);
}
int NetServer::lwsCallback(libwebsocket_context *context, libwebsocket *wsi,
enum libwebsocket_callback_reasons reason, void *user, void *in, size_t len)
{
/*
* Protocol callback for libwebsockets.
*
* Until we have a reason to support a non-default WebSockets protocol, this handles
* everything: plain HTTP, Open Pixel Control, and WebSockets.
*
* For HTTP, this serves simple static HTML documents which are included at compile-time
* from the 'http' directory. Our web UI interacts with the server via the public WebSockets API.
*/
NetServer *self = (NetServer*) libwebsocket_context_user(context);
Client *client = (Client*) user;
switch (reason) {
case LWS_CALLBACK_CLOSED:
case LWS_CALLBACK_CLOSED_HTTP:
if (client && client->opcBuffer) {
free(client->opcBuffer);
client->opcBuffer = NULL;
}
self->mClients.erase(wsi);
break;
case LWS_CALLBACK_ESTABLISHED:
self->mClients.insert(wsi);
break;
case LWS_CALLBACK_HTTP:
return self->httpBegin(context, wsi, *client, (const char*) in);
case LWS_CALLBACK_HTTP_FILE_COMPLETION:
// Only serve one file per connect
return -1;
case LWS_CALLBACK_HTTP_WRITEABLE:
return self->httpWrite(context, wsi, *client);
case LWS_CALLBACK_SOCKET_READ:
// Low-level socket read. We may trap these for OPC protocol handling.
if (client->state != CLIENT_STATE_HTTP) {
return self->opcRead(context, wsi, *client, (uint8_t*)in, len);
}
break;
case LWS_CALLBACK_RECEIVE:
// WebSockets data received
return self->wsRead(context, wsi, *client, (uint8_t*)in, len);
default:
break;
}
return 0;
}
int NetServer::opcRead(libwebsocket_context *context, libwebsocket *wsi,
Client &client, uint8_t *in, size_t len)
{
/*
* Open Pixel Control packet dispatch, and protocol detection.
*
* Store the new packet in our protocol buffer. This should be large enough to never overflow,
* since our OPC buffer is large enough for two packets and the OPC max packet size is much larger
* than the network receive buffer.
*
* If we have no buffered data yet, we can do this without copying.
*/
OPCBuffer *opcb;
uint8_t *buffer;
unsigned bufferLength;
// Allocate the buffer we use for OPC reassembly and protocol-detect.
if (client.opcBuffer == NULL) {
opcb = (OPCBuffer*) malloc(sizeof *client.opcBuffer);
if (opcb == NULL) {
lwsl_err("ERROR: Out of memory allocating OPC reassembly buffer.\n");
return -1;
}
opcb->bufferLength = 0;
client.opcBuffer = opcb;
} else {
opcb = client.opcBuffer;
}
if (len + opcb->bufferLength > sizeof opcb->buffer) {
return -1;
}
if (opcb->bufferLength) {
memcpy(opcb->bufferLength + opcb->buffer, in, len);
buffer = opcb->buffer;
bufferLength = opcb->bufferLength + len;
} else {
buffer = in;
bufferLength = len;
}
if (client.state == CLIENT_STATE_PROTOCOL_DETECT) {
/*
* It's a new connection, and we aren't sure yet whether it's native OPC
* or HTTP / WebSockets. We examine the first four bytes received. If it's
* "GET ", we assume this is HTTP. (Other HTTP methods are not needed).
* If it's anything else, we interpret the connection as native OPC and these
* are the first four bytes of the first OPC packet.
*/
if (bufferLength < 4) {
// Not enough data for protocol detect yet. Save this data for later.
if (buffer != opcb->buffer) {
memcpy(opcb->buffer, buffer, bufferLength);
opcb->bufferLength = bufferLength;
}
// Do not pass this data on to libwebsocket yet
return 1;
}
if (buffer[0] == 'G' && buffer[1] == 'E' && buffer[2] == 'T' && buffer[3] == ' ') {
// Detected HTTP. Convert this to an HTTP client, and let libwebsockets handle
// all data received so far. We can jettison the OPC buffer at this point.
free(client.opcBuffer);
client.opcBuffer = 0;
client.state = CLIENT_STATE_HTTP;
if (libwebsocket_read(context, wsi, buffer, bufferLength) < 0) {
return -1;
}
return 1;
}
// Not HTTP. Handle this as an OPC socket.
client.state = CLIENT_STATE_OPEN_PIXEL_CONTROL;
lwsl_notice("New Open Pixel Control connection\n");
}
// Process any and all complete packets from our buffer
while (1) {
if (bufferLength < OPC::HEADER_BYTES) {
// Still waiting for a header
break;
}
OPC::Message *msg = (OPC::Message*) buffer;
unsigned msgLength = OPC::HEADER_BYTES + msg->length();
if (bufferLength < msgLength) {
// Waiting for more data
break;
}
// Complete packet.
mOpcCallback(*msg, mUserContext);
buffer += msgLength;
bufferLength -= msgLength;
}
// If we have any residual data, save it for later.
if (bufferLength && buffer != opcb->buffer) {
memmove(opcb->buffer, buffer, bufferLength);
}
opcb->bufferLength = bufferLength;
// Don't pass data on to libwebsockets
return 1;
}
bool NetServer::httpPathEqual(const char *a, const char *b)
{
// HTTP path comparison. Stop at '?' or '#', to ignore query/fragment portions.
for (;;) {
char ca = *a;
char cb = *b;
if (ca == '?' || cb == '?' || ca == '#' || cb == '#')
return true;
if (ca != cb)
return false;
if (ca == '\0')
return true;
a++;
b++;
}
}
int NetServer::httpBegin(libwebsocket_context *context, libwebsocket *wsi,
Client &client, const char *path)
{
/*
* We have a new plain HTTP request. Match it against our document list, and send
* back headers for the response.
*
* Note: To keep the size of fcserver down, we compress our HTTP documents.
* We don't bother supporting decompressing them. Instead, we always send
* them back with deflate content-encoding.
*/
HTTPDocument *doc = httpDocumentList;
// Look for this path in the document list. If it isn't found, we'll serve the 404 doc.
while (doc->path && !httpPathEqual(doc->path, path))
doc++;
if (!doc->path) {
lwsl_notice("HTTP document not found, \"%s\"\n", path);
}
char buffer[1024];
int size = snprintf(buffer, sizeof buffer,
"HTTP/1.1 %d %s\r\n"
"Server: %s\r\n"
"Content-Type: %s\r\n"
"Content-Length: %u\r\n"
"Content-Encoding: deflate\r\n"
"Connection: close\r\n"
"\r\n",
doc->path ? 200 : 404,
doc->path ? "OK" : "Not Found",
kFCServerVersion,
doc->contentType,
doc->contentLength
);
if (libwebsocket_write(wsi, (unsigned char*) buffer, size, LWS_WRITE_HTTP) < 0) {
return -1;
}
// Write the body asynchronously
client.httpBody = doc->body;
client.httpLength = doc->contentLength;
libwebsocket_callback_on_writable(context, wsi);
return 0;
}
int NetServer::httpWrite(libwebsocket_context *context, libwebsocket *wsi, Client &client)
{
if (!client.httpBody) {
return -1;
}
do {
if (client.httpLength <= 0) {
// End of document
return -1;
}
int m = libwebsocket_write(wsi, (unsigned char *) client.httpBody, client.httpLength, LWS_WRITE_HTTP);
if (m < 0) {
return -1;
}
client.httpBody += m;
client.httpLength -= m;
} while (!lws_send_pipe_choked(wsi));
libwebsocket_callback_on_writable(context, wsi);
return 0;
}
int NetServer::wsRead(libwebsocket_context *context, libwebsocket *wsi, Client &client, uint8_t *in, size_t len)
{
// If this frame is binary, it's an OPC message. Does it parse?
if (lws_frame_is_binary(wsi)) {
OPC::Message *msg = (OPC::Message*) in;
if (len < OPC::HEADER_BYTES) {
lwsl_notice("NOTICE: Received binary WebSockets packet, but it's too small for an OPC header.\n");
return 0;
}
if (msg->lenLow != 0 || msg->lenHigh != 0) {
lwsl_notice("NOTICE: Received OPC packet over WebSockets with nonzero reserved (length) fields.\n");
}
if (len > sizeof *msg) {
lwsl_notice("NOTICE: Received oversized OPC packet over WebSockets. Truncating.\n");
len = sizeof *msg;
}
msg->setLength(len - OPC::HEADER_BYTES);
mOpcCallback(*msg, mUserContext);
return 0;
}
// Text frames are JSON encoded. Does that parse?
rapidjson::Document message;
message.ParseInsitu<0>((char*) in);
if (message.HasParseError()) {
lwsl_notice("NOTICE: Parse error in received JSON, character %d: %s\n",
int(message.GetErrorOffset()), message.GetParseError());
return 0;
}
if (!message.IsObject()) {
lwsl_notice("NOTICE: Received JSON is not an object {}\n");
return 0;
}
mJsonCallback(wsi, message, mUserContext);
return 0;
}
int NetServer::jsonReply(libwebsocket *wsi, rapidjson::Document &message)
{
jsonBuffer_t buffer;
jsonBufferPrepare(buffer, message);
return jsonBufferSend(buffer, wsi);
}
void NetServer::jsonBufferPrepare(jsonBuffer_t &buffer, rapidjson::Value &value)
{
// Pre-packet padding
rapidjson::PutN<>(buffer, 0, LWS_SEND_BUFFER_PRE_PADDING);
// Write serialized message
rapidjson::Writer<rapidjson::GenericStringBuffer<rapidjson::UTF8<>>> writer(buffer);
value.Accept(writer);
// Post-packet padding
rapidjson::PutN<>(buffer, 0, LWS_SEND_BUFFER_POST_PADDING);
}
int NetServer::jsonBufferSend(jsonBuffer_t &buffer, libwebsocket *wsi)
{
const char *string = buffer.GetString() + LWS_SEND_BUFFER_PRE_PADDING;
size_t len = buffer.Size() - LWS_SEND_BUFFER_PRE_PADDING - LWS_SEND_BUFFER_POST_PADDING;
return libwebsocket_write(wsi, (unsigned char *) string, len, LWS_WRITE_TEXT);
}
void NetServer::flushBroadcastList()
{
// Send any pending broadcast packets. These are enqueued by other threads on a list
// protected by mBroadcastMutex.
mBroadcastMutex.lock();
for (std::vector<jsonBuffer_t*>::iterator buf = mBroadcastList.begin(); buf != mBroadcastList.end(); ++buf) {
for (std::set<libwebsocket*>::iterator cli = mClients.begin(); cli != mClients.end(); ++cli) {
jsonBufferSend(**buf, *cli);
}
delete *buf;
}
mBroadcastList.clear();
mBroadcastMutex.unlock();
}
void NetServer::jsonBroadcast(rapidjson::Document &message)
{
jsonBuffer_t *buffer = new jsonBuffer_t();
jsonBufferPrepare(*buffer, message);
mBroadcastMutex.lock();
mBroadcastList.push_back(buffer);
mBroadcastMutex.unlock();
}