Mutex

proto/connection.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"errors"
 	"io"
+	"sync"
 
 	"github.com/admin-else/strom/data"
 	"github.com/admin-else/strom/event"
@@ -34,12 +35,25 @@ func (u *UnCodablePacket) Decode(r io.Reader) (err error) {
 type Conn struct {
 	RawConn
 	*event.Loop
-	State           proto_base.State
+	state           proto_base.State
+	stateMutex      sync.RWMutex
 	Actor           proto_base.Actor
 	Version         string
 	ProtocolVersion int32
 }
 
+func (c *Conn) SetState(state proto_base.State) {
+	c.stateMutex.Lock()
+	defer c.stateMutex.Unlock()
+	c.state = state
+}
+
+func (c *Conn) State() proto_base.State {
+	c.stateMutex.RLock()
+	defer c.stateMutex.RUnlock()
+	return c.state
+}
+
 func (c *Conn) SetVersion(version string) (err error) {
 	c.Version = version
 	versionData, err := data.LookUpProtocolVersionByName(c.Version)
@@ -55,7 +69,7 @@ func (c *Conn) Send(packet proto_base.EncodeDecodeAble) (err error) {
 	case *UnCodablePacket:
 		err = c.SendRaw(packet.Data)
 	default:
-		i, ok := LookupPacketInfoByTypeLookupPacketInfoByTypeAndMore(packet, c.State)
+		i, ok := LookupPacketInfoByTypeLookupPacketInfoByTypeAndMore(packet, c.State())
 		if !ok {
 			err = BadPacketTypeError
 		}
@@ -83,7 +97,7 @@ func (c *Conn) Receive() (packet proto_base.EncodeDecodeAble, err error) {
 	if err != nil {
 		return
 	}
-	packet, ok := LookUpTypeByPacketInfoAndCopyType(c.Actor.ReceiveDirection(), c.State, id, c.ProtocolVersion)
+	packet, ok := LookUpTypeByPacketInfoAndCopyType(c.Actor.ReceiveDirection(), c.State(), id, c.ProtocolVersion)
 	if !ok {
 		packet = &UnCodablePacket{Err: BadPacketIdError, Data: packetBytes, Direction: c.Actor.ReceiveDirection()}
 		return
@@ -110,10 +124,6 @@ func (c *Conn) OnTick(_ event.Tick) (err error) {
 	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)

proto/connection_test.go

@@ -25,7 +25,7 @@ func NewStateFuzzer(state proto_base.State, actor proto_base.Actor) func(t *test
 		b1 := bytes.NewBuffer(inbytes)
 		b2 := bytes.NewBuffer(nil)
 		conn := server.Servee(nil)
-		conn.State = state
+		conn.SetState(state)
 		conn.Actor = actor
 		conn.R = b1
 		conn.W = b2

proto/raw_conn.go

@@ -21,15 +21,21 @@ type RawConn struct {
 }
 
 func (c *RawConn) SendRaw(rawPacketBytes []byte) (err error) {
+	//fmt.Println("send:", rawPacketBytes)
 	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 int32(len(rawPacketBytes)) >= c.CompressionThreshold {
+			err = proto_base.EncodeVarInt(packetBuffer, int32(len(rawPacketBytes)))
 			if err != nil {
 				return
 			}
-			_, err = zlib.NewWriter(packetBuffer).Write(rawPacketBytes)
+			zW := zlib.NewWriter(packetBuffer)
+			_, err = zW.Write(rawPacketBytes)
+			if err != nil {
+				return
+			}
+			err = zW.Close()
 			if err != nil {
 				return
 			}
@@ -47,11 +53,16 @@ func (c *RawConn) SendRaw(rawPacketBytes []byte) (err error) {
 	} else {
 		packetBytes = rawPacketBytes
 	}
-	err = proto_base.EncodeVarInt(c.W, int32(len(packetBytes)))
+	packetWithLenBuffer := bytes.NewBuffer(nil)
+	err = proto_base.EncodeVarInt(packetWithLenBuffer, int32(len(packetBytes)))
 	if err != nil {
 		return
 	}
-	_, err = c.W.Write(packetBytes)
+	_, err = packetWithLenBuffer.Write(packetBytes)
+	if err != nil {
+		return
+	}
+	_, err = c.W.Write(packetWithLenBuffer.Bytes())
 	return
 }
 
@@ -86,15 +97,16 @@ func (c *RawConn) ReceiveRaw() (packetBytes []byte, err error) {
 			if err != nil {
 				return
 			}
-			defer zReader.Close()
 			packetBytes, err = io.ReadAll(zReader)
 			if err != nil {
 				return
 			}
+			err = zReader.Close()
 		}
 	} else {
 		packetBytes = rawPacketBytes
 	}
+	//fmt.Println("recv:", packetBytes)
 	return
 }