Merge pull request #591 in FS/freeswitch from feature/FS-8293 to master

* commit '1ee2f0abc224fd6830fb8a2ddf120abfd8f641f0':
  add colors to good and bad
  update new pass
  FS-8293 [verto_communicator] - Showing speed in the menu bar if autoBand is true, adding option to test speed before making a call, enabling dedEnc if inboundBandwidth is below dedEncWatermark (3072 by default). FS-8293 [verto_communicator] - Changed info in the top menu bar for a icon+popover, changed autoBand's code in testSpeed method, moved 'Bandwidth settings' to 'Video settings' in settings modal and now hidding video quality selection when autoBand is checked. FS-8293 [verto_communicator] - Better bandwidth info ui. FS-8293 [verto_communicator] - Fixed speed test breaking at splash screen if user isn't logged in.
  Removed unused function.
  FS-8293 [verto_communicator] Implemented speed test in verto communicator.
  always change bw
  tweaks
  WIP not shabby auto vid settings
  first pass, add some funcs to conference and speed test features and fix bugs in ws.c for big payloads

src/mod/endpoints/mod_verto/mod_verto.c

@@ -1844,15 +1844,79 @@ static void client_run(jsock_t *jsock)
 			switch_ssize_t bytes;
 			ws_opcode_t oc;
 			uint8_t *data;
-
+			
 			bytes = ws_read_frame(&jsock->ws, &oc, &data);
-	
+			
 			if (bytes < 0) {
 				die("BAD READ %" SWITCH_SSIZE_T_FMT "\n", bytes);
 				break;
 			}
 
 			if (bytes) {
+				char *s = (char *) data;
+
+				if (*s == '#') {
+					char repl[2048] = "";
+					switch_time_t a, b;
+					
+					if (s[1] == 'S' && s[2] == 'P') {
+
+						if (s[3] == 'U') {
+							int i, size = 0;
+							char *p = s+4;
+							int loops = 0;
+							int rem = 0;
+							int dur = 0, j = 0;
+
+							if (!(size = atoi(p))) {
+								continue;
+							}
+							
+							a = switch_time_now();
+							do {
+								bytes = ws_read_frame(&jsock->ws, &oc, &data);
+								s = (char *) data;
+							} while (bytes && data && s[0] == '#' && s[3] == 'B');
+							b = switch_time_now();
+
+							if (!bytes || !data) continue;
+					
+							if (s[0] != '#') goto nm;
+
+							switch_snprintf(repl, sizeof(repl), "#SPU %ld", (b - a) / 1000);
+							ws_write_frame(&jsock->ws, WSOC_TEXT, repl, strlen(repl));
+							loops = size / 1024;
+							rem = size % 1024;
+							switch_snprintf(repl, sizeof(repl), "#SPB ");
+							memset(repl+4, '.', 1024);
+
+							for (j = 0; j < 10 ; j++) {
+								int ddur = 0;
+								a = switch_time_now();
+								for (i = 0; i < loops; i++) {
+									ws_write_frame(&jsock->ws, WSOC_TEXT, repl, 1024);
+								}
+								if (rem) {
+									ws_write_frame(&jsock->ws, WSOC_TEXT, repl, rem);
+								}
+								b = switch_time_now();
+								ddur += ((b - a) / 1000);
+								dur += ddur;
+
+							}
+
+							dur /= j+1;
+
+							switch_snprintf(repl, sizeof(repl), "#SPD %d", dur);
+							ws_write_frame(&jsock->ws, WSOC_TEXT, repl, strlen(repl));
+						}
+					}
+
+					continue;
+				}
+
+			nm:
+
 				if (process_input(jsock, data, bytes) != SWITCH_STATUS_SUCCESS) {
 					die("Input Error\n");
 				}
@@ -3271,7 +3335,7 @@ static switch_bool_t verto__info_func(const char *method, cJSON *params, jsock_t
 
 static switch_bool_t verto__invite_func(const char *method, cJSON *params, jsock_t *jsock, cJSON **response)
 {
-	cJSON *obj = cJSON_CreateObject(), *screenShare = NULL, *dedEnc = NULL, *mirrorInput, *json_ptr = NULL;
+	cJSON *obj = cJSON_CreateObject(), *screenShare = NULL, *dedEnc = NULL, *mirrorInput, *json_ptr = NULL, *bandwidth = NULL;
 	switch_core_session_t *session = NULL;
 	switch_channel_t *channel;
 	switch_event_t *var_event;
@@ -3281,7 +3345,7 @@ static switch_bool_t verto__invite_func(const char *method, cJSON *params, jsock
 	cJSON *dialog;
 	verto_pvt_t *tech_pvt;
 	char name[512];
-	const char *var, *destination_number, *call_id = NULL, *sdp = NULL, *bandwidth = NULL, 
+	const char *var, *destination_number, *call_id = NULL, *sdp = NULL,
 		*caller_id_name = NULL, *caller_id_number = NULL, *remote_caller_id_name = NULL, *remote_caller_id_number = NULL,*context = NULL;
 	switch_event_header_t *hp;
 	
@@ -3353,15 +3417,19 @@ static switch_bool_t verto__invite_func(const char *method, cJSON *params, jsock
 		switch_channel_set_flag(channel, CF_VIDEO_MIRROR_INPUT);
 	}
 
-	if ((bandwidth = cJSON_GetObjectCstr(dialog, "outgoingBandwidth"))) {
-		if (strcasecmp(bandwidth, "default")) {
-			switch_channel_set_variable(channel, "rtp_video_max_bandwidth_in", bandwidth);
+	if ((bandwidth = cJSON_GetObjectItem(dialog, "outgoingBandwidth"))) {
+		if (!zstr(bandwidth->valuestring) && strcasecmp(bandwidth->valuestring, "default")) {
+			switch_channel_set_variable(channel, "rtp_video_max_bandwidth_in", bandwidth->valuestring);
+		} else if (bandwidth->valueint) {
+			switch_channel_set_variable_printf(channel, "rtp_video_max_bandwidth_in", "%d", bandwidth->valueint);
 		}
 	}
 
-	if ((bandwidth = cJSON_GetObjectCstr(dialog, "incomingBandwidth"))) {
-		if (strcasecmp(bandwidth, "default")) {
-			switch_channel_set_variable(channel, "rtp_video_max_bandwidth_out", bandwidth);
+	if ((bandwidth = cJSON_GetObjectItem(dialog, "incomingBandwidth"))) {
+		if (!zstr(bandwidth->valuestring) && strcasecmp(bandwidth->valuestring, "default")) {
+			switch_channel_set_variable(channel, "rtp_video_max_bandwidth_out", bandwidth->valuestring);
+		} else if (bandwidth->valueint) {
+			switch_channel_set_variable_printf(channel, "rtp_video_max_bandwidth_out", "%d", bandwidth->valueint);
 		}
 	}
 

src/mod/endpoints/mod_verto/ws.c

@@ -264,7 +264,7 @@ int ws_handshake(wsh_t *wsh)
 		}
 	}
 
-	if (bytes > sizeof(wsh->buffer) -1) {
+	if (bytes > wsh->buflen -1) {
 		goto err;
 	}
 
@@ -362,7 +362,7 @@ ssize_t ws_raw_read(wsh_t *wsh, void *data, size_t bytes, int block)
 				}
 			}
 
-		} while (r == -1 && err == SSL_ERROR_WANT_READ && wsh->x < 100);
+		} while (r == -1 && err == SSL_ERROR_WANT_READ && wsh->x < 1000);
 
 		goto end;
 	}
@@ -382,7 +382,7 @@ ssize_t ws_raw_read(wsh_t *wsh, void *data, size_t bytes, int block)
 				ms_sleep(10);
 			}
 		}
-	} while (r == -1 && xp_is_blocking(xp_errno()) && wsh->x < 100);
+	} while (r == -1 && xp_is_blocking(xp_errno()) && wsh->x < 1000);
 	
 	if (wsh->x >= 1000 || (block && wsh->x >= 100)) {
 		r = -1;
@@ -596,7 +596,15 @@ int ws_init(wsh_t *wsh, ws_socket_t sock, SSL_CTX *ssl_ctx, int close_sock, int
 		wsh->close_sock = 1;
 	}
 
-	wsh->buflen = sizeof(wsh->buffer);
+	wsh->buflen = 1024 * 64;
+	wsh->bbuflen = wsh->buflen;
+
+	wsh->buffer = malloc(wsh->buflen);
+	wsh->bbuffer = malloc(wsh->bbuflen);
+	//printf("init %p %ld\n", (void *) wsh->bbuffer, wsh->bbuflen);
+	//memset(wsh->buffer, 0, wsh->buflen);
+	//memset(wsh->bbuffer, 0, wsh->bbuflen);
+
 	wsh->secure = ssl_ctx ? 1 : 0;
 
 	setup_socket(sock);
@@ -644,6 +652,12 @@ void ws_destroy(wsh_t *wsh)
 		SSL_free(wsh->ssl);
 		wsh->ssl = NULL;
 	}
+
+	if (wsh->buffer) free(wsh->buffer);
+	if (wsh->bbuffer) free(wsh->bbuffer);
+
+	wsh->buffer = wsh->bbuffer = NULL;
+
 }
 
 ssize_t ws_close(wsh_t *wsh, int16_t reason) 
@@ -685,6 +699,20 @@ ssize_t ws_close(wsh_t *wsh, int16_t reason)
 	
 }
 
+
+uint64_t hton64(uint64_t val)
+{
+	if (__BYTE_ORDER == __BIG_ENDIAN) return (val);
+	else return __bswap_64(val);
+}
+
+uint64_t ntoh64(uint64_t val)
+{
+	if (__BYTE_ORDER == __BIG_ENDIAN) return (val);
+	else return __bswap_64(val);
+}
+
+
 ssize_t ws_read_frame(wsh_t *wsh, ws_opcode_t *oc, uint8_t **data)
 {
 	
@@ -692,6 +720,10 @@ ssize_t ws_read_frame(wsh_t *wsh, ws_opcode_t *oc, uint8_t **data)
 	char *maskp;
 	int ll = 0;
 	int frag = 0;
+	int blen;
+
+	wsh->body = wsh->bbuffer;
+	wsh->packetlen = 0;
 
  again:
 	need = 2;
@@ -745,12 +777,11 @@ ssize_t ws_read_frame(wsh_t *wsh, ws_opcode_t *oc, uint8_t **data)
 			int fin = (wsh->buffer[0] >> 7) & 1;
 			int mask = (wsh->buffer[1] >> 7) & 1;
 			
-			if (fin) {
-				if (*oc == WSOC_CONTINUATION) {
-					frag = 1;
-				} else {
-					frag = 0;
-				}
+
+			if (!fin && *oc != WSOC_CONTINUATION) {
+				frag = 1;
+			} else if (fin && *oc == WSOC_CONTINUATION) {
+				frag = 0;
 			}
 
 			if (mask) {
@@ -765,23 +796,33 @@ ssize_t ws_read_frame(wsh_t *wsh, ws_opcode_t *oc, uint8_t **data)
 
 			wsh->plen = wsh->buffer[1] & 0x7f;
 			wsh->payload = &wsh->buffer[2];
-
+			
 			if (wsh->plen == 127) {
 				uint64_t *u64;
+				int more = 0;
 
 				need += 8;
 
 				if (need > wsh->datalen) {
 					/* too small - protocol err */
-					*oc = WSOC_CLOSE;
-					return ws_close(wsh, WS_PROTO_ERR);
-				}
+					//*oc = WSOC_CLOSE;
+					//return ws_close(wsh, WS_PROTO_ERR);
 
+					more = ws_raw_read(wsh, wsh->buffer + wsh->datalen, need - wsh->datalen, WS_BLOCK);
+
+					if (more < need - wsh->datalen) {
+						*oc = WSOC_CLOSE;
+						return ws_close(wsh, WS_PROTO_ERR);
+					} else {
+						wsh->datalen += more;
+					}
+
+
+				}
+				
 				u64 = (uint64_t *) wsh->payload;
 				wsh->payload += 8;
-
-				wsh->plen = ntohl((u_long)*u64);
-
+				wsh->plen = ntoh64(*u64);
 			} else if (wsh->plen == 126) {
 				uint16_t *u16;
 
@@ -811,16 +852,30 @@ ssize_t ws_read_frame(wsh_t *wsh, ws_opcode_t *oc, uint8_t **data)
 				return ws_close(wsh, WS_PROTO_ERR);
 			}
 
-			if ((need + wsh->datalen) > (ssize_t)wsh->buflen) {
-				/* too big - Ain't nobody got time fo' dat */
-				*oc = WSOC_CLOSE;
-				return ws_close(wsh, WS_DATA_TOO_BIG);				
+			blen = wsh->body - wsh->bbuffer;
+
+			if (need + blen > (ssize_t)wsh->bbuflen) {
+				void *tmp;
+				
+				wsh->bbuflen = need + blen + wsh->rplen;
+
+				if ((tmp = realloc(wsh->bbuffer, wsh->bbuflen))) {
+					wsh->bbuffer = tmp;
+				} else {
+					abort();
+				}
+
+				wsh->body = wsh->bbuffer + blen;
 			}
 
 			wsh->rplen = wsh->plen - need;
-
+			
+			if (wsh->rplen) {
+				memcpy(wsh->body, wsh->payload, wsh->rplen);
+			}
+			
 			while(need) {
-				ssize_t r = ws_raw_read(wsh, wsh->payload + wsh->rplen, need, WS_BLOCK);
+				ssize_t r = ws_raw_read(wsh, wsh->body + wsh->rplen, need, WS_BLOCK);
 
 				if (r < 1) {
 					/* invalid read - protocol err .. */
@@ -837,28 +892,30 @@ ssize_t ws_read_frame(wsh_t *wsh, ws_opcode_t *oc, uint8_t **data)
 				ssize_t i;
 
 				for (i = 0; i < wsh->datalen; i++) {
-					wsh->payload[i] ^= maskp[i % 4];
+					wsh->body[i] ^= maskp[i % 4];
 				}
 			}
 			
 
 			if (*oc == WSOC_PING) {
-				ws_write_frame(wsh, WSOC_PONG, wsh->payload, wsh->rplen);
+				ws_write_frame(wsh, WSOC_PONG, wsh->body, wsh->rplen);
 				goto again;
 			}
 
+			*(wsh->body+wsh->rplen) = '\0';
+			wsh->packetlen += wsh->rplen;
+			wsh->body += wsh->rplen;
+
 			if (frag) {
 				goto again;
 			}
+
+			*data = (uint8_t *)wsh->bbuffer;
 			
-
-			*(wsh->payload+wsh->rplen) = '\0';
-			*data = (uint8_t *)wsh->payload;
-
-			//printf("READ[%ld][%d]-----------------------------:\n[%s]\n-------------------------------\n", wsh->rplen, *oc, (char *)*data);
+			//printf("READ[%ld][%d]-----------------------------:\n[%s]\n-------------------------------\n", wsh->packetlen, *oc, (char *)*data);
 
 
-			return wsh->rplen;
+			return wsh->packetlen;
 		}
 		break;
 	default:
@@ -871,6 +928,7 @@ ssize_t ws_read_frame(wsh_t *wsh, ws_opcode_t *oc, uint8_t **data)
 	}
 }
 
+#if 0
 ssize_t ws_feed_buf(wsh_t *wsh, void *data, size_t bytes)
 {
 
@@ -885,6 +943,7 @@ ssize_t ws_feed_buf(wsh_t *wsh, void *data, size_t bytes)
 	return bytes;
 }
 
+
 ssize_t ws_send_buf(wsh_t *wsh, ws_opcode_t oc)
 {
 	ssize_t r = 0;
@@ -899,7 +958,7 @@ ssize_t ws_send_buf(wsh_t *wsh, ws_opcode_t oc)
 
 	return r;
 }
-
+#endif 
 
 ssize_t ws_write_frame(wsh_t *wsh, ws_opcode_t oc, void *data, size_t bytes)
 {
@@ -934,7 +993,7 @@ ssize_t ws_write_frame(wsh_t *wsh, ws_opcode_t oc, void *data, size_t bytes)
 		hlen += 8;
 		
 		u64 = (uint64_t *) &hdr[2];
-		*u64 = htonl(bytes);
+		*u64 = hton64(bytes);
 	}
 
 	if (wsh->write_buffer_len < (hlen + bytes + 1)) {

src/mod/endpoints/mod_verto/ws.h

@@ -78,15 +78,18 @@ typedef enum {
 
 typedef struct wsh_s {
 	ws_socket_t sock;
-	char buffer[65536];
-	char wbuffer[65536];
+	char *buffer;
+	char *bbuffer;
+	char *body;
 	char *uri;
 	size_t buflen;
+	size_t bbuflen;
 	ssize_t datalen;
 	ssize_t wdatalen;
 	char *payload;
 	ssize_t plen;
 	ssize_t rplen;
+	ssize_t packetlen;
 	SSL *ssl;
 	int handshake;
 	uint8_t down;