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
Line 
1package jounce
2
3import (
4 "fmt"
5 "log"
6 "net"
7 "sync"
8
9 "gopkg.in/irc.v3"
10)
11
12type Logger interface {
13 Print(v ...interface{})
14 Printf(format string, v ...interface{})
15}
16
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
34type user struct {
35 username string
36 srv *Server
37
38 lock sync.Mutex
39 upstreamConns []*upstreamConn
40 downstreamConns []*downstreamConn
41}
42
43func newUser(srv *Server, username string) *user {
44 return &user{
45 username: username,
46 srv: srv,
47 }
48}
49
50func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
51 u.lock.Lock()
52 for _, uc := range u.upstreamConns {
53 if !uc.registered || uc.closed {
54 continue
55 }
56 f(uc)
57 }
58 u.lock.Unlock()
59}
60
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
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 {
85 return nil, ircError{&irc.Message{
86 Command: irc.ERR_NOSUCHCHANNEL,
87 Params: []string{name, "No such channel"},
88 }}
89 }
90 return channel, nil
91}
92
93type Upstream struct {
94 Addr string
95 Nick string
96 Username string
97 Realname string
98 Channels []string
99}
100
101type Server struct {
102 Hostname string
103 Logger Logger
104 RingCap int
105 Upstreams []Upstream // TODO: per-user
106
107 lock sync.Mutex
108 users map[string]*user
109 downstreamConns []*downstreamConn
110}
111
112func NewServer() *Server {
113 return &Server{
114 Logger: log.New(log.Writer(), "", log.LstdFlags),
115 RingCap: 4096,
116 users: make(map[string]*user),
117 }
118}
119
120func (s *Server) prefix() *irc.Prefix {
121 return &irc.Prefix{Name: s.Hostname}
122}
123
124func (s *Server) Run() {
125 // TODO: multi-user
126 u := newUser(s, "jounce")
127
128 s.lock.Lock()
129 s.users[u.username] = u
130 s.lock.Unlock()
131
132 for i := range s.Upstreams {
133 upstream := &s.Upstreams[i]
134 // TODO: retry connecting
135 go func() {
136 conn, err := connectToUpstream(u, upstream)
137 if err != nil {
138 s.Logger.Printf("failed to connect to upstream server %q: %v", upstream.Addr, err)
139 return
140 }
141
142 conn.register()
143
144 u.lock.Lock()
145 u.upstreamConns = append(u.upstreamConns, conn)
146 u.lock.Unlock()
147
148 if err := conn.readMessages(); err != nil {
149 conn.logger.Printf("failed to handle messages: %v", err)
150 }
151
152 u.lock.Lock()
153 for i, c := range u.upstreamConns {
154 if c == conn {
155 u.upstreamConns = append(u.upstreamConns[:i], u.upstreamConns[i+1:]...)
156 break
157 }
158 }
159 u.lock.Unlock()
160 }()
161 }
162}
163
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
171func (s *Server) Serve(ln net.Listener) error {
172 for {
173 netConn, err := ln.Accept()
174 if err != nil {
175 return fmt.Errorf("failed to accept connection: %v", err)
176 }
177
178 conn := newDownstreamConn(s, netConn)
179 go func() {
180 s.lock.Lock()
181 s.downstreamConns = append(s.downstreamConns, conn)
182 s.lock.Unlock()
183
184 if err := conn.readMessages(); err != nil {
185 conn.logger.Printf("failed to handle messages: %v", err)
186 }
187
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()
196 }()
197 }
198}
Note: See TracBrowser for help on using the repository browser.