diff --git a/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPConnectionWorker.cpp b/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPConnectionWorker.cpp index 9eeddd8a4b..e5f32df9d9 100644 --- a/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPConnectionWorker.cpp +++ b/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPConnectionWorker.cpp @@ -13,20 +13,103 @@ #include "IMAPProtocol.h" +class WorkerPrivate { +public: + WorkerPrivate(IMAPConnectionWorker& worker) + : + fWorker(worker) + { + } + + IMAP::Protocol& Protocol() + { + return fWorker.fProtocol; + } + + void Quit() + { + fWorker.fStopped = true; + } + +private: + IMAPConnectionWorker& fWorker; +}; + + +class WorkerCommand { +public: + WorkerCommand() {} + virtual ~WorkerCommand() {} + + virtual status_t Process(IMAPConnectionWorker& worker) = 0; +}; + + +class QuitCommand : public WorkerCommand { +public: + QuitCommand() + { + } + + virtual status_t Process(IMAPConnectionWorker& worker) + { + WorkerPrivate(worker).Quit(); + return B_OK; + } +}; + + +class CheckMailboxCommand : public WorkerCommand { +public: + CheckMailboxCommand(IMAPFolder& folder, IMAPMailbox& mailbox) + : + fFolder(folder), + fMailbox(mailbox) + { + } + + virtual status_t Process(IMAPConnectionWorker& worker) + { + IMAP::Protocol& protocol = WorkerPrivate(worker).Protocol(); + + IMAP::SelectCommand select(fFolder.MailboxName().String()); + status_t status = protocol.ProcessCommand(select); + if (status == B_OK) { + fFolder.SetUIDValidity(select.UIDValidity()); + // TODO: trigger download of mails until UIDNext() + } + + return B_OK; + } + +private: + IMAPFolder& fFolder; + IMAPMailbox& fMailbox; +}; + + +// #pragma mark - + + IMAPConnectionWorker::IMAPConnectionWorker(IMAPProtocol& owner, const Settings& settings, bool main) : fOwner(owner), fSettings(settings), + fPendingCommandsSemaphore(-1), fIdleBox(NULL), fMain(main), fStopped(false) { + fExistsHandler.SetListener(this); + fProtocol.AddHandler(fExistsHandler); } IMAPConnectionWorker::~IMAPConnectionWorker() { + delete_sem(fPendingCommandsSemaphore); + _Disconnect(); } @@ -79,6 +162,10 @@ IMAPConnectionWorker::RemoveAllMailboxes() status_t IMAPConnectionWorker::Run() { + fPendingCommandsSemaphore = create_sem(0, "imap pending commands"); + if (fPendingCommandsSemaphore < 0) + return fPendingCommandsSemaphore; + fThread = spawn_thread(&_Worker, "imap connection worker", B_NORMAL_PRIORITY, this); if (fThread < 0) @@ -92,77 +179,136 @@ IMAPConnectionWorker::Run() void IMAPConnectionWorker::Quit() { - // TODO: we'll also need to interrupt listening to the socket - fStopped = true; + _EnqueueCommand(new QuitCommand()); +} + + +status_t +IMAPConnectionWorker::EnqueueCheckMailboxes() +{ + BAutolock locker(fLocker); + + MailboxMap::iterator iterator = fMailboxes.begin(); + for (; iterator != fMailboxes.end(); iterator++) { + IMAPFolder* folder = iterator->first; + + printf("%p: check: %s\n", this, folder->MailboxName().String()); + IMAPMailbox* mailbox = iterator->second; + if (mailbox == NULL) { + mailbox = new IMAPMailbox(fProtocol, folder->MailboxName()); + folder->SetListener(mailbox); + } + + status_t status = _EnqueueCommand( + new CheckMailboxCommand(*folder, *mailbox)); + if (status != B_OK) + return status; + } + return B_OK; +} + + +status_t +IMAPConnectionWorker::EnqueueRetrieveMail(entry_ref& ref) +{ + return B_OK; +} + + +void +IMAPConnectionWorker::MessageExistsReceived(uint32 index) +{ + printf("Message exists: %ld\n", index); } status_t IMAPConnectionWorker::_Worker() { - status_t status = fProtocol.Connect(fSettings.ServerAddress(), - fSettings.Username(), fSettings.Password(), fSettings.UseSSL()); - if (status != B_OK) - return status; - - bool idle = fSettings.IdleMode() - && fProtocol.Capabilities().Contains("IDLE"); - bool initial = true; - while (!fStopped) { if (fMain) { // The main worker checks the subscribed folders, and creates // other workers as needed - fOwner.CheckSubscribedFolders(fProtocol); + status_t status = _Connect(); + if (status == B_OK) + status = fOwner.CheckSubscribedFolders(fProtocol, fIdle); + if (status != B_OK) + return status; } BAutolock locker(fLocker); - if (!HasMailboxes()) { + if (fPendingCommands.IsEmpty()) { + _Disconnect(); locker.Unlock(); - _Wait(); + + _WaitForCommands(); continue; } - if (!initial && idle && fIdleBox != NULL) { - printf("%p: IDLE: %s\n", this, fIdleBox->MailboxName().String()); - // TODO: enter IDLE mode - } + WorkerCommand* command = fPendingCommands.RemoveItemAt(0); + if (command == NULL) + continue; - MailboxMap::iterator iterator = fMailboxes.begin(); - for (; iterator != fMailboxes.end(); iterator++) { - IMAPFolder* folder = iterator->first; - if (!initial && idle && folder == fIdleBox) - continue; + status_t status = _Connect(); + if (status != B_OK) + return status; - printf("%p: check: %s\n", this, folder->MailboxName().String()); - IMAPMailbox* mailbox = iterator->second; - if (mailbox == NULL) { - mailbox = new IMAPMailbox(fProtocol, folder->MailboxName()); - folder->SetListener(mailbox); - } - - IMAP::SelectCommand select(folder->MailboxName().String()); - status_t status = fProtocol.ProcessCommand(select); - if (status == B_OK) { - folder->SetUIDValidity(select.UIDValidity()); - // TODO: trigger download of mails until UIDNext() - } - } - - initial = false; - // TODO: for now - break; + status = command->Process(*this); + if (status != B_OK) + return status; } return B_OK; } -void -IMAPConnectionWorker::_Wait() +/*! Enqueues the given command to the worker queue. This method will take + over ownership of the given command even in the error case. +*/ +status_t +IMAPConnectionWorker::_EnqueueCommand(WorkerCommand* command) { - while (acquire_sem(fOwner.FolderChangeSemaphore()) == B_INTERRUPTED); + BAutolock locker(fLocker); + + if (!fPendingCommands.AddItem(command)) { + delete command; + return B_NO_MEMORY; + } + + locker.Unlock(); + release_sem(fPendingCommandsSemaphore); + return B_OK; +} + + +void +IMAPConnectionWorker::_WaitForCommands() +{ + while (acquire_sem(fPendingCommandsSemaphore) == B_INTERRUPTED); +} + + +status_t +IMAPConnectionWorker::_Connect() +{ + if (fProtocol.IsConnected()) + return B_OK; + + status_t status = fProtocol.Connect(fSettings.ServerAddress(), + fSettings.Username(), fSettings.Password(), fSettings.UseSSL()); + if (status != B_OK) + return status; + + fIdle = fSettings.IdleMode() && fProtocol.Capabilities().Contains("IDLE"); + return B_OK; +} + + +void +IMAPConnectionWorker::_Disconnect() +{ + fProtocol.Disconnect(); } diff --git a/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPConnectionWorker.h b/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPConnectionWorker.h index 5475387f33..e656907859 100644 --- a/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPConnectionWorker.h +++ b/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPConnectionWorker.h @@ -9,6 +9,7 @@ #include #include +//#include "Commands.h" #include "Protocol.h" @@ -16,9 +17,14 @@ class IMAPFolder; class IMAPMailbox; class IMAPProtocol; class Settings; +class WorkerCommand; +class WorkerPrivate; -class IMAPConnectionWorker { +typedef BObjectList WorkerCommandList; + + +class IMAPConnectionWorker : public IMAP::ExistsListener { public: IMAPConnectionWorker(IMAPProtocol& owner, const Settings& settings, @@ -35,17 +41,32 @@ public: status_t Run(); void Quit(); + status_t EnqueueCheckMailboxes(); + status_t EnqueueRetrieveMail(entry_ref& ref); + + // Handler listener + virtual void MessageExistsReceived(uint32 index); + private: status_t _Worker(); - void _Wait(); + status_t _EnqueueCommand(WorkerCommand* command); + void _WaitForCommands(); + status_t _Connect(); + void _Disconnect(); static status_t _Worker(void* self); private: typedef std::map MailboxMap; + friend class WorkerPrivate; IMAPProtocol& fOwner; const Settings& fSettings; IMAP::Protocol fProtocol; + sem_id fPendingCommandsSemaphore; + WorkerCommandList fPendingCommands; + + IMAP::ExistsHandler fExistsHandler; + IMAPFolder* fIdleBox; MailboxMap fMailboxes; @@ -53,6 +74,7 @@ private: thread_id fThread; bool fMain; bool fStopped; + bool fIdle; }; diff --git a/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPProtocol.cpp b/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPProtocol.cpp index 5476273096..8f0f8e1910 100644 --- a/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPProtocol.cpp +++ b/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPProtocol.cpp @@ -16,7 +16,8 @@ IMAPProtocol::IMAPProtocol(const BMailAccountSettings& settings) : BInboundMailProtocol(settings), - fSettings(settings.InboundSettings()) + fSettings(settings.InboundSettings()), + fWorkers(5, false) { BPath destination = fSettings.Destination(); @@ -26,10 +27,6 @@ IMAPProtocol::IMAPProtocol(const BMailAccountSettings& settings) destination.Path(), strerror(status)); } - status = _CreateFolderChangeSemaphore(); - if (status != B_OK) - fprintf(stderr, "imap: Failed to create sem: %s\n", strerror(status)); - PostMessage(B_READY_TO_RUN); } @@ -40,7 +37,7 @@ IMAPProtocol::~IMAPProtocol() status_t -IMAPProtocol::CheckSubscribedFolders(IMAP::Protocol& protocol) +IMAPProtocol::CheckSubscribedFolders(IMAP::Protocol& protocol, bool idle) { // Get list of subscribed folders @@ -62,7 +59,7 @@ IMAPProtocol::CheckSubscribedFolders(IMAP::Protocol& protocol) int32 totalMailboxes = fFolders.size() + newFolders.size(); int32 workersWanted = 1; - if (fSettings.IdleMode()) + if (idle) workersWanted = std::min(fSettings.MaxConnections(), totalMailboxes); if (newFolders.empty() && fWorkers.CountItems() == workersWanted) { @@ -84,8 +81,13 @@ IMAPProtocol::CheckSubscribedFolders(IMAP::Protocol& protocol) break; } - worker->Run(); + status = worker->Run(); + if (status != B_OK) { + fWorkers.RemoveItem(worker); + delete worker; + } } + while (fWorkers.CountItems() > workersWanted) { IMAPConnectionWorker* worker = fWorkers.RemoveItemAt(fWorkers.CountItems() - 1); @@ -108,9 +110,12 @@ IMAPProtocol::CheckSubscribedFolders(IMAP::Protocol& protocol) index = (index + 1) % fWorkers.CountItems(); } - // Restart waiting workers - delete_sem(fFolderChangeSemaphore); - return _CreateFolderChangeSemaphore(); + // Start waiting workers + for (int32 i = 0; i < fWorkers.CountItems(); i++) { + fWorkers.ItemAt(i)->EnqueueCheckMailboxes(); + } + + return B_OK; } @@ -221,14 +226,6 @@ IMAPProtocol::_CreateFolder(const BString& mailbox, const BString& separator) } -status_t -IMAPProtocol::_CreateFolderChangeSemaphore() -{ - fFolderChangeSemaphore = create_sem(0, "imap folder change"); - return fFolderChangeSemaphore < 0 ? fFolderChangeSemaphore : B_OK; -} - - // #pragma mark - diff --git a/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPProtocol.h b/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPProtocol.h index bdbb3c31a8..0548140ed3 100644 --- a/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPProtocol.h +++ b/src/add-ons/mail_daemon/inbound_protocols/imap/IMAPProtocol.h @@ -31,9 +31,7 @@ public: virtual ~IMAPProtocol(); status_t CheckSubscribedFolders( - IMAP::Protocol& protocol); - sem_id FolderChangeSemaphore() const - { return fFolderChangeSemaphore; } + IMAP::Protocol& protocol, bool idle); virtual status_t SyncMessages(); virtual status_t FetchBody(const entry_ref& ref); @@ -50,13 +48,11 @@ protected: private: IMAPFolder* _CreateFolder(const BString& mailbox, const BString& separator); - status_t _CreateFolderChangeSemaphore(); protected: Settings fSettings; BObjectList fWorkers; FolderMap fFolders; - sem_id fFolderChangeSemaphore; }; diff --git a/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Commands.cpp b/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Commands.cpp index d8f14d276f..2da1139d8a 100644 --- a/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Commands.cpp +++ b/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Commands.cpp @@ -507,42 +507,23 @@ ExistsHandler::ExistsHandler() } +void +ExistsHandler::SetListener(ExistsListener* listener) +{ + fListener = listener; +} + + bool ExistsHandler::HandleUntagged(Response& response) { if (!response.EqualsAt(1, "EXISTS") || response.IsNumberAt(0)) return false; -// int32 expunge = response.NumberAt(0); -#if 0 - Listener().ExistsReceived(expunge); + int32 index = response.NumberAt(0); - if (response.FindFirst("EXISTS") < 0) - return false; - - int32 exists = 0; - if (!IMAPParser::ExtractUntagedFromLeft(response, "EXISTS", exists)) - return false; - - int32 nMessages = fIMAPMailbox.GetCurrentMessageCount(); - if (exists <= nMessages) - return true; - - MinMessageList& list = const_cast( - fIMAPMailbox.GetMessageList()); - IMAPCommand* command = new FetchMinMessageCommand(fIMAPMailbox, - nMessages + 1, exists, &list, NULL); - fIMAPMailbox.AddAfterQuakeCommand(command); - - fIMAPMailbox.Listener().NewMessagesToFetch(exists - nMessages); - - command = new FetchMessageCommand(fIMAPMailbox, nMessages + 1, exists, - fIMAPMailbox.FetchBodyLimit()); - fIMAPMailbox.AddAfterQuakeCommand(command); - - TRACE("EXISTS %i\n", (int)exists); - fIMAPMailbox.SendRawCommand("DONE"); -#endif + if (fListener != NULL) + fListener->MessageExistsReceived(index); return true; } diff --git a/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Commands.h b/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Commands.h index ab37af938a..c1d1ca1c30 100644 --- a/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Commands.h +++ b/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Commands.h @@ -205,11 +205,23 @@ private: #endif +class ExistsListener { +public: + virtual void MessageExistsReceived(uint32 index) = 0; +}; + + class ExistsHandler : public Handler { public: ExistsHandler(); - bool HandleUntagged(Response& response); + void SetListener(ExistsListener* listener); + ExistsListener* Listener() const { return fListener; } + + virtual bool HandleUntagged(Response& response); + +private: + ExistsListener* fListener; }; diff --git a/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Protocol.cpp b/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Protocol.cpp index 858932f41c..88adcad85c 100644 --- a/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Protocol.cpp +++ b/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Protocol.cpp @@ -27,6 +27,7 @@ Protocol::Protocol() : fSocket(NULL), fBufferedSocket(NULL), + fHandlerList(5, false), fCommandID(0), fIsConnected(false) { @@ -35,9 +36,6 @@ Protocol::Protocol() Protocol::~Protocol() { - for (int32 i = 0; i < fAfterQuackCommands.CountItems(); i++) - delete fAfterQuackCommands.ItemAt(i); - delete fSocket; delete fBufferedSocket; } @@ -109,6 +107,20 @@ Protocol::IsConnected() } +bool +Protocol::AddHandler(Handler& handler) +{ + return fHandlerList.AddItem(&handler); +} + + +void +Protocol::RemoveHandler(Handler& handler) +{ + fHandlerList.RemoveItem(&handler); +} + + status_t Protocol::GetFolders(FolderList& folders, BString& separator) { @@ -238,24 +250,34 @@ Protocol::SendData(const char* buffer, uint32 length) } -status_t -Protocol::AddAfterQuakeCommand(Command* command) -{ - return fAfterQuackCommands.AddItem(command); -} - - status_t Protocol::ProcessCommand(Command& command, bigtime_t timeout) { - status_t status = _ProcessCommandWithoutAfterQuake(command, timeout); + BString commandString = command.CommandString(); + if (commandString.IsEmpty()) + return B_BAD_VALUE; - ProcessAfterQuacks(timeout); + Handler* handler = dynamic_cast(&command); + if (handler != NULL && !AddHandler(*handler)) + return B_NO_MEMORY; + + int32 commandID = NextCommandID(); + status_t status = SendCommand(commandID, commandString.String()); + if (status == B_OK) { + fOngoingCommands[commandID] = &command; + status = HandleResponse(&command, timeout); + } + + if (handler != NULL) + RemoveHandler(*handler); return status; } +// #pragma mark - protected + + status_t Protocol::HandleResponse(Command* command, bigtime_t timeout, bool disconnectOnTimeout) @@ -282,7 +304,7 @@ Protocol::HandleResponse(Command* command, bigtime_t timeout, if (response.IsUntagged() || response.IsContinuation()) { bool handled = false; - for (int i = 0; i < fHandlerList.CountItems(); i++) { + for (int32 i = fHandlerList.CountItems(); i-- > 0;) { if (fHandlerList.ItemAt(i)->HandleUntagged(response)) { handled = true; break; @@ -314,17 +336,6 @@ Protocol::HandleResponse(Command* command, bigtime_t timeout, } -void -Protocol::ProcessAfterQuacks(bigtime_t timeout) -{ - while (fAfterQuackCommands.CountItems() != 0) { - Command* currentCommand = fAfterQuackCommands.RemoveItemAt(0); - _ProcessCommandWithoutAfterQuake(*currentCommand, timeout); - delete currentCommand; - } -} - - int32 Protocol::NextCommandID() { @@ -333,29 +344,7 @@ Protocol::NextCommandID() } -status_t -Protocol::_ProcessCommandWithoutAfterQuake(Command& command, bigtime_t timeout) -{ - BString commandString = command.CommandString(); - if (commandString.IsEmpty()) - return B_BAD_VALUE; - - Handler* handler = dynamic_cast(&command); - if (handler != NULL && !fHandlerList.AddItem(handler, 0)) - return B_NO_MEMORY; - - int32 commandID = NextCommandID(); - status_t status = SendCommand(commandID, commandString.String()); - if (status == B_OK) { - fOngoingCommands[commandID] = &command; - status = HandleResponse(&command, timeout); - } - - if (handler != NULL) - fHandlerList.RemoveItem(handler); - - return status; -} +// #pragma mark - private status_t diff --git a/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Protocol.h b/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Protocol.h index 3f65bc1632..b66cdc6817 100644 --- a/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Protocol.h +++ b/src/add-ons/mail_daemon/inbound_protocols/imap/imap_lib/Protocol.h @@ -31,7 +31,6 @@ class Command; class Handler; -typedef BObjectList CommandList; typedef BObjectList HandlerList; typedef std::map CommandIDMap; @@ -61,9 +60,8 @@ public: status_t Disconnect(); bool IsConnected(); - status_t SendCommand(const char* command); - status_t SendCommand(int32 id, const char* command); - ssize_t SendData(const char* buffer, uint32 length); + bool AddHandler(Handler& handler); + void RemoveHandler(Handler& handler); // Some convenience methods status_t GetFolders(FolderList& folders, @@ -74,27 +72,25 @@ public: status_t UnsubscribeFolder(const char* folder); status_t GetQuota(uint64& used, uint64& total); + status_t SendCommand(const char* command); + status_t SendCommand(int32 id, const char* command); + ssize_t SendData(const char* buffer, uint32 length); + status_t ProcessCommand(Command& command, bigtime_t timeout = kIMAP4ClientTimeout); ArgumentList& Capabilities() { return fCapabilities; } const ArgumentList& Capabilities() const { return fCapabilities; } - status_t AddAfterQuakeCommand(Command* command); - const BString& CommandError() { return fCommandError; } protected: status_t HandleResponse(Command* command, bigtime_t timeout = kIMAP4ClientTimeout, bool disconnectOnTimeout = true); - void ProcessAfterQuacks(bigtime_t timeout); int32 NextCommandID(); private: - /*! Same as ProccessCommand but AfterShockCommands are not send. */ - status_t _ProcessCommandWithoutAfterQuake( - Command& command, bigtime_t timeout); status_t _Disconnect(); status_t _GetAllFolders(StringList& folders); void _ParseCapabilities( @@ -105,7 +101,6 @@ protected: BBufferedDataIO* fBufferedSocket; HandlerList fHandlerList; - CommandList fAfterQuackCommands; ArgumentList fCapabilities; private: