soju/server.go

241 lines
4.5 KiB
Go
Raw Normal View History

2020-02-04 09:46:22 +00:00
package jounce
import (
"fmt"
2020-02-07 10:36:42 +00:00
"log"
2020-02-04 09:46:22 +00:00
"net"
"sync"
"time"
2020-02-04 09:46:22 +00:00
"gopkg.in/irc.v3"
)
// TODO: make configurable
var keepAlivePeriod = time.Minute
var retryConnectMinDelay = time.Minute
func setKeepAlive(c net.Conn) error {
tcpConn, ok := c.(*net.TCPConn)
if !ok {
return fmt.Errorf("cannot enable keep-alive on a non-TCP connection")
}
if err := tcpConn.SetKeepAlive(true); err != nil {
return err
}
return tcpConn.SetKeepAlivePeriod(keepAlivePeriod)
}
2020-02-06 14:50:46 +00:00
type Logger interface {
Print(v ...interface{})
Printf(format string, v ...interface{})
}
2020-02-06 19:25:37 +00:00
type prefixLogger struct {
logger Logger
prefix string
}
var _ Logger = (*prefixLogger)(nil)
func (l *prefixLogger) Print(v ...interface{}) {
v = append([]interface{}{l.prefix}, v...)
l.logger.Print(v...)
}
func (l *prefixLogger) Printf(format string, v ...interface{}) {
v = append([]interface{}{l.prefix}, v...)
l.logger.Printf("%v"+format, v...)
}
type network struct {
Network
user *user
conn *upstreamConn
}
func newNetwork(user *user, record *Network) *network {
return &network{
Network: *record,
user: user,
}
}
func (net *network) run() {
var lastTry time.Time
for {
if dur := time.Now().Sub(lastTry); dur < retryConnectMinDelay {
delay := retryConnectMinDelay - dur
net.user.srv.Logger.Printf("waiting %v before trying to reconnect to %q", delay.Truncate(time.Second), net.Addr)
time.Sleep(delay)
}
lastTry = time.Now()
uc, err := connectToUpstream(net)
if err != nil {
net.user.srv.Logger.Printf("failed to connect to upstream server %q: %v", net.Addr, err)
continue
}
uc.register()
net.user.lock.Lock()
net.conn = uc
net.user.lock.Unlock()
if err := uc.readMessages(); err != nil {
uc.logger.Printf("failed to handle messages: %v", err)
}
uc.Close()
net.user.lock.Lock()
net.conn = nil
net.user.lock.Unlock()
}
}
2020-02-07 10:36:42 +00:00
type user struct {
User
2020-02-07 10:36:42 +00:00
srv *Server
2020-02-07 10:56:36 +00:00
lock sync.Mutex
networks []*network
2020-02-07 10:56:36 +00:00
downstreamConns []*downstreamConn
2020-02-07 10:36:42 +00:00
}
func newUser(srv *Server, record *User) *user {
return &user{
User: *record,
srv: srv,
}
}
2020-02-07 10:46:44 +00:00
func (u *user) forEachUpstream(f func(uc *upstreamConn)) {
u.lock.Lock()
for _, network := range u.networks {
uc := network.conn
if uc == nil || !uc.registered || uc.closed {
2020-02-07 10:46:44 +00:00
continue
}
f(uc)
}
u.lock.Unlock()
}
2020-02-07 10:56:36 +00:00
func (u *user) forEachDownstream(f func(dc *downstreamConn)) {
u.lock.Lock()
for _, dc := range u.downstreamConns {
f(dc)
}
u.lock.Unlock()
}
func (u *user) getNetwork(name string) *network {
for _, network := range u.networks {
if network.Addr == name {
return network
2020-02-07 12:08:27 +00:00
}
}
return nil
2020-02-07 12:08:27 +00:00
}
func (u *user) run() {
networks, err := u.srv.db.ListNetworks(u.Username)
if err != nil {
u.srv.Logger.Printf("failed to list networks for user %q: %v", u.Username, err)
return
}
u.lock.Lock()
for _, record := range networks {
network := newNetwork(u, &record)
u.networks = append(u.networks, network)
go network.run()
}
u.lock.Unlock()
2020-02-06 15:03:07 +00:00
}
type Server struct {
Hostname string
Logger Logger
RingCap int
2020-02-18 15:31:18 +00:00
Debug bool
db *DB
lock sync.Mutex
2020-02-07 10:36:42 +00:00
users map[string]*user
downstreamConns []*downstreamConn
2020-02-07 10:36:42 +00:00
}
func NewServer(db *DB) *Server {
2020-02-07 10:36:42 +00:00
return &Server{
Logger: log.New(log.Writer(), "", log.LstdFlags),
RingCap: 4096,
users: make(map[string]*user),
db: db,
2020-02-07 10:36:42 +00:00
}
2020-02-04 17:56:07 +00:00
}
2020-02-04 09:46:22 +00:00
2020-02-04 17:56:07 +00:00
func (s *Server) prefix() *irc.Prefix {
return &irc.Prefix{Name: s.Hostname}
}
2020-02-04 10:25:53 +00:00
func (s *Server) Run() error {
users, err := s.db.ListUsers()
if err != nil {
return err
}
2020-02-07 10:36:42 +00:00
s.lock.Lock()
for _, record := range users {
s.Logger.Printf("starting bouncer for user %q", record.Username)
u := newUser(s, &record)
s.users[u.Username] = u
2020-02-07 10:36:42 +00:00
go u.run()
2020-02-06 15:03:07 +00:00
}
s.lock.Unlock()
select {}
2020-02-06 15:03:07 +00:00
}
2020-02-07 10:39:56 +00:00
func (s *Server) getUser(name string) *user {
s.lock.Lock()
u := s.users[name]
s.lock.Unlock()
return u
}
2020-02-04 10:25:53 +00:00
func (s *Server) Serve(ln net.Listener) error {
2020-02-04 09:46:22 +00:00
for {
netConn, err := ln.Accept()
2020-02-04 09:46:22 +00:00
if err != nil {
return fmt.Errorf("failed to accept connection: %v", err)
}
setKeepAlive(netConn)
2020-02-18 15:54:06 +00:00
dc := newDownstreamConn(s, netConn)
2020-02-04 09:46:22 +00:00
go func() {
s.lock.Lock()
2020-02-18 15:54:06 +00:00
s.downstreamConns = append(s.downstreamConns, dc)
s.lock.Unlock()
2020-02-18 15:54:06 +00:00
if err := dc.readMessages(); err != nil {
dc.logger.Printf("failed to handle messages: %v", err)
2020-02-04 09:46:22 +00:00
}
2020-02-18 15:54:06 +00:00
dc.Close()
s.lock.Lock()
2020-02-18 15:54:06 +00:00
for i := range s.downstreamConns {
if s.downstreamConns[i] == dc {
s.downstreamConns = append(s.downstreamConns[:i], s.downstreamConns[i+1:]...)
break
}
}
s.lock.Unlock()
2020-02-04 09:46:22 +00:00
}()
}
}