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

static voidas_uv_auth_read(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf){	if (uv_is_closing((uv_handle_t*)stream)) {		return;	}	as_event_command* cmd = as_uv_auth_get_command(stream->data);			if (nread < 0) {		uv_read_stop(stream);		as_error err;		as_error_update(&err, AEROSPIKE_ERR_ASYNC_CONNECTION, "Authenticate socket read failed: %zd", nread);		as_event_socket_error(cmd, &err);		return;	}		cmd->pos += nread;		if (cmd->pos < cmd->len) {		// Read not finished.		return;	}		if (cmd->state == AS_ASYNC_STATE_AUTH_READ_HEADER) {		as_event_set_auth_parse_header(cmd);				if (cmd->len > cmd->capacity) {			uv_read_stop(stream);			as_error err;			as_error_update(&err, AEROSPIKE_ERR_CLIENT, "Authenticate response size is corrupt: %u", cmd->auth_len);			as_event_socket_error(cmd, &err);			return;		}		return;	}		// Done reading authentication data.	uv_read_stop(stream);		// Parse authentication response.	cmd->len -= cmd->auth_len;	uint8_t code = cmd->buf[cmd->len + AS_ASYNC_AUTH_RETURN_CODE];		if (code) {		// Can't authenticate socket, so must close it.		as_error err;		as_error_update(&err, code, "Authentication failed: %s", as_error_string(code));		as_event_socket_error(cmd, &err);		return;	}		cmd->pos = 0;	as_uv_command_write_start(cmd, stream);}

void read_done( uv_stream_t *handle, ssize_t nread, uv_buf_t buf ){    ctx_t *ctx = ( ctx_t *)handle->data;    sql_t *sql = ( sql_t *) malloc( sizeof( sql_t ) );    uv_write_t *wreq = ( uv_write_t *) malloc( sizeof( uv_write_t ) );    int n;    if ( nread < 0 ) { /* Error or EOF */        if ( buf.base ) {            free( buf.base );        }        uv_close( ( uv_handle_t *)&ctx->tcp, close_done );        free( wreq );        return;    }    if ( nread == 0 ) { /* Everything OK, but nothing read. */        free( buf.base );        uv_close( ( uv_handle_t *)&ctx->tcp, close_done );        free( wreq );        return;    }    sql->db = ctx->db;    sql->raw = buf.base;    sql->size = nread;    get_sql( sql );    free( buf.base );    nread = sql->size;    ctx->tcp_nread = nread;    uv_read_stop( ( uv_stream_t *)&ctx->tcp );    wreq->data = ctx;    buf.len =sql->size;    buf.base = sql->raw;    n = uv_write( wreq, ( uv_stream_t *)&ctx->tcp, &buf, 1, write_done );    free( buf.base );    free(sql);}

void httpconnection_on_read(uv_stream_t* handle, ssize_t nread, uv_buf_t buf) {  HTTPConnection* c = static_cast<HTTPConnection*>(handle->data);  if(nread < 0) {    int r = uv_read_stop(handle);    if(r) {      RX_ERROR("> error uv_read_stop: %s", uv_strerror(uv_last_error(handle->loop)));          }    if(buf.base) {      delete buf.base;      buf.base = NULL;    }    uv_err_t err = uv_last_error(handle->loop);    if(err.code != UV_EOF) {      RX_ERROR("> disconnected from server but not correctly: %s",uv_strerror(uv_last_error(handle->loop))) ;    }    r = uv_shutdown(&c->shutdown_req, handle, httpconnection_on_shutdown);    if(r) {      RX_ERROR("> error shutting down client: %s", uv_strerror(uv_last_error(handle->loop)));      RX_ERROR("@ todo should be `delete` the connection here?");    }    return;  }  c->addToInputBuffer(buf.base, nread);  if(buf.base) {    delete[] buf.base;    buf.base = NULL;  }}

// Callback invoked by libuv after it copies the data into the buffer provided// by `alloc_cb`. This is also called on EOF or when `alloc_cb` returns a// 0-length buffer.static void read_cb(uv_stream_t *stream, ssize_t cnt, const uv_buf_t *buf){    RStream *rstream = stream->data;    if (cnt <= 0) {        if (cnt != UV_ENOBUFS) {            // Read error or EOF, either way stop the stream and invoke the callback            // with eof == true            uv_read_stop(stream);            emit_read_event(rstream, true);        }        return;    }    // at this point we're sure that cnt is positive, no error occurred    size_t nread = (size_t) cnt;    // Data was already written, so all we need is to update 'wpos' to reflect    // the space actually used in the buffer.    rstream->wpos += nread;    if (rstream->wpos == rstream->buffer_size) {        // The last read filled the buffer, stop reading for now        rstream_stop(rstream);    }    rstream->reading = false;    emit_read_event(rstream, false);}

void tcp_connection::close(){	if (m_is_closing)		return;	int err;	m_is_closing = true;	// Don't read more.	err = uv_read_stop((uv_stream_t*)m_uv_handle);	if (err)		throw emu_fatalerror("uv_read_stop() failed: %s", uv_strerror(err));	// If there is no error and the peer didn't close its connection side then close gracefully.	if (!m_has_error && !m_is_closed_by_peer)	{		// Use uv_shutdown() so pending data to be written will be sent to the peer		// before closing.		uv_shutdown_t* req = new uv_shutdown_t;		req->data = (void*)this;		err = uv_shutdown(req, (uv_stream_t*)m_uv_handle, (uv_shutdown_cb)on_shutdown);		if (err)			throw emu_fatalerror("uv_shutdown() failed: %s", uv_strerror(err));	}	// Otherwise directly close the socket.	else	{		uv_close((uv_handle_t*)m_uv_handle, (uv_close_cb)on_close);	}}

int luvL_stream_stop(luv_object_t* self) {    if (luvL_object_is_started(self)) {        self->flags &= ~LUV_OSTARTED;        return uv_read_stop(&self->h.stream);    }    return 0;}

static voidremote_recv_cb(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf) {    struct remote_context *remote;    struct client_context *client;    remote = stream->data;    client = remote->client;    if (nread > 0) {        reset_timer(remote);        uv_read_stop(&remote->handle.stream);        int clen = nread + PRIMITIVE_BYTES;        uint8_t *c = remote->buf + HEADER_BYTES;        int rc = crypto_encrypt(c, (uint8_t*)buf->base, nread);        if (!rc) {            forward_to_client(client, c, clen);        } else {            logger_log(LOG_ERR, "invalid tcp packet");            close_client(client);            close_remote(remote);        }    } else if (nread < 0){        if (nread != UV_EOF) {            logger_log(LOG_ERR, "receive from %s failed: %s", client->target_addr, uv_strerror(nread));        }        close_client(client);        close_remote(remote);    }}

void TcpConnection::Close(){	MS_TRACE();	if (this->isClosing)		return;	int err;	this->isClosing = true;	// Don't read more.	err = uv_read_stop((uv_stream_t*)this->uvHandle);	if (err)		MS_ABORT("uv_read_stop() failed: %s", uv_strerror(err));	// If there is no error and the peer didn't close its connection side then close gracefully.	if (!this->hasError && !this->isClosedByPeer)	{		// Use uv_shutdown() so pending data to be written will be sent to the peer		// before closing.		uv_shutdown_t* req = new uv_shutdown_t;		req->data = (void*)this;		err = uv_shutdown(req, (uv_stream_t*)this->uvHandle, (uv_shutdown_cb)on_shutdown);		if (err)			MS_ABORT("uv_shutdown() failed: %s", uv_strerror(err));	}	// Otherwise directly close the socket.	else	{		uv_close((uv_handle_t*)this->uvHandle, (uv_close_cb)on_close);	}}

static voidhandle_http_header_parsed(Connection *conn){  http_request *req = &conn->req;  int content_length = req->content_length;  if (content_length < 0) {    content_length = 0;  }  conn->in_body = true;  bool keep_alive = http_request_is_keep_alive(req);  int num_bytes_in_buffer = conn->read_buffer.size() - req->body_start;  //  // make sure we complete reading in the body  // XXX: Handle Chunked-Encoding  //  if (content_length <= num_bytes_in_buffer) {    // we have read the body completely into the buffer    if (!keep_alive) {      int err = uv_read_stop((uv_stream_t*) conn);      assert(err == 0);    }    handle_http_request(conn);  }}

/** * This one is asynchronously triggered, so as to ensure we don't have any * silly re-entrancy issues. */static void socket_closing_cb(uv_idle_t *idle, int status){    my_sockdata_t *sock = idle->data;    uv_idle_stop(idle);    uv_close((uv_handle_t *)idle, generic_close_cb);    if (sock->pending.read) {        /**         * UV doesn't invoke read callbacks once the handle has been closed         * so we must track this ourselves.         */        lcb_assert(sock->pending.read == 1);        uv_read_stop((uv_stream_t *)&sock->tcp);        sock->pending.read--;        decref_sock(sock);    }#ifdef DEBUG    if (sock->pending.read || sock->pending.write) {        sock_dump_pending(sock);    }#endif    decref_sock(sock);    sock_do_uv_close(sock);    (void)status;}

static voiduv_memcached_get_on_read(uv_stream_t* stream, ssize_t nread, uv_buf_t buf){    uv_memcached_get_req_t* info;    uv_read_stop(stream);    info = (uv_memcached_get_req_t*) stream->data;    stream->data = info->connection;    uv_memcached_conn_pool_release_connection(info->connection);    size_t len = 0;    char   slen[16];    char   key[255];    sscanf(buf.base, "VALUE %s 0 %zu/r/n", key, &len);    sprintf(slen, "%zu", len);    char* data = calloc(len + 1, sizeof(char));    assert(data);    strncpy(data, buf.base + 6 + strlen(key) + 3 + strlen(slen) + 2, len);    data[len] = '/0';    if (len > 0) { // success        info->callback(info->client, 0, data, info->context);    } else {        info->callback(info->client, -1, NULL, info->context);    }    free(buf.base);    free(info);}

static int luv_read_stop(lua_State* L) {  uv_stream_t* handle = luv_check_stream(L, 1);  int ret = uv_read_stop(handle);  if (ret < 0) return luv_error(L, ret);  lua_pushinteger(L, ret);  return 1;}

static void read_cb(uv_stream_t *stream, ssize_t cnt, const uv_buf_t *buf){  // TODO(tarruda): avoid using a growable array for this, refactor the  // algorithm to call `ml_append` directly(skip unecessary copies/resizes)  int i;  ProcessData *pdata = (ProcessData *)stream->data;  if (cnt <= 0) {    if (cnt != UV_ENOBUFS) {      uv_read_stop(stream);      uv_close((uv_handle_t *)stream, NULL);      pdata->exited++;    }    return;  }  for (i = 0; i < cnt; ++i) {    if (pdata->rbuffer[i] == NL) {      // Insert the line      append_ga_line(&pdata->ga);    } else if (pdata->rbuffer[i] == NUL) {      // Translate NUL to NL      ga_append(&pdata->ga, NL);    } else {      // buffer data into the grow array      ga_append(&pdata->ga, pdata->rbuffer[i]);    }  }  windgoto(msg_row, msg_col);  cursor_on();  out_flush();  pdata->reading = false;}

static voidhandle_http_error(Connection *conn) {  // XXX: Write back 500 status code, whatever?  int err = uv_read_stop((uv_stream_t*) conn);  assert(err == 0);  uv_close((uv_handle_t*) conn, on_close_cb);}

void uv__stream_close(uv_stream_t* handle) {#if defined(__APPLE__)  /* Terminate select loop first */  if (handle->select != NULL) {    uv__stream_select_t* s;    s = handle->select;    uv_sem_post(&s->sem);    uv__stream_osx_interrupt_select(handle);    uv_thread_join(&s->thread);    uv_sem_destroy(&s->sem);    uv_mutex_destroy(&s->mutex);    close(s->fake_fd);    close(s->int_fd);    uv_close((uv_handle_t*) &s->async, uv__stream_osx_cb_close);    handle->select = NULL;  }#endif /* defined(__APPLE__) */  uv_read_stop(handle);  uv__io_close(handle->loop, &handle->io_watcher);  close(handle->io_watcher.fd);  handle->io_watcher.fd = -1;  if (handle->accepted_fd >= 0) {    close(handle->accepted_fd);    handle->accepted_fd = -1;  }  assert(!uv__io_active(&handle->io_watcher, UV__POLLIN | UV__POLLOUT));}

void recv_cb(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf){    void **ud_t = stream->data;    LOG_DEBUG("nread: %d %s, EOF_VALUE: %d", (int)nread, buf->base, UV_EOF);    if (nread <0)     {        LOG_DEBUG("error stop strerror: %s, err_name: %s", uv_strerror(nread), uv_err_name(nread));        uv_read_stop(stream);        RECV_CB cb_t = ud_t[0];        void *user_data_t = ud_t[1];        char * buff = ud_t[2];        int len = (int)ud_t[3];        int recved = (int)ud_t[4];        free(ud_t);        stream->data = 0;        cb_t(stream, nread, buff, recved, user_data_t);        return;    }    int tmp = (int)ud_t[4];    tmp += nread;    ud_t[4] = (void*)tmp;}

// connection_terminate: terminate an SSL connection and remove from// event loop. Clean up any allocated memory.void connection_terminate(uv_tcp_t *tcp){  connection_state *state = (connection_state *)tcp->data;  SSL *ssl = state->ssl;  int rc = SSL_shutdown(ssl);  if (rc == 0) {    SSL_shutdown(ssl);  }  rc = uv_read_stop((uv_stream_t *)tcp);  if (rc != 0) {    write_log(1, "Failed to stop TCP read: %s",               error_string(rc));  }  while (state->qr != state->qw) {    free(state->q[state->qr].start);    state->qr += 1;    if (state->qr == QUEUE_LENGTH) {      state->qr = 0;    }  }  *(state->prev) = state->next;  if (state->next) {    state->next->prev = state->prev;  }  uv_close((uv_handle_t *)tcp, close_cb);}

static void read_cb(uv_stream_t* handle,                    ssize_t nread,                    const uv_buf_t* buf) {  uv_pipe_t* p;  uv_pipe_t* inc;  uv_handle_type pending;  unsigned int i;  p = (uv_pipe_t*) handle;  ASSERT(nread >= 0);  while (uv_pipe_pending_count(p) != 0) {    pending = uv_pipe_pending_type(p);    ASSERT(pending == UV_NAMED_PIPE);    ASSERT(incoming_count < ARRAY_SIZE(incoming));    inc = &incoming[incoming_count++];    ASSERT(0 == uv_pipe_init(p->loop, inc, 0));    ASSERT(0 == uv_accept(handle, (uv_stream_t*) inc));  }  if (incoming_count != ARRAY_SIZE(incoming))    return;  ASSERT(0 == uv_read_stop((uv_stream_t*) p));  uv_close((uv_handle_t*) p, close_cb);  for (i = 0; i < ARRAY_SIZE(incoming); i++)    uv_close((uv_handle_t*) &incoming[i], close_cb);}

static voidclient_recv_cb(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf) {    struct client_context *client = stream->data;    struct remote_context *remote = client->remote;    int clen;    if (nread > 0) {        reset_timer(remote);        uv_read_stop(&client->handle.stream);        switch (client->stage) {        case XSTAGE_HANDSHAKE:            if (verify_methods(buf->base, nread)) {                handshake(client);            } else {                logger_log(LOG_ERR, "invalid method packet");                close_client(client);                close_remote(remote);            }            break;        case XSTAGE_REQUEST:            if (verify_request(buf->base, nread)) {                request_start(client, buf->base);            } else {                logger_log(LOG_ERR, "invalid request packet");                close_client(client);                close_remote(remote);            }            break;        case XSTAGE_FORWARD:            clen = nread + PRIMITIVE_BYTES;            uint8_t *c = client->buf + HEADER_BYTES;            int rc = crypto_encrypt(c, (uint8_t*)buf->base, nread);            if (rc) {                logger_log(LOG_ERR, "encrypt failed");                close_client(client);                close_remote(remote);            }            forward_to_remote(remote, c, clen);            break;        default:            break;        }    } else if (nread < 0) {        if (nread != UV_EOF) {            char addrbuf[INET6_ADDRSTRLEN + 1] = {0};            uint16_t port = ip_name(&client->addr, addrbuf, sizeof addrbuf);            logger_log(LOG_ERR, "receive from %s:%d failed: %s", addrbuf, port, uv_strerror(nread));        }        close_client(client);        close_remote(remote);    }}

void uv__stream_close(uv_stream_t* handle) {  unsigned int i;  uv__stream_queued_fds_t* queued_fds;  uv__io_close(handle->loop, &handle->io_watcher);  uv_read_stop(handle);  uv__handle_stop(handle);  if (handle->io_watcher.fd != -1) {    /* Don't close stdio file descriptors.  Nothing good comes from it. */    if (handle->io_watcher.fd > STDERR_FILENO)      uv__close(handle->io_watcher.fd);    handle->io_watcher.fd = -1;  }  if (handle->accepted_fd != -1) {    uv__close(handle->accepted_fd);    handle->accepted_fd = -1;  }  /* Close all queued fds */  if (handle->queued_fds != NULL) {    queued_fds = (uv__stream_queued_fds_t*)handle->queued_fds;    for (i = 0; i < queued_fds->offset; i++) {      uv__close(queued_fds->fds[i]);    }    free(handle->queued_fds);    handle->queued_fds = NULL;  }  assert(!uv__io_active(&handle->io_watcher, UV__POLLIN | UV__POLLOUT));}

bud_client_error_t bud_client_throttle(bud_client_t* client,                                       bud_client_side_t* side,                                       ringbuffer* buf) {  int err;  bud_client_side_t* opposite;  if (!ringbuffer_is_full(buf))    return bud_client_ok(side);  opposite = side == &client->frontend ? &client->backend : &client->frontend;  if (opposite->reading != kBudProgressRunning)    goto done;  DBG(opposite, "throttle, buffer full: %ld", ringbuffer_size(buf));  err = uv_read_stop((uv_stream_t*) &opposite->tcp);  if (err != 0) {    NOTICE(opposite,           "uv_read_stop failed: %d - /"%s/"",           err,           uv_strerror(err));    return bud_client_error(bud_error_num(kBudErrClientReadStop, err), side);  }  opposite->reading = kBudProgressNone;done:  return bud_client_error(bud_error(kBudErrClientThrottle), side);}

static void eof_timer_cb(uv_timer_t* timer, int status) {  uv_pipe_t* pipe = (uv_pipe_t*) timer->data;  uv_loop_t* loop = timer->loop;  assert(status == 0); /* timers can't fail */  assert(pipe->type == UV_NAMED_PIPE);  /* This should always be true, since we start the timer only */  /* in uv_pipe_queue_read after successfully calling ReadFile, */  /* or in uv_process_pipe_shutdown_req if a read is pending, */  /* and we always immediately stop the timer in */  /* uv_process_pipe_read_req. */  assert(pipe->flags & UV_HANDLE_READ_PENDING);  /* If there are many packets coming off the iocp then the timer callback */  /* may be called before the read request is coming off the queue. */  /* Therefore we check here if the read request has completed but will */  /* be processed later. */  if ((pipe->flags & UV_HANDLE_READ_PENDING) &&      HasOverlappedIoCompleted(&pipe->read_req.overlapped)) {    return;  }  /* Force both ends off the pipe. */  CloseHandle(pipe->handle);  pipe->handle = INVALID_HANDLE_VALUE;  /* Stop reading, so the pending read that is going to fail will */  /* not be reported to the user. */  uv_read_stop((uv_stream_t*) pipe);  /* Report the eof and update flags. This will get reported even if the */  /* user stopped reading in the meantime. TODO: is that okay? */  uv_pipe_read_eof(loop, pipe, uv_null_buf_);}

/* _http_conn_dead(): free http connection, close fd.*/static void_http_conn_dead(u3_hcon *hon_u){  // uL(fprintf(uH, "connection dead: %d/n", hon_u->coq_l));  uv_read_stop((uv_stream_t*) &(hon_u->wax_u));  uv_close((uv_handle_t*) &(hon_u->wax_u), _http_conn_free);}

/// Stops watching for events from a `RStream` instance.////// @param rstream The `RStream` instancevoid rstream_stop(RStream *rstream){  if (rstream->file_type == UV_FILE) {    uv_idle_stop(rstream->fread_idle);  } else {    uv_read_stop(rstream->stream);  }}

int io_stop_read(uv_handle_t *handle){	if (handle->type == UV_UDP) {		return uv_udp_recv_stop((uv_udp_t *)handle);	} else {		return uv_read_stop((uv_stream_t *)handle);	}}

void UvTcpServerStream::CloseClient(){	if (m_client.get())	{		uv_read_stop(reinterpret_cast<uv_stream_t*>(m_client.get()));		UvClose(std::move(m_client));	}}

static void timer_cb(uv_timer_t* handle, int status) {  uv_buf_t buf = uv_buf_init("PING", 4);  ASSERT(0 == uv_write(&write_req,                       (uv_stream_t*) &tcp_handle,                       &buf,                       1,                       write_cb));  ASSERT(0 == uv_read_stop((uv_stream_t*) &tcp_handle));}

static void uv_pipe_read_error(uv_pipe_t* handle, int error, uv_buf_t buf) {  /* If there is an eof timer running, we don't need it any more, */  /* so discard it. */  eof_timer_destroy(handle);  uv_read_stop((uv_stream_t*) handle);  uv_set_sys_error(error);  handle->read_cb((uv_stream_t*)handle, -1, buf);}

voidlcb_luv_read_stop(lcb_luv_socket_t sock){    if (sock->read.readhead_active == 0) {        return;    }    uv_read_stop((uv_stream_t*)&sock->tcp);    sock->read.readhead_active = 0;    lcb_luv_socket_unref(sock);}

static int couv_read_stop(lua_State *L) {  uv_stream_t *handle;  int r;  handle = couvL_checkudataclass(L, 1, COUV_STREAM_MTBL_NAME);  r = uv_read_stop(handle);  if (r < 0) {    luaL_error(L, couvL_uv_lasterrname(couv_loop(L)));  }  return 0;}