X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fevent_client.c;h=57a6440102f12ca86d58b83aba96c0f1d1ad76f7;hb=192056a6112602350c1f8a73eae2e134a31c7ba2;hp=11aabf96388ca72d464f219e59e4b8882eb5e590;hpb=ddd73c55e1c30f256b7a539d0a994232fc6bf318;p=citadel.git diff --git a/citadel/event_client.c b/citadel/event_client.c index 11aabf963..57a644010 100644 --- a/citadel/event_client.c +++ b/citadel/event_client.c @@ -67,22 +67,119 @@ #include "event_client.h" -extern citthread_mutex_t EventQueueMutex; -extern HashList *InboundEventQueue; -extern struct ev_loop *event_base; -extern ev_async AddJob; -extern ev_async ExitEventLoop; - -static void -IO_abort_shutdown_callback(struct ev_loop *loop, ev_cleanup *watcher, int revents) +static void IO_abort_shutdown_callback(struct ev_loop *loop, ev_cleanup *watcher, int revents) { - CtdlLogPrintf(CTDL_DEBUG, "EVENT Q: %s\n", __FUNCTION__); + syslog(LOG_DEBUG, "EVENT Q: %s\n", __FUNCTION__); AsyncIO *IO = watcher->data; + assert(IO->ShutdownAbort); IO->ShutdownAbort(IO); } + + +/*-------------------------------------------------------------------------------- + * Server DB IO + */ +extern int evdb_count; +extern pthread_mutex_t DBEventQueueMutex; +extern HashList *DBInboundEventQueue; +extern struct ev_loop *event_db; +extern ev_async DBAddJob; +extern ev_async DBExitEventLoop; + +eNextState QueueDBOperation(AsyncIO *IO, IO_CallBack CB) +{ + IOAddHandler *h; + int i; + + h = (IOAddHandler*)malloc(sizeof(IOAddHandler)); + h->IO = IO; + h->EvAttch = CB; + ev_cleanup_init(&IO->db_abort_by_shutdown, + IO_abort_shutdown_callback); + IO->db_abort_by_shutdown.data = IO; + ev_cleanup_start(event_db, &IO->db_abort_by_shutdown); + + pthread_mutex_lock(&DBEventQueueMutex); + syslog(LOG_DEBUG, "DBEVENT Q\n"); + i = ++evdb_count ; + Put(DBInboundEventQueue, IKEY(i), h, NULL); + pthread_mutex_unlock(&DBEventQueueMutex); + + ev_async_send (event_db, &DBAddJob); + syslog(LOG_DEBUG, "DBEVENT Q Done.\n"); + return eDBQuery; +} + +void ShutDownDBCLient(AsyncIO *IO) +{ + CitContext *Ctx =IO->CitContext; + become_session(Ctx); + + syslog(LOG_DEBUG, "DBEVENT\n"); + ev_cleanup_stop(event_db, &IO->db_abort_by_shutdown); + + assert(IO->Terminate); + IO->Terminate(IO); + + Ctx->state = CON_IDLE; + Ctx->kill_me = 1; +} + +void +DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents) +{ + AsyncIO *IO = watcher->data; + syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__); + become_session(IO->CitContext); -int QueueEventContext(AsyncIO *IO, IO_CallBack CB) + ev_idle_stop(event_db, &IO->db_unwind_stack); + + assert(IO->NextDBOperation); + switch (IO->NextDBOperation(IO)) + { + case eDBQuery: + break; + case eSendDNSQuery: + case eReadDNSReply: + case eConnect: + case eSendReply: + case eSendMore: + case eSendFile: + case eReadMessage: + case eReadMore: + case eReadPayload: + case eReadFile: + ev_cleanup_stop(loop, &IO->db_abort_by_shutdown); + break; + case eTerminateConnection: + case eAbort: + ShutDownDBCLient(IO); + } +} + +eNextState NextDBOperation(AsyncIO *IO, IO_CallBack CB) +{ + IO->NextDBOperation = CB; + ev_idle_init(&IO->db_unwind_stack, + DB_PerformNext); + IO->db_unwind_stack.data = IO; + ev_idle_start(event_db, &IO->db_unwind_stack); + return eDBQuery; +} + +/*-------------------------------------------------------------------------------- + * Client IO + */ +extern int evbase_count; +extern pthread_mutex_t EventQueueMutex; +extern HashList *InboundEventQueue; +extern struct ev_loop *event_base; +extern ev_async AddJob; +extern ev_async ExitEventLoop; + + +eNextState QueueEventContext(AsyncIO *IO, IO_CallBack CB) { IOAddHandler *h; int i; @@ -95,23 +192,48 @@ int QueueEventContext(AsyncIO *IO, IO_CallBack CB) IO->abort_by_shutdown.data = IO; ev_cleanup_start(event_base, &IO->abort_by_shutdown); - citthread_mutex_lock(&EventQueueMutex); - CtdlLogPrintf(CTDL_DEBUG, "EVENT Q\n"); - i = GetCount(InboundEventQueue); + pthread_mutex_lock(&EventQueueMutex); + syslog(LOG_DEBUG, "EVENT Q\n"); + i = ++evbase_count; Put(InboundEventQueue, IKEY(i), h, NULL); - citthread_mutex_unlock(&EventQueueMutex); + pthread_mutex_unlock(&EventQueueMutex); ev_async_send (event_base, &AddJob); - CtdlLogPrintf(CTDL_DEBUG, "EVENT Q Done.\n"); - return 0; + syslog(LOG_DEBUG, "EVENT Q Done.\n"); + return eSendReply; } +extern eNextState evcurl_handle_start(AsyncIO *IO); + +eNextState QueueCurlContext(AsyncIO *IO) +{ + IOAddHandler *h; + int i; + + h = (IOAddHandler*)malloc(sizeof(IOAddHandler)); + h->IO = IO; + h->EvAttch = evcurl_handle_start; + + pthread_mutex_lock(&EventQueueMutex); + syslog(LOG_DEBUG, "EVENT Q\n"); + i = ++evbase_count; + Put(InboundEventQueue, IKEY(i), h, NULL); + pthread_mutex_unlock(&EventQueueMutex); + + ev_async_send (event_base, &AddJob); + syslog(LOG_DEBUG, "EVENT Q Done.\n"); + return eSendReply; +} int ShutDownEventQueue(void) { - citthread_mutex_lock(&EventQueueMutex); + pthread_mutex_lock(&DBEventQueueMutex); + ev_async_send (event_db, &DBExitEventLoop); + pthread_mutex_unlock(&DBEventQueueMutex); + + pthread_mutex_lock(&EventQueueMutex); ev_async_send (EV_DEFAULT_ &ExitEventLoop); - citthread_mutex_unlock(&EventQueueMutex); + pthread_mutex_unlock(&EventQueueMutex); return 0; } @@ -123,25 +245,30 @@ void FreeAsyncIOContents(AsyncIO *IO) } -void ShutDownCLient(AsyncIO *IO) +void StopClientWatchers(AsyncIO *IO) { - CtdlLogPrintf(CTDL_DEBUG, "EVENT x %d\n", IO->SendBuf.fd); - - ev_cleanup_stop(event_base, &IO->abort_by_shutdown); - ev_timer_stop(event_base, &IO->conn_fail); ev_io_stop(event_base, &IO->conn_event); ev_idle_stop(event_base, &IO->unwind_stack); - if (IO->SendBuf.fd != 0) - { - ev_io_stop(event_base, &IO->send_event); - ev_io_stop(event_base, &IO->recv_event); - ev_timer_stop (event_base, &IO->rw_timeout); - close(IO->SendBuf.fd); - IO->SendBuf.fd = 0; - IO->RecvBuf.fd = 0; - } + ev_io_stop(event_base, &IO->send_event); + ev_io_stop(event_base, &IO->recv_event); + ev_timer_stop (event_base, &IO->rw_timeout); + close(IO->SendBuf.fd); + IO->SendBuf.fd = 0; + IO->RecvBuf.fd = 0; +} + +void ShutDownCLient(AsyncIO *IO) +{ + CitContext *Ctx =IO->CitContext; + become_session(Ctx); + + syslog(LOG_DEBUG, "EVENT x %d\n", IO->SendBuf.fd); + + ev_cleanup_stop(event_base, &IO->abort_by_shutdown); + StopClientWatchers(IO); + if (IO->DNSChannel != NULL) { ares_destroy(IO->DNSChannel); ev_io_stop(event_base, &IO->dns_recv_event); @@ -149,32 +276,46 @@ void ShutDownCLient(AsyncIO *IO) IO->DNSChannel = NULL; } assert(IO->Terminate); - become_session(IO->CitContext); IO->Terminate(IO); - + Ctx->state = CON_IDLE; + Ctx->kill_me = 1; } eReadState HandleInbound(AsyncIO *IO) { + const char *Err = NULL; eReadState Finished = eBufferNotEmpty; become_session(IO->CitContext); - while ((Finished == eBufferNotEmpty) && (IO->NextState == eReadMessage)){ + while ((Finished == eBufferNotEmpty) && + ((IO->NextState == eReadMessage)|| + (IO->NextState == eReadMore)|| + (IO->NextState == eReadFile)|| + (IO->NextState == eReadPayload))) + { if (IO->RecvBuf.nBlobBytesWanted != 0) { } else { /* Reading lines... */ //// lex line reply in callback, or do it ourselves. as nnn-blabla means continue reading in SMTP - if (IO->LineReader) + if ((IO->NextState == eReadFile) && + (Finished == eBufferNotEmpty)) + { + Finished = WriteIOBAlreadyRead(&IO->IOB, &Err); + if (Finished == eReadSuccess) + { + IO->NextState = eSendReply; + } + } + else if (IO->LineReader) Finished = IO->LineReader(IO); else Finished = StrBufChunkSipLine(IO->IOBuf, &IO->RecvBuf); switch (Finished) { case eMustReadMore: /// read new from socket... - return Finished; break; case eBufferNotEmpty: /* shouldn't happen... */ case eReadSuccess: /// done for now... @@ -194,17 +335,34 @@ eReadState HandleInbound(AsyncIO *IO) } } - - if ((IO->NextState == eSendReply) || - (IO->NextState == eSendMore)) - { + switch (IO->NextState) { + case eSendFile: + ev_io_start(event_base, &IO->send_event); + break; + case eSendReply: + case eSendMore: assert(IO->SendDone); IO->NextState = IO->SendDone(IO); ev_io_start(event_base, &IO->send_event); - } - else if ((IO->NextState == eTerminateConnection) || - (IO->NextState == eAbort) ) + break; + case eReadPayload: + case eReadMore: + case eReadFile: + ev_io_start(event_base, &IO->recv_event); + break; + case eTerminateConnection: +//////TODOxxxx + break; + case eAbort: ShutDownCLient(IO); + break; + case eSendDNSQuery: + case eReadDNSReply: + case eDBQuery: + case eConnect: + case eReadMessage: + break; + } return Finished; } @@ -214,41 +372,52 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents) { int rc; AsyncIO *IO = watcher->data; + const char *errmsg = NULL; become_session(IO->CitContext); - rc = StrBuf_write_one_chunk_callback(watcher->fd, 0/*TODO*/, &IO->SendBuf); - - if (rc == 0) - { #ifdef BIGBAD_IODBG - { - int rv = 0; - char fn [SIZ]; - FILE *fd; - const char *pch = ChrPtr(IO->SendBuf.Buf); - const char *pchh = IO->SendBuf.ReadWritePointer; - long nbytes; - - if (pchh == NULL) - pchh = pch; - - nbytes = StrLength(IO->SendBuf.Buf) - (pchh - pch); - snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d", "smtpev", IO->SendBuf.fd); + { + int rv = 0; + char fn [SIZ]; + FILE *fd; + const char *pch = ChrPtr(IO->SendBuf.Buf); + const char *pchh = IO->SendBuf.ReadWritePointer; + long nbytes; - fd = fopen(fn, "a+"); - fprintf(fd, "Read: BufSize: %ld BufContent: [", - nbytes); - rv = fwrite(pchh, nbytes, 1, fd); - fprintf(fd, "]\n"); + if (pchh == NULL) + pchh = pch; - - fclose(fd); + nbytes = StrLength(IO->SendBuf.Buf) - (pchh - pch); + snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d", + ((CitContext*)(IO->CitContext))->ServiceName, + IO->SendBuf.fd); + + fd = fopen(fn, "a+"); + fprintf(fd, "Send: BufSize: %ld BufContent: [", + nbytes); + rv = fwrite(pchh, nbytes, 1, fd); + if (!rv) printf("failed to write debug to %s!\n", fn); + fprintf(fd, "]\n"); +#endif + switch (IO->NextState) { + case eSendFile: + rc = FileSendChunked(&IO->IOB, &errmsg); + if (rc < 0) + StrBufPlain(IO->ErrMsg, errmsg, -1); + break; + default: + rc = StrBuf_write_one_chunk_callback(watcher->fd, 0/*TODO*/, &IO->SendBuf); } + +#ifdef BIGBAD_IODBG + fprintf(fd, "Sent: BufSize: %d bytes.\n", rc); + fclose(fd); + } #endif + if (rc == 0) + { ev_io_stop(event_base, &IO->send_event); switch (IO->NextState) { - case eSendReply: - break; case eSendMore: assert(IO->SendDone); IO->NextState = IO->SendDone(IO); @@ -260,7 +429,42 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents) ev_io_start(event_base, &IO->send_event); } break; + case eSendFile: + if (IO->IOB.ChunkSendRemain > 0) { + ev_io_start(event_base, &IO->recv_event); + } else { + assert(IO->ReadDone); + IO->NextState = IO->ReadDone(IO); + switch(IO->NextState) { + case eSendDNSQuery: + case eReadDNSReply: + case eDBQuery: + case eConnect: + break; + case eSendReply: + case eSendMore: + case eSendFile: + ev_io_start(event_base, &IO->send_event); + break; + case eReadMessage: + case eReadMore: + case eReadPayload: + case eReadFile: + break; + case eTerminateConnection: + case eAbort: + break; + } + } + break; + case eSendReply: + if (StrBufCheckBuffer(&IO->SendBuf) != eReadSuccess) + break; + IO->NextState = eReadMore; + case eReadMore: case eReadMessage: + case eReadPayload: + case eReadFile: if (StrBufCheckBuffer(&IO->RecvBuf) == eBufferNotEmpty) { HandleInbound(IO); } @@ -269,6 +473,13 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents) } break; + case eDBQuery: + /* we now live in another queue, so we have to unregister. */ + ev_cleanup_stop(loop, &IO->abort_by_shutdown); + break; + case eSendDNSQuery: + case eReadDNSReply: + case eConnect: case eTerminateConnection: case eAbort: break; @@ -285,14 +496,22 @@ set_start_callback(struct ev_loop *loop, AsyncIO *IO, int revents) { switch(IO->NextState) { + case eReadMore: case eReadMessage: + case eReadFile: ev_io_start(event_base, &IO->recv_event); break; case eSendReply: case eSendMore: + case eReadPayload: + case eSendFile: become_session(IO->CitContext); IO_send_callback(loop, &IO->send_event, revents); break; + case eDBQuery: + case eSendDNSQuery: + case eReadDNSReply: + case eConnect: case eTerminateConnection: case eAbort: /// TODO: WHUT? @@ -301,13 +520,22 @@ set_start_callback(struct ev_loop *loop, AsyncIO *IO, int revents) } static void -IO_Timout_callback(struct ev_loop *loop, ev_timer *watcher, int revents) +IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents) { AsyncIO *IO = watcher->data; ev_timer_stop (event_base, &IO->rw_timeout); become_session(IO->CitContext); + if (IO->SendBuf.fd != 0) + { + ev_io_stop(event_base, &IO->send_event); + ev_io_stop(event_base, &IO->recv_event); + ev_timer_stop (event_base, &IO->rw_timeout); + close(IO->SendBuf.fd); + IO->SendBuf.fd = IO->RecvBuf.fd = 0; + } + assert(IO->Timeout); switch (IO->Timeout(IO)) { @@ -317,13 +545,23 @@ IO_Timout_callback(struct ev_loop *loop, ev_timer *watcher, int revents) break; } } + static void IO_connfail_callback(struct ev_loop *loop, ev_timer *watcher, int revents) { AsyncIO *IO = watcher->data; ev_timer_stop (event_base, &IO->conn_fail); - ev_io_stop(loop, &IO->conn_event); + + if (IO->SendBuf.fd != 0) + { + ev_io_stop(loop, &IO->conn_event); + ev_io_stop(event_base, &IO->send_event); + ev_io_stop(event_base, &IO->recv_event); + ev_timer_stop (event_base, &IO->rw_timeout); + close(IO->SendBuf.fd); + IO->SendBuf.fd = IO->RecvBuf.fd = 0; + } become_session(IO->CitContext); assert(IO->ConnFail); @@ -336,6 +574,32 @@ IO_connfail_callback(struct ev_loop *loop, ev_timer *watcher, int revents) } } + +static void +IO_connfailimmediate_callback(struct ev_loop *loop, ev_idle *watcher, int revents) +{ + AsyncIO *IO = watcher->data; + + ev_idle_stop (event_base, &IO->conn_fail_immediate); + + if (IO->SendBuf.fd != 0) + { + close(IO->SendBuf.fd); + IO->SendBuf.fd = IO->RecvBuf.fd = 0; + } + become_session(IO->CitContext); + + assert(IO->ConnFail); + switch (IO->ConnFail(IO)) + { + case eAbort: + ShutDownCLient(IO); + default: + break; + + } +} + static void IO_connestd_callback(struct ev_loop *loop, ev_io *watcher, int revents) { @@ -348,10 +612,58 @@ IO_connestd_callback(struct ev_loop *loop, ev_io *watcher, int revents) static void IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents) { + const char *errmsg; ssize_t nbytes; AsyncIO *IO = watcher->data; - nbytes = StrBuf_read_one_chunk_callback(watcher->fd, 0 /*TODO */, &IO->RecvBuf); + switch (IO->NextState) { + case eReadFile: + nbytes = FileRecvChunked(&IO->IOB, &errmsg); + if (nbytes < 0) + StrBufPlain(IO->ErrMsg, errmsg, -1); + else + { + if (IO->IOB.ChunkSendRemain == 0) + { + IO->NextState = eSendReply; + } + else + return; + } + break; + default: + nbytes = StrBuf_read_one_chunk_callback(watcher->fd, 0 /*TODO */, &IO->RecvBuf); + break; + } + +#ifdef BIGBAD_IODBG + { + int rv = 0; + char fn [SIZ]; + FILE *fd; + const char *pch = ChrPtr(IO->RecvBuf.Buf); + const char *pchh = IO->RecvBuf.ReadWritePointer; + long nbytes; + + if (pchh == NULL) + pchh = pch; + + nbytes = StrLength(IO->RecvBuf.Buf) - (pchh - pch); + snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d", + ((CitContext*)(IO->CitContext))->ServiceName, + IO->SendBuf.fd); + + fd = fopen(fn, "a+"); + fprintf(fd, "Read: BufSize: %ld BufContent: [", + nbytes); + rv = fwrite(pchh, nbytes, 1, fd); + if (!rv) printf("failed to write debug to %s!\n", fn); + fprintf(fd, "]\n"); + + + fclose(fd); + } +#endif if (nbytes > 0) { HandleInbound(IO); } else if (nbytes == 0) { @@ -367,6 +679,10 @@ IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents) return; } else if (nbytes == -1) { /// TODO: FD is gone. kick it. sock_buff_invoke_free(sb, errno); + syslog(LOG_DEBUG, + "EVENT: Socket Invalid! %s \n", + strerror(errno)); + ShutDownCLient(IO); return; } } @@ -375,10 +691,17 @@ void IO_postdns_callback(struct ev_loop *loop, ev_idle *watcher, int revents) { AsyncIO *IO = watcher->data; - CtdlLogPrintf(CTDL_DEBUG, "event: %s\n", __FUNCTION__); + syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__); become_session(IO->CitContext); - - IO->DNSQuery->PostDNS(IO); + assert(IO->DNSFail); + assert(IO->DNSQuery->PostDNS); + switch (IO->DNSQuery->PostDNS(IO)) + { + case eAbort: + IO->DNSFail(IO); + default: + break; + } } eNextState event_connect_socket(AsyncIO *IO, double conn_timeout, double first_rw_timeout) @@ -393,24 +716,23 @@ eNextState event_connect_socket(AsyncIO *IO, double conn_timeout, double first_r IPPROTO_TCP); if (IO->SendBuf.fd < 0) { - CtdlLogPrintf(CTDL_ERR, "EVENT: socket() failed: %s\n", strerror(errno)); + syslog(LOG_ERR, "EVENT: socket() failed: %s\n", strerror(errno)); StrBufPrintf(IO->ErrMsg, "Failed to create socket: %s", strerror(errno)); -// freeaddrinfo(res); return eAbort; } fdflags = fcntl(IO->SendBuf.fd, F_GETFL); if (fdflags < 0) { - CtdlLogPrintf(CTDL_DEBUG, - "EVENT: unable to get socket flags! %s \n", - strerror(errno)); + syslog(LOG_DEBUG, + "EVENT: unable to get socket flags! %s \n", + strerror(errno)); StrBufPrintf(IO->ErrMsg, "Failed to get socket flags: %s", strerror(errno)); return eAbort; } fdflags = fdflags | O_NONBLOCK; if (fcntl(IO->SendBuf.fd, F_SETFL, fdflags) < 0) { - CtdlLogPrintf(CTDL_DEBUG, - "EVENT: unable to set socket nonblocking flags! %s \n", - strerror(errno)); + syslog(LOG_DEBUG, + "EVENT: unable to set socket nonblocking flags! %s \n", + strerror(errno)); StrBufPrintf(IO->ErrMsg, "Failed to set socket flags: %s", strerror(errno)); close(IO->SendBuf.fd); IO->SendBuf.fd = IO->RecvBuf.fd = -1; @@ -426,21 +748,25 @@ eNextState event_connect_socket(AsyncIO *IO, double conn_timeout, double first_r ev_timer_init(&IO->conn_fail, IO_connfail_callback, conn_timeout, 0); IO->conn_fail.data = IO; - ev_timer_init(&IO->rw_timeout, IO_Timout_callback, first_rw_timeout, 0); + ev_timer_init(&IO->rw_timeout, IO_Timeout_callback, first_rw_timeout, 0); IO->rw_timeout.data = IO; + + /* Bypass it like this: IO->Addr.sin_addr.s_addr = inet_addr("127.0.0.1"); */ +/// ((struct sockaddr_in)IO->ConnectMe->Addr).sin_addr.s_addr = inet_addr("127.0.0.1"); if (IO->ConnectMe->IPv6) rc = connect(IO->SendBuf.fd, &IO->ConnectMe->Addr, sizeof(struct sockaddr_in6)); else rc = connect(IO->SendBuf.fd, (struct sockaddr_in *)&IO->ConnectMe->Addr, sizeof(struct sockaddr_in)); if (rc >= 0){ -//// freeaddrinfo(res); + syslog(LOG_DEBUG, "connect() immediate success.\n"); set_start_callback(event_base, IO, 0); ev_timer_start(event_base, &IO->rw_timeout); return IO->NextState; } else if (errno == EINPROGRESS) { + syslog(LOG_DEBUG, "connect() have to wait now.\n"); ev_io_init(&IO->conn_event, IO_connestd_callback, IO->SendBuf.fd, EV_READ|EV_WRITE); IO->conn_event.data = IO; @@ -450,11 +776,14 @@ eNextState event_connect_socket(AsyncIO *IO, double conn_timeout, double first_r return IO->NextState; } else { - CtdlLogPrintf(CTDL_ERR, "connect() failed: %s\n", strerror(errno)); + ev_idle_init(&IO->conn_fail_immediate, + IO_connfailimmediate_callback); + IO->conn_fail_immediate.data = IO; + ev_idle_start(event_base, &IO->conn_fail_immediate); + + syslog(LOG_ERR, "connect() failed: %s\n", strerror(errno)); StrBufPrintf(IO->ErrMsg, "Failed to connect: %s", strerror(errno)); - assert(IO->ConnFail); - IO->ConnFail(IO); - return eAbort; + return IO->NextState; } return IO->NextState; } @@ -482,3 +811,21 @@ eNextState InitEventIO(AsyncIO *IO, } return event_connect_socket(IO, conn_timeout, first_rw_timeout); } + +eNextState ReAttachIO(AsyncIO *IO, + void *pData, + int ReadFirst) +{ + IO->Data = pData; + become_session(IO->CitContext); + ev_cleanup_start(event_base, &IO->abort_by_shutdown); + if (ReadFirst) { + IO->NextState = eReadMessage; + } + else { + IO->NextState = eSendReply; + } + set_start_callback(event_base, IO, 0); + + return IO->NextState; +}