source: code/trunk/user.go@ 377

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

Make user.stop block

This allows callers to wait until the user has been stopped.
File size: 12.8 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 eventStop struct{}
49
50type networkHistory struct {
51 offlineClients map[string]uint64 // indexed by client name
52 ring *Ring // can be nil if there are no offline clients
53}
54
55type network struct {
56 Network
57 user *user
58 stopped chan struct{}
59
60 conn *upstreamConn
61 channels map[string]*Channel
62 history map[string]*networkHistory // indexed by entity
63 offlineClients map[string]struct{} // indexed by client name
64 lastError error
65}
66
67func newNetwork(user *user, record *Network, channels []Channel) *network {
68 m := make(map[string]*Channel, len(channels))
69 for _, ch := range channels {
70 ch := ch
71 m[ch.Name] = &ch
72 }
73
74 return &network{
75 Network: *record,
76 user: user,
77 stopped: make(chan struct{}),
78 channels: m,
79 history: make(map[string]*networkHistory),
80 offlineClients: make(map[string]struct{}),
81 }
82}
83
84func (net *network) forEachDownstream(f func(*downstreamConn)) {
85 net.user.forEachDownstream(func(dc *downstreamConn) {
86 if dc.network != nil && dc.network != net {
87 return
88 }
89 f(dc)
90 })
91}
92
93func (net *network) isStopped() bool {
94 select {
95 case <-net.stopped:
96 return true
97 default:
98 return false
99 }
100}
101
102func (net *network) run() {
103 var lastTry time.Time
104 for {
105 if net.isStopped() {
106 return
107 }
108
109 if dur := time.Now().Sub(lastTry); dur < retryConnectMinDelay {
110 delay := retryConnectMinDelay - dur
111 net.user.srv.Logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
112 time.Sleep(delay)
113 }
114 lastTry = time.Now()
115
116 uc, err := connectToUpstream(net)
117 if err != nil {
118 net.user.srv.Logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
119 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to connect: %v", err)}
120 continue
121 }
122
123 uc.register()
124 if err := uc.runUntilRegistered(); err != nil {
125 uc.logger.Printf("failed to register: %v", err)
126 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to register: %v", err)}
127 uc.Close()
128 continue
129 }
130
131 // TODO: this is racy with net.stopped. If the network is stopped
132 // before the user goroutine receives eventUpstreamConnected, the
133 // connection won't be closed.
134 net.user.events <- eventUpstreamConnected{uc}
135 if err := uc.readMessages(net.user.events); err != nil {
136 uc.logger.Printf("failed to handle messages: %v", err)
137 net.user.events <- eventUpstreamError{uc, fmt.Errorf("failed to handle messages: %v", err)}
138 }
139 uc.Close()
140 net.user.events <- eventUpstreamDisconnected{uc}
141 }
142}
143
144func (net *network) stop() {
145 if !net.isStopped() {
146 close(net.stopped)
147 }
148
149 if net.conn != nil {
150 net.conn.Close()
151 }
152}
153
154func (net *network) createUpdateChannel(ch *Channel) error {
155 if current, ok := net.channels[ch.Name]; ok {
156 ch.ID = current.ID // update channel if it already exists
157 }
158 if err := net.user.srv.db.StoreChannel(net.ID, ch); err != nil {
159 return err
160 }
161 prev := net.channels[ch.Name]
162 net.channels[ch.Name] = ch
163
164 if prev != nil && prev.Detached != ch.Detached {
165 history := net.history[ch.Name]
166 if ch.Detached {
167 net.user.srv.Logger.Printf("network %q: detaching channel %q", net.GetName(), ch.Name)
168 net.forEachDownstream(func(dc *downstreamConn) {
169 net.offlineClients[dc.clientName] = struct{}{}
170 if history != nil {
171 history.offlineClients[dc.clientName] = history.ring.Cur()
172 }
173
174 dc.SendMessage(&irc.Message{
175 Prefix: dc.prefix(),
176 Command: "PART",
177 Params: []string{dc.marshalEntity(net, ch.Name), "Detach"},
178 })
179 })
180 } else {
181 net.user.srv.Logger.Printf("network %q: attaching channel %q", net.GetName(), ch.Name)
182
183 var uch *upstreamChannel
184 if net.conn != nil {
185 uch = net.conn.channels[ch.Name]
186 }
187
188 net.forEachDownstream(func(dc *downstreamConn) {
189 dc.SendMessage(&irc.Message{
190 Prefix: dc.prefix(),
191 Command: "JOIN",
192 Params: []string{dc.marshalEntity(net, ch.Name)},
193 })
194
195 if uch != nil {
196 forwardChannel(dc, uch)
197 }
198
199 if history != nil {
200 dc.sendNetworkHistory(net)
201 }
202 })
203 }
204 }
205
206 return nil
207}
208
209func (net *network) deleteChannel(name string) error {
210 if err := net.user.srv.db.DeleteChannel(net.ID, name); err != nil {
211 return err
212 }
213 delete(net.channels, name)
214 return nil
215}
216
217type user struct {
218 User
219 srv *Server
220
221 events chan event
222 done chan struct{}
223
224 networks []*network
225 downstreamConns []*downstreamConn
226
227 // LIST commands in progress
228 pendingLISTs []pendingLIST
229}
230
231type pendingLIST struct {
232 downstreamID uint64
233 // list of per-upstream LIST commands not yet sent or completed
234 pendingCommands map[int64]*irc.Message
235}
236
237func newUser(srv *Server, record *User) *user {
238 return &user{
239 User: *record,
240 srv: srv,
241 events: make(chan event, 64),
242 done: make(chan struct{}),
243 }
244}
245
246func (u *user) forEachNetwork(f func(*network)) {
247 for _, network := range u.networks {
248 f(network)
249 }
250}
251
252func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
253 for _, network := range u.networks {
254 if network.conn == nil {
255 continue
256 }
257 f(network.conn)
258 }
259}
260
261func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
262 for _, dc := range u.downstreamConns {
263 f(dc)
264 }
265}
266
267func (u *user) getNetwork(name string) *network {
268 for _, network := range u.networks {
269 if network.Addr == name {
270 return network
271 }
272 if network.Name != "" && network.Name == name {
273 return network
274 }
275 }
276 return nil
277}
278
279func (u *user) getNetworkByID(id int64) *network {
280 for _, net := range u.networks {
281 if net.ID == id {
282 return net
283 }
284 }
285 return nil
286}
287
288func (u *user) run() {
289 defer close(u.done)
290
291 networks, err := u.srv.db.ListNetworks(u.Username)
292 if err != nil {
293 u.srv.Logger.Printf("failed to list networks for user %q: %v", u.Username, err)
294 return
295 }
296
297 for _, record := range networks {
298 record := record
299 channels, err := u.srv.db.ListChannels(record.ID)
300 if err != nil {
301 u.srv.Logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
302 continue
303 }
304
305 network := newNetwork(u, &record, channels)
306 u.networks = append(u.networks, network)
307
308 go network.run()
309 }
310
311 for e := range u.events {
312 switch e := e.(type) {
313 case eventUpstreamConnected:
314 uc := e.uc
315
316 uc.network.conn = uc
317
318 uc.updateAway()
319
320 uc.forEachDownstream(func(dc *downstreamConn) {
321 dc.updateSupportedCaps()
322 sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
323
324 dc.updateNick()
325 })
326 uc.network.lastError = nil
327 case eventUpstreamDisconnected:
328 u.handleUpstreamDisconnected(e.uc)
329 case eventUpstreamConnectionError:
330 net := e.net
331
332 stopped := false
333 select {
334 case <-net.stopped:
335 stopped = true
336 default:
337 }
338
339 if !stopped && (net.lastError == nil || net.lastError.Error() != e.err.Error()) {
340 net.forEachDownstream(func(dc *downstreamConn) {
341 sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
342 })
343 }
344 net.lastError = e.err
345 case eventUpstreamError:
346 uc := e.uc
347
348 uc.forEachDownstream(func(dc *downstreamConn) {
349 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
350 })
351 uc.network.lastError = e.err
352 case eventUpstreamMessage:
353 msg, uc := e.msg, e.uc
354 if uc.isClosed() {
355 uc.logger.Printf("ignoring message on closed connection: %v", msg)
356 break
357 }
358 if err := uc.handleMessage(msg); err != nil {
359 uc.logger.Printf("failed to handle message %q: %v", msg, err)
360 }
361 case eventDownstreamConnected:
362 dc := e.dc
363
364 if err := dc.welcome(); err != nil {
365 dc.logger.Printf("failed to handle new registered connection: %v", err)
366 break
367 }
368
369 u.downstreamConns = append(u.downstreamConns, dc)
370
371 u.forEachUpstream(func(uc *upstreamConn) {
372 uc.updateAway()
373 })
374
375 dc.updateSupportedCaps()
376 case eventDownstreamDisconnected:
377 dc := e.dc
378
379 for i := range u.downstreamConns {
380 if u.downstreamConns[i] == dc {
381 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
382 break
383 }
384 }
385
386 // Save history if we're the last client with this name
387 skipHistory := make(map[*network]bool)
388 u.forEachDownstream(func(conn *downstreamConn) {
389 if dc.clientName == conn.clientName {
390 skipHistory[conn.network] = true
391 }
392 })
393
394 dc.forEachNetwork(func(net *network) {
395 if skipHistory[net] || skipHistory[nil] {
396 return
397 }
398
399 net.offlineClients[dc.clientName] = struct{}{}
400 for target, history := range net.history {
401 if ch, ok := net.channels[target]; ok && ch.Detached {
402 continue
403 }
404 history.offlineClients[dc.clientName] = history.ring.Cur()
405 }
406 })
407
408 u.forEachUpstream(func(uc *upstreamConn) {
409 uc.updateAway()
410 })
411 case eventDownstreamMessage:
412 msg, dc := e.msg, e.dc
413 if dc.isClosed() {
414 dc.logger.Printf("ignoring message on closed connection: %v", msg)
415 break
416 }
417 err := dc.handleMessage(msg)
418 if ircErr, ok := err.(ircError); ok {
419 ircErr.Message.Prefix = dc.srv.prefix()
420 dc.SendMessage(ircErr.Message)
421 } else if err != nil {
422 dc.logger.Printf("failed to handle message %q: %v", msg, err)
423 dc.Close()
424 }
425 case eventStop:
426 u.forEachDownstream(func(dc *downstreamConn) {
427 dc.Close()
428 })
429 for _, n := range u.networks {
430 n.stop()
431 }
432 return
433 default:
434 u.srv.Logger.Printf("received unknown event type: %T", e)
435 }
436 }
437}
438
439func (u *user) handleUpstreamDisconnected(uc *upstreamConn) {
440 uc.network.conn = nil
441
442 for _, ml := range uc.messageLoggers {
443 if err := ml.Close(); err != nil {
444 uc.logger.Printf("failed to close message logger: %v", err)
445 }
446 }
447
448 uc.endPendingLISTs(true)
449
450 uc.forEachDownstream(func(dc *downstreamConn) {
451 dc.updateSupportedCaps()
452 })
453
454 if uc.network.lastError == nil {
455 uc.forEachDownstream(func(dc *downstreamConn) {
456 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
457 })
458 }
459}
460
461func (u *user) addNetwork(network *network) {
462 u.networks = append(u.networks, network)
463 go network.run()
464}
465
466func (u *user) removeNetwork(network *network) {
467 network.stop()
468
469 u.forEachDownstream(func(dc *downstreamConn) {
470 if dc.network != nil && dc.network == network {
471 dc.Close()
472 }
473 })
474
475 for i, net := range u.networks {
476 if net == network {
477 u.networks = append(u.networks[:i], u.networks[i+1:]...)
478 return
479 }
480 }
481
482 panic("tried to remove a non-existing network")
483}
484
485func (u *user) createNetwork(record *Network) (*network, error) {
486 if record.ID != 0 {
487 panic("tried creating an already-existing network")
488 }
489
490 network := newNetwork(u, record, nil)
491 err := u.srv.db.StoreNetwork(u.Username, &network.Network)
492 if err != nil {
493 return nil, err
494 }
495
496 u.addNetwork(network)
497
498 return network, nil
499}
500
501func (u *user) updateNetwork(record *Network) (*network, error) {
502 if record.ID == 0 {
503 panic("tried updating a new network")
504 }
505
506 network := u.getNetworkByID(record.ID)
507 if network == nil {
508 panic("tried updating a non-existing network")
509 }
510
511 if err := u.srv.db.StoreNetwork(u.Username, record); err != nil {
512 return nil, err
513 }
514
515 // Most network changes require us to re-connect to the upstream server
516
517 channels := make([]Channel, 0, len(network.channels))
518 for _, ch := range network.channels {
519 channels = append(channels, *ch)
520 }
521
522 updatedNetwork := newNetwork(u, record, channels)
523
524 // If we're currently connected, disconnect and perform the necessary
525 // bookkeeping
526 if network.conn != nil {
527 network.stop()
528 // Note: this will set network.conn to nil
529 u.handleUpstreamDisconnected(network.conn)
530 }
531
532 // Patch downstream connections to use our fresh updated network
533 u.forEachDownstream(func(dc *downstreamConn) {
534 if dc.network != nil && dc.network == network {
535 dc.network = updatedNetwork
536 }
537 })
538
539 // We need to remove the network after patching downstream connections,
540 // otherwise they'll get closed
541 u.removeNetwork(network)
542
543 // This will re-connect to the upstream server
544 u.addNetwork(updatedNetwork)
545
546 return updatedNetwork, nil
547}
548
549func (u *user) deleteNetwork(id int64) error {
550 network := u.getNetworkByID(id)
551 if network == nil {
552 panic("tried deleting a non-existing network")
553 }
554
555 if err := u.srv.db.DeleteNetwork(network.ID); err != nil {
556 return err
557 }
558
559 u.removeNetwork(network)
560 return nil
561}
562
563func (u *user) updatePassword(hashed string) error {
564 u.User.Password = hashed
565 return u.srv.db.StoreUser(&u.User)
566}
567
568func (u *user) stop() {
569 u.events <- eventStop{}
570 <-u.done
571}
Note: See TracBrowser for help on using the repository browser.