source: code/trunk/user.go@ 210

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

Simplify ring consumer goroutine

Since network.history is now only accessed from the user goroutine, a
lock becomes unnecessary.
File size: 6.3 KB
Line 
1package soju
2
3import (
4 "sync"
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 eventUpstreamConnected struct {
18 uc *upstreamConn
19}
20
21type eventUpstreamDisconnected struct {
22 uc *upstreamConn
23}
24
25type eventDownstreamMessage struct {
26 msg *irc.Message
27 dc *downstreamConn
28}
29
30type eventDownstreamConnected struct {
31 dc *downstreamConn
32}
33
34type eventDownstreamDisconnected struct {
35 dc *downstreamConn
36}
37
38type network struct {
39 Network
40 user *user
41 ring *Ring
42 stopped chan struct{}
43
44 history map[string]uint64
45
46 lock sync.Mutex
47 conn *upstreamConn
48}
49
50func newNetwork(user *user, record *Network) *network {
51 return &network{
52 Network: *record,
53 user: user,
54 ring: NewRing(user.srv.RingCap),
55 stopped: make(chan struct{}),
56 history: make(map[string]uint64),
57 }
58}
59
60func (net *network) run() {
61 var lastTry time.Time
62 for {
63 select {
64 case <-net.stopped:
65 return
66 default:
67 // This space is intentionally left blank
68 }
69
70 if dur := time.Now().Sub(lastTry); dur < retryConnectMinDelay {
71 delay := retryConnectMinDelay - dur
72 net.user.srv.Logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
73 time.Sleep(delay)
74 }
75 lastTry = time.Now()
76
77 uc, err := connectToUpstream(net)
78 if err != nil {
79 net.user.srv.Logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
80 continue
81 }
82
83 uc.register()
84 if err := uc.runUntilRegistered(); err != nil {
85 uc.logger.Printf("failed to register: %v", err)
86 uc.Close()
87 continue
88 }
89
90 net.user.events <- eventUpstreamConnected{uc}
91 if err := uc.readMessages(net.user.events); err != nil {
92 uc.logger.Printf("failed to handle messages: %v", err)
93 }
94 uc.Close()
95 net.user.events <- eventUpstreamDisconnected{uc}
96 }
97}
98
99func (net *network) upstream() *upstreamConn {
100 net.lock.Lock()
101 defer net.lock.Unlock()
102 return net.conn
103}
104
105func (net *network) Stop() {
106 select {
107 case <-net.stopped:
108 return
109 default:
110 close(net.stopped)
111 }
112
113 if uc := net.upstream(); uc != nil {
114 uc.Close()
115 }
116}
117
118type user struct {
119 User
120 srv *Server
121
122 events chan event
123
124 networks []*network
125 downstreamConns []*downstreamConn
126
127 // LIST commands in progress
128 pendingLISTs []pendingLIST
129}
130
131type pendingLIST struct {
132 downstreamID uint64
133 // list of per-upstream LIST commands not yet sent or completed
134 pendingCommands map[int64]*irc.Message
135}
136
137func newUser(srv *Server, record *User) *user {
138 return &user{
139 User: *record,
140 srv: srv,
141 events: make(chan event, 64),
142 }
143}
144
145func (u *user) forEachNetwork(f func(*network)) {
146 for _, network := range u.networks {
147 f(network)
148 }
149}
150
151func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
152 for _, network := range u.networks {
153 uc := network.upstream()
154 if uc == nil {
155 continue
156 }
157 f(uc)
158 }
159}
160
161func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
162 for _, dc := range u.downstreamConns {
163 f(dc)
164 }
165}
166
167func (u *user) getNetwork(name string) *network {
168 for _, network := range u.networks {
169 if network.Addr == name {
170 return network
171 }
172 if network.Name != "" && network.Name == name {
173 return network
174 }
175 }
176 return nil
177}
178
179func (u *user) run() {
180 networks, err := u.srv.db.ListNetworks(u.Username)
181 if err != nil {
182 u.srv.Logger.Printf("failed to list networks for user %q: %v", u.Username, err)
183 return
184 }
185
186 for _, record := range networks {
187 network := newNetwork(u, &record)
188 u.networks = append(u.networks, network)
189
190 go network.run()
191 }
192
193 for e := range u.events {
194 switch e := e.(type) {
195 case eventUpstreamConnected:
196 uc := e.uc
197
198 uc.network.lock.Lock()
199 uc.network.conn = uc
200 uc.network.lock.Unlock()
201
202 uc.updateAway()
203 case eventUpstreamDisconnected:
204 uc := e.uc
205
206 uc.network.lock.Lock()
207 uc.network.conn = nil
208 uc.network.lock.Unlock()
209
210 for _, log := range uc.logs {
211 log.file.Close()
212 }
213
214 uc.endPendingLISTs(true)
215 case eventUpstreamMessage:
216 msg, uc := e.msg, e.uc
217 if uc.isClosed() {
218 uc.logger.Printf("ignoring message on closed connection: %v", msg)
219 break
220 }
221 if err := uc.handleMessage(msg); err != nil {
222 uc.logger.Printf("failed to handle message %q: %v", msg, err)
223 }
224 case eventDownstreamConnected:
225 dc := e.dc
226
227 if err := dc.welcome(); err != nil {
228 dc.logger.Printf("failed to handle new registered connection: %v", err)
229 break
230 }
231
232 u.downstreamConns = append(u.downstreamConns, dc)
233
234 u.forEachUpstream(func(uc *upstreamConn) {
235 uc.updateAway()
236 })
237 case eventDownstreamDisconnected:
238 dc := e.dc
239
240 for net, rc := range dc.ringConsumers {
241 seq := rc.Close()
242 net.history[dc.clientName] = seq
243 }
244
245 for i := range u.downstreamConns {
246 if u.downstreamConns[i] == dc {
247 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
248 break
249 }
250 }
251
252 u.forEachUpstream(func(uc *upstreamConn) {
253 uc.updateAway()
254 })
255 case eventDownstreamMessage:
256 msg, dc := e.msg, e.dc
257 if dc.isClosed() {
258 dc.logger.Printf("ignoring message on closed connection: %v", msg)
259 break
260 }
261 err := dc.handleMessage(msg)
262 if ircErr, ok := err.(ircError); ok {
263 ircErr.Message.Prefix = dc.srv.prefix()
264 dc.SendMessage(ircErr.Message)
265 } else if err != nil {
266 dc.logger.Printf("failed to handle message %q: %v", msg, err)
267 dc.Close()
268 }
269 default:
270 u.srv.Logger.Printf("received unknown event type: %T", e)
271 }
272 }
273}
274
275func (u *user) createNetwork(net *Network) (*network, error) {
276 if net.ID != 0 {
277 panic("tried creating an already-existing network")
278 }
279
280 network := newNetwork(u, net)
281 err := u.srv.db.StoreNetwork(u.Username, &network.Network)
282 if err != nil {
283 return nil, err
284 }
285
286 u.forEachDownstream(func(dc *downstreamConn) {
287 if dc.network == nil {
288 dc.runNetwork(network, false)
289 }
290 })
291
292 u.networks = append(u.networks, network)
293
294 go network.run()
295 return network, nil
296}
297
298func (u *user) deleteNetwork(id int64) error {
299 for i, net := range u.networks {
300 if net.ID != id {
301 continue
302 }
303
304 if err := u.srv.db.DeleteNetwork(net.ID); err != nil {
305 return err
306 }
307
308 u.forEachDownstream(func(dc *downstreamConn) {
309 if dc.network != nil && dc.network == net {
310 dc.Close()
311 }
312 })
313
314 net.Stop()
315 net.ring.Close()
316 u.networks = append(u.networks[:i], u.networks[i+1:]...)
317 return nil
318 }
319
320 panic("tried deleting a non-existing network")
321}
Note: See TracBrowser for help on using the repository browser.