自学教程：C++ uv_read_start函数代码示例

H_CONNECTION KUVSocket::Accept(uv_stream_t* pListen){	H_CONNECTION hAccept = INVALID_HANDLER;	if (!pListen)	{		sprintf_s(g_ErrorMsg, "KUVSocket::Accept Fail, pListen is empty/n%s", g_ErrorMsg);		return hAccept;	}	ConnectionUserData* pData = (ConnectionUserData*)pListen->data;	if (!pData)	{		sprintf_s(g_ErrorMsg, "KUVSocket::Accept Fail, the user data of uv_listen is empty/n%s", g_ErrorMsg);		return hAccept;	}	H_CONNECTION handle = pData->Conn;	if (handle < 0 || handle > m_nIdGen)	{		sprintf_s(g_ErrorMsg, "KUVSocket::Accept Fail, the handle of uv_listen is empty/n%s", g_ErrorMsg);		return hAccept;	}	uv_tcp_t* pAccept = CreateSocket(hAccept);	if (!pAccept)	{		sprintf_s(g_ErrorMsg, "KUVSocket::Accept Fail, CreateSocket Error/n%s", g_ErrorMsg);		return hAccept;	}	m_nError = uv_accept(pListen, (uv_stream_t*)pAccept);	if (m_nError == 0)	{		fprintf_s(stdout, "/n Accpet some one connection");		this->OnAccepted(handle, hAccept, m_nError);		m_nError = uv_read_start((uv_stream_t*)pAccept, AllocReadBuffer, OnConnectionRead);	}	else 	{		this->OnAccepted(handle, INVALID_HANDLER, m_nError);		Close(hAccept);		hAccept = INVALID_HANDLER;	}	return hAccept;}

static voidon_connection(uv_stream_t* server, int status) {  uv_stream_t* stream;  int r;  if (status != 0) {    fprintf(stderr, "Connect error: %s: %s/n", uv_err_name(status), uv_strerror(status));    return;  }  stream = malloc(sizeof(uv_tcp_t));  if (stream == NULL) {    fprintf(stderr, "Allocate error: %s/n", strerror(errno));    return;  }  r = uv_tcp_init(loop, (uv_tcp_t*) stream);  if (r) {    fprintf(stderr, "Socket creation error: %s: %s/n", uv_err_name(r), uv_strerror(r));    return;  }  r = uv_tcp_simultaneous_accepts((uv_tcp_t*) stream, 1);  if (r) {    fprintf(stderr, "Flag error: %s: %s/n", uv_err_name(r), uv_strerror(r));    return;  }  r = uv_accept(server, stream);  if (r) {    fprintf(stderr, "Accept error: %s: %s/n", uv_err_name(r), uv_strerror(r));    return;  }  r = uv_tcp_nodelay((uv_tcp_t*) stream, 1);  if (r) {    fprintf(stderr, "Flag error: %s: %s/n", uv_err_name(r), uv_strerror(r));    return;  }  r = uv_read_start(stream, on_alloc, on_read);  if (r) {    fprintf(stderr, "Read error: %s: %s/n", uv_err_name(r), uv_strerror(r));    uv_close((uv_handle_t*) stream, on_close);  }}

void KUVSocket::OnConnectionConnected(uv_connect_t* req, int status){	uv_stream_t* pHandle = req->handle;	ConnectionUserData* pData = (ConnectionUserData*)req->handle->data;	KUVSocket* pSocket = pData->ServerSocket;	free(req);	req = NULL;	if (status != 0)	{		pSocket->OnConnected(pData->Conn, false);		pSocket->OnError(pData->Conn, status);		return;	}	uv_read_start((uv_stream_t*)pHandle, AllocReadBuffer, OnConnectionRead);	pSocket->OnConnected(pData->Conn, true);}

// read cb may fire multiple times.int lua_uv_stream_read_start(lua_State * L) {	uv_stream_t * s = lua_generic_object<uv_stream_t>(L, 1);	luaL_checktype(L, 2, LUA_TFUNCTION);		// bind ptr to callback function:	lua_pushlightuserdata(L, s);	lua_pushvalue(L, 2);	// func	lua_pushvalue(L, 1);	// stream	lua_pushcclosure(L, lua_uv_stream_read_cb_closure, 2);	// closure(func, timer)	//Lua(L).dump("read start"); 	// registry[s] = closure(func, stream)	lua_rawset(L, LUA_REGISTRYINDEX);		uv_read_start(s, lua_uv_alloc, lua_uv_stream_read_cb);	lua_uv_ok(L);	lua_settop(L, 1);	return 1;	// return stream}

/** * The callback function while server socket listening. It accepts connection * from client and reads request data. */static void on_connect(uv_stream_t *server, int status){    if (status == -1)        return;    uv_tcp_t *client = (uv_tcp_t *) malloc(sizeof(uv_tcp_t));    uv_tcp_init(uv_default_loop(), client);    if (uv_accept(server, (uv_stream_t *) client) == 0)    {        client->data = server->data;        uv_read_start((uv_stream_t *) client, alloc_buffer, on_request);    }    else    {        uv_close((uv_handle_t*) client, NULL);    }}

int ws_read_start(ws_connect_t *ptr, ws_cb_malloc cb_malloc, ws_cb_read cb_read, void *cb_read_obj){    if (WS_ONLINE == ptr->state)    {        ptr->cb.cb_malloc = cb_malloc;                ptr->cb.cb_read = cb_read;        ptr->cb.obj_read = cb_read_obj;                uv_read_start((uv_stream_t *)&ptr->connect, cb_read_malloc, cb_read_operation);    }	else    {        return 1;    }    	return 0;}

void durotanOnConnection(uv_stream_t* durotan, int status) {    if(status == -1) ERROR("durotan on connect", status);    client = (uv_tcp_t*)malloc(sizeof(uv_tcp_t));    uv_tcp_init(loop, client);    int err = uv_accept(durotan, (uv_stream_t*) client);    if(err) {        uv_close((uv_handle_t*) client, durotanOnClose);        ERROR("durotan accept", err);    }    // Buffer for incoming messages.    client->data = calloc(BUFFER_SIZE, sizeof(char));    printf("durotan: got connection/n");    uv_read_start((uv_stream_t*) client, durotanOnAllocate, durotanOnRead);}

static void connect_cb(uv_stream_t* listener, int status){	int n;	if (status) {		SHOW_UV_ERROR(listener->loop);		return;	}	server_ctx *ctx = calloc(1, sizeof(server_ctx));	ctx->handshake_buffer = calloc(1, HANDSHAKE_BUFFER_SIZE);	if (!ctx || !ctx->handshake_buffer)		FATAL("malloc() failed!");	ctx->client.data = ctx;	ctx->remote.data = ctx;		make_encryptor(&crypto, &ctx->encoder, 0, NULL);	n = uv_tcp_init(listener->loop, &ctx->client);	if (n)		SHOW_UV_ERROR_AND_EXIT(listener->loop);	n = uv_accept(listener, (uv_stream_t *)(void *)&ctx->client);	if (n)		SHOW_UV_ERROR_AND_EXIT(listener->loop);	n = uv_tcp_nodelay(&ctx->client, 1);	if (n)		SHOW_UV_ERROR_AND_EXIT(listener->loop);	#ifdef KEEPALIVE_TIMEOUT	n = uv_tcp_keepalive(&ctx->client, 1, KEEPALIVE_TIMEOUT);	if (n)		SHOW_UV_ERROR_AND_EXIT(listener->loop);	#endif /* KEEPALIVE_TIMEOUT */	n = uv_read_start((uv_stream_t *)(void *)&ctx->client, client_handshake_alloc_cb, client_handshake_read_cb);	if (n)		SHOW_UV_ERROR_AND_EXIT(listener->loop);	LOGCONN(&ctx->client, "Accepted connection from %s");}

static void connect_cb(uv_req_t* req, int status) {  uv_buf send_bufs[CHUNKS_PER_WRITE];  uv_handle_t* handle;  int i, j, r;  ASSERT(req != NULL);  ASSERT(status == 0);  handle = req->handle;  connect_cb_called++;  free(req);  /* Write a lot of data */  for (i = 0; i < WRITES; i++) {    for (j = 0; j < CHUNKS_PER_WRITE; j++) {      send_bufs[j].len = CHUNK_SIZE;      send_bufs[j].base = send_buffer + bytes_sent;      bytes_sent += CHUNK_SIZE;    }    req = (uv_req_t*)malloc(sizeof *req);    ASSERT(req != NULL);    uv_req_init(req, handle, write_cb);    r = uv_write(req, (uv_buf*)&send_bufs, CHUNKS_PER_WRITE);    ASSERT(r == 0);  }  /* Shutdown on drain. FIXME: dealloc req? */  req = (uv_req_t*) malloc(sizeof(uv_req_t));  ASSERT(req != NULL);  uv_req_init(req, handle, shutdown_cb);  r = uv_shutdown(req);  ASSERT(r == 0);  /* Start reading */  req = (uv_req_t*)malloc(sizeof *req);  ASSERT(req != NULL);  uv_req_init(req, handle, read_cb);  r = uv_read_start(handle, read_cb);  ASSERT(r == 0);}

static void connection_cb(uv_stream_t* stream, int status) {  conn_rec* conn;  int r;  ASSERT(status == 0);  ASSERT(stream == (uv_stream_t*)&tcp_server);  conn = malloc(sizeof *conn);  ASSERT(conn != NULL);  r = uv_tcp_init(stream->loop, &conn->handle);  ASSERT(r == 0);  r = uv_accept(stream, (uv_stream_t*)&conn->handle);  ASSERT(r == 0);  r = uv_read_start((uv_stream_t*)&conn->handle, alloc_cb, read_cb);  ASSERT(r == 0);}

int main(int argc, char **argv) {    loop = uv_default_loop();    uv_pipe_init(loop, &stdin_pipe, 0);    uv_pipe_open(&stdin_pipe, 0);    uv_pipe_init(loop, &stdout_pipe, 0);    uv_pipe_open(&stdout_pipe, 1);        uv_fs_t file_req;    int fd = uv_fs_open(loop, &file_req, argv[1], O_CREAT | O_RDWR, 0644, NULL);    uv_pipe_init(loop, &file_pipe, 0);    uv_pipe_open(&file_pipe, fd);    uv_read_start((uv_stream_t*)&stdin_pipe, alloc_buffer, read_stdin);    uv_run(loop, UV_RUN_DEFAULT);    return 0;}

static void connect_cb(uv_connect_t * req, int status){	int r;	uv_stream_t *stream;	stream = req->handle;#if USE_PIPE	printf("pipe connected..../n");#else	printf("tcp connected..../n");#endif	// do no write	//r = uv_write(&pmyWr->write_req, stream, &pmyWr->buf, 1, write_cb);	//printf("wr: r=%d/n",r);	/* Start reading */	r = uv_read_start(stream, alloc_cb, tcp_read_cb);}

static void connect_cb(uv_connect_t *req, int status) {  ASSERT(status == 0);  ASSERT(req == &connect_req);  /* Start reading from our connection so we can receive the EOF.  */  uv_read_start((uv_stream_t*)&tcp, alloc_cb, read_cb);  /*   * Write the letter 'Q' to gracefully kill the echo-server. This will not   * effect our connection.   */  uv_write(&write_req, (uv_stream_t*) &tcp, &qbuf, 1, NULL);  /* Shutdown our end of the connection.  */  uv_shutdown(&shutdown_req, (uv_stream_t*) &tcp, shutdown_cb);  called_connect_cb++;  ASSERT(called_shutdown_cb == 0);}

static void tcp_connect_cb(uv_connect_t *connect_req, int status){    ls_tcp_t *client = containerof(connect_req->handle, ls_tcp_t, handle);    uv_tcp_t *handle = &client->handle;    lua_State *l = ls_default_state();    lua_State *nl;    uv_loop_t *loop = uv_default_loop();    ls_free(l, connect_req);    if (ls_object_is_waited(&client->wait_object))    {        int ref = client->wait_object.mthread_ref;        client->wait_object.mthread_ref = LUA_NOREF;        ls_getref(l, ref);        nl = lua_tothread(l, -1);        lua_pop(l, 1);        if (nl)        {            ls_clear_waiting(nl);            if (status)            {                uv_close((uv_handle_t*)handle, tcp_close_cb);                ls_last_error_resume(nl, loop);            }            else            {                if (uv_read_start((uv_stream_t*)handle, tcp_alloc_cb, tcp_read_cb))                {                    uv_close((uv_handle_t*)handle, tcp_close_cb);                    ls_last_error_resume(nl, loop);                }                else                {                    lua_pushboolean(nl, 1);                    new_tcp_connection_udata(nl, client);                    ls_resume(nl, 2);                }            }        }        ls_unref(l, ref);    }}

void tcp_new_connection_cb(uv_stream_t* uv_server, int status) {  if (uv_server != (uv_stream_t*)&_uv_server) return;  if (status != 0) return;  printf("connected/n");  client_t* client = malloc(sizeof(client_t));    uv_tcp_init(_uv_loop, &client->handle);  printf("%p/n", client);  int error = uv_accept((uv_stream_t*)&_uv_server, (uv_stream_t*)&client->handle);  if (error == 0) {    client->handle.data = client;        uv_read_start((uv_stream_t*)&client->handle, alloc_buffer_cb, tcp_read_cb);  } else {    uv_close((uv_handle_t*)&client->handle, tcp_close_cb);  }}

void TCPClient::ConnectCallback(uv_connect_t * connect_request, int status){    auto tcp_client = (TCPClient *) connect_request->data;    if (status < 0)    {        LogError(LOG_NETWORKING, "Connecting to server - failure, reason: (%s)", uv_err_name(status));        NetworkManager::_state = NetworkState::Disconnected;    }    else    {        LogTrace(LOG_NETWORKING, "Connecting to server - success");        NetworkManager::_state = NetworkState::Connected;        uv_read_start((uv_stream_t *) tcp_client, AllocBuffer, IncomingData);    }    free(connect_request);}

static void uv_on_tcp_accept(uv_stream_t *tcpacc, int status){	debug("%s(status=%d)/n", __FUNCTION__, status);	int ret;	outlet_t *ol = tcpacc->data;	assert(ol);	if (status < 0)		return;	acc_pend_t *pend = get_free_pending(ol);	if (pend == 0)		return;  // fatal error?	uv_tcp_t *conn = malloc(sizeof(uv_tcp_t));	if (!conn)		goto pend_cleanup;	ret = uv_tcp_init(uv_default_loop(), conn);	if (ret)		goto pend_cleanup;	conn->data = pend;	ret = uv_accept(tcpacc, (uv_stream_t *)conn);	if (ret)		goto pend_cleanup;	ret = uv_read_start((uv_stream_t *)conn, on_alloc, uv_on_tcpacc_recv);	if (ret)		goto pend_cleanup;	pend->tcp = conn;	try_to_bake(ol, pend);	return;pend_cleanup:	if (conn)		uv_close((uv_handle_t *)conn, NULL);	reuse_pending(&ol->free_pends, pend);	return;}

void conn_done( uv_stream_t *listener, int status ){    int n;    ctx_t *ctx = ( ctx_t *) malloc( sizeof( ctx_t ) ); //创建实例，返回实例指针    ctx->tcp.data = ctx;    ctx->isChunked = 0;    ctx->db = ( sqlite3 *)listener->data;    n = uv_tcp_init( listener->loop, &ctx->tcp );    if ( n ) {        return;    }    n = uv_accept( listener, ( uv_stream_t *)&ctx->tcp );    if ( n ) {        return;    }    n = uv_read_start( ( uv_stream_t *)&ctx->tcp, alloc_buffer, read_done );    if ( n ) {        return;    }}

/** * the static callback called by the C-level routines in uv. it's main job is to call a C++ level callback to do the work with higher layers * * from uv.h * typedef void (*uv_connect_cb)(uv_connect_t* req, int status); */voidUVEventLoop::OnConnect(uv_connect_t *req, int status){	DEBUG_OUT("OnConnect() ... " << status);	UVReader* ocCBp=nullptr;	if (req) {		ocCBp=static_cast<UVReader*>(req->data);	}	if (ocCBp) {		ocCBp->client->socket->data = ocCBp;		if (status >= 0) {			status = uv_read_start((uv_stream_t*)ocCBp->client->socket, AllocBuffer, OnRead);		}		if (ocCBp->connectCB) {			(*ocCBp->connectCB)(req, status);		}//		ocCBp->client->socket->data = ocCBp->readerCB;//		ocCBp->disposed = true;	}}

static void timer_cb(uv_timer_t* handle, int status) {  int r;  ASSERT(handle == &timer);  ASSERT(status == 0);  ASSERT(nested == 0 && "timer_cb must be called from a fresh stack");  puts("Timeout complete. Now read data...");  nested++;  if (uv_read_start((uv_stream_t*)&client, alloc_cb, read_cb)) {    FATAL("uv_read_start failed");  }  nested--;  timer_cb_called++;  r = uv_close((uv_handle_t*)handle, close_cb);  ASSERT(r == 0);}

static void connection_cb(uv_handle_t* s, int status) {  uv_tcp_t* tcp;  int r;  ASSERT(&server == (uv_tcp_t*)s);  ASSERT(status == 0);  tcp = malloc(sizeof(uv_tcp_t));  uv_tcp_init(tcp);  r = uv_accept(s, (uv_stream_t*)tcp);  ASSERT(r == 0);  r = uv_read_start((uv_stream_t*)tcp, buf_alloc, read_cb);  ASSERT(r == 0);  read_sockets++;  max_read_sockets++;}

static int start_read(lcb_io_opt_t iobase,                      lcb_sockdata_t *sockbase,                      lcb_io_read_cb callback){    my_sockdata_t *sock = (my_sockdata_t *)sockbase;    my_iops_t *io = (my_iops_t *)iobase;    int ret;    sock->cur_iov = 0;    sock->tcp.callback = callback;    ret = uv_read_start((uv_stream_t *)&sock->tcp.t, alloc_cb, read_cb);    set_last_error(io, ret);    if (ret == 0) {        SOCK_INCR_PENDING(sock, read);        incref_sock(sock);    }    return ret;}

int main (int argc, char *argv[]){   UNUSED (argc);   UNUSED (argv);   uv_loop_t *loop = uv_default_loop ();#ifdef HF_BASE_APP   HF::UID::URI *uid = new HF::UID::URI ("hf://base.example.com");#endif#ifdef HF_NODE_APP   if (argc < 2)   {      std::cout << "usage : " << argv[0] << " [number]" << std::endl;      exit (-1);   }   std::stringstream ss;   ss << "hf://node.example.com/" << argv[1];   HF::UID::URI *uid = new HF::UID::URI (ss.str ());#endif   HF::Application::Initialize (transport);   transport.uid (uid);   HF::Application::Handle ("?");   uv_pipe_init (loop, &stdin_pipe, 0);   uv_pipe_open (&stdin_pipe, 0);   uv_read_start ((uv_stream_t *) &stdin_pipe, alloc_buffer, read_stdin);   uv_run (loop, UV_RUN_DEFAULT);   HF::Application::Save ();   return 0;}

    void client::onConnect(uv_connect_t* conn_req_, int status_)    {        if (status_ == -1)        {            uv_close((uv_handle_t*)conn_req_->handle, onClose);            throw std::runtime_error(uv_err_name(status_));        }        data_t* data = (data_t*)conn_req_->data;        uv_write_t* req;        uv_buf_t buf;        int r;        std::string request = data->req->toString();        buf = uv_buf_init((char*)request.c_str(), request.length());        req = (uv_write_t*)malloc(sizeof *req);        if (req == NULL)        {            throw std::runtime_error("Error malloc.");        }        r = uv_write(req, conn_req_->handle, &buf, 1, onWrite);        if (r)        {            throw std::runtime_error(uv_err_name(r));        }        conn_req_->handle->data = data;        r = uv_read_start(conn_req_->handle, onAlloc, onRead);        if (r)        {            throw std::runtime_error(uv_err_name(r));        }    }

static void _uv_on_connect(uv_connect_t* conn, int status) {    uvx_client_t* xclient = (uvx_client_t*) conn->data;    xclient->uvclient.data = xclient;	if(status == 0) {		if(xclient->config.log_out)            fprintf(xclient->config.log_out, "[uvx-client] %s connect to server ok/n", xclient->config.name);		assert(conn->handle == (uv_stream_t*) &xclient->uvclient);		xclient->uvserver = (uv_tcp_t*) conn->handle;        if(xclient->config.on_conn_ok)            xclient->config.on_conn_ok(xclient);		uv_read_start(conn->handle, uvx__on_alloc_buf, uvx__on_client_read);	} else {		xclient->uvserver = NULL;		if(xclient->config.log_err)            fprintf(xclient->config.log_err, "/n!!! [uvx-client] %s connect to server failed: %s/n", xclient->config.name, uv_strerror(status));        if(xclient->config.on_conn_fail)            xclient->config.on_conn_fail(xclient);		_uvx_client_close(xclient); // will try reconnect on next on_heartbeat	}}

static void on_connect(uv_stream_t *server, int status) {  int r;  CHECK(status, "connecting");  debug("connecting req");  // the tcp handle points to our sws_handle_req_t which can store a bit extra info  sws_handle_req_t *handle_req = malloc(sizeof(sws_handle_req_t));  sws_handle_req_init(handle_req, id++);  r = uv_accept(server, (uv_stream_t*) &handle_req->handle);  if (r) {    log_err("error accepting connection %d", r);    uv_close((uv_handle_t*) handle_req, NULL);  } else {    // read the req into the tcp socket to cause it to get parsed    // once the headers are in we'll get called back the first time (see on_headers_complete)    // for now we assume no body since this is just a static webserver    uv_read_start((uv_stream_t*) handle_req, alloc_cb, on_req_read);  }}

void on_new_connection(uv_stream_t *server, int status){	uv_tcp_t *client;	if (status < 0)	{		fprintf(stderr, "New connection error %s/n", uv_strerror(status));		return;	}	client = (uv_tcp_t*)malloc(sizeof(uv_tcp_t));	uv_tcp_init(loop, client);	if (uv_accept(server, (uv_stream_t*)client) == 0)	{		uv_read_start((uv_stream_t*)client, alloc_buffer, echo_read);	}	else	{		uv_close((uv_handle_t*)client, NULL);	}}

voidon_connection(uv_stream_t *server, int status){     uv_tcp_t *client = malloc(sizeof(uv_tcp_t));     if (status == -1) {	  /* error */     }     uv_tcp_init(loop, client);     if (uv_accept(server, (uv_stream_t *)client) == 0) {	  int r = uv_read_start((uv_stream_t *)client, alloc_buffer, on_read);	  if (r) {	       /* error */	  }     } else {	  uv_close((uv_handle_t *)client, NULL);     }}

static void on_connect(ws_connect_t *ptr){	//print_info();	ptr->state = WS_CONNECT;	uv_read_start((uv_stream_t *)&ptr->connect, cb_pre_malloc, cb_read_over);	if(WS_CLIENT == ptr->type)	{		// step1 websocket握手		ptr->write_buf[0] = uv_buf_init(str_client_req, strlen(str_client_req));		uv_write(&(ptr->write_req[0]), (uv_stream_t*)&ptr->connect, ptr->write_buf, 1, cb_write);	}    else    {            }}

int async_read(uv_stream_t *const stream, size_t const size, uv_buf_t *const out) {	if(!stream) return UV_EINVAL;	if(!out) return UV_EINVAL;	async_state state[1];	state->thread = async_active();	state->size = size;	state->status = 0;	*state->buf = uv_buf_init(NULL, 0);	stream->data = state;	int rc = uv_read_start(stream, alloc_cb, read_cb);	if(rc < 0) return rc;	rc = async_yield_cancelable();	uv_read_stop(stream);	if(rc < 0) {		free(state->buf->base);		return rc;	}	out->base = state->buf->base;	out->len = state->buf->len;	return state->status;}