Function TestBrokerCleanup

pkg/streamclient/streambroker_test.go:260–294

(t *testing.T)

}

func TestBrokerCleanup(t *testing.T) {
	broker1, broker2 := setupBrokerPair()

	reader, meta := broker1.CreateStreamReader("reader1", "writer1", 1024)
	writer, err := broker2.CreateStreamWriter(meta)
	if err != nil {
		t.Fatalf("CreateStreamWriter failed: %v", err)
	}

	testData := []byte("cleanup test")
	writer.Write(testData)

	buf := make([]byte, 1024)
	reader.Read(buf)

	writer.Close()

	time.Sleep(100 * time.Millisecond)

	broker1.lock.Lock()
	_, readerExists := broker1.readers[meta.Id]
	broker1.lock.Unlock()

	if readerExists {
		t.Fatal("Reader should have been cleaned up")
	}

	broker2.lock.Lock()
	_, writerExists := broker2.writers[meta.Id]
	broker2.lock.Unlock()

	if writerExists {
		t.Fatal("Writer should have been cleaned up")
	}
}