1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
package session
import (
"context"
"crypto/ed25519"
"crypto/rand"
"log"
"net"
"sync"
"time"
"github.com/kyren223/eko/internal/packet"
"github.com/kyren223/eko/pkg/assert"
"github.com/kyren223/eko/pkg/snowflake"
)
type SessionManager interface {
AddSession(session *Session)
RemoveSession(id snowflake.ID)
Session(id snowflake.ID) *Session
UseSessions(f func(map[snowflake.ID]*Session))
Node() *snowflake.Node
}
type Session struct {
manager SessionManager
addr *net.TCPAddr
cancel context.CancelFunc
writeQueue chan packet.Packet
issuedTime time.Time
challenge []byte
PubKey ed25519.PublicKey
id snowflake.ID
mu sync.Mutex
}
func NewSession(
manager SessionManager,
addr *net.TCPAddr, cancel context.CancelFunc,
id snowflake.ID, pubKey ed25519.PublicKey,
) *Session {
session := &Session{
manager: manager,
addr: addr,
cancel: cancel,
writeQueue: make(chan packet.Packet, 10),
issuedTime: time.Time{},
challenge: make([]byte, 32),
PubKey: pubKey,
id: id,
mu: sync.Mutex{},
}
session.Challenge() // Make sure an initial nonce is generated
return session
}
func (s *Session) Addr() *net.TCPAddr {
return s.addr
}
func (s *Session) ID() snowflake.ID {
return s.id
}
func (s *Session) Manager() SessionManager {
return s.manager
}
func (s *Session) Challenge() []byte {
s.mu.Lock()
defer s.mu.Unlock()
if time.Since(s.issuedTime) > time.Minute {
s.issuedTime = time.Now()
_, err := rand.Read(s.challenge)
assert.NoError(err, "random should always produce a value")
}
return s.challenge
}
func (s *Session) Write(ctx context.Context, pkt packet.Packet) bool {
select {
case s.writeQueue <- pkt:
return true
case <-ctx.Done():
return false
}
}
func (s *Session) Read(ctx context.Context) (packet.Packet, bool) {
select {
case pkt := <-s.writeQueue:
return pkt, true
case <-ctx.Done():
return packet.Packet{}, false
}
}
func (s *Session) Close() {
timeout := 10 * time.Millisecond
ctx, cancel := context.WithTimeout(context.Background(), timeout)
payload := &packet.Error{
Error: "new connection from another location, closing this one",
PktType: packet.PacketError,
}
pkt := packet.NewPacket(packet.NewMsgPackEncoder(payload))
s.Write(ctx, pkt)
cancel()
// Add some delay before canceling to let the writer enough time to
// actually write that into the connection
// HACK: consider just giving session the private connection and to
// write directly so we don't have to wait
time.Sleep(100 * time.Millisecond)
log.Println(s.addr, "closed due to new connection from another location")
s.cancel()
}
|
