source: code/trunk/upstream.go@ 68

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

Fix PING handlers, again
File size: 8.1 KB
Line 
1package jounce
2
3import (
4 "crypto/tls"
5 "fmt"
6 "io"
7 "net"
8 "strconv"
9 "strings"
10 "time"
11
12 "gopkg.in/irc.v3"
13)
14
15type upstreamChannel struct {
16 Name string
17 conn *upstreamConn
18 Topic string
19 TopicWho string
20 TopicTime time.Time
21 Status channelStatus
22 modes modeSet
23 Members map[string]membership
24 complete bool
25}
26
27type upstreamConn struct {
28 upstream *Upstream
29 logger Logger
30 net net.Conn
31 irc *irc.Conn
32 srv *Server
33 user *user
34 messages chan<- *irc.Message
35 ring *Ring
36
37 serverName string
38 availableUserModes string
39 availableChannelModes string
40 channelModesWithParam string
41
42 registered bool
43 nick string
44 closed bool
45 modes modeSet
46 channels map[string]*upstreamChannel
47 history map[string]uint64
48}
49
50func connectToUpstream(u *user, upstream *Upstream) (*upstreamConn, error) {
51 logger := &prefixLogger{u.srv.Logger, fmt.Sprintf("upstream %q: ", upstream.Addr)}
52 logger.Printf("connecting to server")
53
54 netConn, err := tls.Dial("tcp", upstream.Addr, nil)
55 if err != nil {
56 return nil, fmt.Errorf("failed to dial %q: %v", upstream.Addr, err)
57 }
58
59 setKeepAlive(netConn)
60
61 msgs := make(chan *irc.Message, 64)
62 uc := &upstreamConn{
63 upstream: upstream,
64 logger: logger,
65 net: netConn,
66 irc: irc.NewConn(netConn),
67 srv: u.srv,
68 user: u,
69 messages: msgs,
70 ring: NewRing(u.srv.RingCap),
71 channels: make(map[string]*upstreamChannel),
72 history: make(map[string]uint64),
73 }
74
75 go func() {
76 for msg := range msgs {
77 if uc.srv.Debug {
78 uc.logger.Printf("sent: %v", msg)
79 }
80 if err := uc.irc.WriteMessage(msg); err != nil {
81 uc.logger.Printf("failed to write message: %v", err)
82 }
83 }
84 if err := uc.net.Close(); err != nil {
85 uc.logger.Printf("failed to close connection: %v", err)
86 } else {
87 uc.logger.Printf("connection closed")
88 }
89 }()
90
91 return uc, nil
92}
93
94func (uc *upstreamConn) Close() error {
95 if uc.closed {
96 return fmt.Errorf("upstream connection already closed")
97 }
98 close(uc.messages)
99 uc.closed = true
100 return nil
101}
102
103func (uc *upstreamConn) getChannel(name string) (*upstreamChannel, error) {
104 ch, ok := uc.channels[name]
105 if !ok {
106 return nil, fmt.Errorf("unknown channel %q", name)
107 }
108 return ch, nil
109}
110
111func (uc *upstreamConn) handleMessage(msg *irc.Message) error {
112 switch msg.Command {
113 case "PING":
114 uc.SendMessage(&irc.Message{
115 Command: "PONG",
116 Params: msg.Params,
117 })
118 return nil
119 case "MODE":
120 var name, modeStr string
121 if err := parseMessageParams(msg, &name, &modeStr); err != nil {
122 return err
123 }
124
125 if name == msg.Prefix.Name { // user mode change
126 if name != uc.nick {
127 return fmt.Errorf("received MODE message for unknow nick %q", name)
128 }
129 return uc.modes.Apply(modeStr)
130 } else { // channel mode change
131 ch, err := uc.getChannel(name)
132 if err != nil {
133 return err
134 }
135 if err := ch.modes.Apply(modeStr); err != nil {
136 return err
137 }
138 }
139
140 uc.user.forEachDownstream(func(dc *downstreamConn) {
141 dc.SendMessage(msg)
142 })
143 case "NOTICE":
144 uc.logger.Print(msg)
145 case irc.RPL_WELCOME:
146 uc.registered = true
147 uc.logger.Printf("connection registered")
148
149 for _, ch := range uc.upstream.Channels {
150 uc.SendMessage(&irc.Message{
151 Command: "JOIN",
152 Params: []string{ch},
153 })
154 }
155 case irc.RPL_MYINFO:
156 if err := parseMessageParams(msg, nil, &uc.serverName, nil, &uc.availableUserModes, &uc.availableChannelModes); err != nil {
157 return err
158 }
159 if len(msg.Params) > 5 {
160 uc.channelModesWithParam = msg.Params[5]
161 }
162 case "NICK":
163 var newNick string
164 if err := parseMessageParams(msg, &newNick); err != nil {
165 return err
166 }
167
168 if msg.Prefix.Name == uc.nick {
169 uc.logger.Printf("changed nick from %q to %q", uc.nick, newNick)
170 uc.nick = newNick
171 }
172
173 for _, ch := range uc.channels {
174 if membership, ok := ch.Members[msg.Prefix.Name]; ok {
175 delete(ch.Members, msg.Prefix.Name)
176 ch.Members[newNick] = membership
177 }
178 }
179
180 uc.user.forEachDownstream(func(dc *downstreamConn) {
181 dc.SendMessage(msg)
182 })
183 case "JOIN":
184 var channels string
185 if err := parseMessageParams(msg, &channels); err != nil {
186 return err
187 }
188
189 for _, ch := range strings.Split(channels, ",") {
190 if msg.Prefix.Name == uc.nick {
191 uc.logger.Printf("joined channel %q", ch)
192 uc.channels[ch] = &upstreamChannel{
193 Name: ch,
194 conn: uc,
195 Members: make(map[string]membership),
196 }
197 } else {
198 ch, err := uc.getChannel(ch)
199 if err != nil {
200 return err
201 }
202 ch.Members[msg.Prefix.Name] = 0
203 }
204 }
205
206 uc.user.forEachDownstream(func(dc *downstreamConn) {
207 dc.SendMessage(msg)
208 })
209 case "PART":
210 var channels string
211 if err := parseMessageParams(msg, &channels); err != nil {
212 return err
213 }
214
215 for _, ch := range strings.Split(channels, ",") {
216 if msg.Prefix.Name == uc.nick {
217 uc.logger.Printf("parted channel %q", ch)
218 delete(uc.channels, ch)
219 } else {
220 ch, err := uc.getChannel(ch)
221 if err != nil {
222 return err
223 }
224 delete(ch.Members, msg.Prefix.Name)
225 }
226 }
227
228 uc.user.forEachDownstream(func(dc *downstreamConn) {
229 dc.SendMessage(msg)
230 })
231 case irc.RPL_TOPIC, irc.RPL_NOTOPIC:
232 var name, topic string
233 if err := parseMessageParams(msg, nil, &name, &topic); err != nil {
234 return err
235 }
236 ch, err := uc.getChannel(name)
237 if err != nil {
238 return err
239 }
240 if msg.Command == irc.RPL_TOPIC {
241 ch.Topic = topic
242 } else {
243 ch.Topic = ""
244 }
245 case "TOPIC":
246 var name string
247 if err := parseMessageParams(msg, nil, &name); err != nil {
248 return err
249 }
250 ch, err := uc.getChannel(name)
251 if err != nil {
252 return err
253 }
254 if len(msg.Params) > 1 {
255 ch.Topic = msg.Params[1]
256 } else {
257 ch.Topic = ""
258 }
259 case rpl_topicwhotime:
260 var name, who, timeStr string
261 if err := parseMessageParams(msg, nil, &name, &who, &timeStr); err != nil {
262 return err
263 }
264 ch, err := uc.getChannel(name)
265 if err != nil {
266 return err
267 }
268 ch.TopicWho = who
269 sec, err := strconv.ParseInt(timeStr, 10, 64)
270 if err != nil {
271 return fmt.Errorf("failed to parse topic time: %v", err)
272 }
273 ch.TopicTime = time.Unix(sec, 0)
274 case irc.RPL_NAMREPLY:
275 var name, statusStr, members string
276 if err := parseMessageParams(msg, nil, &statusStr, &name, &members); err != nil {
277 return err
278 }
279 ch, err := uc.getChannel(name)
280 if err != nil {
281 return err
282 }
283
284 status, err := parseChannelStatus(statusStr)
285 if err != nil {
286 return err
287 }
288 ch.Status = status
289
290 for _, s := range strings.Split(members, " ") {
291 membership, nick := parseMembershipPrefix(s)
292 ch.Members[nick] = membership
293 }
294 case irc.RPL_ENDOFNAMES:
295 var name string
296 if err := parseMessageParams(msg, nil, &name); err != nil {
297 return err
298 }
299 ch, err := uc.getChannel(name)
300 if err != nil {
301 return err
302 }
303
304 if ch.complete {
305 return fmt.Errorf("received unexpected RPL_ENDOFNAMES")
306 }
307 ch.complete = true
308
309 uc.user.forEachDownstream(func(dc *downstreamConn) {
310 forwardChannel(dc, ch)
311 })
312 case "PRIVMSG":
313 uc.ring.Produce(msg)
314 case irc.RPL_YOURHOST, irc.RPL_CREATED:
315 // Ignore
316 case irc.RPL_LUSERCLIENT, irc.RPL_LUSEROP, irc.RPL_LUSERUNKNOWN, irc.RPL_LUSERCHANNELS, irc.RPL_LUSERME:
317 // Ignore
318 case irc.RPL_MOTDSTART, irc.RPL_MOTD, irc.RPL_ENDOFMOTD:
319 // Ignore
320 case rpl_localusers, rpl_globalusers:
321 // Ignore
322 case irc.RPL_STATSVLINE, irc.RPL_STATSPING, irc.RPL_STATSBLINE, irc.RPL_STATSDLINE:
323 // Ignore
324 default:
325 uc.logger.Printf("unhandled upstream message: %v", msg)
326 }
327 return nil
328}
329
330func (uc *upstreamConn) register() {
331 uc.nick = uc.upstream.Nick
332 uc.SendMessage(&irc.Message{
333 Command: "NICK",
334 Params: []string{uc.upstream.Nick},
335 })
336 uc.SendMessage(&irc.Message{
337 Command: "USER",
338 Params: []string{uc.upstream.Username, "0", "*", uc.upstream.Realname},
339 })
340}
341
342func (uc *upstreamConn) readMessages() error {
343 for {
344 msg, err := uc.irc.ReadMessage()
345 if err == io.EOF {
346 break
347 } else if err != nil {
348 return fmt.Errorf("failed to read IRC command: %v", err)
349 }
350
351 if uc.srv.Debug {
352 uc.logger.Printf("received: %v", msg)
353 }
354
355 if err := uc.handleMessage(msg); err != nil {
356 uc.logger.Printf("failed to handle message %q: %v", msg, err)
357 }
358 }
359
360 return nil
361}
362
363func (uc *upstreamConn) SendMessage(msg *irc.Message) {
364 uc.messages <- msg
365}
Note: See TracBrowser for help on using the repository browser.