limit queue

packet-queue.go

@@ -47,7 +47,7 @@ func NewPacketQueue(connection net.Conn) *PacketQueue {
 			l := make([]PacketWithData, 0, len(queue.Packets))
 
 			for _, packet := range queue.Packets {
-				if !packet.Consumed {
+				if !packet.Consumed && packet.Added.After(ts.Add(-time.Second)) {
 					l = append(l, packet)
 				}
 			}

packet-queue_test.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"testing"
+	"time"
+)
+
+func TestQueue(t *testing.T) {
+	localPool := NewGrainLocalPool(100, time.Minute, func(id CartId) (*CartGrain, error) {
+		return &CartGrain{
+			Id:              id,
+			storageMessages: []Message{},
+			Items:           []CartItem{},
+			TotalPrice:      0,
+		}, nil
+	})
+	pool, err := NewSyncedPool(localPool, "localhost", nil)
+	if err != nil {
+		t.Errorf("Error creating pool: %v", err)
+	}
+
+	err = pool.AddRemote("localhost")
+	if err != nil {
+		t.Errorf("Error adding remote: %v", err)
+	}
+	r := pool.remotes[0]
+
+	if len(r.PacketQueue.Packets) != 1 {
+		t.Errorf("Expected 1 packet, got %d", len(r.PacketQueue.Packets))
+	}
+
+}

synced-pool.go

@@ -422,6 +422,10 @@ func (p *SyncedPool) addRemoteHost(address string, remote *RemoteHost) error {
 		ids := remote.GetCartMappings()
 		p.mu.Lock()
 		for _, id := range ids {
+			if p.local.grains[id] != nil {
+				log.Printf("Grain %s already exists locally, deleting\n", id)
+				delete(p.local.grains, id)
+			}
 			p.remoteIndex[id] = remote.Pool
 		}
 		p.mu.Unlock()