* Anthony Minessale II <anthm@freeswitch.org>
* John Wehle <john@feith.com>
* Garmt Boekholt <garmt@cimico.com>
+ * Seven Du <dujinfang@gmail.com>
*
* mod_xml_rpc.c -- XML RPC
*
#include <../lib/abyss/src/token.h>
#include <../lib/abyss/src/http.h>
#include <../lib/abyss/src/session.h>
+#include "ws.h"
SWITCH_MODULE_LOAD_FUNCTION(mod_xml_rpc_load);
SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_xml_rpc_shutdown);
switch_bool_t virtual_host;
TServer abyssServer;
xmlrpc_registry *registryP;
+ switch_bool_t enable_websocket;
} globals;
SWITCH_DECLARE_GLOBAL_STRING_FUNC(set_global_realm, globals.realm);
default_domain = val;
} else if (!strcasecmp(var, "virtual-host")) {
globals.virtual_host = switch_true(val);
+ } else if (!strcasecmp(var, "enable-websocket")) {
+ globals.enable_websocket = switch_true(val);
}
}
}
return rval;
}
+void stop_hook_event_handler(switch_event_t *event) {
+ char *json;
+ wsh_t *wsh = (TSession *)event->bind_user_data;
+
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "got websocket::stophook, closing\n");
+ wsh->down++;
+}
+
+void event_handler(switch_event_t *event) {
+ char *json;
+ wsh_t *wsh = (TSession *)event->bind_user_data;
+ switch_event_serialize_json(event, &json);
+ ws_write_frame(wsh, WSOC_TEXT, json, strlen(json));
+ free(json);
+}
+
+#define MAX_EVENT_BIND_SLOTS SWITCH_EVENT_ALL
+
+abyss_bool websocket_hook(TSession *r)
+{
+ wsh_t wsh;
+ int ret;
+ int i;
+ ws_opcode_t opcode;
+ uint8_t *data;
+ switch_event_node_t *nodes[MAX_EVENT_BIND_SLOTS];
+ int node_count = 0;
+ char *p;
+ char *key = TableFind(&r->requestHeaderFields, "sec-websocket-key");
+ char *version = TableFind(&r->requestHeaderFields, "sec-websocket-version");
+ char *proto = TableFind(&r->requestHeaderFields, "sec-websocket-protocol");
+ char *upgrade = TableFind(&r->requestHeaderFields, "connection");
+
+ if (!key || !version || !proto || !upgrade) return FALSE;
+ if (strncasecmp(upgrade, "Upgrade", 7) || strncasecmp(proto, "websocket", 9)) return FALSE;
+
+ for (i = 0; i < r->requestHeaderFields.size; ++i) {
+ TTableItem * const fieldP = &r->requestHeaderFields.item[i];
+ const char * const fieldValue = fieldP->value;
+
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "headers %s: %s\n", fieldP->name, fieldValue);
+ }
+
+ ret = ws_init(&wsh, r, NULL, 0);
+ if (ret != 0) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "websocket error %d\n", ret);
+ return FALSE;
+ }
+
+ while(!wsh.down && !wsh.handshake) {
+ ret = ws_handshake_kvp(&wsh, key, version, proto);
+ if (ret < 0) wsh.down = 1;
+ }
+
+ if (ret != 0) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "handshake error %d\n", ret);
+ return FALSE;
+ }
+
+ if (switch_event_bind_removable("websocket", SWITCH_EVENT_CUSTOM, "websocket::stophook", stop_hook_event_handler, &wsh, &nodes[node_count++]) != SWITCH_STATUS_SUCCESS) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Can't bind!\n");
+ node_count--;
+ }
+
+ while (!wsh.down) {
+ int bytes = ws_read_frame(&wsh, &opcode, &data);
+
+ if (bytes < 0) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "%d %s\n", opcode, (char *)data);
+ switch_yield(1000);
+ continue;
+ }
+
+ switch (opcode) {
+ case WSOC_CLOSE:
+ ws_close(&wsh, 1000);
+ break;
+ case WSOC_CONTINUATION:
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "continue\n");
+ continue;
+ case WSOC_TEXT:
+ p = data;
+ if (!p) continue;
+ if (!strncasecmp(data, "event ", 6)) {
+ switch_event_types_t type;
+ char *subclass;
+
+ if (node_count == MAX_EVENT_BIND_SLOTS - 1) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "cannot subscribe more than %d events\n", node_count);
+ continue;
+ }
+ p += 6;
+ if (p = strchr(p, ' ')) p++;
+ if (!strncasecmp(p, "json ", 5)) {
+ p += 5;
+ } else if (!strncasecmp(p, "xml ", 4)) {
+ p += 4;
+ } else if (!strncasecmp(p, "plain ", 6)) {
+ p += 6;
+ }
+ if (!*p) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_WARNING, "missing event type in [%s]\n", data);
+ break;
+ } else {
+ }
+ if (subclass = strchr(p, ' ')) {
+ *subclass++ = '\0';
+ if (!*subclass) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "missing subclass\n");
+ continue;
+ }
+ } else {
+ subclass = SWITCH_EVENT_SUBCLASS_ANY;
+ }
+
+ if (switch_name_event(p, &type) != SWITCH_STATUS_SUCCESS) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Unknown event %s\n", p);
+ continue;
+ }
+
+ if (switch_event_bind_removable("websocket", type, subclass, event_handler, &wsh, &nodes[node_count++]) != SWITCH_STATUS_SUCCESS) {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "Can't bind!\n");
+ node_count--;
+ continue;
+ } else {
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_INFO, "Bind %s\n", data);
+ }
+
+ }
+ break;
+ default:
+ break;
+ }
+ }
+
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_ERROR, "wsh.down = %d, node_count = %d\n", wsh.down, node_count);
+
+ switch_yield(2000);
+ while (--node_count >= 0) switch_event_unbind(&nodes[node_count]);
+
+ return FALSE;
+}
+
abyss_bool auth_hook(TSession * r)
{
char *domain_name, *e;
abyss_bool ret = FALSE;
+ if (globals.enable_websocket && !strncmp(r->requestInfo.uri, "/socket", 7)) {
+ // Chrome has no Authorization support yet
+ // https://code.google.com/p/chromium/issues/detail?id=123862
+ return websocket_hook(r);
+ }
+
if (!strncmp(r->requestInfo.uri, "/domains/", 9)) {
domain_name = strdup(r->requestInfo.uri + 9);
switch_assert(domain_name);
ServerAddHandler(&globals.abyssServer, auth_hook);
ServerSetKeepaliveTimeout(&globals.abyssServer, 5);
- switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Starting HTTP Port %d, DocRoot [%s]\n", globals.port, SWITCH_GLOBAL_dirs.htdocs_dir);
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "Starting HTTP Port %d, DocRoot [%s]%s\n",
+ globals.port, SWITCH_GLOBAL_dirs.htdocs_dir, globals.enable_websocket ? " with websocket." : "");
ServerRun(&globals.abyssServer);
switch_yield(1000000);
return SWITCH_STATUS_TERM;
}
+void stop_all_websockets()
+{
+ switch_event_t *event;
+ if (switch_event_create_subclass(&event, SWITCH_EVENT_CUSTOM, "websocket::stophook") != SWITCH_STATUS_SUCCESS) {
+ switch_log_printf(SWITCH_CHANNEL_LOG,SWITCH_LOG_ERROR, "Failed to create event!\n");
+ }
+ switch_event_add_header_string(event, SWITCH_STACK_BOTTOM, "stop", "now");
+ switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "stopping all websockets ...\n");
+ if (switch_event_fire(&event) != SWITCH_STATUS_SUCCESS) {
+ switch_log_printf(SWITCH_CHANNEL_LOG,SWITCH_LOG_ERROR, "Failed to fire the event!\n");
+ switch_event_destroy(&event);
+ return false;
+ }
+}
+
/* upon module unload */
SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_xml_rpc_shutdown)
{
+ /* Cann't find a way to stop the websockets, use this for a workaround before finding the real one that works */
+ stop_all_websockets();
+
/* this makes the worker thread (ServerRun) stop */
ServerTerminate(&globals.abyssServer);
#endif
-int ws_handshake(wsh_t *wsh)
+int ws_handshake_kvp(wsh_t *wsh, char *key, char *version, char *proto)
{
- char key[256] = "";
- char version[5] = "";
- char proto[256] = "";
char uri[256] = "";
char input[256] = "";
unsigned char output[SHA1_HASH_SIZE] = "";
issize_t bytes;
char *p, *e = 0;
- if (wsh->sock == ws_sock_invalid) {
+ if (!wsh->tsession) {
return -3;
}
- while((bytes = ws_raw_read(wsh, wsh->buffer + wsh->datalen, wsh->buflen - wsh->datalen)) > 0) {
- wsh->datalen += bytes;
- if (strstr(wsh->buffer, "\r\n\r\n") || strstr(wsh->buffer, "\n\n")) {
- break;
- }
- }
-
- if (bytes > sizeof(wsh->buffer)) {
+ if (!*key || !*version || !*proto) {
goto err;
}
- *(wsh->buffer+bytes) = '\0';
-
- if (strncasecmp(wsh->buffer, "GET ", 4)) {
- goto err;
- }
-
- p = wsh->buffer + 4;
-
- e = strchr(p, ' ');
- if (!e) {
- goto err;
- }
-
- strncpy(uri, p, e-p);
-
- cheezy_get_var(wsh->buffer, "Sec-WebSocket-Key", key, sizeof(key));
- cheezy_get_var(wsh->buffer, "Sec-WebSocket-Version", version, sizeof(version));
- cheezy_get_var(wsh->buffer, "Sec-WebSocket-Protocol", proto, sizeof(proto));
-
- if (!*key) {
- goto err;
- }
-
snprintf(input, sizeof(input), "%s%s", key, WEBSOCKET_GUID);
sha1_digest(output, input);
b64encode((unsigned char *)output, SHA1_HASH_SIZE, (unsigned char *)b64, sizeof(b64));
b64,
proto);
-
ws_raw_write(wsh, respond, strlen(respond));
wsh->handshake = 1;
{
issize_t r;
int x = 0;
+ TConn *conn = wsh->tsession->connP;
+#if 0
if (wsh->ssl) {
do {
r = SSL_read(wsh->ssl, data, bytes);
return r;
}
-
- do {
- r = recv(wsh->sock, data, bytes, 0);
-#ifndef _MSC_VER
- if (x++) usleep(10000);
-#else
- if (x++) Sleep(10);
#endif
- } while (r == -1 && (errno == EAGAIN || errno == EINTR || errno == EWOULDBLOCK ||
- errno == 35 || errno == 730035 || errno == 2 || errno == 60) && x < 100);
-
- if (x >= 100) {
- r = -1;
+
+ if (!wsh->handshake) {
+ r = wsh->tsession->connP->buffersize;
+ memcpy(data, conn->buffer.b, r);
+ printf("%s\n", conn->buffer.t);
+ ConnReadInit(conn);
+ return r;
+ } else {
+ const char *readError = NULL;
+
+ // printf(" pos=%d size=%d need=%d\n", conn->bufferpos, conn->buffersize, bytes);
+
+ r = conn->buffersize - conn->bufferpos;
+
+ if (r < 0) {
+ printf("348 Read Error %d!\n", r);
+ return 0;
+ } else if (r == 0) {
+ ConnRead(conn, 2, NULL, NULL, &readError);
+
+ if (readError) {
+ // printf("354 Read Error %s\n", readError);
+ xmlrpc_strfree(readError);
+ return 0;
+ }
+
+ r = conn->buffersize - conn->bufferpos;
+ }
+
+ if (r <= bytes) {
+ memcpy(data, conn->buffer.b + conn->bufferpos, r);
+ // ConnReadInit(conn);
+ conn->bufferpos = conn->buffersize;
+ ConnReadInit(conn);
+ return r;
+ } else {
+ memcpy(data, conn->buffer.b + conn->bufferpos, bytes);
+ conn->bufferpos += bytes;
+ return bytes;
+ }
+
}
-
+
return r;
}
return r;
}
- do {
- r = send(wsh->sock, data, bytes, 0);
- } while (r == -1 && (errno == EAGAIN || errno == EINTR));
+ if (ConnWrite(wsh->tsession->connP, data, bytes)) {
+ return bytes;
+ } else {
+ return 0;
+ }
//if (r<0) {
//printf("wRITE FAIL: %s\n", strerror(errno));
#endif
-
-int ws_init(wsh_t *wsh, ws_socket_t sock, SSL_CTX *ssl_ctx, int close_sock)
+int ws_init(wsh_t *wsh, ws_tsession_t *tsession, SSL_CTX *ssl_ctx, int close_sock)
{
memset(wsh, 0, sizeof(*wsh));
- wsh->sock = sock;
+ wsh->tsession = tsession;
if (!ssl_ctx) {
ssl_ctx = globals.ssl_ctx;
wsh->buflen = sizeof(wsh->buffer);
wsh->secure = ssl_ctx ? 1 : 0;
- setup_socket(sock);
+ // setup_socket(sock);
if (wsh->secure) {
int code;
}
+/*
while (!wsh->down && !wsh->handshake) {
int r = ws_handshake(wsh);
return -1;
}
}
+*/
if (wsh->down) {
return -1;
}
if ((wsh->datalen = ws_raw_read(wsh, wsh->buffer, 14)) < need) {
- if ((wsh->datalen += ws_raw_read(wsh, wsh->buffer + wsh->datalen, 14 - wsh->datalen)) < need) {
- /* too small - protocol err */
+ while (!wsh->down && (wsh->datalen += ws_raw_read(wsh, wsh->buffer + wsh->datalen, 14 - wsh->datalen)) < need) ;
+
+ if (0 && (wsh->datalen += ws_raw_read(wsh, wsh->buffer + wsh->datalen, 14 - wsh->datalen)) < need) {
+ /* too small - protocol err */
return ws_close(wsh, WS_PROTO_ERR);
}
}
projects / thirdparty / freeswitch.git / commitdiff
