source: code/trunk/server.go@ 59

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

Fix issues related to Ring

  • RingConsumer is now used directly in the goroutine responsible for writing downstream messages. This allows the ring buffer not to be consumed on write error.
  • RingConsumer now has a channel attached. This allows PRIVMSG messages to always use RingConsumer, instead of also directly pushing messages to all downstream connections.
  • Multiple clients with the same history name are now supported.
  • Ring is now protected by a mutex
File size: 3.7 KB
RevLine 
[1]1package jounce
2
3import (
4 "fmt"
[37]5 "log"
[1]6 "net"
[24]7 "sync"
[1]8
9 "gopkg.in/irc.v3"
10)
11
[9]12type Logger interface {
13 Print(v ...interface{})
14 Printf(format string, v ...interface{})
15}
16
[21]17type prefixLogger struct {
18 logger Logger
19 prefix string
20}
21
22var _ Logger = (*prefixLogger)(nil)
23
24func (l *prefixLogger) Print(v ...interface{}) {
25 v = append([]interface{}{l.prefix}, v...)
26 l.logger.Print(v...)
27}
28
29func (l *prefixLogger) Printf(format string, v ...interface{}) {
30 v = append([]interface{}{l.prefix}, v...)
31 l.logger.Printf("%v"+format, v...)
32}
33
[37]34type user struct {
35 username string
36 srv *Server
37
[40]38 lock sync.Mutex
39 upstreamConns []*upstreamConn
40 downstreamConns []*downstreamConn
[37]41}
42
[57]43func newUser(srv *Server, username string) *user {
44 return &user{
45 username: username,
46 srv: srv,
47 }
48}
49
[39]50func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
51 u.lock.Lock()
52 for _, uc := range u.upstreamConns {
[41]53 if !uc.registered || uc.closed {
[39]54 continue
55 }
56 f(uc)
57 }
58 u.lock.Unlock()
59}
60
[40]61func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
62 u.lock.Lock()
63 for _, dc := range u.downstreamConns {
64 f(dc)
65 }
66 u.lock.Unlock()
67}
68
[46]69func (u *user) getChannel(name string) (*upstreamChannel, error) {
70 var channel *upstreamChannel
71 var err error
72 u.forEachUpstream(func(uc *upstreamConn) {
73 if err != nil {
74 return
75 }
76 if ch, ok := uc.channels[name]; ok {
77 if channel != nil {
78 err = fmt.Errorf("ambiguous channel name %q", name)
79 } else {
80 channel = ch
81 }
82 }
83 })
84 if channel == nil {
[49]85 return nil, ircError{&irc.Message{
86 Command: irc.ERR_NOSUCHCHANNEL,
87 Params: []string{name, "No such channel"},
88 }}
[46]89 }
90 return channel, nil
91}
92
[10]93type Upstream struct {
94 Addr string
95 Nick string
96 Username string
97 Realname string
[19]98 Channels []string
[10]99}
100
101type Server struct {
102 Hostname string
103 Logger Logger
[50]104 RingCap int
[10]105 Upstreams []Upstream // TODO: per-user
[22]106
[24]107 lock sync.Mutex
[37]108 users map[string]*user
[22]109 downstreamConns []*downstreamConn
[10]110}
111
[37]112func NewServer() *Server {
113 return &Server{
[50]114 Logger: log.New(log.Writer(), "", log.LstdFlags),
115 RingCap: 4096,
116 users: make(map[string]*user),
[37]117 }
118}
119
[5]120func (s *Server) prefix() *irc.Prefix {
121 return &irc.Prefix{Name: s.Hostname}
122}
123
[10]124func (s *Server) Run() {
[37]125 // TODO: multi-user
[57]126 u := newUser(s, "jounce")
[37]127
128 s.lock.Lock()
129 s.users[u.username] = u
130 s.lock.Unlock()
131
[10]132 for i := range s.Upstreams {
[12]133 upstream := &s.Upstreams[i]
[10]134 // TODO: retry connecting
[12]135 go func() {
[37]136 conn, err := connectToUpstream(u, upstream)
[23]137 if err != nil {
138 s.Logger.Printf("failed to connect to upstream server %q: %v", upstream.Addr, err)
139 return
[12]140 }
[29]141
[44]142 conn.register()
143
[37]144 u.lock.Lock()
145 u.upstreamConns = append(u.upstreamConns, conn)
146 u.lock.Unlock()
[29]147
[23]148 if err := conn.readMessages(); err != nil {
149 conn.logger.Printf("failed to handle messages: %v", err)
150 }
[29]151
[37]152 u.lock.Lock()
153 for i, c := range u.upstreamConns {
[29]154 if c == conn {
[37]155 u.upstreamConns = append(u.upstreamConns[:i], u.upstreamConns[i+1:]...)
[29]156 break
157 }
158 }
[37]159 u.lock.Unlock()
[12]160 }()
[10]161 }
162}
163
[38]164func (s *Server) getUser(name string) *user {
165 s.lock.Lock()
166 u := s.users[name]
167 s.lock.Unlock()
168 return u
169}
170
[3]171func (s *Server) Serve(ln net.Listener) error {
[1]172 for {
[22]173 netConn, err := ln.Accept()
[1]174 if err != nil {
175 return fmt.Errorf("failed to accept connection: %v", err)
176 }
177
[22]178 conn := newDownstreamConn(s, netConn)
[1]179 go func() {
[24]180 s.lock.Lock()
181 s.downstreamConns = append(s.downstreamConns, conn)
182 s.lock.Unlock()
[29]183
[22]184 if err := conn.readMessages(); err != nil {
[23]185 conn.logger.Printf("failed to handle messages: %v", err)
[1]186 }
[29]187
[24]188 s.lock.Lock()
189 for i, c := range s.downstreamConns {
190 if c == conn {
191 s.downstreamConns = append(s.downstreamConns[:i], s.downstreamConns[i+1:]...)
192 break
193 }
194 }
195 s.lock.Unlock()
[1]196 }()
197 }
198}
Note: See TracBrowser for help on using the repository browser.