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

cleanup
All checks were successful
Build and Publish / BuildAndDeploy (push) Successful in 1m54s
Details

Browse Source
This commit is contained in:
matst80
2024-11-10 11:13:14 +01:00
parent e9e8a1518f
commit 05a7dc50cf
2 changed files with 19 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
rpc-pool.go
View File

@@ -26,8 +26,8 @@ func ToCartId(id string) CartId {
} }
type RemoteGrain struct { type RemoteGrain struct {
connection net.Conn net.Conn
queue *PacketQueue *PacketQueue
Id CartId Id CartId
Address string Address string
} }
@@ -49,8 +49,8 @@ func (g *RemoteGrain) Connect() error {
if err != nil { if err != nil {
return err return err
} }
g.connection = client g.Conn = client
g.queue = NewPacketQueue(client) g.PacketQueue = NewPacketQueue(client)
} }
return nil return nil
} }
@@ -64,7 +64,7 @@ func (g *RemoteGrain) HandleMessage(message *Message, isReplay bool) ([]byte, er
if err != nil { if err != nil {
return nil, err return nil, err
} }
packet, err := g.queue.Expect(ResponseBody, time.Second) packet, err := g.Expect(ResponseBody, time.Second)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -86,7 +86,7 @@ func (g *RemoteGrain) GetCurrentState() ([]byte, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
packet, err := g.queue.Expect(ResponseBody, time.Second) packet, err := g.Expect(ResponseBody, time.Second)
if err != nil { if err != nil {
return nil, err return nil, err
} }

18
synced-pool.go
View File

@@ -20,11 +20,11 @@ type Quorum interface {
} }
type RemoteHost struct { type RemoteHost struct {
net.Conn
*PacketQueue
Host string Host string
MissedPings int MissedPings int
Pool *RemoteGrainPool Pool *RemoteGrainPool
connection net.Conn
queue *PacketQueue
} }
type SyncedPool struct { type SyncedPool struct {
@@ -239,7 +239,7 @@ func NewPacketQueue(connection net.Conn) *PacketQueue {
queue.mu.Lock() queue.mu.Lock()
for i, packet := range queue.Packets { for i, packet := range queue.Packets {
if time.Since(packet.Added) < time.Second*5 { if time.Since(packet.Added) < time.Second {
stillInQueue := queue.Packets[i:] stillInQueue := queue.Packets[i:]
log.Printf("DEBUG: Requeueing %v packets\n", stillInQueue) log.Printf("DEBUG: Requeueing %v packets\n", stillInQueue)
queue.Packets = stillInQueue queue.Packets = stillInQueue
@@ -378,7 +378,7 @@ func (h *RemoteHost) Negotiate(knownHosts []string) ([]string, error) {
if err != nil { if err != nil {
return nil, err return nil, err
} }
packet, err := h.queue.Expect(RemoteNegotiate, time.Second) packet, err := h.Expect(RemoteNegotiate, time.Second)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -395,7 +395,7 @@ func (g *RemoteHost) GetCartMappings() []CartId {
log.Printf("Error getting mappings: %v\n", err) log.Printf("Error getting mappings: %v\n", err)
return nil return nil
} }
packet, err := g.queue.Expect(CartIdsResponse, time.Second*3) packet, err := g.Expect(CartIdsResponse, time.Second*3)
if err != nil { if err != nil {
log.Printf("Error getting mappings: %v\n", err) log.Printf("Error getting mappings: %v\n", err)
return nil return nil
@@ -434,7 +434,7 @@ func (r *RemoteHost) ConfirmChange(id CartId, host string) error {
if err != nil { if err != nil {
return err return err
} }
_, err = r.queue.Expect(AckChange, time.Second) _, err = r.Expect(AckChange, time.Second)
if err != nil { if err != nil {
return err return err
@@ -459,7 +459,7 @@ func DoPing(host *RemoteHost) error {
SendPacket(host.connection, Ping, func(w io.Writer) error { SendPacket(host.connection, Ping, func(w io.Writer) error {
return nil return nil
}) })
_, err := host.queue.Expect(Pong, time.Second) _, err := host.Expect(Pong, time.Second)
if err != nil { if err != nil {
return err return err
@@ -511,8 +511,8 @@ func (p *SyncedPool) AddRemote(address string) error {
pool := NewRemoteGrainPool(address) pool := NewRemoteGrainPool(address)
remote := RemoteHost{ remote := RemoteHost{
connection: connection, Conn: connection,
queue: NewPacketQueue(connection), PacketQueue: NewPacketQueue(connection),
Pool: pool, Pool: pool,
Host: address, Host: address,
} }