Refactor event system

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
 }
 

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 {