Added Copyright header.
[src/xds/xds-server.git] / lib / syncthing / stEvent.go
1 /*
2  * Copyright (C) 2017 "IoT.bzh"
3  * Author Sebastien Douheret <sebastien@iot.bzh>
4  *
5  * Licensed under the Apache License, Version 2.0 (the "License");
6  * you may not use this file except in compliance with the License.
7  * You may obtain a copy of the License at
8  *
9  *   http://www.apache.org/licenses/LICENSE-2.0
10  *
11  * Unless required by applicable law or agreed to in writing, software
12  * distributed under the License is distributed on an "AS IS" BASIS,
13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14  * See the License for the specific language governing permissions and
15  * limitations under the License.
16  */
17
18 package st
19
20 import (
21         "encoding/json"
22         "fmt"
23         "os"
24         "strconv"
25         "strings"
26         "time"
27
28         "github.com/Sirupsen/logrus"
29 )
30
31 // Events .
32 type Events struct {
33         MonitorTime time.Duration
34         Debug       bool
35
36         stop  chan bool
37         st    *SyncThing
38         log   *logrus.Logger
39         cbArr map[string][]cbMap
40 }
41
42 type Event struct {
43         Type string            `json:"type"`
44         Time time.Time         `json:"time"`
45         Data map[string]string `json:"data"`
46 }
47
48 type EventsCBData map[string]interface{}
49 type EventsCB func(ev Event, cbData *EventsCBData)
50
51 const (
52         EventFolderCompletion string = "FolderCompletion"
53         EventFolderSummary    string = "FolderSummary"
54         EventFolderPaused     string = "FolderPaused"
55         EventFolderResumed    string = "FolderResumed"
56         EventFolderErrors     string = "FolderErrors"
57         EventStateChanged     string = "StateChanged"
58 )
59
60 var EventsAll string = EventFolderCompletion + "|" +
61         EventFolderSummary + "|" +
62         EventFolderPaused + "|" +
63         EventFolderResumed + "|" +
64         EventFolderErrors + "|" +
65         EventStateChanged
66
67 type STEvent struct {
68         // Per-subscription sequential event ID. Named "id" for backwards compatibility with the REST API
69         SubscriptionID int `json:"id"`
70         // Global ID of the event across all subscriptions
71         GlobalID int                    `json:"globalID"`
72         Time     time.Time              `json:"time"`
73         Type     string                 `json:"type"`
74         Data     map[string]interface{} `json:"data"`
75 }
76
77 type cbMap struct {
78         id       int
79         cb       EventsCB
80         filterID string
81         data     *EventsCBData
82 }
83
84 // NewEventListener Create a new instance of Event listener
85 func (s *SyncThing) NewEventListener() *Events {
86         _, dbg := os.LookupEnv("XDS_DEBUG_STEVENTS") // set to add more debug log
87         return &Events{
88                 MonitorTime: 100, // in Milliseconds
89                 Debug:       dbg,
90                 stop:        make(chan bool, 1),
91                 st:          s,
92                 log:         s.log,
93                 cbArr:       make(map[string][]cbMap),
94         }
95 }
96
97 // Start starts event monitoring loop
98 func (e *Events) Start() error {
99         go e.monitorLoop()
100         return nil
101 }
102
103 // Stop stops event monitoring loop
104 func (e *Events) Stop() {
105         e.stop <- true
106 }
107
108 // Register Add a listener on an event
109 func (e *Events) Register(evName string, cb EventsCB, filterID string, data *EventsCBData) (int, error) {
110         if evName == "" || !strings.Contains(EventsAll, evName) {
111                 return -1, fmt.Errorf("Unknown event name")
112         }
113         if data == nil {
114                 data = &EventsCBData{}
115         }
116
117         cbList := []cbMap{}
118         if _, ok := e.cbArr[evName]; ok {
119                 cbList = e.cbArr[evName]
120         }
121
122         id := len(cbList)
123         (*data)["id"] = strconv.Itoa(id)
124
125         e.cbArr[evName] = append(cbList, cbMap{id: id, cb: cb, filterID: filterID, data: data})
126
127         return id, nil
128 }
129
130 // UnRegister Remove a listener event
131 func (e *Events) UnRegister(evName string, id int) error {
132         cbKey, ok := e.cbArr[evName]
133         if !ok {
134                 return fmt.Errorf("No event registered to such name")
135         }
136
137         // FIXME - NOT TESTED
138         if id >= len(cbKey) {
139                 return fmt.Errorf("Invalid id")
140         } else if id == len(cbKey) {
141                 e.cbArr[evName] = cbKey[:id-1]
142         } else {
143                 e.cbArr[evName] = cbKey[id : id+1]
144         }
145
146         return nil
147 }
148
149 // GetEvents returns the Syncthing events
150 func (e *Events) getEvents(since int) ([]STEvent, error) {
151         var data []byte
152         ev := []STEvent{}
153         url := "events"
154         if since != -1 {
155                 url += "?since=" + strconv.Itoa(since)
156         }
157         if err := e.st.client.HTTPGet(url, &data); err != nil {
158                 return ev, err
159         }
160         err := json.Unmarshal(data, &ev)
161         return ev, err
162 }
163
164 // Loop to monitor Syncthing events
165 func (e *Events) monitorLoop() {
166         e.log.Infof("Event monitoring running...")
167         since := 0
168         cntErrConn := 0
169         cntErrRetry := 1
170         for {
171                 select {
172                 case <-e.stop:
173                         e.log.Infof("Event monitoring exited")
174                         return
175
176                 case <-time.After(e.MonitorTime * time.Millisecond):
177
178                         if !e.st.Connected {
179                                 cntErrConn++
180                                 time.Sleep(time.Second)
181                                 if cntErrConn > cntErrRetry {
182                                         e.log.Error("ST Event monitor: ST connection down")
183                                         cntErrConn = 0
184                                         cntErrRetry *= 2
185                                         if _, err := e.getEvents(since); err == nil {
186                                                 e.st.Connected = true
187                                                 cntErrRetry = 1
188                                                 // XXX - should we reset since value ?
189                                                 goto readEvent
190                                         }
191                                 }
192                                 continue
193                         }
194
195                 readEvent:
196                         stEvArr, err := e.getEvents(since)
197                         if err != nil {
198                                 e.log.Errorf("Syncthing Get Events: %v", err)
199                                 e.st.Connected = false
200                                 continue
201                         }
202
203                         // Process events
204                         for _, stEv := range stEvArr {
205                                 since = stEv.SubscriptionID
206                                 if e.Debug {
207                                         e.log.Warnf("ST EVENT: %d %s\n  %v", stEv.GlobalID, stEv.Type, stEv)
208                                 }
209
210                                 cbKey, ok := e.cbArr[stEv.Type]
211                                 if !ok {
212                                         continue
213                                 }
214
215                                 evData := Event{
216                                         Type: stEv.Type,
217                                         Time: stEv.Time,
218                                 }
219
220                                 // Decode Events
221                                 // FIXME: re-define data struct for each events
222                                 // instead of map of string and use JSON marshing/unmarshing
223                                 fID := ""
224                                 evData.Data = make(map[string]string)
225                                 switch stEv.Type {
226
227                                 case EventFolderCompletion:
228                                         fID = convString(stEv.Data["folder"])
229                                         evData.Data["completion"] = convFloat64(stEv.Data["completion"])
230
231                                 case EventFolderSummary:
232                                         fID = convString(stEv.Data["folder"])
233                                         evData.Data["needBytes"] = convInt64(stEv.Data["needBytes"])
234                                         evData.Data["state"] = convString(stEv.Data["state"])
235
236                                 case EventFolderPaused, EventFolderResumed:
237                                         fID = convString(stEv.Data["id"])
238                                         evData.Data["label"] = convString(stEv.Data["label"])
239
240                                 case EventFolderErrors:
241                                         fID = convString(stEv.Data["folder"])
242                                         // TODO decode array evData.Data["errors"] = convString(stEv.Data["errors"])
243
244                                 case EventStateChanged:
245                                         fID = convString(stEv.Data["folder"])
246                                         evData.Data["from"] = convString(stEv.Data["from"])
247                                         evData.Data["to"] = convString(stEv.Data["to"])
248
249                                 default:
250                                         e.log.Warnf("Unsupported event type")
251                                 }
252
253                                 if fID != "" {
254                                         evData.Data["id"] = fID
255                                 }
256
257                                 // Call all registered callbacks
258                                 for _, c := range cbKey {
259                                         if e.Debug {
260                                                 e.log.Warnf("EVENT CB fID=%s, filterID=%s", fID, c.filterID)
261                                         }
262                                         // Call when filterID is not set or when it matches
263                                         if c.filterID == "" || (fID != "" && fID == c.filterID) {
264                                                 c.cb(evData, c.data)
265                                         }
266                                 }
267                         }
268                 }
269         }
270 }
271
272 func convString(d interface{}) string {
273         return d.(string)
274 }
275
276 func convFloat64(d interface{}) string {
277         return strconv.FormatFloat(d.(float64), 'f', -1, 64)
278 }
279
280 func convInt64(d interface{}) string {
281         return strconv.FormatInt(d.(int64), 10)
282 }