diff --git a/client/client.go b/client/client.go index 5502caf..0fc4a5c 100644 --- a/client/client.go +++ b/client/client.go @@ -15,7 +15,7 @@ func Connect(addr string) (ret *proto.Conn, err error) { return } ret.SetState(proto_base.Handshaking) - ret.CompressionThreshold = -1 + ret.SetCompressionThreshold(-1) ret.Actor = proto_base.Client ret.Conn, err = net.Dial("tcp", addr) if err != nil { diff --git a/client/login.go b/client/login.go index e9b1211..642dc20 100644 --- a/client/login.go +++ b/client/login.go @@ -80,7 +80,7 @@ func (s *LoginClient) OnStart(_ event.Start) (err error) { } func (s *LoginClient) OnCompress(compress *v1_21_8.LoginToClientPacketCompress) (err error) { - s.CompressionThreshold = compress.Threshold + s.SetCompressionThreshold(compress.Threshold) return } diff --git a/event/event_bus.go b/event/event_bus.go index 3025dcb..8d1d641 100644 --- a/event/event_bus.go +++ b/event/event_bus.go @@ -2,16 +2,27 @@ package event import ( + "context" "errors" "fmt" "log/slog" "reflect" "strings" + "time" +) + +const ( + LoopsPerSecond = 100 // i dunno seems reasonable + TimePerLoop = time.Second / LoopsPerSecond ) type Loop struct { handlerFunctions map[reflect.Type][]reflect.Value Handlers []any + + Ctx context.Context + ErrChan chan error + cancel context.CancelFunc } type Unhandled struct { @@ -107,15 +118,18 @@ func (l *Loop) Fire(event any) (err error) { return } -func (l *Loop) Start() (err error) { +func (l *Loop) startLoop() (err error) { _ = *l // exit early on nil loop l.UpdateHandlerFunctions() + err = l.Fire(Start{}) for err == nil { + tickStartTime := time.Now() err = l.Fire(Tick{}) if err != nil { break } + time.Sleep(TimePerLoop - time.Since(tickStartTime)) } if errors.Is(err, HandlerDoneErr{}) { err = errors.Unwrap(err) @@ -126,3 +140,14 @@ func (l *Loop) Start() (err error) { } return } + +func (l *Loop) Start() (err error) { + l.ErrChan = make(chan error) + l.Ctx, l.cancel = context.WithCancel(context.Background()) + go func() { + l.ErrChan <- l.startLoop() + }() + err = <-l.ErrChan + l.cancel() + return +} diff --git a/proto/connection.go b/proto/connection.go index 7fbbf72..9167383 100644 --- a/proto/connection.go +++ b/proto/connection.go @@ -16,6 +16,7 @@ import ( var BadPacketTypeError = errors.New("bad packet type") var BadPacketIdError = errors.New("bad packet id") var PacketNotFullyDecodedError = errors.New("packet not fully decoded") +var ErrContextDone = errors.New("context done") // UnCodablePacket represents a packet that could not be decoded. type UnCodablePacket struct { @@ -42,6 +43,25 @@ type Conn struct { Actor proto_base.Actor Version string ProtocolVersion int32 + + receiveRoutine bool + packetCh chan proto_base.EncodeDecodeAble +} + +func (c *Conn) ReceiveJob() { + for { + select { + case <-c.Ctx.Done(): + return + default: + packet, err := c.Receive() + if err != nil { + c.ErrChan <- err + return + } + c.packetCh <- packet + } + } } func (c *Conn) SetState(state proto_base.State) { @@ -62,10 +82,17 @@ func (c *Conn) SetVersion(version string) (err error) { if err != nil { return } - c.ProtocolVersion = int32(versionData.Version) + c.ProtocolVersion = versionData.Version return } +func (c *Conn) ActivateReceiveRoutine() { + c.receiveRoutine = true + c.packetCh = make(chan proto_base.EncodeDecodeAble, 100) // i dunno how many to buffer 100 seems reasonable + go c.ReceiveJob() + +} + func (c *Conn) Send(packet proto_base.EncodeDecodeAble) (err error) { slog.Debug("send", "actor", c.Actor, "packet", fmt.Sprintf("%#v", packet)) switch packet := packet.(type) { @@ -120,12 +147,25 @@ func (c *Conn) Receive() (packet proto_base.EncodeDecodeAble, err error) { } func (c *Conn) OnTick(_ event.Tick) (err error) { - packet, err := c.Receive() - if err != nil { - return + var packet proto_base.EncodeDecodeAble + if c.receiveRoutine { + select { + case packet = <-c.packetCh: + slog.Debug("receive", "actor", c.Actor, "packet", fmt.Sprintf("%#v", packet)) + err = c.Loop.Fire(packet) + case <-c.Ctx.Done(): + err = ErrContextDone + return + default: + } + } else { + packet, err = c.Receive() + if err != nil { + return + } + slog.Debug("receive", "actor", c.Actor, "packet", fmt.Sprintf("%#v", packet)) + err = c.Loop.Fire(packet) } - slog.Debug("receive", "actor", c.Actor, "packet", fmt.Sprintf("%#v", packet)) - err = c.Loop.Fire(packet) return } diff --git a/proto/raw_conn.go b/proto/raw_conn.go index 638a66d..0491a20 100644 --- a/proto/raw_conn.go +++ b/proto/raw_conn.go @@ -8,6 +8,7 @@ import ( "errors" "io" "net" + "sync" "github.com/admin-else/strom/crypto" "github.com/admin-else/strom/proto_base" @@ -15,17 +16,31 @@ import ( type RawConn struct { net.Conn - R io.Reader - W io.Writer - CompressionThreshold int32 + R io.Reader + W io.Writer + compressionThreshold int32 + CompressionThresholdMut sync.RWMutex +} + +func (c *RawConn) SetCompressionThreshold(threshold int32) { + c.CompressionThresholdMut.Lock() + defer c.CompressionThresholdMut.Unlock() + c.compressionThreshold = threshold +} + +func (c *RawConn) GetCompressionThreshold() int32 { + c.CompressionThresholdMut.RLock() + defer c.CompressionThresholdMut.RUnlock() + return c.compressionThreshold } func (c *RawConn) SendRaw(rawPacketBytes []byte) (err error) { //fmt.Println("send:", rawPacketBytes) var packetBytes []byte - if c.CompressionThreshold > 0 { + threshold := c.GetCompressionThreshold() + if threshold > 0 { packetBuffer := bytes.NewBuffer(nil) - if int32(len(rawPacketBytes)) >= c.CompressionThreshold { + if int32(len(rawPacketBytes)) >= threshold { err = proto_base.EncodeVarInt(packetBuffer, int32(len(rawPacketBytes))) if err != nil { return @@ -80,7 +95,7 @@ func (c *RawConn) ReceiveRaw() (packetBytes []byte, err error) { return } rawPacketBuffer := bytes.NewBuffer(rawPacketBytes) - if c.CompressionThreshold > 0 { + if c.GetCompressionThreshold() > 0 { var packetLen int32 packetLen, err = proto_base.DecodeVarInt(rawPacketBuffer) if err != nil { diff --git a/server/login_server.go b/server/login_server.go index d1e3f84..e34eff5 100644 --- a/server/login_server.go +++ b/server/login_server.go @@ -90,10 +90,6 @@ func (l *LoginServer) SetCompressionThreshold(threshold int32) (err error) { func (l *LoginServer) OnLoginStart(packet *v1_21_8.LoginToServerPacketLoginStart) (err error) { l.Requested = NameAndUUID{packet.Username, packet.PlayerUUID} - //err = l.SetCompressionThreshold(l.CompressionThreshold) - //if err != nil { - // return - //} if l.Given == nil { l.Given = &l.Requested } @@ -124,7 +120,7 @@ func (l *LoginServer) OnCycle(_ event.Tick) (err error) { return } -func (s *LoginServer) OnClose(_ event.Close) (err error) { +func (l *LoginServer) OnClose(_ event.Close) (err error) { return } diff --git a/server/server.go b/server/server.go index 85ede09..9ab5fe8 100644 --- a/server/server.go +++ b/server/server.go @@ -14,7 +14,7 @@ func Servee(c net.Conn) (ret *proto.Conn) { ret = &proto.Conn{} _ = ret.SetVersion("1.21.8") // cant fail ret.SetState(proto_base.Handshaking) - ret.CompressionThreshold = -1 + ret.SetCompressionThreshold(-1) ret.Actor = proto_base.Servee ret.Conn = c ret.R = c