Split db_mysql_live into two modules so other modules can make use of the asynchronous command interface

1 files changed, 137 insertions, 0 deletions

diff --git a/modules/extra/m_async_commands.cpp b/modules/extra/m_async_commands.cpp
new file mode 100644
index 000000000..ddc660ea8
--- /dev/null
+++ b/modules/extra/m_async_commands.cpp
@@ -0,0 +1,137 @@
+#include "module.h"
+#include "async_commands.h"
+
+static Pipe *me;
+static CommandMutex *current_command = NULL;
+static std::list<CommandMutex *> commands;
+/* Mutex held by the core when it is processing. Used by threads to halt the core */
+static Mutex main_mutex;
+
+class AsynchCommandMutex : public CommandMutex
+{
+ public:
+	AsynchCommandMutex() : CommandMutex()
+	{
+		commands.push_back(this);
+		current_command = this;
+	}
+
+	~AsynchCommandMutex()
+	{
+		std::list<CommandMutex *>::iterator it = std::find(commands.begin(), commands.end(), this);
+		if (it != commands.end())
+			commands.erase(it);
+		if (this == current_command)
+			current_command = NULL;
+	}
+
+	void Run()
+	{
+		User *u = this->source.u;
+		BotInfo *bi = this->source.owner;
+
+		if (!command->permission.empty() && !u->Account()->HasCommand(command->permission))
+		{
+			u->SendMessage(bi, LanguageString::ACCESS_DENIED);
+			Log(LOG_COMMAND, "denied", bi) << "Access denied for user " << u->GetMask() << " with command " << command;
+		}
+		else
+		{
+			CommandReturn ret = command->Execute(source, params);
+
+			if (ret == MOD_CONT)
+			{
+				FOREACH_MOD(I_OnPostCommand, OnPostCommand(source, command, params));
+			}
+		}
+
+		main_mutex.Unlock();
+	}
+
+	void Lock()
+	{
+		this->processing = true;
+		me->Notify();
+		this->mutex.Lock();
+	}
+
+	void Unlock()
+	{
+		this->processing = false;
+		main_mutex.Unlock();
+	}
+};
+
+
+class ModuleAsynchCommands : public Module, public Pipe, public AsynchCommandsService
+{
+ public:
+	ModuleAsynchCommands(const Anope::string &modname, const Anope::string &creator) : Module(modname, creator), Pipe(), AsynchCommandsService(this, "asynch_commands") 
+	{
+		me = this;
+
+		this->SetPermanent(true);
+
+		main_mutex.Lock();
+
+		Implementation i[] = { I_OnPreCommand };
+		ModuleManager::Attach(i, this, 1);
+
+		ModuleManager::RegisterService(this);
+	}
+
+	EventReturn OnPreCommand(CommandSource &source, Command *command, const std::vector<Anope::string> &params)
+	{
+		AsynchCommandMutex *cm = new AsynchCommandMutex();
+		try
+		{
+			cm->mutex.Lock();
+			cm->command = command;
+			cm->source = source;
+			cm->params = params;
+
+			// Give processing to the command thread
+			Log(LOG_DEBUG_2) << "Waiting for command thread " << cm->command->name << " from " << source.u->nick;
+			threadEngine.Start(cm);
+			main_mutex.Lock();
+
+			return EVENT_STOP;
+		}
+		catch (const CoreException &ex)
+		{
+			delete cm;
+			Log() << "Unable to thread for command: " << ex.GetReason();
+		}
+
+		return EVENT_CONTINUE;
+	}
+
+	void OnNotify()
+	{
+		for (std::list<CommandMutex *>::iterator it = commands.begin(), it_end = commands.end(); it != it_end; ++it)
+		{
+			CommandMutex *cm = *it;
+
+			// Thread engine will pick this up later
+			if (cm->GetExitState() || !cm->processing)
+				continue;
+
+			Log(LOG_DEBUG_2) << "Waiting for command thread " << cm->command->name << " from " << cm->source.u->nick;
+			current_command = cm;
+
+			// Unlock to give processing back to the command thread
+			cm->mutex.Unlock();
+			// Relock to regain processing once the command thread hangs for any reason
+			main_mutex.Lock();
+
+			current_command = NULL;
+		}
+	}
+
+	CommandMutex *CurrentCommand()
+	{
+		return current_command;
+	}
+};
+
+MODULE_INIT(ModuleAsynchCommands)