implement count windows

[quentin-quix]

Implements:

• sdf.tumbling_count_window
• sdf.hopping_count_window
• sdf.sliding_count_window

Those windows divide messages into fixed-size batch of messages. They allow performing stateful aggregations like sum, reduce, etc. on top of a fixed number of messages and emit results downstream.

app = Application()
sdf = app.dataframe(...)

sdf = (
    # Define a tumbling window of 10 messages
    sdf.tumbling_count_window(count=10)

    # Specify the aggregation function
    .sum()

    # Specify how the results should be emitted downstream.
    # "current()" will emit results as they come for each updated window,
    # possibly producing multiple messages per key-window pair
    # "final()" will emit windows only when they are closed and cannot
    # receive any updates anymore.
    .current()
)