anygatecloud/strom
2
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: modularize connection handling and streamline event loop structure

Browse Source
This commit is contained in:
adman
2026-01-05 12:40:43 +01:00
parent 83a3b0f886
commit 253a7020f9
11 changed files with 176 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/modules/config_ignorer.go
+1 -1
View File
@@ -29,7 +29,7 @@ func (c *ConfigIgnorer) OnFinish(_ v1_21_8.ConfigurationToClientPacketFinishConf
return
}
c.State = proto_base.Play
err = event.HandlerDone
err = event.ErrHandlerDone
return
}
client/modules/login.go
+3 -18
View File
@@ -1,8 +1,6 @@
package modules
import (
"crypto/aes"
"crypto/cipher"
"crypto/rand"
"crypto/rsa"
"crypto/x509"
@@ -34,7 +32,7 @@ func (s *LoginClient) Default(event event.Unhandled) (err error) {
return
}
func (s *LoginClient) OnCycle(_ event.OnLoopCycle) (err error) {
func (s *LoginClient) OnCycle(_ event.Tick) (err error) {
return
}
@@ -109,20 +107,7 @@ func (s *LoginClient) OnEncrypt(packet v1_21_8.LoginToClientPacketEncryptionBegi
if err != nil {
return
}
var b cipher.Block
b, err = aes.NewCipher(sharedSecret)
if err != nil {
return
}
s.R = cipher.StreamReader{
S: crypto.NewCFB8Decrypt(b, sharedSecret),
R: s.Conn,
}
s.W = cipher.StreamWriter{
S: crypto.NewCFB8Encrypt(b, sharedSecret),
W: s.Conn,
}
err = s.SetSecret(sharedSecret)
return
}
@@ -134,7 +119,7 @@ func (s *LoginClient) OnSuccess(success *v1_21_8.LoginToClientPacketSuccess) (er
return
}
s.State = proto_base.Configuration
err = event.HandlerDone
err = event.ErrHandlerDone
return
}
client/modules/status.go
+1 -1
View File
@@ -49,6 +49,6 @@ func (s *StatusClient) OnStart(_ event.OnStart) (err error) {
func (s *StatusClient) OnStatus(p v1_21_8.StatusToClientPacketServerInfo) (err error) {
s.Status = p.Response
err = event.HandlerDone
err = event.ErrHandlerDone
return
}
event/event_bus.go
+22 -3
View File
@@ -11,6 +11,7 @@ import (
type Loop struct {
HandlerFunctions map[reflect.Type][]reflect.Value
Handlers []any
HandlerIndex int
}
type Unhandled struct {
@@ -23,7 +24,7 @@ type Anything struct {
type (
OnStart struct{}
OnLoopCycle struct{}
Tick struct{}
)
// ErrHandlerDone will stop the event loop.
@@ -32,9 +33,9 @@ var ErrHandlerDone = errors.New("handler done")
// ErrDontForward will stop the event from being forwarded to other handlers, including anything handlers.
var ErrDontForward = errors.New("dont forward")
func (l *Loop) FindHandlers(insts []any) {
func (l *Loop) FindHandlers() {
l.HandlerFunctions = make(map[reflect.Type][]reflect.Value)
for _, inst := range insts {
for _, inst := range l.Handlers {
t := reflect.TypeOf(inst)
v := reflect.ValueOf(inst)
for i := 0; i < t.NumMethod(); i++ {
@@ -53,6 +54,7 @@ func (l *Loop) FindHandlers(insts []any) {
}
}
// FireFound triggers all handlers registered for the specified event type and returns if handlers were found or an error occurred.
func (l *Loop) FireFound(event any) (found bool, err error) {
var handlers []reflect.Value
if handlers, found = l.HandlerFunctions[reflect.TypeOf(event)]; found {
@@ -68,6 +70,7 @@ func (l *Loop) FireFound(event any) (found bool, err error) {
return
}
// Fire dispatches the provided event to all appropriate handlers and invokes fallback handlers if the event is unhandled.
func (l *Loop) Fire(event any) (err error) {
found, err := l.FireFound(event)
if err != nil {
@@ -85,3 +88,19 @@ func (l *Loop) Fire(event any) (err error) {
_, err = l.FireFound(Anything{event})
return
}
func (l *Loop) Start() (err error) {
_ = *l // exit early on nil connection
l.FindHandlers()
err = l.Fire(OnStart{})
for err == nil {
err = l.Fire(Tick{})
if err != nil {
break
}
}
if errors.Is(err, ErrHandlerDone) {
err = nil
}
return
}
proto/connection.go
+18 -107
View File
@@ -2,11 +2,8 @@ package proto
import (
"bytes"
"compress/zlib"
"errors"
"io"
"net"
"reflect"
"github.com/admin-else/strom/data"
"github.com/admin-else/strom/event"
@@ -35,16 +32,12 @@ func (u *UnCodablePacket) Decode(r io.Reader) (err error) {
}
type Conn struct {
net.Conn
R io.Reader
W io.Writer
CompressionThreshold int32
RawConn
*event.Loop
State proto_base.State
Actor proto_base.Actor
Version string
ProtocolVersion int32
Handlers map[reflect.Type][]reflect.Value
}
func (c *Conn) SetVersion(version string) (err error) {
@@ -79,83 +72,6 @@ func (c *Conn) Send(packet proto_base.EncodeDecodeAble) (err error) {
}
return
}
func (c *Conn) SendRaw(rawPacketBytes []byte) (err error) {
var packetBytes []byte
if c.CompressionThreshold > 0 {
packetBuffer := bytes.NewBuffer(nil)
if int32(len(packetBytes)) >= c.CompressionThreshold {
err = proto_base.EncodeVarInt(packetBuffer, int32(len(packetBytes)))
if err != nil {
return
}
_, err = zlib.NewWriter(packetBuffer).Write(rawPacketBytes)
if err != nil {
return
}
} else {
err = proto_base.EncodeVarInt(packetBuffer, 0)
if err != nil {
return
}
_, err = packetBuffer.Write(rawPacketBytes)
if err != nil {
return
}
}
packetBytes = packetBuffer.Bytes()
} else {
packetBytes = rawPacketBytes
}
err = proto_base.EncodeVarInt(c.W, int32(len(packetBytes)))
if err != nil {
return
}
_, err = c.W.Write(packetBytes)
return
}
func (c *Conn) ReceiveRaw() (packetBytes []byte, err error) {
rawPacketLen, err := proto_base.DecodeVarInt(c.R)
if err != nil {
return
}
rawPacketBytes, err := io.ReadAll(io.LimitReader(c.R, int64(rawPacketLen)))
if err != nil {
return
}
if len(rawPacketBytes) != int(rawPacketLen) {
err = errors.New("bad packet length")
return
}
rawPacketBuffer := bytes.NewBuffer(rawPacketBytes)
if c.CompressionThreshold > 0 {
var packetLen int32
packetLen, err = proto_base.DecodeVarInt(rawPacketBuffer)
if err != nil {
return
}
if packetLen == 0 {
packetBytes, err = io.ReadAll(rawPacketBuffer)
if err != nil {
return
}
} else {
var zReader io.ReadCloser
zReader, err = zlib.NewReader(rawPacketBuffer)
if err != nil {
return
}
defer zReader.Close()
packetBytes, err = io.ReadAll(zReader)
if err != nil {
return
}
}
} else {
packetBytes = rawPacketBytes
}
return
}
func (c *Conn) Receive() (packet proto_base.EncodeDecodeAble, err error) {
packetBytes, err := c.ReceiveRaw()
@@ -185,28 +101,23 @@ func (c *Conn) Receive() (packet proto_base.EncodeDecodeAble, err error) {
return
}
func (c *Conn) StartOne(inst any) (err error) {
return c.Start([]any{inst})
}
func (c *Conn) Start(insts []any) (err error) {
_ = *c // exit early on nil connection
c.Handlers = event.FindHandlers(insts)
err = event.Fire(event.OnStart{}, c.Handlers)
for err == nil {
var packet any
packet, err = c.Receive()
func (c *Conn) OnTick(_ event.Tick) (err error) {
packet, err := c.Receive()
if err != nil {
break
}
err = event.Fire(packet, c.Handlers)
if err != nil {
break
}
err = event.Fire(event.OnLoopCycle{}, c.Handlers)
}
if errors.Is(err, event.HandlerDone) {
err = nil
return
}
err = c.Loop.Fire(packet)
return
}
func (c *Conn) StartOne(handler any) (err error) {
return c.Start(handler)
}
func (c *Conn) Start(handlers ...any) (err error) {
c.Loop = &event.Loop{}
c.Handlers = append(c.Handlers, c)
c.Handlers = append(c.Handlers, handlers...)
err = c.Loop.Start()
return
}
proto/packet_codec.go
+1
View File
@@ -0,0 +1 @@
package proto
proto/packet_test.go
-55
View File
@@ -1,55 +0,0 @@
package proto_test
/*
import (
"bytes"
"math"
"testing"
"github.com/admin-else/strom/proto_generated/v1_21_8"
"github.com/google/go-cmp/cmp"
)
var nanIgnorer32 = cmp.Comparer(func(x, y float32) bool {
return (math.IsNaN(float64(x)) && math.IsNaN(float64(y))) || x == y
})
var nanIgnorer64 = cmp.Comparer(func(x, y float64) bool {
return (math.IsNaN(x) && math.IsNaN(y)) || x == y
})
func FuzzPlayPackets(f *testing.F) {
f.Fuzz(func(t *testing.T, b []byte) {
t.Logf("Fuzz data %q\n", b)
p := v1_21_8.PlayToClientPacket{}
b1 := bytes.NewBuffer(b)
p, err := p.Decode(b1)
if err != nil {
t.Skipf("failed to decode %q because \"%v\" (partial %v)", b, err, p)
return
}
if b1.Len() != 0 {
t.Skipf("there are %d bytes left it should be read entirely", b1.Len())
return
}
t.Logf("decoded %#v\n", p)
b2 := bytes.NewBuffer(nil)
err = p.Encode(b2)
if err != nil {
t.Errorf("failed to encode %q because %v", b, err)
return
}
t.Logf("encoded %q\n", b2.Bytes())
p2 := v1_21_8.PlayToClientPacket{}
p2, err = p2.Decode(b2)
if err != nil {
t.Errorf("failed to decode encoded data: %v (partial %#v)", err, p2)
return
}
if !cmp.Equal(p, p2, nanIgnorer32, nanIgnorer64) {
t.Errorf("decoded and encoded decoded data are not equal: %#v", p2)
}
})
}
*/
proto/raw_conn.go
+116
View File
@@ -0,0 +1,116 @@
package proto
import (
"bytes"
"compress/zlib"
"crypto/aes"
"crypto/cipher"
"errors"
"io"
"net"
"github.com/admin-else/strom/crypto"
"github.com/admin-else/strom/proto_base"
)
type RawConn struct {
net.Conn
R io.Reader
W io.Writer
CompressionThreshold int32
}
func (c *RawConn) SendRaw(rawPacketBytes []byte) (err error) {
var packetBytes []byte
if c.CompressionThreshold > 0 {
packetBuffer := bytes.NewBuffer(nil)
if int32(len(packetBytes)) >= c.CompressionThreshold {
err = proto_base.EncodeVarInt(packetBuffer, int32(len(packetBytes)))
if err != nil {
return
}
_, err = zlib.NewWriter(packetBuffer).Write(rawPacketBytes)
if err != nil {
return
}
} else {
err = proto_base.EncodeVarInt(packetBuffer, 0)
if err != nil {
return
}
_, err = packetBuffer.Write(rawPacketBytes)
if err != nil {
return
}
}
packetBytes = packetBuffer.Bytes()
} else {
packetBytes = rawPacketBytes
}
err = proto_base.EncodeVarInt(c.W, int32(len(packetBytes)))
if err != nil {
return
}
_, err = c.W.Write(packetBytes)
return
}
func (c *RawConn) ReceiveRaw() (packetBytes []byte, err error) {
rawPacketLen, err := proto_base.DecodeVarInt(c.R)
if err != nil {
return
}
rawPacketBytes, err := io.ReadAll(io.LimitReader(c.R, int64(rawPacketLen)))
if err != nil {
return
}
if len(rawPacketBytes) != int(rawPacketLen) {
err = errors.New("bad packet length")
return
}
rawPacketBuffer := bytes.NewBuffer(rawPacketBytes)
if c.CompressionThreshold > 0 {
var packetLen int32
packetLen, err = proto_base.DecodeVarInt(rawPacketBuffer)
if err != nil {
return
}
if packetLen == 0 {
packetBytes, err = io.ReadAll(rawPacketBuffer)
if err != nil {
return
}
} else {
var zReader io.ReadCloser
zReader, err = zlib.NewReader(rawPacketBuffer)
if err != nil {
return
}
defer zReader.Close()
packetBytes, err = io.ReadAll(zReader)
if err != nil {
return
}
}
} else {
packetBytes = rawPacketBytes
}
return
}
func (c *RawConn) SetSecret(sharedSecret []byte) (err error) {
var b cipher.Block
b, err = aes.NewCipher(sharedSecret)
if err != nil {
return
}
c.R = cipher.StreamReader{
S: crypto.NewCFB8Decrypt(b, sharedSecret),
R: c.Conn,
}
c.W = cipher.StreamWriter{
S: crypto.NewCFB8Encrypt(b, sharedSecret),
W: c.Conn,
}
return
}
proto_base/base_types.go
+4 -5
View File
@@ -81,14 +81,13 @@ type (
var ToDoError = errors.New("to do")
var BadTypeError = errors.New("bad type")
func (r *RestBuffer) Encode(w io.Writer) (err error) {
_, err = w.Write(*r)
func (b *RestBuffer) Encode(w io.Writer) (err error) {
_, err = w.Write(*b)
return
}
func (ret *RestBuffer) Decode(r io.Reader) (err error) {
b, err := io.ReadAll(r)
*ret = b
func (b *RestBuffer) Decode(r io.Reader) (err error) {
*b, err = io.ReadAll(r)
return
}
server/server_modules/login_server.go
+2 -2
View File
@@ -71,7 +71,7 @@ func (l *LoginServer) OnLoginStart(packet *v1_21_8.LoginToServerPacketLoginStart
func (l *LoginServer) OnLoginAcknowledged(_ *v1_21_8.LoginToServerPacketLoginAcknowledged) (err error) {
l.State = proto_base.Configuration
err = event.HandlerDone
err = event.ErrHandlerDone
return
}
@@ -85,7 +85,7 @@ func (l *LoginServer) OnStart(_ event.OnStart) (err error) {
return
}
func (l *LoginServer) OnCycle(_ event.OnLoopCycle) (err error) {
func (l *LoginServer) OnCycle(_ event.Tick) (err error) {
return
}
tools/uncodeable_packet_hunter_proxy/main.go
+2 -2
View File
@@ -30,7 +30,7 @@ func (p *ProxyClient) OnDefault(event event.Unhandled) (err error) {
return
}
func (p *ProxyClient) OnCycle(_ event.OnLoopCycle) (err error) {
func (p *ProxyClient) OnCycle(_ event.Tick) (err error) {
return
}
@@ -61,7 +61,7 @@ func (p *Proxy) OnDefault(event event.Unhandled) (err error) {
return
}
func (p *Proxy) OnCycle(_ event.OnLoopCycle) (err error) {
func (p *Proxy) OnCycle(_ event.Tick) (err error) {
return
}