IMAP: The connection workers are now started.

* On SyncMessages(), the main connection worker is started if it is not already
  running (it may only run already in idle mode).
* This will then list the subscribed folders, and create as many more connection
  workers as allowed and make sense.
* Finally, it will distribute the folders to the workers which don't do anything
  further yet.
This commit is contained in:
Axel Dörfler
2015-01-06 15:25:05 +01:00
parent bfe83adcb2
commit adbe8fc944
6 changed files with 215 additions and 43 deletions
@@ -1,16 +1,23 @@
/*
* Copyright 2011, Axel Dörfler, [email protected].
* Copyright 2011-2013, Axel Dörfler, [email protected].
* Distributed under the terms of the MIT License.
*/
#include "IMAPConnectionWorker.h"
#include <Autolock.h>
IMAPConnectionWorker::IMAPConnectionWorker(IMAP::Protocol& protocol,
StringList& mailboxes)
#include "IMAPProtocol.h"
IMAPConnectionWorker::IMAPConnectionWorker(IMAPProtocol& owner,
const Settings& settings, bool main)
:
fProtocol(protocol)
fOwner(owner),
fSettings(settings),
fMain(main),
fStopped(false)
{
}
@@ -20,8 +27,55 @@ IMAPConnectionWorker::~IMAPConnectionWorker()
}
uint32
IMAPConnectionWorker::CountMailboxes() const
{
BAutolock locker(const_cast<IMAPConnectionWorker*>(this)->fLocker);
return (fIdleBox.IsEmpty() ? 0 : 1) + fOtherBoxes.size();
}
void
IMAPConnectionWorker::AddMailbox(const BString& name)
{
BAutolock locker(fLocker);
if (fSettings.IdleMode() && fIdleBox.IsEmpty()) {
fIdleBox = name;
} else if (fSettings.IdleMode() && name == "INBOX") {
// Prefer to have the INBOX in idle mode over other mail boxes
fOtherBoxes.push_back(fIdleBox);
fIdleBox = name;
} else
fOtherBoxes.push_back(name);
}
void
IMAPConnectionWorker::RemoveMailbox(const BString& name)
{
BAutolock locker(fLocker);
if (fSettings.IdleMode() && fIdleBox == name) {
if (!fOtherBoxes.empty()) {
fIdleBox = fOtherBoxes[0];
fOtherBoxes.erase(fOtherBoxes.begin());
} else
fIdleBox.SetTo(NULL);
} else {
StringList::iterator iterator = fOtherBoxes.begin();
for (; iterator != fOtherBoxes.end(); iterator++) {
if (*iterator == name) {
fOtherBoxes.erase(iterator);
break;
}
}
}
}
status_t
IMAPConnectionWorker::Start(bool usePush)
IMAPConnectionWorker::Start()
{
fThread = spawn_thread(&_Worker, "imap connection worker",
B_NORMAL_PRIORITY, this);
@@ -36,18 +90,50 @@ IMAPConnectionWorker::Start(bool usePush)
void
IMAPConnectionWorker::Stop()
{
// TODO: we'll also need to interrupt listening to the socket
fStopped = true;
}
status_t
IMAPConnectionWorker::_Worker()
{
status_t status = fProtocol.Connect(fSettings.ServerAddress(),
fSettings.Username(), fSettings.Password(), fSettings.UseSSL());
if (status != B_OK)
return status;
while (!fStopped) {
if (fMain) {
// The main worker checks the subscribed folders, and creates
// other workers as needed
fOwner.CheckSubscribedFolders(fProtocol);
}
BAutolock locker(fLocker);
if (!fIdleBox.IsEmpty())
printf("%p: IDLE: %s\n", this, fIdleBox.String());
StringList::iterator iterator = fOtherBoxes.begin();
for (; iterator != fOtherBoxes.end(); iterator++) {
printf("%p: check: %s\n", this, iterator->String());
}
// TODO: for now
break;
}
return B_OK;
}
/*static*/ status_t
IMAPConnectionWorker::_Worker(void* self)
IMAPConnectionWorker::_Worker(void* _self)
{
return ((IMAPConnectionWorker*)self)->_Worker();
IMAPConnectionWorker* self = (IMAPConnectionWorker*)_self;
status_t status = self->_Worker();
delete self;
return status;
}
@@ -1,5 +1,5 @@
/*
* Copyright 2011, Axel Dörfler, [email protected].
* Copyright 2011-2013, Axel Dörfler, [email protected].
* Distributed under the terms of the MIT License.
*/
#ifndef IMAP_CONNECTION_WORKER_H
@@ -12,13 +12,24 @@
#include "Protocol.h"
class IMAPProtocol;
class Settings;
class IMAPConnectionWorker {
public:
IMAPConnectionWorker(IMAP::Protocol& protocol,
StringList& mailboxes);
IMAPConnectionWorker(IMAPProtocol& owner,
const Settings& settings,
bool main = false);
virtual ~IMAPConnectionWorker();
status_t Start(bool usePush);
uint32 CountMailboxes() const;
void AddMailbox(const BString& name);
void RemoveMailbox(const BString& name);
bool IsMain() const { return fMain; }
status_t Start();
void Stop();
private:
@@ -26,12 +37,16 @@ private:
static status_t _Worker(void* self);
private:
IMAP::Protocol& fProtocol;
IMAPProtocol& fOwner;
const Settings& fSettings;
IMAP::Protocol fProtocol;
BString fIdleBox;
StringList fOtherBoxes;
BLocker fLocker;
thread_id fThread;
bool fMain;
bool fStopped;
};
@@ -16,8 +16,6 @@ IMAPProtocol::IMAPProtocol(const BMailAccountSettings& settings)
BInboundMailProtocol(settings),
fSettings(settings.InboundSettings())
{
puts("IMAP protocol started");
BPath destination = fSettings.Destination();
status_t status = create_directory(destination.Path(), 0755);
@@ -35,11 +33,77 @@ IMAPProtocol::~IMAPProtocol()
}
status_t
IMAPProtocol::CheckSubscribedFolders(IMAP::Protocol& protocol)
{
// Get list of subscribed folders
StringList folders;
status_t status = protocol.GetSubscribedFolders(folders);
if (status != B_OK)
return status;
// Determine how many new mailboxes we have
StringList::iterator iterator = folders.begin();
for (; iterator != folders.end(); iterator++) {
if (fKnownMailboxes.find(*iterator) != fKnownMailboxes.end())
iterator = folders.erase(iterator);
}
if (fSettings.IdleMode()) {
// Create connection workers as allowed
int32 totalMailboxes = fKnownMailboxes.size() + folders.size();
while (fWorkers.CountItems() < fSettings.MaxConnections()
&& fWorkers.CountItems() < totalMailboxes) {
IMAPConnectionWorker* worker = new IMAPConnectionWorker(*this,
fSettings);
if (!fWorkers.AddItem(worker)) {
delete worker;
break;
}
worker->Start();
}
}
// Distribute the new mailboxes to the existing workers
int32 index = 0;
while (!folders.empty()) {
BString folder = folders[0];
folders.erase(folders.begin());
fWorkers.ItemAt(index)->AddMailbox(folder);
fKnownMailboxes.insert(folder);
index = (index + 1) % fWorkers.CountItems();
}
return B_OK;
}
status_t
IMAPProtocol::SyncMessages()
{
puts("IMAP: sync");
return B_ERROR;
if (fWorkers.IsEmpty()) {
// Create main (and possibly initial) connection worker
IMAPConnectionWorker* worker = new IMAPConnectionWorker(*this,
fSettings, true);
if (!fWorkers.AddItem(worker)) {
delete worker;
return B_NO_MEMORY;
}
return worker->Start();
}
return B_OK;
}
@@ -82,6 +146,10 @@ IMAPProtocol::MessageReceived(BMessage* message)
case B_READY_TO_RUN:
ReadyToRun();
break;
default:
BInboundMailProtocol::MessageReceived(message);
break;
}
}
@@ -89,14 +157,9 @@ IMAPProtocol::MessageReceived(BMessage* message)
void
IMAPProtocol::ReadyToRun()
{
// Determine how many connection workers we'll need
// TODO: in passive mode, this should be done on every sync
IMAP::Protocol protocol;
status_t status = protocol.Connect(fSettings.ServerAddress(),
fSettings.Username(), fSettings.Password(), fSettings.UseSSL());
if (status != B_OK)
return;
puts("IMAP: ready to run!");
if (fSettings.IdleMode())
SyncMessages();
}
@@ -15,6 +15,11 @@
class IMAPConnectionWorker;
namespace IMAP {
class Protocol;
}
typedef std::set<BString> StringSet;
@@ -24,6 +29,9 @@ public:
const BMailAccountSettings& settings);
virtual ~IMAPProtocol();
status_t CheckSubscribedFolders(
IMAP::Protocol& protocol);
virtual status_t SyncMessages();
virtual status_t FetchBody(const entry_ref& ref);
virtual status_t MarkMessageAsRead(const entry_ref& ref,
@@ -1,5 +1,5 @@
/*
* Copyright 2010-2012, Haiku Inc. All Rights Reserved.
* Copyright 2010-2013, Haiku Inc. All Rights Reserved.
* Copyright 2010 Clemens Zeidler. All rights reserved.
*
* Distributed under the terms of the MIT License.
@@ -118,11 +118,11 @@ Protocol::GetFolders(FolderList& folders)
return status;
StringList subscribedFolders;
status = _GetSubscribedFolders(subscribedFolders);
status = GetSubscribedFolders(subscribedFolders);
if (status != B_OK)
return status;
for (unsigned int i = 0; i < allFolders.size(); i++) {
for (size_t i = 0; i < allFolders.size(); i++) {
FolderEntry entry;
entry.folder = allFolders[i];
for (unsigned int a = 0; a < subscribedFolders.size(); a++) {
@@ -136,9 +136,9 @@ Protocol::GetFolders(FolderList& folders)
}
// you could be subscribed to a folder which not exist currently, add them:
for (unsigned int a = 0; a < subscribedFolders.size(); a++) {
for (size_t a = 0; a < subscribedFolders.size(); a++) {
bool isInlist = false;
for (unsigned int i = 0; i < allFolders.size(); i++) {
for (size_t i = 0; i < allFolders.size(); i++) {
if (subscribedFolders[a] == allFolders[i]) {
isInlist = true;
break;
@@ -157,6 +157,19 @@ Protocol::GetFolders(FolderList& folders)
}
status_t
Protocol::GetSubscribedFolders(StringList& folders)
{
ListCommand command(NULL, true);
status_t status = ProcessCommand(command);
if (status != B_OK)
return status;
folders = command.FolderList();
return status;
}
status_t
Protocol::SubscribeFolder(const char* folder)
{
@@ -371,19 +384,6 @@ Protocol::_GetAllFolders(StringList& folders)
}
status_t
Protocol::_GetSubscribedFolders(StringList& folders)
{
ListCommand command(NULL, true);
status_t status = ProcessCommand(command);
if (status != B_OK)
return status;
folders = command.FolderList();
return status;
}
void
Protocol::_ParseCapabilities(const ArgumentList& arguments)
{
@@ -1,5 +1,5 @@
/*
* Copyright 2001-2011, Haiku Inc. All Rights Reserved.
* Copyright 2001-2013, Haiku Inc. All Rights Reserved.
* Copyright 2001-2002 Dr. Zoidberg Enterprises. All rights reserved.
* Copyright 2010 Clemens Zeidler. All rights reserved.
*
@@ -67,6 +67,7 @@ public:
// Some convenience methods
status_t GetFolders(FolderList& folders);
status_t GetSubscribedFolders(StringList& folders);
status_t SubscribeFolder(const char* folder);
status_t UnsubscribeFolder(const char* folder);
status_t GetQuota(uint64& used, uint64& total);
@@ -94,7 +95,6 @@ private:
Command& command, bigtime_t timeout);
status_t _Disconnect();
status_t _GetAllFolders(StringList& folders);
status_t _GetSubscribedFolders(StringList& folders);
void _ParseCapabilities(
const ArgumentList& arguments);