source: code/trunk/user.go@ 274

Last change on this file since 274 was 267, checked in by contact, 5 years ago

Add network.channels, remove DB.GetChannel

Store the list of configured channels in the network data structure.
This removes the need for a database lookup and will be useful for
detached channels.
File size: 9.2 KB
Line 
1package soju
2
3import (
4 "fmt"
5 "time"
6
7 "gopkg.in/irc.v3"
8)
9
10type event interface{}
11
12type eventUpstreamMessage struct {
13 msg *irc.Message
14 uc *upstreamConn
15}
16
17type eventUpstreamConnectionError struct {
18 net *network
19 err error
20}
21
22type eventUpstreamConnected struct {
23 uc *upstreamConn
24}
25
26type eventUpstreamDisconnected struct {
27 uc *upstreamConn
28}
29
30type eventUpstreamError struct {
31 uc *upstreamConn
32 err error
33}
34
35type eventDownstreamMessage struct {
36 msg *irc.Message
37 dc *downstreamConn
38}
39
40type eventDownstreamConnected struct {
41 dc *downstreamConn
42}
43
44type eventDownstreamDisconnected struct {
45 dc *downstreamConn
46}
47
48type networkHistory struct {
49 offlineClients map[string]uint64 // indexed by client name
50 ring *Ring // can be nil if there are no offline clients
51}
52
53type network struct {
54 Network
55 user *user
56 stopped chan struct{}
57
58 conn *upstreamConn
59 channels map[string]*Channel
60 history map[string]*networkHistory // indexed by entity
61 offlineClients map[string]struct{} // indexed by client name
62 lastError error
63}
64
65func newNetwork(user *user, record *Network, channels []Channel) *network {
66 m := make(map[string]*Channel, len(channels))
67 for _, ch := range channels {
68 m[ch.Name] = &ch
69 }
70
71 return &network{
72 Network: *record,
73 user: user,
74 stopped: make(chan struct{}),
75 channels: m,
76 history: make(map[string]*networkHistory),
77 offlineClients: make(map[string]struct{}),
78 }
79}
80
81func (net *network) forEachDownstream(f func(*downstreamConn)) {
82 net.user.forEachDownstream(func(dc *downstreamConn) {
83 if dc.network != nil && dc.network != net {
84 return
85 }
86 f(dc)
87 })
88}
89
90func (net *network) run() {
91 var lastTry time.Time
92 for {
93 select {
94 case <-net.stopped:
95 return
96 default:
97 // This space is intentionally left blank
98 }
99
100 if dur := time.Now().Sub(lastTry); dur < retryConnectMinDelay {
101 delay := retryConnectMinDelay - dur
102 net.user.srv.Logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
103 time.Sleep(delay)
104 }
105 lastTry = time.Now()
106
107 uc, err := connectToUpstream(net)
108 if err != nil {
109 net.user.srv.Logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
110 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to connect: %v", err)}
111 continue
112 }
113
114 uc.register()
115 if err := uc.runUntilRegistered(); err != nil {
116 uc.logger.Printf("failed to register: %v", err)
117 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to register: %v", err)}
118 uc.Close()
119 continue
120 }
121
122 net.user.events <- eventUpstreamConnected{uc}
123 if err := uc.readMessages(net.user.events); err != nil {
124 uc.logger.Printf("failed to handle messages: %v", err)
125 net.user.events <- eventUpstreamError{uc, fmt.Errorf("failed to handle messages: %v", err)}
126 }
127 uc.Close()
128 net.user.events <- eventUpstreamDisconnected{uc}
129 }
130}
131
132func (net *network) upstream() *upstreamConn {
133 return net.conn
134}
135
136func (net *network) Stop() {
137 select {
138 case <-net.stopped:
139 return
140 default:
141 close(net.stopped)
142 }
143
144 if uc := net.upstream(); uc != nil {
145 uc.Close()
146 }
147}
148
149func (net *network) createUpdateChannel(ch *Channel) error {
150 if current, ok := net.channels[ch.Name]; ok {
151 ch.ID = current.ID // update channel if it already exists
152 }
153 if err := net.user.srv.db.StoreChannel(net.ID, ch); err != nil {
154 return err
155 }
156 net.channels[ch.Name] = ch
157 return nil
158}
159
160func (net *network) deleteChannel(name string) error {
161 if err := net.user.srv.db.DeleteChannel(net.ID, name); err != nil {
162 return err
163 }
164 delete(net.channels, name)
165 return nil
166}
167
168type user struct {
169 User
170 srv *Server
171
172 events chan event
173
174 networks []*network
175 downstreamConns []*downstreamConn
176
177 // LIST commands in progress
178 pendingLISTs []pendingLIST
179}
180
181type pendingLIST struct {
182 downstreamID uint64
183 // list of per-upstream LIST commands not yet sent or completed
184 pendingCommands map[int64]*irc.Message
185}
186
187func newUser(srv *Server, record *User) *user {
188 return &user{
189 User: *record,
190 srv: srv,
191 events: make(chan event, 64),
192 }
193}
194
195func (u *user) forEachNetwork(f func(*network)) {
196 for _, network := range u.networks {
197 f(network)
198 }
199}
200
201func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
202 for _, network := range u.networks {
203 uc := network.upstream()
204 if uc == nil {
205 continue
206 }
207 f(uc)
208 }
209}
210
211func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
212 for _, dc := range u.downstreamConns {
213 f(dc)
214 }
215}
216
217func (u *user) getNetwork(name string) *network {
218 for _, network := range u.networks {
219 if network.Addr == name {
220 return network
221 }
222 if network.Name != "" && network.Name == name {
223 return network
224 }
225 }
226 return nil
227}
228
229func (u *user) run() {
230 networks, err := u.srv.db.ListNetworks(u.Username)
231 if err != nil {
232 u.srv.Logger.Printf("failed to list networks for user %q: %v", u.Username, err)
233 return
234 }
235
236 for _, record := range networks {
237 channels, err := u.srv.db.ListChannels(record.ID)
238 if err != nil {
239 u.srv.Logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
240 }
241
242 network := newNetwork(u, &record, channels)
243 u.networks = append(u.networks, network)
244
245 go network.run()
246 }
247
248 for e := range u.events {
249 switch e := e.(type) {
250 case eventUpstreamConnected:
251 uc := e.uc
252
253 uc.network.conn = uc
254
255 uc.updateAway()
256
257 uc.forEachDownstream(func(dc *downstreamConn) {
258 sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
259 })
260 uc.network.lastError = nil
261 case eventUpstreamDisconnected:
262 uc := e.uc
263
264 uc.network.conn = nil
265
266 for _, ml := range uc.messageLoggers {
267 if err := ml.Close(); err != nil {
268 uc.logger.Printf("failed to close message logger: %v", err)
269 }
270 }
271
272 uc.endPendingLISTs(true)
273
274 if uc.network.lastError == nil {
275 uc.forEachDownstream(func(dc *downstreamConn) {
276 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
277 })
278 }
279 case eventUpstreamConnectionError:
280 net := e.net
281
282 if net.lastError == nil || net.lastError.Error() != e.err.Error() {
283 net.forEachDownstream(func(dc *downstreamConn) {
284 sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
285 })
286 }
287 net.lastError = e.err
288 case eventUpstreamError:
289 uc := e.uc
290
291 uc.forEachDownstream(func(dc *downstreamConn) {
292 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
293 })
294 uc.network.lastError = e.err
295 case eventUpstreamMessage:
296 msg, uc := e.msg, e.uc
297 if uc.isClosed() {
298 uc.logger.Printf("ignoring message on closed connection: %v", msg)
299 break
300 }
301 if err := uc.handleMessage(msg); err != nil {
302 uc.logger.Printf("failed to handle message %q: %v", msg, err)
303 }
304 case eventDownstreamConnected:
305 dc := e.dc
306
307 if err := dc.welcome(); err != nil {
308 dc.logger.Printf("failed to handle new registered connection: %v", err)
309 break
310 }
311
312 u.downstreamConns = append(u.downstreamConns, dc)
313
314 u.forEachUpstream(func(uc *upstreamConn) {
315 uc.updateAway()
316 })
317 case eventDownstreamDisconnected:
318 dc := e.dc
319
320 for i := range u.downstreamConns {
321 if u.downstreamConns[i] == dc {
322 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
323 break
324 }
325 }
326
327 // Save history if we're the last client with this name
328 skipHistory := make(map[*network]bool)
329 u.forEachDownstream(func(conn *downstreamConn) {
330 if dc.clientName == conn.clientName {
331 skipHistory[conn.network] = true
332 }
333 })
334
335 dc.forEachNetwork(func(net *network) {
336 if skipHistory[net] || skipHistory[nil] {
337 return
338 }
339
340 net.offlineClients[dc.clientName] = struct{}{}
341 for _, history := range net.history {
342 history.offlineClients[dc.clientName] = history.ring.Cur()
343 }
344 })
345
346 u.forEachUpstream(func(uc *upstreamConn) {
347 uc.updateAway()
348 })
349 case eventDownstreamMessage:
350 msg, dc := e.msg, e.dc
351 if dc.isClosed() {
352 dc.logger.Printf("ignoring message on closed connection: %v", msg)
353 break
354 }
355 err := dc.handleMessage(msg)
356 if ircErr, ok := err.(ircError); ok {
357 ircErr.Message.Prefix = dc.srv.prefix()
358 dc.SendMessage(ircErr.Message)
359 } else if err != nil {
360 dc.logger.Printf("failed to handle message %q: %v", msg, err)
361 dc.Close()
362 }
363 default:
364 u.srv.Logger.Printf("received unknown event type: %T", e)
365 }
366 }
367}
368
369func (u *user) createNetwork(net *Network) (*network, error) {
370 if net.ID != 0 {
371 panic("tried creating an already-existing network")
372 }
373
374 network := newNetwork(u, net, nil)
375 err := u.srv.db.StoreNetwork(u.Username, &network.Network)
376 if err != nil {
377 return nil, err
378 }
379
380 u.networks = append(u.networks, network)
381
382 go network.run()
383 return network, nil
384}
385
386func (u *user) deleteNetwork(id int64) error {
387 for i, net := range u.networks {
388 if net.ID != id {
389 continue
390 }
391
392 if err := u.srv.db.DeleteNetwork(net.ID); err != nil {
393 return err
394 }
395
396 u.forEachDownstream(func(dc *downstreamConn) {
397 if dc.network != nil && dc.network == net {
398 dc.Close()
399 }
400 })
401
402 net.Stop()
403 u.networks = append(u.networks[:i], u.networks[i+1:]...)
404 return nil
405 }
406
407 panic("tried deleting a non-existing network")
408}
409
410func (u *user) updatePassword(hashed string) error {
411 u.User.Password = hashed
412 return u.srv.db.UpdatePassword(&u.User)
413}
Note: See TracBrowser for help on using the repository browser.