from fluxt import Fluxt
from fluxt.storage import LevelStore


# create a streaming app
fluxt = Fluxt(name='basic_reduce')

word_count = fluxt.State('word_count', default=0,
                store=LevelStore('/tmp/data'))


@fluxt.operation()
def tokenize(event, output):
    for word in event.lower().split():
        output.send(word)


@fluxt.operation(state=word_count)
def count(event, output, state):
    state[event] += 1
    output.send((event, state[event]))


@fluxt.stream()
def word_count_processor(datastream):
    events = ['welcome', 'to', 'fluxt!',
                'The', 'python', 'streaming framework']

    datastream.source_from_collection(events)

    datastream.pipeline(tokenize, count)

    datastream.print()


if __name__ == '__main__':
    # run the fluxt app
    fluxt.run()