Datastream has a lot in common with the CSP module.
Although they both were designed for I/O processing, there are several important distinctions:
Extremely low: stream can be started with 1 virtual call, short-circuit evaluation optimizes performance
No short-circuit evaluation, overhead is higher
Fast, but slower than Datastream
Small pieces of data
Medium-sized objects, ByteBufs
Simple and convenient
To provide maximum efficiency, our framework widely utilizes combinations of CSP and Datastream. For this purpose,
ChannelSupplier, ChannelConsumer, StreamSupplier and StreamConsumer have transformWith() methods and special
Transformer interfaces. Using them, you can seamlessly transform channels into other channels or datastreams and vice
versa, creating chains of such transformations.
We’ve measured Datastream performance (StreamSupplier streams 100M Integer objects to StreamConsumer scenario)
and received the following result:
We’ve also measured TCP server performance that uses both Datastream and CSP and got the average result of 47495905 requests per second.
You can add Datastream module to your project by inserting dependency in pom.xml:
Simple Supplier -
shows how to create a simple custom Supplier and stream some data to Consumer.
To run the examples, you need to clone DataKernel from GitHub:
$ git clone https://github.com/softindex/datakernel And import it as a Maven project. Check out branch v3.1. Before running the examples, build the project.
These examples are located at datakernel -> examples -> core -> datastream.
When you run SupplierExample, you’ll see the following output:
This output represents the data that our custom StreamSupplier provided to StreamConsumer. Let’s have a
look at the implementation:
When you run ConsumerExample, you’ll see the following output:
ConsumerExample extends AbstractStreamConsumer and just prints out received data. The stream process is managed with
overridden methods onStarted(), onEndOfStream() and onError():
TransformerExample shows how to create a custom StreamTransformer which takes strings from input stream and
transforms them to their length if it is less than defined MAX_LENGTH.
First, we define AbstractStreamConsumer and AbstractStreamSupplier:
Now we define the main method, which creates a supplier of test data, an instance of TransformerExample and StreamConsumerToList.
Next, we define the sequence of transformation and output:
So, if you run the example, you’ll receive the following output:
Built-in Stream Nodes
BuiltinStreamNodesExample demonstrates some simple examples of utilizing built-in datastream nodes. If you
run the example, you’ll receive the following output:
The first line is a result of StreamMapper:
The next three lines of the output are results of utilizing StreamSharder:
The last line of the output is a result of utilizing StreamFilter, which filters numbers and leaves only odd ones: