如何使用 Libevent 检测客户端连接到服务器
How to detect a client gets connected to the server using Libevent
我正在编写使用 libevent 设置本地主机服务器和本地主机客户端的程序。我需要知道客户端何时连接到服务器。 libevent为事件回调提供了六种事件信号，分别是BEV_EVENT_READING、BEV_EVENT_WRITING、BEV_EVENT_ERROR、BEV_EVENT_TIMEOUT、BEV_EVENT_EOF和BEV_EVENT_CONNECTED。基本上我想让客户端进入事件循环并继续等待服务器在线。如果我只是 运行 没有服务器在线的客户端程序，我有时会同时获得 BEV_EVENT_READING 和 BEV_EVENT_CONNECTED 标志。 BEV_EVENT_READING 的错误是 connection refused，这是有道理的，因为服务器处于脱机状态。然而，在这种情况下，标志 BEV_EVENT_CONNECTED 也被设置，即使我调度它，程序也会立即离开。下面是代码片段
address.sin_family = AF_INET;
address.sin_addr.s_addr = inet_addr("127.0.0.1");
address.sin_port = htons(7777);
buffev_ptr = bufferevent_socket_new(_evbase_ptr, -1, BEV_OPT_CLOSE_ON_FREE);
bufferevent_setcb(buffev_ptr, myread, mywrite, mysignal, NULL);
bufferevent_enable(buffev_ptr, EV_READ | EV_WRITE);
bufferevent_socket_connect(buffev_ptr, (sockaddr_pt)&address, sizeof(address));
那么mysignal这个回调，到底应该怎么写才能应对client连接上的情况呢？
void_t Client::mysignal(buffev_pt bev, short flag, void_pt user_data) {
// An event occurred during a read operation on the bufferevent.
if(flag & BEV_EVENT_READING) {
}
// An event occurred during a write operation on the bufferevent.
if(flag & BEV_EVENT_WRITING) {
}
// An error occurred during a bufferevent operation.
if(flag & BEV_EVENT_ERROR) {
}
// A timeout expired on the bufferevent.
if(flag & BEV_EVENT_TIMEOUT) {
}
// We got an end-of-line indication on the bufferevent.
if(flag & BEV_EVENT_EOF) {
}
// We finished a requested connection on the bufferevent.
if(flag & BEV_EVENT_CONNECTED) {
}
}
如果我运行客户端程序没有服务器在线，BEV_EVENT_CONNECTED和BEV_EVENT_ERROR可能会同时发生。我假设 BEV_EVENT_CONNECTED 只会在客户端连接到服务器时设置，但似乎不会。
这是一个很好的示例页面：http://www.wangafu.net/~nickm/libevent-book/Ref6_bufferevent.html
一些事情：bufferevent_socket_connect 可能会立即失败 [旁注：套接字是非阻塞的]。那没有正确处理。 有可能同时获得 CONNECT 和 ERROR。我已经在您的代码中添加了一些内容并对其进行了注释[请原谅不必要的样式清理]。
您的代码只是处理事件而没有跟踪任何 "state" 信息，因此我还添加了一个 "state" 结构，您可以使用它来确定如何处理传入的事件确定连接是否发生以及它是否可用。远非完美，但可能有所帮助：
// connection state control
struct privdata {
u32 priv_evseen; // events previously seen
struct trace_buffer *priv_trace; // trace buffer to aid debug (e.g.)
...
};
int
start_connection(...)
{
struct privdata *data;
int err;
address.sin_family = AF_INET;
address.sin_addr.s_addr = inet_addr("127.0.0.1");
address.sin_port = htons(7777);
buffev_ptr = bufferevent_socket_new(_evbase_ptr, -1, BEV_OPT_CLOSE_ON_FREE);
data = calloc(1,sizeof(struct privdata));
bufferevent_setcb(buffev_ptr, myread, mywrite, mysignal, data);
bufferevent_enable(buffev_ptr, EV_READ | EV_WRITE);
// NOTE: this can fail immediately
err = bufferevent_socket_connect(buffev_ptr,
(sockaddr_pt) &address,sizeof(address));
// NOTE: if it fails above, the connection attempt will _not_ be retried by
// libevent -- we _must_ tear down and recreate from scratch (i.e. libevent
// has already released the socket it tried to connect with)
//
// we'll have to decide how/when/if we want to reenter this function to
// try again (e.g. server is offline and we requeue externally to call us
// [say] every 5 minutes, with message "server offline -- will retry in
// 5 minutes")
if (err < 0) {
err = errno;
event_base_free(_evbase_ptr);
free(data);
errno = err;
return -1;
}
event_base_dispatch(base);
return 0;
}
void_t
Client::mysignal(buffev_pt bev, short flag, void_pt user_data)
{
struct privdata *priv;
priv = user_data;
// NOTE: it _is_ possible to get an event that has _both_ CONNNECTED and
// ERROR because when the connection occurs, libevent issues a getsockopt
// to check for errors
//
// we can use our private data to determine if an ERROR occurs at the
// connection point (e.g. priv_evseen does _not_ have CONNECTED set) or
// later during normal socket operation
do {
// wait for connection
// NOTE: this should also handle other errors like timeout
if ((priv->priv_evseen & CONNECTED) == 0) {
switch (flag & (ERROR | CONNECTED)) {
case CONNECTED: // got clean connection -- puppy is happy!
break;
case ERROR: // got error first -- CONNECTED event unlikely
break;
case (CONNECTED | ERROR): // connected with error -- how bad is it?
break;
case 0: // huh? -- the only excuse ...
if (flag & TIMEOUT)
...
break;
}
break;
}
// events that occur during normal operation ...
// An error occurred during a bufferevent operation.
if (flag & ERROR) {
}
// A timeout expired on the bufferevent.
if (flag & TIMEOUT) {
}
// We got an end-of-line indication on the bufferevent.
if (flag & EOF) {
}
// We finished a requested connection on the bufferevent.
// NOTE: this should _never_ happen now
if (flag & CONNECTED) {
}
// An event occurred during a read operation on the bufferevent.
if (flag & READING) {
}
// An event occurred during a write operation on the bufferevent.
if (flag & WRITING) {
}
} while (0);
// remember types we've seen before
priv->priv_seen |= flag;
}
以下是一些相关的 libevent 来源：
int
bufferevent_socket_connect(struct bufferevent *bev, struct sockaddr *sa, int socklen)
{
struct bufferevent_private *bufev_p = EVUTIL_UPCAST(bev, struct bufferevent_private, bev);
evutil_socket_t fd;
int r = 0;
int result = -1;
int ownfd = 0;
_bufferevent_incref_and_lock(bev);
if (!bufev_p)
goto done;
fd = bufferevent_getfd(bev);
if (fd < 0) {
if (!sa)
goto done;
fd = socket(sa->sa_family, SOCK_STREAM, 0);
if (fd < 0)
goto done;
if (evutil_make_socket_nonblocking(fd) < 0)
goto done;
ownfd = 1;
}
if (sa) {
#ifdef WIN32
if (bufferevent_async_can_connect(bev)) {
bufferevent_setfd(bev, fd);
r = bufferevent_async_connect(bev, fd, sa, socklen);
if (r < 0)
goto freesock;
bufev_p->connecting = 1;
result = 0;
goto done;
}
else
#endif
r = evutil_socket_connect(&fd, sa, socklen);
if (r < 0)
goto freesock;
}
#ifdef WIN32
/* ConnectEx() isn't always around, even when IOCP is enabled. Here, we borrow the socket object's write handler to fall back on a non-blocking connect() when ConnectEx() is unavailable. */
if (BEV_IS_ASYNC(bev)) {
event_assign(&bev->ev_write, bev->ev_base, fd, EV_WRITE | EV_PERSIST, bufferevent_writecb, bev);
}
#endif
bufferevent_setfd(bev, fd);
if (r == 0) {
if (!be_socket_enable(bev, EV_WRITE)) {
bufev_p->connecting = 1;
result = 0;
goto done;
}
}
else if (r == 1) {
/* The connect succeeded already. How very BSD of it. */
result = 0;
bufev_p->connecting = 1;
event_active(&bev->ev_write, EV_WRITE, 1);
}
else {
/* The connect failed already. How very BSD of it. */
bufev_p->connection_refused = 1;
bufev_p->connecting = 1;
result = 0;
event_active(&bev->ev_write, EV_WRITE, 1);
}
goto done;
freesock:
_bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);
if (ownfd)
evutil_closesocket(fd);
/* do something about the error? */
done:
_bufferevent_decref_and_unlock(bev);
return result;
}
/* XXX we should use an enum here. */
/* 2 for connection refused, 1 for connected, 0 for not yet, -1 for error. */
int
evutil_socket_connect(evutil_socket_t * fd_ptr, struct sockaddr *sa, int socklen)
{
int made_fd = 0;
if (*fd_ptr < 0) {
if ((*fd_ptr = socket(sa->sa_family, SOCK_STREAM, 0)) < 0)
goto err;
made_fd = 1;
if (evutil_make_socket_nonblocking(*fd_ptr) < 0) {
goto err;
}
}
if (connect(*fd_ptr, sa, socklen) < 0) {
int e = evutil_socket_geterror(*fd_ptr);
if (EVUTIL_ERR_CONNECT_RETRIABLE(e))
return 0;
if (EVUTIL_ERR_CONNECT_REFUSED(e))
return 2;
goto err;
}
else {
return 1;
}
err:
if (made_fd) {
evutil_closesocket(*fd_ptr);
*fd_ptr = -1;
}
return -1;
}
/* Check whether a socket on which we called connect() is done
connecting. Return 1 for connected, 0 for not yet, -1 for error. In the
error case, set the current socket errno to the error that happened during
the connect operation. */
int
evutil_socket_finished_connecting(evutil_socket_t fd)
{
int e;
ev_socklen_t elen = sizeof(e);
if (getsockopt(fd, SOL_SOCKET, SO_ERROR, (void *) &e, &elen) < 0)
return -1;
if (e) {
if (EVUTIL_ERR_CONNECT_RETRIABLE(e))
return 0;
EVUTIL_SET_SOCKET_ERROR(e);
return -1;
}
return 1;
}
struct bufferevent *
bufferevent_socket_new(struct event_base *base, evutil_socket_t fd, int options)
{
struct bufferevent_private *bufev_p;
struct bufferevent *bufev;
#ifdef WIN32
if (base && event_base_get_iocp(base))
return bufferevent_async_new(base, fd, options);
#endif
if ((bufev_p = mm_calloc(1, sizeof(struct bufferevent_private))) == NULL)
return NULL;
if (bufferevent_init_common(bufev_p, base, &bufferevent_ops_socket, options) < 0) {
mm_free(bufev_p);
return NULL;
}
bufev = &bufev_p->bev;
evbuffer_set_flags(bufev->output, EVBUFFER_FLAG_DRAINS_TO_FD);
event_assign(&bufev->ev_read, bufev->ev_base, fd, EV_READ | EV_PERSIST, bufferevent_readcb, bufev);
event_assign(&bufev->ev_write, bufev->ev_base, fd, EV_WRITE | EV_PERSIST, bufferevent_writecb, bufev);
evbuffer_add_cb(bufev->output, bufferevent_socket_outbuf_cb, bufev);
evbuffer_freeze(bufev->input, 0);
evbuffer_freeze(bufev->output, 1);
return bufev;
}
static void
bufferevent_writecb(evutil_socket_t fd, short event, void *arg)
{
struct bufferevent *bufev = arg;
struct bufferevent_private *bufev_p = EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
int res = 0;
short what = BEV_EVENT_WRITING;
int connected = 0;
ev_ssize_t atmost = -1;
_bufferevent_incref_and_lock(bufev);
if (event == EV_TIMEOUT) {
/* Note that we only check for event==EV_TIMEOUT. If event==EV_TIMEOUT|EV_WRITE, we can safely ignore the timeout, since a read has occurred */
what |= BEV_EVENT_TIMEOUT;
goto error;
}
if (bufev_p->connecting) {
int c = evutil_socket_finished_connecting(fd);
/* we need to fake the error if the connection was refused immediately - usually connection to localhost on BSD */
if (bufev_p->connection_refused) {
bufev_p->connection_refused = 0;
c = -1;
}
if (c == 0)
goto done;
bufev_p->connecting = 0;
if (c < 0) {
event_del(&bufev->ev_write);
event_del(&bufev->ev_read);
_bufferevent_run_eventcb(bufev, BEV_EVENT_ERROR);
goto done;
}
else {
connected = 1;
#ifdef WIN32
if (BEV_IS_ASYNC(bufev)) {
event_del(&bufev->ev_write);
bufferevent_async_set_connected(bufev);
_bufferevent_run_eventcb(bufev, BEV_EVENT_CONNECTED);
goto done;
}
#endif
_bufferevent_run_eventcb(bufev, BEV_EVENT_CONNECTED);
if (!(bufev->enabled & EV_WRITE) || bufev_p->write_suspended) {
event_del(&bufev->ev_write);
goto done;
}
}
}
atmost = _bufferevent_get_write_max(bufev_p);
if (bufev_p->write_suspended)
goto done;
if (evbuffer_get_length(bufev->output)) {
evbuffer_unfreeze(bufev->output, 1);
res = evbuffer_write_atmost(bufev->output, fd, atmost);
evbuffer_freeze(bufev->output, 1);
if (res == -1) {
int err = evutil_socket_geterror(fd);
if (EVUTIL_ERR_RW_RETRIABLE(err))
goto reschedule;
what |= BEV_EVENT_ERROR;
}
else if (res == 0) {
/* eof case XXXX Actually, a 0 on write doesn't indicate an EOF. An ECONNRESET might be more typical. */
what |= BEV_EVENT_EOF;
}
if (res <= 0)
goto error;
_bufferevent_decrement_write_buckets(bufev_p, res);
}
if (evbuffer_get_length(bufev->output) == 0) {
event_del(&bufev->ev_write);
}
/*
* Invoke the user callback if our buffer is drained or below the
* low watermark.
*/
if ((res || !connected) && evbuffer_get_length(bufev->output) <= bufev->wm_write.low) {
_bufferevent_run_writecb(bufev);
}
goto done;
reschedule:
if (evbuffer_get_length(bufev->output) == 0) {
event_del(&bufev->ev_write);
}
goto done;
error:
bufferevent_disable(bufev, EV_WRITE);
_bufferevent_run_eventcb(bufev, what);
done:
_bufferevent_decref_and_unlock(bufev);
}
我正在编写使用 libevent 设置本地主机服务器和本地主机客户端的程序。我需要知道客户端何时连接到服务器。 libevent为事件回调提供了六种事件信号，分别是BEV_EVENT_READING、BEV_EVENT_WRITING、BEV_EVENT_ERROR、BEV_EVENT_TIMEOUT、BEV_EVENT_EOF和BEV_EVENT_CONNECTED。基本上我想让客户端进入事件循环并继续等待服务器在线。如果我只是 运行 没有服务器在线的客户端程序，我有时会同时获得 BEV_EVENT_READING 和 BEV_EVENT_CONNECTED 标志。 BEV_EVENT_READING 的错误是 connection refused，这是有道理的，因为服务器处于脱机状态。然而，在这种情况下，标志 BEV_EVENT_CONNECTED 也被设置，即使我调度它，程序也会立即离开。下面是代码片段
address.sin_family = AF_INET;
address.sin_addr.s_addr = inet_addr("127.0.0.1");
address.sin_port = htons(7777);
buffev_ptr = bufferevent_socket_new(_evbase_ptr, -1, BEV_OPT_CLOSE_ON_FREE);
bufferevent_setcb(buffev_ptr, myread, mywrite, mysignal, NULL);
bufferevent_enable(buffev_ptr, EV_READ | EV_WRITE);
bufferevent_socket_connect(buffev_ptr, (sockaddr_pt)&address, sizeof(address));
那么mysignal这个回调，到底应该怎么写才能应对client连接上的情况呢？
void_t Client::mysignal(buffev_pt bev, short flag, void_pt user_data) {
// An event occurred during a read operation on the bufferevent.
if(flag & BEV_EVENT_READING) {
}
// An event occurred during a write operation on the bufferevent.
if(flag & BEV_EVENT_WRITING) {
}
// An error occurred during a bufferevent operation.
if(flag & BEV_EVENT_ERROR) {
}
// A timeout expired on the bufferevent.
if(flag & BEV_EVENT_TIMEOUT) {
}
// We got an end-of-line indication on the bufferevent.
if(flag & BEV_EVENT_EOF) {
}
// We finished a requested connection on the bufferevent.
if(flag & BEV_EVENT_CONNECTED) {
}
}
如果我运行客户端程序没有服务器在线，BEV_EVENT_CONNECTED和BEV_EVENT_ERROR可能会同时发生。我假设 BEV_EVENT_CONNECTED 只会在客户端连接到服务器时设置，但似乎不会。
这是一个很好的示例页面：http://www.wangafu.net/~nickm/libevent-book/Ref6_bufferevent.html
一些事情：bufferevent_socket_connect 可能会立即失败 [旁注：套接字是非阻塞的]。那没有正确处理。 有可能同时获得 CONNECT 和 ERROR。我已经在您的代码中添加了一些内容并对其进行了注释[请原谅不必要的样式清理]。
您的代码只是处理事件而没有跟踪任何 "state" 信息，因此我还添加了一个 "state" 结构，您可以使用它来确定如何处理传入的事件确定连接是否发生以及它是否可用。远非完美，但可能有所帮助：
// connection state control
struct privdata {
u32 priv_evseen; // events previously seen
struct trace_buffer *priv_trace; // trace buffer to aid debug (e.g.)
...
};
int
start_connection(...)
{
struct privdata *data;
int err;
address.sin_family = AF_INET;
address.sin_addr.s_addr = inet_addr("127.0.0.1");
address.sin_port = htons(7777);
buffev_ptr = bufferevent_socket_new(_evbase_ptr, -1, BEV_OPT_CLOSE_ON_FREE);
data = calloc(1,sizeof(struct privdata));
bufferevent_setcb(buffev_ptr, myread, mywrite, mysignal, data);
bufferevent_enable(buffev_ptr, EV_READ | EV_WRITE);
// NOTE: this can fail immediately
err = bufferevent_socket_connect(buffev_ptr,
(sockaddr_pt) &address,sizeof(address));
// NOTE: if it fails above, the connection attempt will _not_ be retried by
// libevent -- we _must_ tear down and recreate from scratch (i.e. libevent
// has already released the socket it tried to connect with)
//
// we'll have to decide how/when/if we want to reenter this function to
// try again (e.g. server is offline and we requeue externally to call us
// [say] every 5 minutes, with message "server offline -- will retry in
// 5 minutes")
if (err < 0) {
err = errno;
event_base_free(_evbase_ptr);
free(data);
errno = err;
return -1;
}
event_base_dispatch(base);
return 0;
}
void_t
Client::mysignal(buffev_pt bev, short flag, void_pt user_data)
{
struct privdata *priv;
priv = user_data;
// NOTE: it _is_ possible to get an event that has _both_ CONNNECTED and
// ERROR because when the connection occurs, libevent issues a getsockopt
// to check for errors
//
// we can use our private data to determine if an ERROR occurs at the
// connection point (e.g. priv_evseen does _not_ have CONNECTED set) or
// later during normal socket operation
do {
// wait for connection
// NOTE: this should also handle other errors like timeout
if ((priv->priv_evseen & CONNECTED) == 0) {
switch (flag & (ERROR | CONNECTED)) {
case CONNECTED: // got clean connection -- puppy is happy!
break;
case ERROR: // got error first -- CONNECTED event unlikely
break;
case (CONNECTED | ERROR): // connected with error -- how bad is it?
break;
case 0: // huh? -- the only excuse ...
if (flag & TIMEOUT)
...
break;
}
break;
}
// events that occur during normal operation ...
// An error occurred during a bufferevent operation.
if (flag & ERROR) {
}
// A timeout expired on the bufferevent.
if (flag & TIMEOUT) {
}
// We got an end-of-line indication on the bufferevent.
if (flag & EOF) {
}
// We finished a requested connection on the bufferevent.
// NOTE: this should _never_ happen now
if (flag & CONNECTED) {
}
// An event occurred during a read operation on the bufferevent.
if (flag & READING) {
}
// An event occurred during a write operation on the bufferevent.
if (flag & WRITING) {
}
} while (0);
// remember types we've seen before
priv->priv_seen |= flag;
}
以下是一些相关的 libevent 来源：
int
bufferevent_socket_connect(struct bufferevent *bev, struct sockaddr *sa, int socklen)
{
struct bufferevent_private *bufev_p = EVUTIL_UPCAST(bev, struct bufferevent_private, bev);
evutil_socket_t fd;
int r = 0;
int result = -1;
int ownfd = 0;
_bufferevent_incref_and_lock(bev);
if (!bufev_p)
goto done;
fd = bufferevent_getfd(bev);
if (fd < 0) {
if (!sa)
goto done;
fd = socket(sa->sa_family, SOCK_STREAM, 0);
if (fd < 0)
goto done;
if (evutil_make_socket_nonblocking(fd) < 0)
goto done;
ownfd = 1;
}
if (sa) {
#ifdef WIN32
if (bufferevent_async_can_connect(bev)) {
bufferevent_setfd(bev, fd);
r = bufferevent_async_connect(bev, fd, sa, socklen);
if (r < 0)
goto freesock;
bufev_p->connecting = 1;
result = 0;
goto done;
}
else
#endif
r = evutil_socket_connect(&fd, sa, socklen);
if (r < 0)
goto freesock;
}
#ifdef WIN32
/* ConnectEx() isn't always around, even when IOCP is enabled. Here, we borrow the socket object's write handler to fall back on a non-blocking connect() when ConnectEx() is unavailable. */
if (BEV_IS_ASYNC(bev)) {
event_assign(&bev->ev_write, bev->ev_base, fd, EV_WRITE | EV_PERSIST, bufferevent_writecb, bev);
}
#endif
bufferevent_setfd(bev, fd);
if (r == 0) {
if (!be_socket_enable(bev, EV_WRITE)) {
bufev_p->connecting = 1;
result = 0;
goto done;
}
}
else if (r == 1) {
/* The connect succeeded already. How very BSD of it. */
result = 0;
bufev_p->connecting = 1;
event_active(&bev->ev_write, EV_WRITE, 1);
}
else {
/* The connect failed already. How very BSD of it. */
bufev_p->connection_refused = 1;
bufev_p->connecting = 1;
result = 0;
event_active(&bev->ev_write, EV_WRITE, 1);
}
goto done;
freesock:
_bufferevent_run_eventcb(bev, BEV_EVENT_ERROR);
if (ownfd)
evutil_closesocket(fd);
/* do something about the error? */
done:
_bufferevent_decref_and_unlock(bev);
return result;
}
/* XXX we should use an enum here. */
/* 2 for connection refused, 1 for connected, 0 for not yet, -1 for error. */
int
evutil_socket_connect(evutil_socket_t * fd_ptr, struct sockaddr *sa, int socklen)
{
int made_fd = 0;
if (*fd_ptr < 0) {
if ((*fd_ptr = socket(sa->sa_family, SOCK_STREAM, 0)) < 0)
goto err;
made_fd = 1;
if (evutil_make_socket_nonblocking(*fd_ptr) < 0) {
goto err;
}
}
if (connect(*fd_ptr, sa, socklen) < 0) {
int e = evutil_socket_geterror(*fd_ptr);
if (EVUTIL_ERR_CONNECT_RETRIABLE(e))
return 0;
if (EVUTIL_ERR_CONNECT_REFUSED(e))
return 2;
goto err;
}
else {
return 1;
}
err:
if (made_fd) {
evutil_closesocket(*fd_ptr);
*fd_ptr = -1;
}
return -1;
}
/* Check whether a socket on which we called connect() is done
connecting. Return 1 for connected, 0 for not yet, -1 for error. In the
error case, set the current socket errno to the error that happened during
the connect operation. */
int
evutil_socket_finished_connecting(evutil_socket_t fd)
{
int e;
ev_socklen_t elen = sizeof(e);
if (getsockopt(fd, SOL_SOCKET, SO_ERROR, (void *) &e, &elen) < 0)
return -1;
if (e) {
if (EVUTIL_ERR_CONNECT_RETRIABLE(e))
return 0;
EVUTIL_SET_SOCKET_ERROR(e);
return -1;
}
return 1;
}
struct bufferevent *
bufferevent_socket_new(struct event_base *base, evutil_socket_t fd, int options)
{
struct bufferevent_private *bufev_p;
struct bufferevent *bufev;
#ifdef WIN32
if (base && event_base_get_iocp(base))
return bufferevent_async_new(base, fd, options);
#endif
if ((bufev_p = mm_calloc(1, sizeof(struct bufferevent_private))) == NULL)
return NULL;
if (bufferevent_init_common(bufev_p, base, &bufferevent_ops_socket, options) < 0) {
mm_free(bufev_p);
return NULL;
}
bufev = &bufev_p->bev;
evbuffer_set_flags(bufev->output, EVBUFFER_FLAG_DRAINS_TO_FD);
event_assign(&bufev->ev_read, bufev->ev_base, fd, EV_READ | EV_PERSIST, bufferevent_readcb, bufev);
event_assign(&bufev->ev_write, bufev->ev_base, fd, EV_WRITE | EV_PERSIST, bufferevent_writecb, bufev);
evbuffer_add_cb(bufev->output, bufferevent_socket_outbuf_cb, bufev);
evbuffer_freeze(bufev->input, 0);
evbuffer_freeze(bufev->output, 1);
return bufev;
}
static void
bufferevent_writecb(evutil_socket_t fd, short event, void *arg)
{
struct bufferevent *bufev = arg;
struct bufferevent_private *bufev_p = EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
int res = 0;
short what = BEV_EVENT_WRITING;
int connected = 0;
ev_ssize_t atmost = -1;
_bufferevent_incref_and_lock(bufev);
if (event == EV_TIMEOUT) {
/* Note that we only check for event==EV_TIMEOUT. If event==EV_TIMEOUT|EV_WRITE, we can safely ignore the timeout, since a read has occurred */
what |= BEV_EVENT_TIMEOUT;
goto error;
}
if (bufev_p->connecting) {
int c = evutil_socket_finished_connecting(fd);
/* we need to fake the error if the connection was refused immediately - usually connection to localhost on BSD */
if (bufev_p->connection_refused) {
bufev_p->connection_refused = 0;
c = -1;
}
if (c == 0)
goto done;
bufev_p->connecting = 0;
if (c < 0) {
event_del(&bufev->ev_write);
event_del(&bufev->ev_read);
_bufferevent_run_eventcb(bufev, BEV_EVENT_ERROR);
goto done;
}
else {
connected = 1;
#ifdef WIN32
if (BEV_IS_ASYNC(bufev)) {
event_del(&bufev->ev_write);
bufferevent_async_set_connected(bufev);
_bufferevent_run_eventcb(bufev, BEV_EVENT_CONNECTED);
goto done;
}
#endif
_bufferevent_run_eventcb(bufev, BEV_EVENT_CONNECTED);
if (!(bufev->enabled & EV_WRITE) || bufev_p->write_suspended) {
event_del(&bufev->ev_write);
goto done;
}
}
}
atmost = _bufferevent_get_write_max(bufev_p);
if (bufev_p->write_suspended)
goto done;
if (evbuffer_get_length(bufev->output)) {
evbuffer_unfreeze(bufev->output, 1);
res = evbuffer_write_atmost(bufev->output, fd, atmost);
evbuffer_freeze(bufev->output, 1);
if (res == -1) {
int err = evutil_socket_geterror(fd);
if (EVUTIL_ERR_RW_RETRIABLE(err))
goto reschedule;
what |= BEV_EVENT_ERROR;
}
else if (res == 0) {
/* eof case XXXX Actually, a 0 on write doesn't indicate an EOF. An ECONNRESET might be more typical. */
what |= BEV_EVENT_EOF;
}
if (res <= 0)
goto error;
_bufferevent_decrement_write_buckets(bufev_p, res);
}
if (evbuffer_get_length(bufev->output) == 0) {
event_del(&bufev->ev_write);
}
/*
* Invoke the user callback if our buffer is drained or below the
* low watermark.
*/
if ((res || !connected) && evbuffer_get_length(bufev->output) <= bufev->wm_write.low) {
_bufferevent_run_writecb(bufev);
}
goto done;
reschedule:
if (evbuffer_get_length(bufev->output) == 0) {
event_del(&bufev->ev_write);
}
goto done;
error:
bufferevent_disable(bufev, EV_WRITE);
_bufferevent_run_eventcb(bufev, what);
done:
_bufferevent_decref_and_unlock(bufev);
}