con_ws_read(agooCon c) {
    ssize_t	cnt;
    uint8_t	*b;
    uint8_t	op;
    long	mlen;

    if (NULL != c->req) {
	cnt = recv(c->sock, c->req->msg + c->bcnt, c->req->mlen - c->bcnt, 0);
    } else {
	cnt = recv(c->sock, c->buf + c->bcnt, sizeof(c->buf) - c->bcnt - 1, 0);
    }
    c->timeout = dtime() + CON_TIMEOUT;
    if (0 >= cnt) {
	// If nothing read then no need to complain. Just close.
	if (0 < c->bcnt) {
	    if (0 == cnt) {
		agoo_log_cat(&agoo_warn_cat, "Nothing to read. Client closed socket on connection %llu.", (unsigned long long)c->id);
	    } else {
		char	msg[1024];
		int	len = snprintf(msg, sizeof(msg) - 1, "Failed to read WebSocket message. %s.", strerror(errno));

		push_error(c->up, msg, len);
		agoo_log_cat(&agoo_warn_cat, "Failed to read WebSocket message. %s.", strerror(errno));
	    }
	}
	return true;
    }
    c->bcnt += cnt;
    while (true) {
	if (NULL == c->req) {
	    if (c->bcnt < 2) {
		return false; // Try again.
	    }
	    b = (uint8_t*)c->buf;
	    if (0 >= (mlen = agoo_ws_calc_len(c, b, c->bcnt))) {
		return (mlen < 0);
	    }
	    op = 0x0F & *b;
	    switch (op) {
	    case AGOO_WS_OP_TEXT:
	    case AGOO_WS_OP_BIN:
		if (NULL != c->gsub) {
		    // GraphQL subscriptions do not accept input on the
		    // connection.
		    break;
		}
		if (agoo_ws_create_req(c, mlen)) {
		    return true;
		}
		break;
	    case AGOO_WS_OP_CLOSE:
		return true;
	    case AGOO_WS_OP_PING:
		if (mlen == (long)c->bcnt) {
		    agoo_ws_pong(c);
		    c->bcnt = 0;
		}
		break;
	    case AGOO_WS_OP_PONG:
		// ignore
		if (mlen == (long)c->bcnt) {
		    c->bcnt = 0;
		}
		break;
	    case AGOO_WS_OP_CONT:
	    default: {
		char	msg[1024];
		int	len = snprintf(msg, sizeof(msg) - 1, "WebSocket op 0x%02x not supported on %llu.",
				       op, (unsigned long long)c->id);

		push_error(c->up, msg, len);
		agoo_log_cat(&agoo_error_cat, "WebSocket op 0x%02x not supported on %llu.", op, (unsigned long long)c->id);
		return true;
	    }
	    }
	}
	if (NULL != c->req) {
	    mlen = c->req->mlen;
	    c->req->mlen = agoo_ws_decode(c->req->msg, c->req->mlen);
	    if (mlen <= (long)c->bcnt) {
		if (agoo_debug_cat.on) {
		    if (AGOO_ON_MSG == c->req->method) {
			agoo_log_cat(&agoo_debug_cat, "WebSocket message on %llu: %s", (unsigned long long)c->id, c->req->msg);
		    } else {
			agoo_log_cat(&agoo_debug_cat, "WebSocket binary message on %llu", (unsigned long long)c->id);
		    }
		}
	    }
	    agoo_upgraded_ref(c->up);
	    agoo_queue_push(&agoo_server.eval_queue, (void*)c->req);
	    if (mlen < (long)c->bcnt) {
		memmove(c->buf, c->buf + mlen, c->bcnt - mlen);
		c->bcnt -= mlen;
	    } else {
		c->bcnt = 0;
		*c->buf = '\0';
		c->req = NULL;
		break;
	    }
	    c->req = NULL;
	    continue;
	}
	break;
    }
    return false;
}
con_ws_read(agooCon c) {
    ssize_t	cnt;
    uint8_t	*b;
    uint8_t	op;
    long	mlen;

    if (NULL != c->req) {
	cnt = recv(c->sock, c->req->msg + c->bcnt, c->req->mlen - c->bcnt, 0);
    } else {
	cnt = recv(c->sock, c->buf + c->bcnt, sizeof(c->buf) - c->bcnt - 1, 0);
    }
    c->timeout = dtime() + CON_TIMEOUT;
    if (0 >= cnt) {
	 
	if (0 < c->bcnt) {
	    if (0 == cnt) {
		agoo_log_cat(&agoo_warn_cat, "Nothing to read. Client closed socket on connection %llu.", (unsigned long long)c->id);
	    } else {
		char	msg[1024];
		int	len = snprintf(msg, sizeof(msg) - 1, "Failed to read WebSocket message. %s.", strerror(errno));

		push_error(c->up, msg, len);
		agoo_log_cat(&agoo_warn_cat, "Failed to read WebSocket message. %s.", strerror(errno));
	    }
	}
	return true;
    }
    c->bcnt += cnt;
    while (true) {
	if (NULL == c->req) {
	    if (c->bcnt < 2) {
		return false;  
	    }
	    b = (uint8_t*)c->buf;
	    if (0 >= (mlen = agoo_ws_calc_len(c, b, c->bcnt))) {
		return (mlen < 0);
	    }
	    op = 0x0F & *b;
	    switch (op) {
	    case AGOO_WS_OP_TEXT:
	    case AGOO_WS_OP_BIN:
		if (NULL != c->gsub) {
		     
		     
		    break;
		}
		if (agoo_ws_create_req(c, mlen)) {
		    return true;
		}
		break;
	    case AGOO_WS_OP_CLOSE:
		return true;
	    case AGOO_WS_OP_PING:
		if (mlen == (long)c->bcnt) {
		    agoo_ws_pong(c);
		    c->bcnt = 0;
		}
		break;
	    case AGOO_WS_OP_PONG:
		 
		if (mlen == (long)c->bcnt) {
		    c->bcnt = 0;
		}
		break;
	    case AGOO_WS_OP_CONT:
	    default: {
		char	msg[1024];
		int	len = snprintf(msg, sizeof(msg) - 1, "WebSocket op 0x%02x not supported on %llu.",
				       op, (unsigned long long)c->id);

		push_error(c->up, msg, len);
		agoo_log_cat(&agoo_error_cat, "WebSocket op 0x%02x not supported on %llu.", op, (unsigned long long)c->id);
		return true;
	    }
	    }
	}
	if (NULL != c->req) {
	    mlen = c->req->mlen;
	    c->req->mlen = agoo_ws_decode(c->req->msg, c->req->mlen);
	    if (mlen <= (long)c->bcnt) {
		if (agoo_debug_cat.on) {
		    if (AGOO_ON_MSG == c->req->method) {
			agoo_log_cat(&agoo_debug_cat, "WebSocket message on %llu: %s", (unsigned long long)c->id, c->req->msg);
		    } else {
			agoo_log_cat(&agoo_debug_cat, "WebSocket binary message on %llu", (unsigned long long)c->id);
		    }
		}
	    }
	    agoo_upgraded_ref(c->up);
	    agoo_queue_push(&agoo_server.eval_queue, (void*)c->req);
	    if (mlen < (long)c->bcnt) {
		memmove(c->buf, c->buf + mlen, c->bcnt - mlen);
		c->bcnt -= mlen;
	    } else {
		c->bcnt = 0;
		*c->buf = '\0';
		c->req = NULL;
		break;
	    }
	    c->req = NULL;
	    continue;
	}
	break;
    }
    return false;
}