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