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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
|
package sfive
import "time"
type appendManyToken struct {
data []StatisticsData
response chan bool
}
type queryStatsResult struct {
stats StatsResult
err error
}
type queryStatsToken struct {
filter *StatsFilter
response chan queryStatsResult
}
type getUpdatesResult struct {
values []StatisticsData
err error
}
type getUpdatesAfterToken struct {
id int
response chan getUpdatesResult
}
type getUpdatesToken struct {
filter *StatsFilter
response chan getUpdatesResult
}
type StatsSinkServer struct {
store sqliteStore
quit chan bool
done chan bool
appendData chan StatisticsData
appendManyData chan appendManyToken // chan []StatisticsData
getStatsChan chan queryStatsToken
getUpdatesAfterChan chan getUpdatesAfterToken
getUpdatesChan chan getUpdatesToken
}
func (self StatsSinkServer) appendActor() {
defer func() { self.done <- true }()
for {
select {
case <-self.quit:
return
case value := <-self.appendData:
var err error
for tryNum := 0; tryNum < 5; tryNum++ {
err = self.store.Append(value)
if err != nil {
time.Sleep(1 * time.Second)
} else {
break
}
}
if err != nil {
s5l.Printf("failed to store data: %v\n", err)
}
case token := <-self.appendManyData:
err := self.store.AppendMany(token.data)
if err != nil {
s5l.Printf("failed to store many data: %v\n", err)
token.response <- false
} else {
token.response <- true
}
case token := <-self.getStatsChan:
stats, err := self.store.GetStats(token.filter)
token.response <- queryStatsResult{stats, err}
case token := <-self.getUpdatesAfterChan:
values, err := self.store.GetUpdatesAfter(token.id)
token.response <- getUpdatesResult{values, err}
case token := <-self.getUpdatesChan:
values, err := self.store.GetUpdates(token.filter)
token.response <- getUpdatesResult{values, err}
}
}
}
func (self StatsSinkServer) getUpdatesAfterInvoke(id int) (values []StatisticsData, err error) {
token := getUpdatesAfterToken{id: id, response: make(chan getUpdatesResult, 1)}
defer close(token.response)
self.getUpdatesAfterChan <- token
res := <-token.response
values = res.values
err = res.err
return
}
func (self StatsSinkServer) getUpdatesInvoke(filter *StatsFilter) (values []StatisticsData, err error) {
token := getUpdatesToken{filter: filter, response: make(chan getUpdatesResult, 1)}
defer close(token.response)
self.getUpdatesChan <- token
res := <-token.response
values = res.values
err = res.err
return
}
func (self StatsSinkServer) getStatsInvoke(filter *StatsFilter) (StatsResult, error) {
token := queryStatsToken{filter: filter, response: make(chan queryStatsResult, 1)}
defer close(token.response)
self.getStatsChan <- token
res := <-token.response
return res.stats, res.err
}
func (self StatsSinkServer) Close() {
self.quit <- true
<-self.done
close(self.quit)
close(self.done)
close(self.appendData)
close(self.appendManyData)
close(self.getStatsChan)
close(self.getUpdatesAfterChan)
close(self.getUpdatesChan)
self.store.Close()
}
func NewServer(mysql bool, dbPath string) (server *StatsSinkServer, err error) {
// TODO read configuration and create instance with correct settings
server = new(StatsSinkServer)
server.store, err = NewStore(mysql, dbPath)
if err != nil {
return
}
server.quit = make(chan bool)
server.done = make(chan bool)
server.appendData = make(chan StatisticsData, 5)
server.appendManyData = make(chan appendManyToken, 5)
server.getStatsChan = make(chan queryStatsToken, 5)
server.getUpdatesAfterChan = make(chan getUpdatesAfterToken, 1)
server.getUpdatesChan = make(chan getUpdatesToken, 3)
go server.appendActor()
return
}
|
