Function acquireSource

pkg/acquisition/acquisition.go:581–612

(
	ctx context.Context,
	source types.DataSource,
	name string,
	output chan pipeline.Event,
	acquisTomb *tomb.Tomb,
)

}

func acquireSource(
	ctx context.Context,
	source types.DataSource,
	name string,
	output chan pipeline.Event,
	acquisTomb *tomb.Tomb,
) error {
	if source.GetMode() == configuration.CAT_MODE {
		if s, ok := source.(types.BatchFetcher); ok {
			// s.Logger.Info("Start OneShot")
			return runBatchFetcher(ctx, s, output, acquisTomb)
		}

		if s, ok := source.(types.Fetcher); ok {
			// s.Logger.Info("Start OneShotAcquisition")
			return s.OneShotAcquisition(ctx, output, acquisTomb)
		}

		return fmt.Errorf("%s: cat mode is set but OneShotAcquisition is not supported", source.GetName())
	}

	if s, ok := source.(types.Tailer); ok {
		// s.Logger.Info("Streaming Acquisition")
		return s.StreamingAcquisition(ctx, output, acquisTomb)
	}

	if s, ok := source.(types.RestartableStreamer); ok {
		return runRestartableStream(ctx, s, name, output, acquisTomb)
	}

	return fmt.Errorf("%s: tail mode is set but the datasource does not support streaming acquisition", source.GetName())
}

func StartAcquisition(
	ctx context.Context,