source: code/trunk/upstream.go@ 67

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

Enable TCP keep-alive on all connections
File size: 8.4 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":
[66]114 var from, to string
115 if len(msg.Params) >= 1 {
116 from = msg.Params[0]
117 }
118 if len(msg.Params) >= 2 {
119 to = msg.Params[1]
120 }
121
122 if to != "" && to != uc.srv.Hostname {
123 return fmt.Errorf("invalid PING destination %q", to)
124 }
125
126 params := []string{uc.srv.Hostname}
127 if from != "" {
128 params = append(params, from)
129 }
[60]130 uc.SendMessage(&irc.Message{
[13]131 Command: "PONG",
[66]132 Params: params,
[60]133 })
[33]134 return nil
[17]135 case "MODE":
[43]136 var name, modeStr string
137 if err := parseMessageParams(msg, &name, &modeStr); err != nil {
138 return err
[17]139 }
[35]140
141 if name == msg.Prefix.Name { // user mode change
[55]142 if name != uc.nick {
[35]143 return fmt.Errorf("received MODE message for unknow nick %q", name)
144 }
[55]145 return uc.modes.Apply(modeStr)
[35]146 } else { // channel mode change
[55]147 ch, err := uc.getChannel(name)
[35]148 if err != nil {
149 return err
150 }
151 if err := ch.modes.Apply(modeStr); err != nil {
152 return err
153 }
[46]154 }
[35]155
[55]156 uc.user.forEachDownstream(func(dc *downstreamConn) {
[54]157 dc.SendMessage(msg)
[46]158 })
[18]159 case "NOTICE":
[55]160 uc.logger.Print(msg)
[14]161 case irc.RPL_WELCOME:
[55]162 uc.registered = true
163 uc.logger.Printf("connection registered")
[19]164
[55]165 for _, ch := range uc.upstream.Channels {
[60]166 uc.SendMessage(&irc.Message{
[19]167 Command: "JOIN",
168 Params: []string{ch},
[60]169 })
[19]170 }
[16]171 case irc.RPL_MYINFO:
[55]172 if err := parseMessageParams(msg, nil, &uc.serverName, nil, &uc.availableUserModes, &uc.availableChannelModes); err != nil {
[43]173 return err
[16]174 }
175 if len(msg.Params) > 5 {
[55]176 uc.channelModesWithParam = msg.Params[5]
[16]177 }
[42]178 case "NICK":
[43]179 var newNick string
180 if err := parseMessageParams(msg, &newNick); err != nil {
181 return err
[42]182 }
183
[55]184 if msg.Prefix.Name == uc.nick {
185 uc.logger.Printf("changed nick from %q to %q", uc.nick, newNick)
186 uc.nick = newNick
[42]187 }
188
[55]189 for _, ch := range uc.channels {
[42]190 if membership, ok := ch.Members[msg.Prefix.Name]; ok {
191 delete(ch.Members, msg.Prefix.Name)
192 ch.Members[newNick] = membership
193 }
194 }
195
[55]196 uc.user.forEachDownstream(func(dc *downstreamConn) {
[54]197 dc.SendMessage(msg)
[42]198 })
[19]199 case "JOIN":
[43]200 var channels string
201 if err := parseMessageParams(msg, &channels); err != nil {
202 return err
[19]203 }
[34]204
[43]205 for _, ch := range strings.Split(channels, ",") {
[55]206 if msg.Prefix.Name == uc.nick {
207 uc.logger.Printf("joined channel %q", ch)
208 uc.channels[ch] = &upstreamChannel{
[34]209 Name: ch,
[55]210 conn: uc,
[34]211 Members: make(map[string]membership),
212 }
213 } else {
[55]214 ch, err := uc.getChannel(ch)
[34]215 if err != nil {
216 return err
217 }
218 ch.Members[msg.Prefix.Name] = 0
[19]219 }
220 }
[34]221
[55]222 uc.user.forEachDownstream(func(dc *downstreamConn) {
[54]223 dc.SendMessage(msg)
[40]224 })
[34]225 case "PART":
[43]226 var channels string
227 if err := parseMessageParams(msg, &channels); err != nil {
228 return err
[34]229 }
230
[43]231 for _, ch := range strings.Split(channels, ",") {
[55]232 if msg.Prefix.Name == uc.nick {
233 uc.logger.Printf("parted channel %q", ch)
234 delete(uc.channels, ch)
[34]235 } else {
[55]236 ch, err := uc.getChannel(ch)
[34]237 if err != nil {
238 return err
239 }
240 delete(ch.Members, msg.Prefix.Name)
241 }
242 }
243
[55]244 uc.user.forEachDownstream(func(dc *downstreamConn) {
[54]245 dc.SendMessage(msg)
[40]246 })
[19]247 case irc.RPL_TOPIC, irc.RPL_NOTOPIC:
[43]248 var name, topic string
249 if err := parseMessageParams(msg, nil, &name, &topic); err != nil {
250 return err
[19]251 }
[55]252 ch, err := uc.getChannel(name)
[19]253 if err != nil {
254 return err
255 }
256 if msg.Command == irc.RPL_TOPIC {
[43]257 ch.Topic = topic
[19]258 } else {
259 ch.Topic = ""
260 }
261 case "TOPIC":
[43]262 var name string
263 if err := parseMessageParams(msg, nil, &name); err != nil {
264 return err
[19]265 }
[55]266 ch, err := uc.getChannel(name)
[19]267 if err != nil {
268 return err
269 }
270 if len(msg.Params) > 1 {
271 ch.Topic = msg.Params[1]
272 } else {
273 ch.Topic = ""
274 }
275 case rpl_topicwhotime:
[43]276 var name, who, timeStr string
277 if err := parseMessageParams(msg, nil, &name, &who, &timeStr); err != nil {
278 return err
[19]279 }
[55]280 ch, err := uc.getChannel(name)
[19]281 if err != nil {
282 return err
283 }
[43]284 ch.TopicWho = who
285 sec, err := strconv.ParseInt(timeStr, 10, 64)
[19]286 if err != nil {
287 return fmt.Errorf("failed to parse topic time: %v", err)
288 }
289 ch.TopicTime = time.Unix(sec, 0)
290 case irc.RPL_NAMREPLY:
[43]291 var name, statusStr, members string
292 if err := parseMessageParams(msg, nil, &statusStr, &name, &members); err != nil {
293 return err
[19]294 }
[55]295 ch, err := uc.getChannel(name)
[19]296 if err != nil {
297 return err
298 }
299
[43]300 status, err := parseChannelStatus(statusStr)
[19]301 if err != nil {
302 return err
303 }
304 ch.Status = status
305
[43]306 for _, s := range strings.Split(members, " ") {
[19]307 membership, nick := parseMembershipPrefix(s)
308 ch.Members[nick] = membership
309 }
310 case irc.RPL_ENDOFNAMES:
[43]311 var name string
312 if err := parseMessageParams(msg, nil, &name); err != nil {
313 return err
[25]314 }
[55]315 ch, err := uc.getChannel(name)
[25]316 if err != nil {
317 return err
318 }
319
[34]320 if ch.complete {
321 return fmt.Errorf("received unexpected RPL_ENDOFNAMES")
322 }
[25]323 ch.complete = true
[27]324
[55]325 uc.user.forEachDownstream(func(dc *downstreamConn) {
[27]326 forwardChannel(dc, ch)
[40]327 })
[36]328 case "PRIVMSG":
[55]329 uc.ring.Produce(msg)
[16]330 case irc.RPL_YOURHOST, irc.RPL_CREATED:
[14]331 // Ignore
332 case irc.RPL_LUSERCLIENT, irc.RPL_LUSEROP, irc.RPL_LUSERUNKNOWN, irc.RPL_LUSERCHANNELS, irc.RPL_LUSERME:
333 // Ignore
334 case irc.RPL_MOTDSTART, irc.RPL_MOTD, irc.RPL_ENDOFMOTD:
335 // Ignore
336 case rpl_localusers, rpl_globalusers:
337 // Ignore
338 case irc.RPL_STATSVLINE, irc.RPL_STATSPING, irc.RPL_STATSBLINE, irc.RPL_STATSDLINE:
339 // Ignore
[13]340 default:
[55]341 uc.logger.Printf("unhandled upstream message: %v", msg)
[13]342 }
[14]343 return nil
[13]344}
345
[55]346func (uc *upstreamConn) register() {
347 uc.nick = uc.upstream.Nick
[60]348 uc.SendMessage(&irc.Message{
[13]349 Command: "NICK",
[55]350 Params: []string{uc.upstream.Nick},
[60]351 })
352 uc.SendMessage(&irc.Message{
[13]353 Command: "USER",
[55]354 Params: []string{uc.upstream.Username, "0", "*", uc.upstream.Realname},
[60]355 })
[44]356}
[13]357
[55]358func (uc *upstreamConn) readMessages() error {
[13]359 for {
[55]360 msg, err := uc.irc.ReadMessage()
[13]361 if err == io.EOF {
362 break
363 } else if err != nil {
364 return fmt.Errorf("failed to read IRC command: %v", err)
365 }
366
[64]367 if uc.srv.Debug {
368 uc.logger.Printf("received: %v", msg)
369 }
370
[55]371 if err := uc.handleMessage(msg); err != nil {
372 uc.logger.Printf("failed to handle message %q: %v", msg, err)
[13]373 }
374 }
375
[45]376 return nil
[13]377}
[60]378
379func (uc *upstreamConn) SendMessage(msg *irc.Message) {
380 uc.messages <- msg
381}
Note: See TracBrowser for help on using the repository browser.