* Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved.
* You can use this software according to the terms and conditions of the Mulan PSL v2.
* You may obtain a copy of Mulan PSL v2 at:
* http://license.coscl.org.cn/MulanPSL2
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PSL v2 for more details.
*/
#ifndef MAILBOX_PROCESSOR_H
#define MAILBOX_PROCESSOR_H
#include <memory>
#include <utility>
#include <vector>
#include <string>
#include "TaskMailbox.h"
#include "MailboxDefaultAction.h"
#include "StreamTaskActionExecutor.h"
#include "MailboxExecutor.h"
#include "PeriodTimer.h"
namespace omnistream {
class MailboxProcessor {
public:
MailboxProcessor();
explicit MailboxProcessor(std::unique_ptr<MailboxDefaultAction> mailboxDefaultAction);
MailboxProcessor(
std::unique_ptr<MailboxDefaultAction> mailboxDefaultAction,
std::shared_ptr<StreamTaskActionExecutor> actionExecutor);
MailboxProcessor(
std::unique_ptr<MailboxDefaultAction> mailboxDefaultAction,
TaskMailbox* mailbox,
std::shared_ptr<StreamTaskActionExecutor> actionExecutor);
~MailboxProcessor();
std::shared_ptr<MailboxExecutor> getMainMailboxExecutor();
std::shared_ptr<MailboxExecutor> getMailboxExecutor(int priority);
inline void prepareClose();
void close();
inline void drain();
void runMailboxLoop();
void suspend();
bool runMailboxStep();
bool isMailboxThread();
void reportThrowable(std::exception_ptr throwable);
void allActionsCompleted();
bool isDefaultActionAvailable()
{
return suspendedDefaultAction == nullptr;
}
bool isMailboxLoopRunning();
bool hasMail();
std::string toString() const;
void sendPoisonMail(std::shared_ptr<ThrowingRunnable> mail);
void sendPoisonMail(std::shared_ptr<ThrowingRunnable> mail, const std::string& descriptionFormat);
void sendControlMail(
std::shared_ptr<ThrowingRunnable> mail,
const std::string& descriptionFormat,
const std::vector<std::string>& descriptionArgs = {});
bool processMail(TaskMailbox* mailbox, bool singleStep);
bool processMailsWhenDefaultActionUnavailable();
bool processMailsNonBlocking(bool singleStep);
void maybePauseIdleTimer();
void maybeRestartIdleTimer();
std::shared_ptr<MailboxDefaultAction::Suspension> suspendDefaultAction(
std::shared_ptr<PeriodTimer> suspensionTimer);
bool isNextLoopPossible() const
{
return !suspended;
}
class DefaultActionSuspension : public MailboxDefaultAction::Suspension {
public:
explicit DefaultActionSuspension(
MailboxProcessor* mailboxProcessor, std::shared_ptr<PeriodTimer> suspensionTimer)
: mailboxProcessor_(mailboxProcessor),
suspensionTimer_(std::move(suspensionTimer))
{
}
void resume() override;
std::shared_ptr<PeriodTimer> getSuspensionTimer()
{
return suspensionTimer_;
};
private:
MailboxProcessor* mailboxProcessor_;
std::shared_ptr<PeriodTimer> suspensionTimer_;
};
class lambdaHelper {
public:
explicit lambdaHelper(MailboxProcessor* mailbox_processor) : processor_(mailbox_processor) {};
MailboxProcessor* processor_;
void suspended(bool suspended)
{
processor_->suspended = suspended;
}
void mailboxLoopRunning(bool mailboxLoopRunning)
{
processor_->mailboxLoopRunning = mailboxLoopRunning;
}
void resume()
{
processor_->suspendedDefaultAction = nullptr;
}
TaskMailbox* getMailbox()
{
return processor_->mailbox_;
}
MailboxProcessor* getOwner()
{
return processor_;
}
};
private:
TaskMailbox* mailbox_ = nullptr;
std::unique_ptr<MailboxDefaultAction> mailboxDefaultAction;
bool suspended;
bool mailboxLoopRunning;
std::shared_ptr<DefaultActionSuspension> suspendedDefaultAction;
std::shared_ptr<StreamTaskActionExecutor> actionExecutor;
int workaround = 0;
};
}
#endif