source: code/trunk/downstream.go@ 51

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

Add names to consumers
File size: 7.4 KB
RevLine 
[13]1package jounce
2
3import (
4 "fmt"
5 "io"
6 "net"
[39]7 "strings"
[13]8
9 "gopkg.in/irc.v3"
10)
11
12type ircError struct {
13 Message *irc.Message
14}
15
16func newUnknownCommandError(cmd string) ircError {
17 return ircError{&irc.Message{
18 Command: irc.ERR_UNKNOWNCOMMAND,
19 Params: []string{
20 "*",
21 cmd,
22 "Unknown command",
23 },
24 }}
25}
26
27func newNeedMoreParamsError(cmd string) ircError {
28 return ircError{&irc.Message{
29 Command: irc.ERR_NEEDMOREPARAMS,
30 Params: []string{
31 "*",
32 cmd,
33 "Not enough parameters",
34 },
35 }}
36}
37
38func (err ircError) Error() string {
39 return err.Message.String()
40}
41
42type downstreamConn struct {
[26]43 net net.Conn
44 irc *irc.Conn
45 srv *Server
46 logger Logger
47 messages chan<- *irc.Message
[22]48
[13]49 registered bool
[37]50 user *user
[13]51 closed bool
52 nick string
53 username string
54 realname string
55}
56
[22]57func newDownstreamConn(srv *Server, netConn net.Conn) *downstreamConn {
[26]58 msgs := make(chan *irc.Message, 64)
59 conn := &downstreamConn{
60 net: netConn,
61 irc: irc.NewConn(netConn),
62 srv: srv,
63 logger: &prefixLogger{srv.Logger, fmt.Sprintf("downstream %q: ", netConn.RemoteAddr())},
64 messages: msgs,
[22]65 }
[26]66
67 go func() {
68 for msg := range msgs {
69 if err := conn.irc.WriteMessage(msg); err != nil {
70 conn.logger.Printf("failed to write message: %v", err)
71 }
72 }
[45]73 if err := conn.net.Close(); err != nil {
74 conn.logger.Printf("failed to close connection: %v", err)
75 } else {
76 conn.logger.Printf("connection closed")
77 }
[26]78 }()
79
80 return conn
[22]81}
82
[27]83func (c *downstreamConn) prefix() *irc.Prefix {
84 return &irc.Prefix{
85 Name: c.nick,
86 User: c.username,
87 // TODO: fill the host?
88 }
89}
90
[22]91func (c *downstreamConn) readMessages() error {
92 c.logger.Printf("new connection")
93
94 for {
95 msg, err := c.irc.ReadMessage()
96 if err == io.EOF {
97 break
98 } else if err != nil {
99 return fmt.Errorf("failed to read IRC command: %v", err)
100 }
101
102 err = c.handleMessage(msg)
103 if ircErr, ok := err.(ircError); ok {
104 ircErr.Message.Prefix = c.srv.prefix()
[26]105 c.messages <- ircErr.Message
[22]106 } else if err != nil {
107 return fmt.Errorf("failed to handle IRC command %q: %v", msg.Command, err)
108 }
109
110 if c.closed {
111 return nil
112 }
113 }
114
[45]115 return nil
[22]116}
117
[13]118func (c *downstreamConn) Close() error {
[26]119 if c.closed {
120 return fmt.Errorf("downstream connection already closed")
121 }
[40]122
123 if u := c.user; u != nil {
124 u.lock.Lock()
125 for i := range u.downstreamConns {
126 if u.downstreamConns[i] == c {
127 u.downstreamConns = append(u.downstreamConns[:i], u.downstreamConns[i+1:]...)
128 }
129 }
130 u.lock.Unlock()
[51]131
132 // TODO: figure out a better way to advance the ring buffer consumer cursor
133 u.forEachUpstream(func(uc *upstreamConn) {
134 // TODO: let clients specify the ring buffer name in their username
135 uc.ring.Consumer("").Reset()
136 })
[13]137 }
[40]138
[26]139 close(c.messages)
[13]140 c.closed = true
[40]141
[45]142 return nil
[13]143}
144
145func (c *downstreamConn) handleMessage(msg *irc.Message) error {
146 switch msg.Command {
[28]147 case "QUIT":
148 return c.Close()
[13]149 case "PING":
150 // TODO: handle params
[31]151 c.messages <- &irc.Message{
152 Prefix: c.srv.prefix(),
[13]153 Command: "PONG",
154 Params: []string{c.srv.Hostname},
[31]155 }
[26]156 return nil
[13]157 default:
158 if c.registered {
159 return c.handleMessageRegistered(msg)
160 } else {
161 return c.handleMessageUnregistered(msg)
162 }
163 }
164}
165
166func (c *downstreamConn) handleMessageUnregistered(msg *irc.Message) error {
167 switch msg.Command {
168 case "NICK":
[43]169 if err := parseMessageParams(msg, &c.nick); err != nil {
170 return err
[13]171 }
172 case "USER":
[43]173 var username string
174 if err := parseMessageParams(msg, &username, nil, nil, &c.realname); err != nil {
175 return err
[13]176 }
[43]177 c.username = "~" + username
[13]178 default:
[22]179 c.logger.Printf("unhandled message: %v", msg)
[13]180 return newUnknownCommandError(msg.Command)
181 }
182 if c.username != "" && c.nick != "" {
183 return c.register()
184 }
185 return nil
186}
187
188func (c *downstreamConn) register() error {
[39]189 u := c.srv.getUser(strings.TrimPrefix(c.username, "~"))
[38]190 if u == nil {
[39]191 c.logger.Printf("failed authentication: unknown username %q", c.username)
[37]192 c.messages <- &irc.Message{
193 Prefix: c.srv.prefix(),
194 Command: irc.ERR_PASSWDMISMATCH,
195 Params: []string{"*", "Invalid username or password"},
196 }
197 return nil
198 }
199
[13]200 c.registered = true
[37]201 c.user = u
[13]202
[40]203 u.lock.Lock()
204 u.downstreamConns = append(u.downstreamConns, c)
205 u.lock.Unlock()
206
[31]207 c.messages <- &irc.Message{
208 Prefix: c.srv.prefix(),
[13]209 Command: irc.RPL_WELCOME,
210 Params: []string{c.nick, "Welcome to jounce, " + c.nick},
[31]211 }
212 c.messages <- &irc.Message{
213 Prefix: c.srv.prefix(),
[13]214 Command: irc.RPL_YOURHOST,
215 Params: []string{c.nick, "Your host is " + c.srv.Hostname},
[31]216 }
217 c.messages <- &irc.Message{
218 Prefix: c.srv.prefix(),
[13]219 Command: irc.RPL_CREATED,
[47]220 Params: []string{c.nick, "Who cares when the server was created?"},
[31]221 }
222 c.messages <- &irc.Message{
223 Prefix: c.srv.prefix(),
[13]224 Command: irc.RPL_MYINFO,
[28]225 Params: []string{c.nick, c.srv.Hostname, "jounce", "aiwroO", "OovaimnqpsrtklbeI"},
[31]226 }
227 c.messages <- &irc.Message{
228 Prefix: c.srv.prefix(),
[13]229 Command: irc.ERR_NOMOTD,
230 Params: []string{c.nick, "No MOTD"},
[31]231 }
[13]232
[39]233 u.forEachUpstream(func(uc *upstreamConn) {
[30]234 // TODO: fix races accessing upstream connection data
235 for _, ch := range uc.channels {
236 if ch.complete {
237 forwardChannel(c, ch)
238 }
239 }
[50]240
[51]241 // TODO: let clients specify the ring buffer name in their username
242 consumer := uc.ring.Consumer("")
[50]243 for {
[51]244 // TODO: these messages will get lost if the connection is closed
[50]245 msg := consumer.Consume()
246 if msg == nil {
247 break
248 }
249 c.messages <- msg
250 }
[39]251 })
[50]252
[13]253 return nil
254}
255
256func (c *downstreamConn) handleMessageRegistered(msg *irc.Message) error {
257 switch msg.Command {
[42]258 case "USER":
[13]259 return ircError{&irc.Message{
260 Command: irc.ERR_ALREADYREGISTERED,
[42]261 Params: []string{c.nick, "You may not reregister"},
[13]262 }}
[42]263 case "NICK":
264 c.user.forEachUpstream(func(uc *upstreamConn) {
265 uc.messages <- msg
266 })
[48]267 case "JOIN":
268 var name string
269 if err := parseMessageParams(msg, &name); err != nil {
270 return err
271 }
272
273 if ch, _ := c.user.getChannel(name); ch != nil {
274 break // already joined
275 }
276
277 // TODO: extract network name from channel name
278 return ircError{&irc.Message{
279 Command: irc.ERR_NOSUCHCHANNEL,
280 Params: []string{name, "Channel name ambiguous"},
281 }}
[49]282 case "PART":
283 var name string
284 if err := parseMessageParams(msg, &name); err != nil {
285 return err
286 }
287
288 ch, err := c.user.getChannel(name)
289 if err != nil {
290 return err
291 }
292
293 ch.conn.messages <- msg
294 // TODO: remove channel from upstream config
[46]295 case "MODE":
296 var name string
297 if err := parseMessageParams(msg, &name); err != nil {
298 return err
299 }
300
301 var modeStr string
302 if len(msg.Params) > 1 {
303 modeStr = msg.Params[1]
304 }
305
306 if msg.Prefix.Name != name {
307 ch, err := c.user.getChannel(name)
308 if err != nil {
309 return err
310 }
311
312 if modeStr != "" {
313 ch.conn.messages <- msg
314 } else {
315 c.messages <- &irc.Message{
316 Prefix: c.srv.prefix(),
317 Command: irc.RPL_CHANNELMODEIS,
318 Params: []string{ch.Name, string(ch.modes)},
319 }
320 }
321 } else {
322 if name != c.nick {
323 return ircError{&irc.Message{
324 Command: irc.ERR_USERSDONTMATCH,
325 Params: []string{c.nick, "Cannot change mode for other users"},
326 }}
327 }
328
329 if modeStr != "" {
330 c.user.forEachUpstream(func(uc *upstreamConn) {
331 uc.messages <- msg
332 })
333 } else {
334 c.messages <- &irc.Message{
335 Prefix: c.srv.prefix(),
336 Command: irc.RPL_UMODEIS,
337 Params: []string{""}, // TODO
338 }
339 }
340 }
[13]341 default:
[22]342 c.logger.Printf("unhandled message: %v", msg)
[13]343 return newUnknownCommandError(msg.Command)
344 }
[42]345 return nil
[13]346}
Note: See TracBrowser for help on using the repository browser.