Properly send info between main and dl/parsing threads

TODO.txt

@@ -8,7 +8,6 @@ Important Features:
 
 
 Important improvements:
-* Stop locking during updates
 * Stop memory accumulation
 	* A few MB accumulate on every update
 	* ime it's than .1 MB per feed― but if you have several feeds, it can
@@ -16,7 +15,6 @@ Important improvements:
 	  and updates every 30 minutes, it got to about 200MB after a week or
 	  so.
 	* _Huge_ problem, but I haven't had luck figuring it out yet…
-* Proper queue list (see Multiple downloads at once?)
 * No hardcoded paths
 * General input sanitization
 * File error-handling
@@ -29,7 +27,6 @@ Important improvements:
 
 
 Nice Improvements:
-* 'About' window
 * Save window positions
 	* Main window
 	* Subscription-edit window
@@ -37,6 +34,8 @@ Nice Improvements:
 * Store feeds in kEnqueueFeeds/etc messages as paths? hmm
 	* This way, if the user edits the feed after it is enqueued but before
 	processing, the changes will be applied.
+* Move the provisional BList download queue to a BJob + JobQueue system
+	* This might add some more flexibility
 * Multiple downloads at once?
 	* Array of thread_ids for multiple DL threads?
 		* In which case, it'd probably make sense to only spin them

src/App.cpp

@@ -55,6 +55,7 @@ App::MessageReceived(BMessage* msg)
 	switch (msg->what)
 	{
 		case kUpdateSubscribed:
+		case kControllerCheck:
 		{
 			fFeedController->MessageReceived(msg);
 			break;

src/FeedController.cpp

@@ -7,6 +7,7 @@
 
 #include <Directory.h>
 #include <Message.h>
+#include <MessageRunner.h>
 #include <Notification.h>
 
 #include <iostream>
@@ -19,13 +20,16 @@
 
 FeedController::FeedController()
 	:
+	fMainThread(find_thread(NULL)),
 	fDownloadThread(0),
-	fParseThread(0)
+	fParseThread(0),
+	fDownloadQueue(new BList()),
+	fMessageRunner(new BMessageRunner(be_app, BMessage(kControllerCheck), 50000, -1))
 {
 	fDownloadThread = spawn_thread(_DownloadLoop, "here, eat this",
-		B_NORMAL_PRIORITY, NULL);
+		B_NORMAL_PRIORITY, &fMainThread);
 	fParseThread = spawn_thread(_ParseLoop, "oki tnx nomnomnom",
-		B_NORMAL_PRIORITY, NULL);
+		B_NORMAL_PRIORITY, &fMainThread);
 	resume_thread(fDownloadThread);
 	resume_thread(fParseThread);
 }
@@ -51,7 +55,7 @@ FeedController::MessageReceived(BMessage* msg)
 
 			while (msg->HasData("feeds", B_RAW_TYPE, i)) {
 				msg->FindData("feeds", B_RAW_TYPE, i, &data, &size);
-				send_data(fDownloadThread, msg->what, data, size);
+				_EnqueueFeed((Feed*)data);
 				i++;
 			}
 			break;
@@ -60,10 +64,7 @@ FeedController::MessageReceived(BMessage* msg)
 		{
 			BList subFeeds = SubscribedFeeds();
 			for (int i = 0; i < subFeeds.CountItems(); i++) {
-				BMessage getFeed(kEnqueueFeed);
+				_EnqueueFeed((Feed*)subFeeds.ItemAt(i));
-				getFeed.AddData("feeds", B_RAW_TYPE, subFeeds.ItemAt(i),
-					sizeof(Feed));
-				((App*)be_app)->MessageReceived(&getFeed);
 			}
 			break;
 		}
@@ -71,25 +72,10 @@ FeedController::MessageReceived(BMessage* msg)
 		{
 			break;
 		}
-		case kDownloadComplete:
+		case kControllerCheck:
 		{
-			int i = 0;
+			_ProcessQueueItem();
-			const void* data;
+			_CheckStatus();
-			ssize_t size = sizeof(Feed);
-
-			while (msg->HasData("feeds", B_RAW_TYPE, i)) {
-				msg->FindData("feeds", B_RAW_TYPE, i, &data, &size);
-
-				if (((Feed*)data)->IsUpdated() == true)
-					send_data(fParseThread, msg->what, data, size);
-				else {
-					BMessage complete(kParseComplete);
-					complete.AddString("feed_name", ((Feed*)data)->GetTitle());
-					complete.AddInt32("entry_count", 0);
-					((App*)be_app)->MessageReceived(&complete);
-				}
-				i++;
-			}
 			break;
 		}
 		default:
@@ -114,29 +100,96 @@ FeedController::SubscribedFeeds()
 }
 
 
-int32
+void
-FeedController::_DownloadLoop(void* ignored)
+FeedController::_EnqueueFeed(Feed* feed)
 {
+	fMessageRunner->SetCount(-1);
+	fDownloadQueue->AddItem(feed);
+}
+
+
+void
+FeedController::_ProcessQueueItem()
+{
+	if (has_data(fDownloadThread) && !fDownloadQueue->IsEmpty()) {
+		Feed* buffer = (Feed*)(fDownloadQueue->RemoveItem(0));
+		send_data(fDownloadThread, 0, (void*)buffer, sizeof(Feed));
+
+		BMessage downloadInit = BMessage(kDownloadStart);
+		downloadInit.AddString("feed", buffer->GetTitle());
+		((App*)be_app)->MessageReceived(&downloadInit);
+	}
+}
+
+
+void
+FeedController::_CheckStatus()
+{
+	thread_id sender;
+
+	while (has_data(find_thread(NULL))) {
+		Feed* feedBuffer = (Feed*)malloc(sizeof(Feed));
+		int32 code = receive_data(&sender, (void*)feedBuffer, sizeof(Feed));
+
+		switch (code)
+		{
+			case kDownloadComplete: 
+			{
+				BMessage complete = BMessage(kDownloadComplete);
+				complete.AddString("feed_url",
+					feedBuffer->GetXmlUrl().UrlString());
+				((App*)be_app)->MessageReceived(&complete);
+
+				send_data(fParseThread, 0, (void*)feedBuffer, sizeof(Feed));
+				break;
+			}
+			case kDownloadFail:
+			{
+				BMessage failure = BMessage(kDownloadFail);
+				failure.AddString("feed_url",
+					feedBuffer->GetXmlUrl().UrlString());
+				((App*)be_app)->MessageReceived(&failure);
+				break;
+			}
+			case kParseComplete:
+			{
+				BMessage complete = BMessage(kParseComplete);
+				complete.AddString("feed_name", feedBuffer->GetTitle());
+				complete.AddInt32("entry_count", feedBuffer->GetNewEntries().CountItems());
+				((App*)be_app)->MessageReceived(&complete);
+				break;
+			}
+			case kParseFail:
+			{
+				BMessage failure = BMessage(kParseFail);
+				failure.AddString("feed_url", feedBuffer->GetXmlUrl().UrlString());
+				((App*)be_app)->MessageReceived(&failure);
+				break;
+			}
+		}
+	}
+}
+
+
+int32
+FeedController::_DownloadLoop(void* data)
+{
+	thread_id main = *((thread_id*)data);
 	thread_id sender;
 	Feed* feedBuffer = (Feed*)malloc(sizeof(Feed));
 
-	while (receive_data(&sender, (void*)feedBuffer, sizeof(Feed)) != 0) {
+
+	while (true) {
+		int32 code = receive_data(&sender, (void*)feedBuffer, sizeof(Feed));
+
 		std::cout << "Downloading feed from "
 			<< feedBuffer->GetXmlUrl().UrlString() << "…\n";
 
-		BMessage downloadInit = BMessage(kDownloadStart);
-		downloadInit.AddString("feed", feedBuffer->GetTitle());
-		((App*)be_app)->MessageReceived(&downloadInit);
-
 		if (feedBuffer->Fetch()) {
-			BMessage downloaded = BMessage(kDownloadComplete);
+			send_data(main, kDownloadComplete, (void*)feedBuffer, sizeof(Feed));
-			downloaded.AddData("feeds", B_RAW_TYPE, feedBuffer, sizeof(Feed));
-			((App*)be_app)->MessageReceived(&downloaded);
 		}
 		else {
-			BMessage failure = BMessage(kDownloadFail);
+			send_data(main, kDownloadFail, (void*)feedBuffer, sizeof(Feed));
-			failure.AddString("feed_url", feedBuffer->GetXmlUrl().UrlString());
-			((App*)be_app)->MessageReceived(&failure);
 		}
 	}
 	delete(feedBuffer);
@@ -145,12 +198,15 @@ FeedController::_DownloadLoop(void* ignored)
 
 
 int32
-FeedController::_ParseLoop(void* ignored)
+FeedController::_ParseLoop(void* data)
 {
+	thread_id main = *((thread_id*)data);
 	thread_id sender;
 	Feed* feedBuffer = (Feed*)malloc(sizeof(Feed));
 
-	while (receive_data(&sender, (void*)feedBuffer, sizeof(Feed)) != 0) {
+	while (true) {
+		int32 code = receive_data(&sender, (void*)feedBuffer, sizeof(Feed));
+
 		BList entries;
 		BString feedTitle;
 		BUrl feedUrl = feedBuffer->GetXmlUrl();
@@ -181,15 +237,10 @@ FeedController::_ParseLoop(void* ignored)
 
 
 		if (feedBuffer->IsAtom() || feedBuffer->IsRss()) {
-			BMessage complete = BMessage(kParseComplete);
+			send_data(main, kParseComplete, (void*)feedBuffer, sizeof(Feed));
-			complete.AddString("feed_name", feedTitle);
-			complete.AddInt32("entry_count", entries.CountItems());
-			((App*)be_app)->MessageReceived(&complete);
 		}
 		else {
-			BMessage failure = BMessage(kParseFail);
+			send_data(main, kParseFail, (void*)feedBuffer, sizeof(Feed));
-			failure.AddString("feed_url", feedUrl.UrlString());
-			((App*)be_app)->MessageReceived(&failure);
 		}
 	}
 

src/FeedController.h

@@ -10,6 +10,8 @@
 
 class BList;
 class BMessage;
+class BMessageRunner;
+class Feed;
 
 
 enum
@@ -21,7 +23,8 @@ enum
 	kDownloadFail		= 'fdlf',
 	kParseComplete		= 'fpec',
 	kParseFail			= 'fpef',
-	kUpdateSubscribed	= 'fups'
+	kUpdateSubscribed	= 'fups',
+	kControllerCheck	= 'coch'
 };
 
 
@@ -31,14 +34,23 @@ public:
 	~FeedController();
 
 	void MessageReceived(BMessage* msg);
+
 	static BList SubscribedFeeds();
 
 private:
-	static int32 _DownloadLoop(void* ignored);
+	static int32 _DownloadLoop(void* data);
-	static int32 _ParseLoop(void* ignored);
+	static int32 _ParseLoop(void* data);
 
+	void _EnqueueFeed(Feed* feed);
+	void _ProcessQueueItem();
+	void _CheckStatus();
+
+	thread_id	fMainThread;
 	thread_id	fDownloadThread;
 	thread_id	fParseThread;
+
+	BList*			fDownloadQueue;
+	BMessageRunner*	fMessageRunner;
 };