Function TestMemoryTube

fs/contube/memory_test.go:28–87

(t *testing.T)

)

func TestMemoryTube(t *testing.T) {

	ctx, cancel := context.WithCancel(context.Background())
	tubeFactory := NewMemoryQueueFactory(ctx)
	memoryQueueFactory := tubeFactory.(*MemoryQueueFactory)

	var wg sync.WaitGroup
	var events []Record

	topics := []string{"topic1", "topic2", "topic3"}
	source, err := memoryQueueFactory.NewSourceTube(ctx, (&SourceQueueConfig{Topics: topics,
		SubName: "consume-" + strconv.Itoa(rand.Int())}).ToConfigMap())
	if err != nil {
		t.Fatal(err)
	}

	for i, v := range topics {
		wg.Add(1)
		sink, err := memoryQueueFactory.NewSinkTube(ctx, (&SinkQueueConfig{Topic: v}).ToConfigMap())
		if err != nil {
			t.Fatal(err)
		}
		go func(i int) {
			defer wg.Done()
			defer close(sink)
			sink <- NewRecordImpl([]byte{byte(i + 1)}, func() {})
		}(i)
	}

	wg.Add(1)
	go func() {
		defer wg.Done()
		for {
			select {
			case event := <-source:
				events = append(events, event)
				if len(events) == len(topics) {
					return
				}
			default:
				continue
			}
		}
	}()

	wg.Wait()
	cancel()

	// Give enough time to ensure that the goroutine execution within NewSource Tube and NewSinkTube is complete and
	// the released queue is successful.
	time.Sleep(100 * time.Millisecond)

	// assert the memoryQueueFactory.queues is empty.
	memoryQueueFactory.mu.Lock()
	if len(memoryQueueFactory.queues) != 0 {
		t.Fatal("MemoryQueueFactory.queues is not empty")
	}
	memoryQueueFactory.mu.Unlock()