ReadCapture reconstructs the event stream from the capture file.
(ctx context.Context, args []string)
| 307 | |
| 308 | // ReadCapture reconstructs the event stream from the capture file. |
| 309 | func (f *App) ReadCapture(ctx context.Context, args []string) error { |
| 310 | if f.reader == nil { |
| 311 | panic("reader is nil") |
| 312 | } |
| 313 | fltr, err := filter.NewFromCLIWithAllAccessors(args) |
| 314 | if err != nil { |
| 315 | return err |
| 316 | } |
| 317 | f.hsnap, f.psnap, err = f.reader.RecoverSnapshotters() |
| 318 | if err != nil { |
| 319 | return err |
| 320 | } |
| 321 | |
| 322 | if f.config.IsFilamentSet() { |
| 323 | f.filament, err = filament.New(f.config.Filament.Name, f.psnap, f.hsnap, f.config) |
| 324 | if err != nil { |
| 325 | return err |
| 326 | } |
| 327 | if f.filament.Filter() != nil { |
| 328 | // filament filter overrides CLI filter |
| 329 | f.reader.SetFilter(f.filament.Filter()) |
| 330 | } else if fltr != nil { |
| 331 | f.reader.SetFilter(fltr) |
| 332 | } |
| 333 | // returns the channel where events are read from the cap |
| 334 | evts, errs := f.reader.Read(ctx) |
| 335 | go func() { |
| 336 | defer f.filament.Close() |
| 337 | err = f.filament.Run(evts, errs) |
| 338 | if err != nil { |
| 339 | log.Errorf("filament failed: %v", err) |
| 340 | f.stop() |
| 341 | } |
| 342 | }() |
| 343 | } else { |
| 344 | if fltr != nil { |
| 345 | f.reader.SetFilter(fltr) |
| 346 | } |
| 347 | // use the channels where events are read |
| 348 | // from the capture as aggregator source |
| 349 | evts, errs := f.reader.Read(ctx) |
| 350 | f.agg, err = aggregator.NewBuffered( |
| 351 | evts, |
| 352 | errs, |
| 353 | f.config.Aggregator, |
| 354 | f.config.Output, |
| 355 | f.config.Transformers, |
| 356 | f.config.Alertsenders, |
| 357 | ) |
| 358 | if err != nil { |
| 359 | return err |
| 360 | } |
| 361 | } |
| 362 | |
| 363 | return api.StartServer(f.config) |
| 364 | } |
| 365 | |
| 366 | // Wait waits for the app to receive the termination signal. |
no test coverage detected