mats/go-cart-actor
1
0
Fork 0
Code Issues Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
matst80
2024-11-11 19:47:44 +01:00
parent e63a92213b
commit 3eaa42b615
4 changed files with 33 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
cart-packet-queue.go
View File

@@ -23,10 +23,12 @@ type CartPacketQueue struct {
//connection net.Conn //connection net.Conn
} }
const cartCap = 150
func NewCartPacketQueue(connection net.Conn) *CartPacketQueue { func NewCartPacketQueue(connection net.Conn) *CartPacketQueue {
queue := &CartPacketQueue{ queue := &CartPacketQueue{
Packets: make([]CartPacketWithData, 0), Packets: make([]CartPacketWithData, 0, cartCap),
//connection: connection, //connection: connection,
} }
go func() { go func() {
@@ -55,14 +57,13 @@ func NewCartPacketQueue(connection net.Conn) *CartPacketQueue {
func (p *CartPacketQueue) HandleData(t uint32, id CartId, data []byte) { func (p *CartPacketQueue) HandleData(t uint32, id CartId, data []byte) {
ts := time.Now() ts := time.Now()
cap := 150 l := make([]CartPacketWithData, 0, cartCap)
l := make([]CartPacketWithData, 0, cap)
p.mu.RLock() p.mu.RLock()
breakAt := ts.Add(-time.Millisecond * 250) breakAt := ts.Add(-time.Millisecond * 250)
for _, packet := range p.Packets { for _, packet := range p.Packets {
if !packet.Consumed && packet.Added.After(breakAt) { if !packet.Consumed && packet.Added.After(breakAt) {
l = append(l, packet) l = append(l, packet)
if len(l) >= cap { if len(l) >= cartCap {
break break
} }
} }
@@ -89,7 +90,7 @@ func (p *CartPacketQueue) Expect(messageType uint32, id CartId, timeToWait time.
} }
p.mu.RLock() p.mu.RLock()
for _, packet := range p.Packets { for _, packet := range p.Packets {
if packet.MessageType == messageType && packet.Id == id && packet.Added.After(start) { if !packet.Consumed && packet.MessageType == messageType && packet.Id == id && packet.Added.After(start) {
packet.Consumed = true packet.Consumed = true
p.mu.RUnlock() p.mu.RUnlock()
return &packet, nil return &packet, nil

12
packet-queue.go
View File

@@ -22,10 +22,12 @@ type PacketQueue struct {
//connection net.Conn //connection net.Conn
} }
const cap = 150
func NewPacketQueue(connection net.Conn) *PacketQueue { func NewPacketQueue(connection net.Conn) *PacketQueue {
queue := &PacketQueue{ queue := &PacketQueue{
Packets: make([]PacketWithData, 0), Packets: make([]PacketWithData, 0, cap),
//connection: connection, //connection: connection,
} }
go func() { go func() {
@@ -53,7 +55,6 @@ func NewPacketQueue(connection net.Conn) *PacketQueue {
func (p *PacketQueue) HandleData(t uint32, data []byte) { func (p *PacketQueue) HandleData(t uint32, data []byte) {
ts := time.Now() ts := time.Now()
cap := 150
l := make([]PacketWithData, 0, cap) l := make([]PacketWithData, 0, cap)
p.mu.RLock() p.mu.RLock()
breakAt := ts.Add(-time.Millisecond * 250) breakAt := ts.Add(-time.Millisecond * 250)
@@ -85,14 +86,13 @@ func (p *PacketQueue) Expect(messageType uint32, timeToWait time.Duration) (*Pac
return nil, fmt.Errorf("timeout waiting for message type %d", messageType) return nil, fmt.Errorf("timeout waiting for message type %d", messageType)
} }
p.mu.RLock() p.mu.RLock()
defer p.mu.RUnlock()
for _, packet := range p.Packets { for _, packet := range p.Packets {
if packet.MessageType == messageType && packet.Added.After(start) { if !packet.Consumed && packet.MessageType == messageType && packet.Added.After(start) {
packet.Consumed = true packet.Consumed = true
p.mu.RUnlock()
return &packet, nil return &packet, nil
} }
} }
p.mu.RUnlock() time.Sleep(time.Millisecond * 4)
time.Sleep(time.Millisecond * 2)
} }
} }

24
packet.go
View File

@@ -43,16 +43,16 @@ func GetPacketData(conn io.Reader, len uint64) ([]byte, error) {
return data, err return data, err
} }
func ReceivePacket(conn io.Reader) (uint32, []byte, error) { // func ReceivePacket(conn io.Reader) (uint32, []byte, error) {
var packet Packet // var packet Packet
err := ReadPacket(conn, &packet) // err := ReadPacket(conn, &packet)
if err != nil { // if err != nil {
return 0, nil, err // return 0, nil, err
} // }
data, err := GetPacketData(conn, packet.DataLength) // data, err := GetPacketData(conn, packet.DataLength)
if err != nil { // if err != nil {
return 0, nil, err // return 0, nil, err
} // }
return packet.MessageType, data, nil // return packet.MessageType, data, nil
} // }

14
tcp-mux-server.go
View File

@@ -90,8 +90,9 @@ func (m *TCPServerMux) handleFunction(connection net.Conn, messageType uint32, d
func (m *TCPServerMux) HandleConnection(connection net.Conn) error { func (m *TCPServerMux) HandleConnection(connection net.Conn) error {
defer connection.Close() defer connection.Close()
var packet Packet
for { for {
messageType, data, err := ReceivePacket(connection) err := ReadPacket(connection, &packet)
if err != nil { if err != nil {
if err == io.EOF { if err == io.EOF {
return nil return nil
@@ -99,18 +100,21 @@ func (m *TCPServerMux) HandleConnection(connection net.Conn) error {
log.Printf("Error receiving packet: %v\n", err) log.Printf("Error receiving packet: %v\n", err)
return err return err
} }
data, err := GetPacketData(connection, packet.DataLength)
status, err := m.handleListener(messageType, data) if err != nil {
log.Printf("Error receiving packet data: %v\n", err)
}
status, err := m.handleListener(packet.MessageType, data)
if err != nil { if err != nil {
log.Printf("Error handling listener: %v\n", err) log.Printf("Error handling listener: %v\n", err)
} }
if !status { if !status {
status, err = m.handleFunction(connection, messageType, data) status, err = m.handleFunction(connection, packet.MessageType, data)
if err != nil { if err != nil {
log.Printf("Error handling function: %v\n", err) log.Printf("Error handling function: %v\n", err)
} }
if !status { if !status {
log.Printf("Unknown message type: %d\n", messageType) log.Printf("Unknown message type: %d\n", packet.MessageType)
} }
} }
} }