/
event_processor.h
129 lines (102 loc) · 4.43 KB
/
event_processor.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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
// Copyright (c) 2011, François Saint-Jacques
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
// * Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
// * Redistributions in binary form must reproduce the above copyright
// notice, this list of conditions and the following disclaimer in the
// documentation and/or other materials provided with the distribution.
// * Neither the name of the disruptor-- nor the
// names of its contributors may be used to endorse or promote products
// derived from this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL FRANÇOIS SAINT-JACQUES BE LIABLE FOR ANY
// DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
// (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
// LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
// ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#ifndef DISRUPTOR_EVENT_PROCESSOR_H_ // NOLINT
#define DISRUPTOR_EVENT_PROCESSOR_H_ // NOLINT
#include <exception>
#include "disruptor/ring_buffer.h"
namespace disruptor {
template <typename T>
class NoOpEventProcessor : public EventProcessorInterface<T> {
public:
NoOpEventProcessor(RingBuffer<T>* ring_buffer) :
ring_buffer_(ring_buffer) { }
virtual Sequence* GetSequence() {
return ring_buffer_->GetSequencePtr();
}
virtual void Halt() {}
virtual void Run() {}
private:
RingBuffer<T>* ring_buffer_;
};
template <typename T>
class BatchEventProcessor : public EventProcessorInterface<T> {
public:
BatchEventProcessor(RingBuffer<T>* ring_buffer,
SequenceBarrierInterface* sequence_barrier,
EventHandlerInterface<T>* event_handler,
ExceptionHandlerInterface<T>* exception_handler) :
running_(false),
ring_buffer_(ring_buffer),
sequence_barrier_(sequence_barrier),
event_handler_(event_handler),
exception_handler_(exception_handler) {}
virtual Sequence* GetSequence() { return &sequence_; }
virtual void Halt() {
running_.store(false);
sequence_barrier_->Alert();
}
virtual void Run() {
if (running_.load())
throw std::runtime_error("Thread is already running");
running_.store(true);
sequence_barrier_->ClearAlert();
event_handler_->OnStart();
T* event = NULL;
int64_t next_sequence = sequence_.sequence() + 1L;
while (true) {
try {
int64_t avalaible_sequence = \
sequence_barrier_->WaitFor(next_sequence);
while (next_sequence <= avalaible_sequence) {
event = ring_buffer_->Get(next_sequence);
event_handler_->OnEvent(next_sequence,
next_sequence == avalaible_sequence, event);
next_sequence++;
}
sequence_.set_sequence(next_sequence - 1L);
} catch(const AlertException& e) {
if (!running_.load())
break;
} catch(const std::exception& e) {
exception_handler_->Handle(e, next_sequence, event);
sequence_.set_sequence(next_sequence);
next_sequence++;
}
}
event_handler_->OnShutdown();
running_.store(false);
}
void operator()() { Run(); }
private:
std::atomic<bool> running_;
Sequence sequence_;
RingBuffer<T>* ring_buffer_;
SequenceBarrierInterface* sequence_barrier_;
EventHandlerInterface<T>* event_handler_;
ExceptionHandlerInterface<T>* exception_handler_;
DISALLOW_COPY_AND_ASSIGN(BatchEventProcessor);
};
}; // namespace disruptor
#endif // DISRUPTOR_EVENT_PROCESSOR_H_ NOLINT