Method NewSourceTube

fs/contube/memory.go:80–115

(ctx context.Context, configMap ConfigMap)

}

func (f *MemoryQueueFactory) NewSourceTube(ctx context.Context, configMap ConfigMap) (<-chan Record, error) {
	config := SourceQueueConfig{}
	if err := configMap.ToConfigStruct(&config); err != nil {
		return nil, err
	}
	result := make(chan Record)

	var wg sync.WaitGroup
	for _, topic := range config.Topics {
		t := topic
		wg.Add(1)
		go func() {
			<-ctx.Done()
			f.release(t)
		}()
		c := f.getOrCreateChan(t)
		go func() {
			defer wg.Done()
			for {
				select {
				case <-ctx.Done():
					return
				case event := <-c:
					result <- event
				}
			}
		}()
	}

	go func() {
		wg.Wait()
		close(result)
	}()

	return result, nil
}

func (f *MemoryQueueFactory) NewSinkTube(ctx context.Context, configMap ConfigMap) (chan<- Record, error) {
	config := SinkQueueConfig{}