Streaming components in Scala for building Microservices, Event Sourcing, Event Logging, CQRS, and Reactive systems. Functions and pre-built stream stages for Akka Streams, HTTP, Actors, Kafka, Cassandra, Algebird, and Avro. The ACK in a SMACK stack
It's meant to be used as
- a more adaptable alternative to frameworks and products
- examples for learning how to use these components and work with streams
- templates for other components or custom alternatives
- is in your product
This Readme covers download, setup and configuration, unit and integration testing and adding it to your project. Documentation is on the website, Scaladocs have examples and lower level descriptions.
Add dependency in build.sbt
libraryDependencies += "com.github.garyaiki" % "dendrites_2.12" % "0.6.3"
Add dependency in pom.xml
`<!-- https://mvnrepository.com/artifact/com.github.garyaiki/dendrites_2.12 -->
<dependency>
<groupId>com.github.garyaiki</groupId>
<artifactId>dendrites_2.12</artifactId>
<version>0.6.3</version>
</dependency>`
Requires Java 8 download jdk8 and Scala 2.12 download scala. Clone or download dendrites from Github, see cloning-a-repository
sbt version, download, documentation
In a terminal window cd to the dendrites directory.
$ sbt launch sbt, returns > prompt
> help list available help
> update Resolves and optionally retrieves dependencies.
> clean Deletes files produced by the build under target directory
> compile Compiles sources under src/main/scala
> test:compile Compiles unit test sources under src/test/scala
> it:compile Compiles integration test sources under src/it/scala
> test Executes unit tests.
> testOnly Executes specific unit tests
> testQuick Re-executes failed unit tests
> package Produces the jar.
> doc Generates Scaladocs in /dendrites/target/scala-2.12/api
> scalastyle Run scalastyle on source code under src/main/scala
> test:scalastyle Run scalastyle on source code under src/test/scala
> dependencyTree Prints an ascii tree of all the dependencies to the console
> exit quit sbt
In a terminal window cd to the dendrites directory.
$ mvn dependency:resolve Resolves and optionally retrieves dependencies.
$ mvn clean Deletes files produced by the build under target directory
$ mvn compile Compiles sources under src/main/scala
$ mvn scala:testCompile Compiles unit test sources under src/test/scala
$ mvn test Executes unit tests.
$ mvn install Install the package in local repository, for use as in other projects locally.
$ mvn source:jar Bundle the main sources of the project into a jar archive.
$ mvn scala:doc-jar Generate Scaladocs in a jar
Minimal install and configure for running dendrites integration tests. Kafka's documentation, download Kafka
Extract server files tar -xvf kafka_2.12-1.0.*.*.tar
Optionally, create or replace symbolic link ln -nsf kafka_2.12-1.0.*.* kafka
Edit configuration in install directory /config/server.properties
To advertise the broker in a local install. In Socket Server Settings add
listeners=PLAINTEXT://localhost:9092
In production, log retention hours are 168 (1 week). For development you can shorten them to 1. In Log Retention Policy
log.retention.hours=1
Add dependencies to driver jars. sbt and Maven dependency settings. See kafka-clients Maven repository page
Put a Consumer Config properties file on your classpath. See Kafka's New Consumer Configs documentation and a dendrites example. KafkaSource requires enable.auto.commit=false for stream backpressure to control commit. To pass this configuration to KafkaSource, create a subclass of ConsumerConfig. example
Put a Producer Config properties file on your classpath. See Kafka's Producer Configs documentation example To pass this configuration to KafkaSink, create a subclass of ProducerConfig example
In a Terminal window cd to kafka install directory
bin/zookeeper-server-start.sh config/zookeeper.properties
Zookeeper will run in the foreground
In a 2nd Terminal window in the same directory
bin/kafka-server-start.sh config/server.properties
Kafka server will run in the foreground
In a 3rd Terminal window in the same directory
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic account-topic
This takes a few seconds.
Kafka server must be running and have an account-topic
In dendrites directory terminal window with sbt running
> it:compile
> it:testOnly com.github.garyaiki.dendrites.kafka.KafkaProducerConsumerSpec
> it:testOnly com.github.garyaiki.dendrites.kafka.AvroKafkaProducerConsumerSpec
> it:testOnly com.github.garyaiki.dendrites.kafka.stream.KafkaStreamSpec
Note: KafkaStreamSpec, ShoppingCartCmdSpec creates KafkaConsumer in KafkaSource. KafkaConsumer calls must all be from the same thread. KafkaSource doesn't close KafkaConsumer until the stream it's in completes. To re-run one of these tests, exit and restart sbt.
List topics
bin/kafka-topics.sh --zookeeper localhost:2181 --list
Delete topic, this may not happen right away. The topic _consumerOffsets can't be deleted from the command line. Kafka Tool is a GUI tool and free for personal use.
bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic account-topic
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic shoppingcartcmd-topic
it:testOnly com.github.garyaiki.dendrites.examples.cqrs.shoppingcart.ShoppingCartCmdSpec
bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic shoppingcartcmd-topic
After topic deleted, stop Kafka server
bin/kafka-server-stop.sh
Give Zookeeper a few seconds. Then stop it too
bin/zookeeper-server-stop.sh
Minimal install and configure for running dendrites integration tests, Cassandra's documentation, download Cassandra
Extract files tar -xvf apache-cassandra-3.11.*-bin.tar
Optionally, create or replace symbolic link ln -nsf apache-cassandra-3.11.* cassandra
For development, default configuration can be used.
Add dependencies to cassandra-driver-core, cassandra-driver-mapping, cassandra-driver-extras. Driver jars with sbt and Maven dependency settings. See Datastax Cassandra Maven repository page
dendrites has an example Cassandra configuration in resource.conf under dendrites/cassandra. You can override these in your application.conf file. CassandraConfig is a trait you can extend. PlayListSongConfig is an example.
In a Terminal window cd to cassandra install directory
bin/cassandra -f
It will run in the foreground
In a dendrites directory terminal window with sbt running. These tests teardown their keyspaces, tables, and connections, run them one at a time so ScalaTest doesn't mix them up and report false errors.
> it:compile
> it:testOnly com.github.garyaiki.dendrites.cassandra.BoundStmtSpec
> it:testOnly com.github.garyaiki.dendrites.cassandra.stream.CassandraSongSpec
> it:testOnly com.github.garyaiki.dendrites.cassandra.stream.CassandraPlaylistSpec
> it:testOnly com.github.garyaiki.dendrites.cassandra.stream.CassandraSpec
> it:testOnly com.github.garyaiki.dendrites.examples.cqrs.shoppingcart.cassandra.stream.CassandraShoppingCartSpec
> it:testOnly com.github.garyaiki.dendrites.examples.cqrs.shoppingcart.cassandra.stream.CassandraShoppingCartCmdSpec
> it:testOnly com.github.garyaiki.dendrites.examples.cqrs.shoppingcart.cassandra.stream.CassandraShoppingCartEvtLogSpec
> it:testOnly com.github.garyaiki.dendrites.examples.cqrs.shoppingcart.cmd.stream.ShoppingCartCmdAndEvtSinkSpec
> it:testOnly com.github.garyaiki.dendrites.examples.cqrs.shoppingcart.cmd.stream.ShoppingCartCmdAndEvtSinksSpec
With Kafka shoppingcartcmd-topic
> it:testOnly com.github.garyaiki.dendrites.examples.cqrs.shoppingcart.cmd.stream.ShoppingCartCmdAndEvtSpec
Exit sbt to close its connection to Cassandra. If you kill Cassandra while sbt is still it will keep trying to reconnect to Cassandra.
Then, Ctrl-C in the terminal running Cassandra
Balances Server is an example server that handles requests from integration tests under src/it/scala
These integration tests need Balances Server running to handle HTTP requests.
> bgRun Runs Balances Server an Http server in the background
In the dendrites directory terminal with sbt running
> it:compile
> it:testOnly com.github.garyaiki.dendrites.http.stream.TypedQueryResponseFlowSpec
> it:testOnly com.github.garyaiki.dendrites.examples.account.http.*
Sometimes, some of these tests fail because Balances Server has a default limit of 4 concurrent requests and tests are running in parallel and may not get time on the server. If this happens, re-run the failed tests one at a time.
Ctrl-C in the terminal window running Balances Server
dendrites uses Typesafe Config. Traits with names ending in Config define configuration properties for components. This mediates configuration from usage, so you may use these traits with different configuration sources.
Typesafe Config user guide, Akka config user guide
Configurations in /src/main/resources/reference.conf can be overridden by your application.conf.
The akka section is for logging and has Akka specific logging settings
Under the dendrites section
algebird has default settings for CountMinSketch, DecayedValue, HyperLogLog, and QTree
blocking-dispatcher configures a thread pool to be used when there are blocking calls
checking-balances, money-market-balances, and savings-balances are example http client configurations
kafka is an example topic and timeout configuration
cassandra is an example cluster and keyspace configuration
timer is asychronous and exponential backoff timeout configuration
Logging configuration for running tests and integration tests is in /src/test/resources/logback.xml
Your application Logback configuration is done separately. You can copy logback.xml paste it into your /src/main/resources/ and modify it.