X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fevent_client.c;h=27e8a8146ff9641a651982c06aaed0ddab5ac215;hb=5268c0a9bf8202336b9e3e8b86f45d9d010f2811;hp=f1abcbef7ff9396550809c03be7b3af22edec87e;hpb=36111d9ace0953d536650249f7339a52b0bed77d;p=citadel.git diff --git a/citadel/event_client.c b/citadel/event_client.c index f1abcbef7..27e8a8146 100644 --- a/citadel/event_client.c +++ b/citadel/event_client.c @@ -2,7 +2,7 @@ * * Copyright (c) 1998-2009 by the citadel.org team * - * This program is free software; you can redistribute it and/or modify + * This program is open source software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 3 of the License, or * (at your option) any later version. @@ -69,9 +69,9 @@ static void IO_abort_shutdown_callback(struct ev_loop *loop, ev_cleanup *watcher, int revents) { - syslog(LOG_DEBUG, "EVENT Q: %s\n", __FUNCTION__); - AsyncIO *IO = watcher->data; + EV_syslog(LOG_DEBUG, "EVENT Q: %s\n", __FUNCTION__); + assert(IO->ShutdownAbort); IO->ShutdownAbort(IO); } @@ -101,13 +101,13 @@ eNextState QueueDBOperation(AsyncIO *IO, IO_CallBack CB) ev_cleanup_start(event_db, &IO->db_abort_by_shutdown); pthread_mutex_lock(&DBEventQueueMutex); - syslog(LOG_DEBUG, "DBEVENT Q\n"); + EVM_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"); + EVM_syslog(LOG_DEBUG, "DBEVENT Q Done.\n"); return eDBQuery; } @@ -116,7 +116,7 @@ void ShutDownDBCLient(AsyncIO *IO) CitContext *Ctx =IO->CitContext; become_session(Ctx); - syslog(LOG_DEBUG, "DBEVENT\n"); + EVM_syslog(LOG_DEBUG, "DBEVENT Terminating.\n"); ev_cleanup_stop(event_db, &IO->db_abort_by_shutdown); assert(IO->Terminate); @@ -130,7 +130,7 @@ void DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents) { AsyncIO *IO = watcher->data; - syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__); + EV_syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__); become_session(IO->CitContext); ev_idle_stop(event_db, &IO->db_unwind_stack); @@ -145,14 +145,18 @@ DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents) 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); + ev_idle_stop(event_db, &IO->db_unwind_stack); + ev_cleanup_stop(loop, &IO->db_abort_by_shutdown); + ShutDownDBCLient(IO); } } @@ -191,26 +195,36 @@ eNextState QueueEventContext(AsyncIO *IO, IO_CallBack CB) ev_cleanup_start(event_base, &IO->abort_by_shutdown); pthread_mutex_lock(&EventQueueMutex); - syslog(LOG_DEBUG, "EVENT Q\n"); + EVM_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"); + EVM_syslog(LOG_DEBUG, "EVENT Q Done.\n"); return eSendReply; } -int ShutDownEventQueue(void) +extern eNextState evcurl_handle_start(AsyncIO *IO); + +eNextState QueueCurlContext(AsyncIO *IO) { - pthread_mutex_lock(&DBEventQueueMutex); - ev_async_send (event_db, &DBExitEventLoop); - pthread_mutex_unlock(&DBEventQueueMutex); + IOAddHandler *h; + int i; + + h = (IOAddHandler*)malloc(sizeof(IOAddHandler)); + h->IO = IO; + h->EvAttch = evcurl_handle_start; pthread_mutex_lock(&EventQueueMutex); - ev_async_send (EV_DEFAULT_ &ExitEventLoop); + EVM_syslog(LOG_DEBUG, "EVENT Q\n"); + i = ++evbase_count; + Put(InboundEventQueue, IKEY(i), h, NULL); pthread_mutex_unlock(&EventQueueMutex); - return 0; + + ev_async_send (event_base, &AddJob); + EVM_syslog(LOG_DEBUG, "EVENT Q Done.\n"); + return eSendReply; } void FreeAsyncIOContents(AsyncIO *IO) @@ -240,16 +254,16 @@ void ShutDownCLient(AsyncIO *IO) CitContext *Ctx =IO->CitContext; become_session(Ctx); - syslog(LOG_DEBUG, "EVENT x %d\n", IO->SendBuf.fd); + EVM_syslog(LOG_DEBUG, "EVENT Terminating \n"); 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); - ev_io_stop(event_base, &IO->dns_send_event); - IO->DNSChannel = NULL; + if (IO->DNS.Channel != NULL) { + ares_destroy(IO->DNS.Channel); + ev_io_stop(event_base, &IO->DNS.recv_event); + ev_io_stop(event_base, &IO->DNS.send_event); + IO->DNS.Channel = NULL; } assert(IO->Terminate); IO->Terminate(IO); @@ -260,6 +274,7 @@ void ShutDownCLient(AsyncIO *IO) eReadState HandleInbound(AsyncIO *IO) { + const char *Err = NULL; eReadState Finished = eBufferNotEmpty; become_session(IO->CitContext); @@ -267,6 +282,7 @@ eReadState HandleInbound(AsyncIO *IO) while ((Finished == eBufferNotEmpty) && ((IO->NextState == eReadMessage)|| (IO->NextState == eReadMore)|| + (IO->NextState == eReadFile)|| (IO->NextState == eReadPayload))) { if (IO->RecvBuf.nBlobBytesWanted != 0) { @@ -274,7 +290,16 @@ eReadState HandleInbound(AsyncIO *IO) } 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); @@ -301,6 +326,9 @@ eReadState HandleInbound(AsyncIO *IO) } switch (IO->NextState) { + case eSendFile: + ev_io_start(event_base, &IO->send_event); + break; case eSendReply: case eSendMore: assert(IO->SendDone); @@ -309,9 +337,12 @@ eReadState HandleInbound(AsyncIO *IO) 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; @@ -331,6 +362,7 @@ 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); #ifdef BIGBAD_IODBG @@ -357,7 +389,15 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents) if (!rv) printf("failed to write debug to %s!\n", fn); fprintf(fd, "]\n"); #endif - rc = StrBuf_write_one_chunk_callback(watcher->fd, 0/*TODO*/, &IO->SendBuf); + 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); @@ -379,6 +419,34 @@ 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; @@ -386,6 +454,7 @@ IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents) case eReadMore: case eReadMessage: case eReadPayload: + case eReadFile: if (StrBufCheckBuffer(&IO->RecvBuf) == eBufferNotEmpty) { HandleInbound(IO); } @@ -419,11 +488,13 @@ 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; @@ -531,10 +602,30 @@ 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; @@ -578,9 +669,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)); + EV_syslog(LOG_DEBUG, + "EVENT: Socket Invalid! %s \n", + strerror(errno)); + ShutDownCLient(IO); return; } } @@ -589,15 +681,22 @@ void IO_postdns_callback(struct ev_loop *loop, ev_idle *watcher, int revents) { AsyncIO *IO = watcher->data; - syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__); + EV_syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__); become_session(IO->CitContext); - assert(IO->DNSFail); - switch (IO->DNSQuery->PostDNS(IO)) + assert(IO->DNS.Fail); + assert(IO->DNS.Query->PostDNS); + switch (IO->DNS.Query->PostDNS(IO)) { case eAbort: - IO->DNSFail(IO); + switch (IO->DNS.Fail(IO)) { + case eAbort: + ShutDownCLient(IO); + default: + break; + + } default: - break; + break; } } @@ -613,23 +712,23 @@ eNextState event_connect_socket(AsyncIO *IO, double conn_timeout, double first_r IPPROTO_TCP); if (IO->SendBuf.fd < 0) { - syslog(LOG_ERR, "EVENT: socket() failed: %s\n", strerror(errno)); + EV_syslog(LOG_ERR, "EVENT: socket() failed: %s\n", strerror(errno)); StrBufPrintf(IO->ErrMsg, "Failed to create socket: %s", strerror(errno)); return eAbort; } fdflags = fcntl(IO->SendBuf.fd, F_GETFL); if (fdflags < 0) { - syslog(LOG_DEBUG, - "EVENT: unable to get socket flags! %s \n", - strerror(errno)); + EV_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) { - syslog(LOG_DEBUG, - "EVENT: unable to set socket nonblocking flags! %s \n", - strerror(errno)); + EV_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; @@ -657,13 +756,13 @@ eNextState event_connect_socket(AsyncIO *IO, double conn_timeout, double first_r rc = connect(IO->SendBuf.fd, (struct sockaddr_in *)&IO->ConnectMe->Addr, sizeof(struct sockaddr_in)); if (rc >= 0){ - syslog(LOG_DEBUG, "connect() immediate success.\n"); + EVM_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"); + EVM_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; @@ -678,7 +777,7 @@ eNextState event_connect_socket(AsyncIO *IO, double conn_timeout, double first_r IO->conn_fail_immediate.data = IO; ev_idle_start(event_base, &IO->conn_fail_immediate); - syslog(LOG_ERR, "connect() failed: %s\n", strerror(errno)); + EV_syslog(LOG_ERR, "connect() failed: %s\n", strerror(errno)); StrBufPrintf(IO->ErrMsg, "Failed to connect: %s", strerror(errno)); return IO->NextState; }