stream-processor

Under the hood, Benthos csv input uses the standard encoding/csv packages's csv.Reader struct.

The current implementation of csv input doesn't allow setting the LazyQuotes field.

We have a use case where we need to set the LazyQuotes field in order to make things work correctly.

It can be very difficult to piece together a reasonably estimate of a history of events from the current workers logs because none of them have timestamps.

So for that end, I think we should add timestamps to the logs.

This has some cons:

1. We can't just use @printf like we have been until now. We need to either include a timestamp in every @printf call (laborious and error prone) or c

An Operator that both filters and maps.

Akin to Rust's own FilterMap but on a Stream rather than Iterator.

let strings = ["1", "two", "NaN", "four", "5"];
let mut app = Application::default()
  .iterator(strings, |conf| {
     conf.set_arcon_time(ArconTime::Process);
  })
  .filter_map(|s| s.parse().ok())
  .b

Integration tests were originally used within generator project as inline compiling was not supported by Fluxtion. Many tests have been migrated to use com.fluxtion.generator.util.MultipleSepTargetInProcessTest.

MultipleSepTargetInProcessTest has the following advantages:

• Supports in-process compiling no need for integration tests
• Parameterized to test both compiled and interpreted v