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

types cleanup
All checks were successful
Build and Publish / BuildAndDeployAmd64 (push) Successful in 28s
Details
Build and Publish / BuildAndDeploy (push) Successful in 2m22s
Details

Browse Source
This commit is contained in:
matst80
2024-11-12 18:17:43 +01:00
parent 3f8cdec9af
commit 7e0f070637
11 changed files with 75 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
cart-packet-queue.go
View File

@@ -10,7 +10,7 @@ import (
type CartPacketQueue struct { type CartPacketQueue struct {
mu sync.RWMutex mu sync.RWMutex
expectedPackages map[uint32]*CartListener expectedPackages map[CartMessage]*CartListener
} }
const CurrentPacketVersion = 2 const CurrentPacketVersion = 2
@@ -20,7 +20,7 @@ type CartListener map[CartId]Listener
func NewCartPacketQueue(connection *PersistentConnection) *CartPacketQueue { func NewCartPacketQueue(connection *PersistentConnection) *CartPacketQueue {
queue := &CartPacketQueue{ queue := &CartPacketQueue{
expectedPackages: make(map[uint32]*CartListener), expectedPackages: make(map[CartMessage]*CartListener),
} }
go queue.HandleConnection(connection) go queue.HandleConnection(connection)
return queue return queue
@@ -34,7 +34,7 @@ func (p *CartPacketQueue) RemoveListeners() {
close(l.Chan) close(l.Chan)
} }
} }
p.expectedPackages = make(map[uint32]*CartListener) p.expectedPackages = make(map[CartMessage]*CartListener)
} }
func (p *CartPacketQueue) HandleConnection(connection *PersistentConnection) error { func (p *CartPacketQueue) HandleConnection(connection *PersistentConnection) error {
@@ -73,7 +73,7 @@ func (p *CartPacketQueue) HandleConnection(connection *PersistentConnection) err
} }
} }
func (p *CartPacketQueue) HandleData(t uint32, id CartId, data CallResult) { func (p *CartPacketQueue) HandleData(t CartMessage, id CartId, data CallResult) {
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
pl, ok := p.expectedPackages[t] pl, ok := p.expectedPackages[t]
@@ -90,7 +90,7 @@ func (p *CartPacketQueue) HandleData(t uint32, id CartId, data CallResult) {
} }
} }
func (p *CartPacketQueue) Expect(messageType uint32, id CartId) <-chan CallResult { func (p *CartPacketQueue) Expect(messageType CartMessage, id CartId) <-chan CallResult {
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
l, ok := p.expectedPackages[messageType] l, ok := p.expectedPackages[messageType]

10
packet-queue.go
View File

@@ -9,7 +9,7 @@ import (
type PacketQueue struct { type PacketQueue struct {
mu sync.RWMutex mu sync.RWMutex
expectedPackages map[uint32]*Listener expectedPackages map[PoolMessage]*Listener
} }
type CallResult struct { type CallResult struct {
@@ -24,7 +24,7 @@ type Listener struct {
func NewPacketQueue(connection *PersistentConnection) *PacketQueue { func NewPacketQueue(connection *PersistentConnection) *PacketQueue {
queue := &PacketQueue{ queue := &PacketQueue{
expectedPackages: make(map[uint32]*Listener), expectedPackages: make(map[PoolMessage]*Listener),
} }
go queue.HandleConnection(connection) go queue.HandleConnection(connection)
return queue return queue
@@ -36,7 +36,7 @@ func (p *PacketQueue) RemoveListeners() {
for _, l := range p.expectedPackages { for _, l := range p.expectedPackages {
close(l.Chan) close(l.Chan)
} }
p.expectedPackages = make(map[uint32]*Listener) p.expectedPackages = make(map[PoolMessage]*Listener)
} }
func (p *PacketQueue) HandleConnection(connection *PersistentConnection) error { func (p *PacketQueue) HandleConnection(connection *PersistentConnection) error {
@@ -74,7 +74,7 @@ func (p *PacketQueue) HandleConnection(connection *PersistentConnection) error {
} }
} }
func (p *PacketQueue) HandleData(t uint32, data CallResult) { func (p *PacketQueue) HandleData(t PoolMessage, data CallResult) {
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
l, ok := p.expectedPackages[t] l, ok := p.expectedPackages[t]
@@ -89,7 +89,7 @@ func (p *PacketQueue) HandleData(t uint32, data CallResult) {
} }
} }
func (p *PacketQueue) Expect(messageType uint32) <-chan CallResult { func (p *PacketQueue) Expect(messageType PoolMessage) <-chan CallResult {
p.mu.Lock() p.mu.Lock()
defer p.mu.Unlock() defer p.mu.Unlock()
l, ok := p.expectedPackages[messageType] l, ok := p.expectedPackages[messageType]

21
packet.go
View File

@@ -5,25 +5,28 @@ import (
"io" "io"
) )
type CartMessage uint32
type PackageVersion uint32
const ( const (
RemoteGetState = uint32(0x01) RemoteGetState = CartMessage(0x01)
RemoteHandleMutation = uint32(0x02) RemoteHandleMutation = CartMessage(0x02)
ResponseBody = uint32(0x03) ResponseBody = CartMessage(0x03)
RemoteGetStateReply = uint32(0x04) RemoteGetStateReply = CartMessage(0x04)
RemoteHandleMutationReply = uint32(0x05) RemoteHandleMutationReply = CartMessage(0x05)
) )
type CartPacket struct { type CartPacket struct {
Version uint32 Version PackageVersion
MessageType uint32 MessageType CartMessage
DataLength uint32 DataLength uint32
StatusCode uint32 StatusCode uint32
Id CartId Id CartId
} }
type Packet struct { type Packet struct {
Version uint32 Version PackageVersion
MessageType uint32 MessageType PoolMessage
DataLength uint32 DataLength uint32
StatusCode uint32 StatusCode uint32
} }

4
rpc-server.go
View File

@@ -34,7 +34,7 @@ func (h *GrainHandler) IsHealthy() bool {
return len(h.pool.grains) < h.pool.PoolSize return len(h.pool.grains) < h.pool.PoolSize
} }
func (h *GrainHandler) RemoteHandleMessageHandler(id CartId, data []byte) (uint32, []byte, error) { func (h *GrainHandler) RemoteHandleMessageHandler(id CartId, data []byte) (CartMessage, []byte, error) {
var msg Message var msg Message
err := ReadMessage(bytes.NewReader(data), &msg) err := ReadMessage(bytes.NewReader(data), &msg)
if err != nil { if err != nil {
@@ -53,7 +53,7 @@ func (h *GrainHandler) RemoteHandleMessageHandler(id CartId, data []byte) (uint3
return RemoteHandleMutationReply, replyData, nil return RemoteHandleMutationReply, replyData, nil
} }
func (h *GrainHandler) RemoteGetStateHandler(id CartId, data []byte) (uint32, []byte, error) { func (h *GrainHandler) RemoteGetStateHandler(id CartId, data []byte) (CartMessage, []byte, error) {
reply, err := h.pool.Get(id) reply, err := h.pool.Get(id)
if err != nil { if err != nil {
return RemoteGetStateReply, nil, err return RemoteGetStateReply, nil, err

28
synced-pool.go
View File

@@ -61,11 +61,11 @@ var (
}) })
) )
func (p *SyncedPool) PongHandler(data []byte) (uint32, []byte, error) { func (p *SyncedPool) PongHandler(data []byte) (PoolMessage, []byte, error) {
return Pong, data, nil return Pong, data, nil
} }
func (p *SyncedPool) GetCartIdHandler(data []byte) (uint32, []byte, error) { func (p *SyncedPool) GetCartIdHandler(data []byte) (PoolMessage, []byte, error) {
ids := make([]string, 0, len(p.local.grains)) ids := make([]string, 0, len(p.local.grains))
for id := range p.local.grains { for id := range p.local.grains {
if p.local.grains[id] == nil { if p.local.grains[id] == nil {
@@ -81,7 +81,7 @@ func (p *SyncedPool) GetCartIdHandler(data []byte) (uint32, []byte, error) {
return CartIdsResponse, []byte(strings.Join(ids, ";")), nil return CartIdsResponse, []byte(strings.Join(ids, ";")), nil
} }
func (p *SyncedPool) NegotiateHandler(data []byte) (uint32, []byte, error) { func (p *SyncedPool) NegotiateHandler(data []byte) (PoolMessage, []byte, error) {
negotiationCount.Inc() negotiationCount.Inc()
log.Printf("Handling negotiation\n") log.Printf("Handling negotiation\n")
for _, host := range p.ExcludeKnown(strings.Split(string(data), ";")) { for _, host := range p.ExcludeKnown(strings.Split(string(data), ";")) {
@@ -95,7 +95,7 @@ func (p *SyncedPool) NegotiateHandler(data []byte) (uint32, []byte, error) {
return RemoteNegotiateResponse, []byte("ok"), nil return RemoteNegotiateResponse, []byte("ok"), nil
} }
func (p *SyncedPool) GrainOwnerChangeHandler(data []byte) (uint32, []byte, error) { func (p *SyncedPool) GrainOwnerChangeHandler(data []byte) (PoolMessage, []byte, error) {
grainSyncCount.Inc() grainSyncCount.Inc()
idAndHostParts := strings.Split(string(data), ";") idAndHostParts := strings.Split(string(data), ";")
@@ -276,16 +276,18 @@ func (p *SyncedPool) RemoveHostMappedCarts(host *RemoteHost) {
} }
} }
type PoolMessage uint32
const ( const (
RemoteNegotiate = uint32(3) RemoteNegotiate = PoolMessage(3)
RemoteGrainChanged = uint32(4) RemoteGrainChanged = PoolMessage(4)
AckChange = uint32(5) AckChange = PoolMessage(5)
//AckError = uint32(6) //AckError = PoolMessage(6)
Ping = uint32(7) Ping = PoolMessage(7)
Pong = uint32(8) Pong = PoolMessage(8)
GetCartIds = uint32(9) GetCartIds = PoolMessage(9)
CartIdsResponse = uint32(10) CartIdsResponse = PoolMessage(10)
RemoteNegotiateResponse = uint32(11) RemoteNegotiateResponse = PoolMessage(11)
) )
func (p *SyncedPool) Negotiate() { func (p *SyncedPool) Negotiate() {

4
tcp-cart-client.go
View File

@@ -47,7 +47,7 @@ func NewCartTCPClient(address string) (*CartTCPClient, error) {
}, nil }, nil
} }
func (m *CartTCPClient) SendPacket(messageType uint32, id CartId, data []byte) error { func (m *CartTCPClient) SendPacket(messageType CartMessage, id CartId, data []byte) error {
err := binary.Write(m.Conn, binary.LittleEndian, CartPacket{ err := binary.Write(m.Conn, binary.LittleEndian, CartPacket{
Version: CurrentPacketVersion, Version: CurrentPacketVersion,
@@ -62,7 +62,7 @@ func (m *CartTCPClient) SendPacket(messageType uint32, id CartId, data []byte) e
return m.HandleConnectionError(err) return m.HandleConnectionError(err)
} }
func (m *CartTCPClient) Call(messageType uint32, id CartId, responseType uint32, data []byte) (*CallResult, error) { func (m *CartTCPClient) Call(messageType CartMessage, id CartId, responseType CartMessage, data []byte) (*CallResult, error) {
packetChan := m.Expect(responseType, id) packetChan := m.Expect(responseType, id)
err := m.SendPacket(messageType, id, data) err := m.SendPacket(messageType, id, data)
if err != nil { if err != nil {

18
tcp-cart-mux-server.go
View File

@@ -37,21 +37,21 @@ func CartListen(address string) (*CartServer, error) {
type TCPCartServerMux struct { type TCPCartServerMux struct {
mu sync.RWMutex mu sync.RWMutex
listeners map[uint32]func(CartId, []byte) error listeners map[CartMessage]func(CartId, []byte) error
functions map[uint32]func(CartId, []byte) (uint32, []byte, error) functions map[CartMessage]func(CartId, []byte) (CartMessage, []byte, error)
} }
func NewCartTCPServerMux() *TCPCartServerMux { func NewCartTCPServerMux() *TCPCartServerMux {
m := &TCPCartServerMux{ m := &TCPCartServerMux{
mu: sync.RWMutex{}, mu: sync.RWMutex{},
listeners: make(map[uint32]func(CartId, []byte) error), listeners: make(map[CartMessage]func(CartId, []byte) error),
functions: make(map[uint32]func(CartId, []byte) (uint32, []byte, error)), functions: make(map[CartMessage]func(CartId, []byte) (CartMessage, []byte, error)),
} }
return m return m
} }
func (m *TCPCartServerMux) handleListener(messageType uint32, id CartId, data []byte) (bool, error) { func (m *TCPCartServerMux) handleListener(messageType CartMessage, id CartId, data []byte) (bool, error) {
m.mu.RLock() m.mu.RLock()
handler, ok := m.listeners[messageType] handler, ok := m.listeners[messageType]
m.mu.RUnlock() m.mu.RUnlock()
@@ -64,7 +64,7 @@ func (m *TCPCartServerMux) handleListener(messageType uint32, id CartId, data []
return false, nil return false, nil
} }
func (m *TCPCartServerMux) handleFunction(connection net.Conn, messageType uint32, id CartId, data []byte) (bool, error) { func (m *TCPCartServerMux) handleFunction(connection net.Conn, messageType CartMessage, id CartId, data []byte) (bool, error) {
m.mu.RLock() m.mu.RLock()
fn, ok := m.functions[messageType] fn, ok := m.functions[messageType]
m.mu.RUnlock() m.mu.RUnlock()
@@ -126,7 +126,7 @@ func (m *TCPCartServerMux) HandleConnection(connection net.Conn) error {
} }
} }
func (m *TCPCartServerMux) HandleData(connection net.Conn, t uint32, id CartId, data []byte) { func (m *TCPCartServerMux) HandleData(connection net.Conn, t CartMessage, id CartId, data []byte) {
status, err := m.handleListener(t, id, data) status, err := m.handleListener(t, id, data)
if err != nil { if err != nil {
log.Printf("Error handling listener: %v\n", err) log.Printf("Error handling listener: %v\n", err)
@@ -142,13 +142,13 @@ func (m *TCPCartServerMux) HandleData(connection net.Conn, t uint32, id CartId,
} }
} }
func (m *TCPCartServerMux) ListenFor(messageType uint32, handler func(CartId, []byte) error) { func (m *TCPCartServerMux) ListenFor(messageType CartMessage, handler func(CartId, []byte) error) {
m.mu.Lock() m.mu.Lock()
m.listeners[messageType] = handler m.listeners[messageType] = handler
m.mu.Unlock() m.mu.Unlock()
} }
func (m *TCPCartServerMux) HandleCall(messageType uint32, handler func(CartId, []byte) (uint32, []byte, error)) { func (m *TCPCartServerMux) HandleCall(messageType CartMessage, handler func(CartId, []byte) (CartMessage, []byte, error)) {
m.mu.Lock() m.mu.Lock()
m.functions[messageType] = handler m.functions[messageType] = handler
m.mu.Unlock() m.mu.Unlock()

21
tcp-cart_test.go
View File

@@ -22,17 +22,17 @@ func TestCartTcpHelpers(t *testing.T) {
messageData = string(data) messageData = string(data)
return nil return nil
}) })
server.HandleCall(666, func(id CartId, data []byte) (uint32, []byte, error) { server.HandleCall(666, func(id CartId, data []byte) (CartMessage, []byte, error) {
log.Printf("Received call: %s\n", string(data)) log.Printf("Received call: %s\n", string(data))
return 3, []byte("Hello, client!"), fmt.Errorf("Det blev fel") return 3, []byte("Hello, client!"), fmt.Errorf("Det blev fel")
}) })
server.HandleCall(2, func(id CartId, data []byte) (uint32, []byte, error) { server.HandleCall(2, func(id CartId, data []byte) (CartMessage, []byte, error) {
log.Printf("Received call: %s\n", string(data)) log.Printf("Received call: %s\n", string(data))
return 3, []byte("Hello, client!"), nil return 3, []byte("Hello, client!"), nil
}) })
server.HandleCall(Ping, func(id CartId, data []byte) (uint32, []byte, error) { // server.HandleCall(Ping, func(id CartId, data []byte) (CartMessage, []byte, error) {
return Pong, nil, nil // return Pong, nil, nil
}) // })
id := ToCartId("kalle") id := ToCartId("kalle")
client.SendPacket(1, id, []byte("Hello, world!")) client.SendPacket(1, id, []byte("Hello, world!"))
answer, err := client.Call(2, id, 3, []byte("Hello, server!")) answer, err := client.Call(2, id, 3, []byte("Hello, server!"))
@@ -46,16 +46,7 @@ func TestCartTcpHelpers(t *testing.T) {
if s.StatusCode != 500 { if s.StatusCode != 500 {
t.Errorf("Expected 500, got %d\n", s.StatusCode) t.Errorf("Expected 500, got %d\n", s.StatusCode)
} }
for i := 0; i < 100; i++ {
_, err = client.Call(Ping, id, Pong, nil)
if err != nil {
t.Errorf("Error calling: %v\n", err)
}
}
_, err = client.Call(Ping, id, Pong, nil)
if err != nil {
t.Errorf("Error calling: %v\n", err)
}
if string(answer.Data) != "Hello, client!" { if string(answer.Data) != "Hello, client!" {
t.Errorf("Expected answer 'Hello, client!', got %s\n", string(answer.Data)) t.Errorf("Expected answer 'Hello, client!', got %s\n", string(answer.Data))
} }

4
tcp-client.go
View File

@@ -92,7 +92,7 @@ func NewTCPClient(address string) (*TCPClient, error) {
}, nil }, nil
} }
func (m *TCPClient) SendPacket(messageType uint32, data []byte) error { func (m *TCPClient) SendPacket(messageType PoolMessage, data []byte) error {
err := binary.Write(m.Conn, binary.LittleEndian, Packet{ err := binary.Write(m.Conn, binary.LittleEndian, Packet{
Version: CurrentPacketVersion, Version: CurrentPacketVersion,
@@ -107,7 +107,7 @@ func (m *TCPClient) SendPacket(messageType uint32, data []byte) error {
return m.HandleConnectionError(err) return m.HandleConnectionError(err)
} }
func (m *TCPClient) Call(messageType uint32, responseType uint32, data []byte) (*CallResult, error) { func (m *TCPClient) Call(messageType PoolMessage, responseType PoolMessage, data []byte) (*CallResult, error) {
packetChan := m.Expect(responseType) packetChan := m.Expect(responseType)
err := m.SendPacket(messageType, data) err := m.SendPacket(messageType, data)
if err != nil { if err != nil {

18
tcp-mux-server.go
View File

@@ -37,21 +37,21 @@ func Listen(address string) (*Server, error) {
type TCPServerMux struct { type TCPServerMux struct {
mu sync.RWMutex mu sync.RWMutex
listeners map[uint32]func(data []byte) error listeners map[PoolMessage]func(data []byte) error
functions map[uint32]func(data []byte) (uint32, []byte, error) functions map[PoolMessage]func(data []byte) (PoolMessage, []byte, error)
} }
func NewTCPServerMux() *TCPServerMux { func NewTCPServerMux() *TCPServerMux {
m := &TCPServerMux{ m := &TCPServerMux{
mu: sync.RWMutex{}, mu: sync.RWMutex{},
listeners: make(map[uint32]func(data []byte) error), listeners: make(map[PoolMessage]func(data []byte) error),
functions: make(map[uint32]func(data []byte) (uint32, []byte, error)), functions: make(map[PoolMessage]func(data []byte) (PoolMessage, []byte, error)),
} }
return m return m
} }
func (m *TCPServerMux) handleListener(messageType uint32, data []byte) (bool, error) { func (m *TCPServerMux) handleListener(messageType PoolMessage, data []byte) (bool, error) {
m.mu.RLock() m.mu.RLock()
handler, ok := m.listeners[messageType] handler, ok := m.listeners[messageType]
m.mu.RUnlock() m.mu.RUnlock()
@@ -64,7 +64,7 @@ func (m *TCPServerMux) handleListener(messageType uint32, data []byte) (bool, er
return false, nil return false, nil
} }
func (m *TCPServerMux) handleFunction(connection net.Conn, messageType uint32, data []byte) (bool, error) { func (m *TCPServerMux) handleFunction(connection net.Conn, messageType PoolMessage, data []byte) (bool, error) {
m.mu.RLock() m.mu.RLock()
function, ok := m.functions[messageType] function, ok := m.functions[messageType]
m.mu.RUnlock() m.mu.RUnlock()
@@ -124,7 +124,7 @@ func (m *TCPServerMux) HandleConnection(connection net.Conn) error {
} }
} }
func (m *TCPServerMux) HandleData(connection net.Conn, t uint32, data []byte) { func (m *TCPServerMux) HandleData(connection net.Conn, t PoolMessage, data []byte) {
// listener := m.listeners[t] // listener := m.listeners[t]
// handler := m.functions[t] // handler := m.functions[t]
status, err := m.handleListener(t, data) status, err := m.handleListener(t, data)
@@ -142,13 +142,13 @@ func (m *TCPServerMux) HandleData(connection net.Conn, t uint32, data []byte) {
} }
} }
func (m *TCPServerMux) ListenFor(messageType uint32, handler func(data []byte) error) { func (m *TCPServerMux) ListenFor(messageType PoolMessage, handler func(data []byte) error) {
m.mu.Lock() m.mu.Lock()
m.listeners[messageType] = handler m.listeners[messageType] = handler
m.mu.Unlock() m.mu.Unlock()
} }
func (m *TCPServerMux) HandleCall(messageType uint32, handler func(data []byte) (uint32, []byte, error)) { func (m *TCPServerMux) HandleCall(messageType PoolMessage, handler func(data []byte) (PoolMessage, []byte, error)) {
m.mu.Lock() m.mu.Lock()
m.functions[messageType] = handler m.functions[messageType] = handler
m.mu.Unlock() m.mu.Unlock()

10
tcp_test.go
View File

@@ -21,11 +21,11 @@ func TestTcpHelpers(t *testing.T) {
messageData = string(data) messageData = string(data)
return nil return nil
}) })
server.HandleCall(2, func(data []byte) (uint32, []byte, error) { server.HandleCall(2, func(data []byte) (PoolMessage, []byte, error) {
log.Printf("Received call: %s\n", string(data)) log.Printf("Received call: %s\n", string(data))
return 3, []byte("Hello, client!"), nil return 3, []byte("Hello, client!"), nil
}) })
server.HandleCall(Ping, func(data []byte) (uint32, []byte, error) { server.HandleCall(Ping, func(data []byte) (PoolMessage, []byte, error) {
return Pong, nil, nil return Pong, nil, nil
}) })
@@ -34,6 +34,12 @@ func TestTcpHelpers(t *testing.T) {
if err != nil { if err != nil {
t.Errorf("Error calling: %v\n", err) t.Errorf("Error calling: %v\n", err)
} }
for i := 0; i < 100; i++ {
_, err = client.Call(Ping, Pong, nil)
if err != nil {
t.Errorf("Error calling: %v\n", err)
}
}
_, err = client.Call(Ping, Pong, nil) _, err = client.Call(Ping, Pong, nil)
if err != nil { if err != nil {
t.Errorf("Error calling: %v\n", err) t.Errorf("Error calling: %v\n", err)