Kool Streams is a simple framework for working with aynchronous events and collections. Kool Streams are inspired by a combination of the Reactive Extensions (Rx), Iteratees and various other similar approaches to dealing with concurrency.
Kool Streams provide a form of asynchronous collection or Event Stream that can be composed and processed like regular collections using the same kind of combinator API folks are familiar with (filter(), flatMap(), map(), fold() etc) but done asynchronously to deal with time delay, network communication, to use threads efficiently and avoid blocking.
- Stream represents a stream of asynchronous events. Its like an asynchronous collection, where events are pushed into a Handler rather than pulled via an iterator.
- Handler is used to process the events from a stream, through you can just pass a function to handle each next event instead
A Stream is then opened via the open() method either passing in a handler in open(Handler) or by passing in a function to handle even next event via open(fn: (T) -> Unit).
When you open() a Stream you get back a Cursor which can be used to close the stream.
A Stream must only invoke a Handler from one thread at once; so a handler does not need to worry about being thread safe.
The lifecycle of events from a Handler perspective is a stream will invoke
- onOpen(Cursor) once before any other events
- onNext(T) zero to many times for each event on the stream after opening
- onError(Throwable) on errors
- onComplete() when its completed
This means that a Handler can choose to close a stream if it knows it has finished processing it. For example if you can use take(n) to limit the size of a stream.
Create event streams from various things:
- java.util.Collection
- bean event listeners (TODO :)
- java.util.Timer via extension functions
- java.util.concurrent.ScheduledExecutorService via extension functions
- SimpleStream
- Apache Camel Endpoints via extension functions on Endpoint
Combine streams with Collection-style combinators
- filter and map values via filter() and map()
- filter distinct values via distinct()
- take a specific amount of events via take(n) or takeWhile(predicate)
Using windows of events for complex event processing types of things
- create a moving fixed window of events via window(size)
- create a moving time window of events via timeWindow(millis)
- group events in a window using groupBy(keyFunction)
Finally you can pipe the events from a Stream to any Apache Camel Endpoint