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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
|
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 getHubIdResult struct {
id string
err error
}
type getHubIdToken struct {
response chan getHubIdResult
}
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
getHubIdChan chan getHubIdToken
}
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}
case token := <-self.getHubIdChan:
storeId, err := self.store.GetStoreId()
token.response <- getHubIdResult{storeId, 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) getHubIdInvoke() (string, error) {
token := getHubIdToken{response: make(chan getHubIdResult, 1)}
defer close(token.response)
self.getHubIdChan <- token
res := <-token.response
return res.id, 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)
close(self.getHubIdChan)
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)
server.getHubIdChan = make(chan getHubIdToken, 1)
go server.appendActor()
return
}
|
