Method NewSinkTube

fs/contube/memory.go:117–140

(ctx context.Context, configMap ConfigMap)

}

func (f *MemoryQueueFactory) NewSinkTube(ctx context.Context, configMap ConfigMap) (chan<- Record, error) {
	config := SinkQueueConfig{}
	if err := configMap.ToConfigStruct(&config); err != nil {
		return nil, err
	}
	c := f.getOrCreateChan(config.Topic)
	wrapperC := make(chan Record)
	go func() {
		defer f.release(config.Topic)
		for {
			select {
			case <-ctx.Done():
				return
			case event, ok := <-wrapperC:
				if !ok {
					return
				}
				event.Commit()
				c <- event
			}
		}
	}()
	return wrapperC, nil
}