source: code/trunk/user.go@ 284

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

Add support for detached channels

Channels can now be detached by leaving them with the reason "detach",
and re-attached by joining them again. Upon detaching the channel is
no longer forwarded to downstream connections. Upon re-attaching the
history buffer is sent.
File size: 10.5 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 ch := ch
69 m[ch.Name] = &ch
70 }
71
72 return &network{
73 Network: *record,
74 user: user,
75 stopped: make(chan struct{}),
76 channels: m,
77 history: make(map[string]*networkHistory),
78 offlineClients: make(map[string]struct{}),
79 }
80}
81
82func (net *network) forEachDownstream(f func(*downstreamConn)) {
83 net.user.forEachDownstream(func(dc *downstreamConn) {
84 if dc.network != nil && dc.network != net {
85 return
86 }
87 f(dc)
88 })
89}
90
91func (net *network) run() {
92 var lastTry time.Time
93 for {
94 select {
95 case <-net.stopped:
96 return
97 default:
98 // This space is intentionally left blank
99 }
100
101 if dur := time.Now().Sub(lastTry); dur < retryConnectMinDelay {
102 delay := retryConnectMinDelay - dur
103 net.user.srv.Logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
104 time.Sleep(delay)
105 }
106 lastTry = time.Now()
107
108 uc, err := connectToUpstream(net)
109 if err != nil {
110 net.user.srv.Logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
111 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to connect: %v", err)}
112 continue
113 }
114
115 uc.register()
116 if err := uc.runUntilRegistered(); err != nil {
117 uc.logger.Printf("failed to register: %v", err)
118 net.user.events <- eventUpstreamConnectionError{net, fmt.Errorf("failed to register: %v", err)}
119 uc.Close()
120 continue
121 }
122
123 net.user.events <- eventUpstreamConnected{uc}
124 if err := uc.readMessages(net.user.events); err != nil {
125 uc.logger.Printf("failed to handle messages: %v", err)
126 net.user.events <- eventUpstreamError{uc, fmt.Errorf("failed to handle messages: %v", err)}
127 }
128 uc.Close()
129 net.user.events <- eventUpstreamDisconnected{uc}
130 }
131}
132
133func (net *network) Stop() {
134 select {
135 case <-net.stopped:
136 return
137 default:
138 close(net.stopped)
139 }
140
141 if net.conn != nil {
142 net.conn.Close()
143 }
144}
145
146func (net *network) createUpdateChannel(ch *Channel) error {
147 if current, ok := net.channels[ch.Name]; ok {
148 ch.ID = current.ID // update channel if it already exists
149 }
150 if err := net.user.srv.db.StoreChannel(net.ID, ch); err != nil {
151 return err
152 }
153 prev := net.channels[ch.Name]
154 net.channels[ch.Name] = ch
155
156 if prev != nil && prev.Detached != ch.Detached {
157 history := net.history[ch.Name]
158 if ch.Detached {
159 net.user.srv.Logger.Printf("network %q: detaching channel %q", net.GetName(), ch.Name)
160 net.forEachDownstream(func(dc *downstreamConn) {
161 net.offlineClients[dc.clientName] = struct{}{}
162 if history != nil {
163 history.offlineClients[dc.clientName] = history.ring.Cur()
164 }
165
166 dc.SendMessage(&irc.Message{
167 Prefix: dc.prefix(),
168 Command: "PART",
169 Params: []string{dc.marshalEntity(net, ch.Name), "Detach"},
170 })
171 })
172 } else {
173 net.user.srv.Logger.Printf("network %q: attaching channel %q", net.GetName(), ch.Name)
174
175 var uch *upstreamChannel
176 if net.conn != nil {
177 uch = net.conn.channels[ch.Name]
178 }
179
180 net.forEachDownstream(func(dc *downstreamConn) {
181 dc.SendMessage(&irc.Message{
182 Prefix: dc.prefix(),
183 Command: "JOIN",
184 Params: []string{dc.marshalEntity(net, ch.Name)},
185 })
186
187 if uch != nil {
188 forwardChannel(dc, uch)
189 }
190
191 if history != nil {
192 dc.sendNetworkHistory(net)
193 }
194 })
195 }
196 }
197
198 return nil
199}
200
201func (net *network) deleteChannel(name string) error {
202 if err := net.user.srv.db.DeleteChannel(net.ID, name); err != nil {
203 return err
204 }
205 delete(net.channels, name)
206 return nil
207}
208
209type user struct {
210 User
211 srv *Server
212
213 events chan event
214
215 networks []*network
216 downstreamConns []*downstreamConn
217
218 // LIST commands in progress
219 pendingLISTs []pendingLIST
220}
221
222type pendingLIST struct {
223 downstreamID uint64
224 // list of per-upstream LIST commands not yet sent or completed
225 pendingCommands map[int64]*irc.Message
226}
227
228func newUser(srv *Server, record *User) *user {
229 return &user{
230 User: *record,
231 srv: srv,
232 events: make(chan event, 64),
233 }
234}
235
236func (u *user) forEachNetwork(f func(*network)) {
237 for _, network := range u.networks {
238 f(network)
239 }
240}
241
242func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
243 for _, network := range u.networks {
244 if network.conn == nil {
245 continue
246 }
247 f(network.conn)
248 }
249}
250
251func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
252 for _, dc := range u.downstreamConns {
253 f(dc)
254 }
255}
256
257func (u *user) getNetwork(name string) *network {
258 for _, network := range u.networks {
259 if network.Addr == name {
260 return network
261 }
262 if network.Name != "" && network.Name == name {
263 return network
264 }
265 }
266 return nil
267}
268
269func (u *user) run() {
270 networks, err := u.srv.db.ListNetworks(u.Username)
271 if err != nil {
272 u.srv.Logger.Printf("failed to list networks for user %q: %v", u.Username, err)
273 return
274 }
275
276 for _, record := range networks {
277 record := record
278 channels, err := u.srv.db.ListChannels(record.ID)
279 if err != nil {
280 u.srv.Logger.Printf("failed to list channels for user %q, network %q: %v", u.Username, record.GetName(), err)
281 }
282
283 network := newNetwork(u, &record, channels)
284 u.networks = append(u.networks, network)
285
286 go network.run()
287 }
288
289 for e := range u.events {
290 switch e := e.(type) {
291 case eventUpstreamConnected:
292 uc := e.uc
293
294 uc.network.conn = uc
295
296 uc.updateAway()
297
298 uc.forEachDownstream(func(dc *downstreamConn) {
299 dc.updateSupportedCaps()
300 sendServiceNOTICE(dc, fmt.Sprintf("connected to %s", uc.network.GetName()))
301 })
302 uc.network.lastError = nil
303 case eventUpstreamDisconnected:
304 uc := e.uc
305
306 uc.network.conn = nil
307
308 for _, ml := range uc.messageLoggers {
309 if err := ml.Close(); err != nil {
310 uc.logger.Printf("failed to close message logger: %v", err)
311 }
312 }
313
314 uc.endPendingLISTs(true)
315
316 uc.forEachDownstream(func(dc *downstreamConn) {
317 dc.updateSupportedCaps()
318 })
319
320 if uc.network.lastError == nil {
321 uc.forEachDownstream(func(dc *downstreamConn) {
322 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s", uc.network.GetName()))
323 })
324 }
325 case eventUpstreamConnectionError:
326 net := e.net
327
328 if net.lastError == nil || net.lastError.Error() != e.err.Error() {
329 net.forEachDownstream(func(dc *downstreamConn) {
330 sendServiceNOTICE(dc, fmt.Sprintf("failed connecting/registering to %s: %v", net.GetName(), e.err))
331 })
332 }
333 net.lastError = e.err
334 case eventUpstreamError:
335 uc := e.uc
336
337 uc.forEachDownstream(func(dc *downstreamConn) {
338 sendServiceNOTICE(dc, fmt.Sprintf("disconnected from %s: %v", uc.network.GetName(), e.err))
339 })
340 uc.network.lastError = e.err
341 case eventUpstreamMessage:
342 msg, uc := e.msg, e.uc
343 if uc.isClosed() {
344 uc.logger.Printf("ignoring message on closed connection: %v", msg)
345 break
346 }
347 if err := uc.handleMessage(msg); err != nil {
348 uc.logger.Printf("failed to handle message %q: %v", msg, err)
349 }
350 case eventDownstreamConnected:
351 dc := e.dc
352
353 if err := dc.welcome(); err != nil {
354 dc.logger.Printf("failed to handle new registered connection: %v", err)
355 break
356 }
357
358 u.downstreamConns = append(u.downstreamConns, dc)
359
360 u.forEachUpstream(func(uc *upstreamConn) {
361 uc.updateAway()
362 })
363
364 dc.updateSupportedCaps()
365 case eventDownstreamDisconnected:
366 dc := e.dc
367
368 for i := range u.downstreamConns {
369 if u.downstreamConns[i] == dc {
370 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
371 break
372 }
373 }
374
375 // Save history if we're the last client with this name
376 skipHistory := make(map[*network]bool)
377 u.forEachDownstream(func(conn *downstreamConn) {
378 if dc.clientName == conn.clientName {
379 skipHistory[conn.network] = true
380 }
381 })
382
383 dc.forEachNetwork(func(net *network) {
384 if skipHistory[net] || skipHistory[nil] {
385 return
386 }
387
388 net.offlineClients[dc.clientName] = struct{}{}
389 for target, history := range net.history {
390 if ch, ok := net.channels[target]; ok && ch.Detached {
391 continue
392 }
393 history.offlineClients[dc.clientName] = history.ring.Cur()
394 }
395 })
396
397 u.forEachUpstream(func(uc *upstreamConn) {
398 uc.updateAway()
399 })
400 case eventDownstreamMessage:
401 msg, dc := e.msg, e.dc
402 if dc.isClosed() {
403 dc.logger.Printf("ignoring message on closed connection: %v", msg)
404 break
405 }
406 err := dc.handleMessage(msg)
407 if ircErr, ok := err.(ircError); ok {
408 ircErr.Message.Prefix = dc.srv.prefix()
409 dc.SendMessage(ircErr.Message)
410 } else if err != nil {
411 dc.logger.Printf("failed to handle message %q: %v", msg, err)
412 dc.Close()
413 }
414 default:
415 u.srv.Logger.Printf("received unknown event type: %T", e)
416 }
417 }
418}
419
420func (u *user) createNetwork(net *Network) (*network, error) {
421 if net.ID != 0 {
422 panic("tried creating an already-existing network")
423 }
424
425 network := newNetwork(u, net, nil)
426 err := u.srv.db.StoreNetwork(u.Username, &network.Network)
427 if err != nil {
428 return nil, err
429 }
430
431 u.networks = append(u.networks, network)
432
433 go network.run()
434 return network, nil
435}
436
437func (u *user) deleteNetwork(id int64) error {
438 for i, net := range u.networks {
439 if net.ID != id {
440 continue
441 }
442
443 if err := u.srv.db.DeleteNetwork(net.ID); err != nil {
444 return err
445 }
446
447 u.forEachDownstream(func(dc *downstreamConn) {
448 if dc.network != nil && dc.network == net {
449 dc.Close()
450 }
451 })
452
453 net.Stop()
454 u.networks = append(u.networks[:i], u.networks[i+1:]...)
455 return nil
456 }
457
458 panic("tried deleting a non-existing network")
459}
460
461func (u *user) updatePassword(hashed string) error {
462 u.User.Password = hashed
463 return u.srv.db.UpdatePassword(&u.User)
464}
Note: See TracBrowser for help on using the repository browser.