A self-hosted orchestration system crafted for power users. composia.xyz
Something went wrong. Try again.
Go
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475package agent
import ( "context" "sync" "testing" "time"
agentv1 "forgejo.alexma.top/alexma233/composia/gen/go/proto/composia/agent/v1")
type recordingExecTunnelStream struct { mu sync.Mutex messages []*agentv1.OpenExecTunnelRequest inSend int maxConcurrent int}
func (stream *recordingExecTunnelStream) Send(message *agentv1.OpenExecTunnelRequest) error { stream.mu.Lock() stream.inSend++ if stream.inSend > stream.maxConcurrent { stream.maxConcurrent = stream.inSend } stream.mu.Unlock()
time.Sleep(time.Millisecond)
stream.mu.Lock() stream.messages = append(stream.messages, message) stream.inSend-- stream.mu.Unlock() return nil}
func TestExecTunnelSenderSerializesConcurrentSends(t *testing.T) { t.Parallel()
stream := &recordingExecTunnelStream{} sender := newExecTunnelSender(context.Background(), stream) var wg sync.WaitGroup for range 25 { wg.Add(1) go func() { defer wg.Done() if err := sender.send(&agentv1.OpenExecTunnelRequest{Kind: execKindOutput}); err != nil { t.Errorf("send exec tunnel message: %v", err) } }() } wg.Wait() deadline := time.Now().Add(time.Second) for { stream.mu.Lock() messageCount := len(stream.messages) stream.mu.Unlock() if messageCount == 25 || time.Now().After(deadline) { break } time.Sleep(time.Millisecond) } if err := sender.close(); err != nil { t.Fatalf("close sender: %v", err) }
stream.mu.Lock() defer stream.mu.Unlock() if stream.maxConcurrent != 1 { t.Fatalf("expected serialized sends, max concurrency was %d", stream.maxConcurrent) } if len(stream.messages) != 25 { t.Fatalf("expected 25 messages, got %d", len(stream.messages)) }}