library/cpp/messagebus/scheduler_actor.h


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

#pragma once

#include "local_tasks.h"

#include <library/cpp/messagebus/actor/actor.h>
#include <library/cpp/messagebus/actor/what_thread_does_guard.h>
#include <library/cpp/messagebus/scheduler/scheduler.h>

#include <util/system/mutex.h>

namespace NBus {
    namespace NPrivate {
        template <typename TThis, typename TTag = NActor::TDefaultTag>
        class TScheduleActor {
            typedef NActor::TActor<TThis, TTag> TActorForMe;

        private:
            TScheduler* const Scheduler;

            TMutex Mutex;

            TInstant ScheduleTime;

        public:
            TLocalTasks Alarm;

        private:
            struct TScheduleItemImpl: public IScheduleItem {
                TIntrusivePtr<TThis> Thiz;

                TScheduleItemImpl(TIntrusivePtr<TThis> thiz, TInstant when)
                    : IScheduleItem(when)
                    , Thiz(thiz)
                {
                }

                void Do() override {
                    {
                        TWhatThreadDoesAcquireGuard<TMutex> guard(Thiz->Mutex, "scheduler actor: acquiring lock for Do");

                        if (Thiz->ScheduleTime == TInstant::Max()) {
                            // was already fired
                            return;
                        }

                        Thiz->ScheduleTime = TInstant::Max();
                    }

                    Thiz->Alarm.AddTask();
                    Thiz->GetActorForMe()->Schedule();
                }
            };

        public:
            TScheduleActor(TScheduler* scheduler)
                : Scheduler(scheduler)
                , ScheduleTime(TInstant::Max())
            {
            }

            /// call Act(TTag) at specified time, unless it is already scheduled at earlier time.
            void ScheduleAt(TInstant when) {
                TWhatThreadDoesAcquireGuard<TMutex> guard(Mutex, "scheduler: acquiring lock for ScheduleAt");

                if (when > ScheduleTime) {
                    // already scheduled
                    return;
                }

                ScheduleTime = when;
                Scheduler->Schedule(new TScheduleItemImpl(GetThis(), when));
            }

        private:
            TThis* GetThis() {
                return static_cast<TThis*>(this);
            }

            TActorForMe* GetActorForMe() {
                return static_cast<TActorForMe*>(GetThis());
            }
        };

    }
}