DEV Community: Oleg Agafonov

Building a DIY Smart Home Device: Get S**t Done

Oleg Agafonov — Sat, 20 Jul 2024 21:38:18 +0000

This year is the first time I wrote down a list of annual goals. One of them was to learn how to solder and build a DIY smart home device. When you set a goal like this, the hardest part is finding a project that will give you enough motivation to get shit done 💩💩💩. Ironically, this phrase, known by every developer, was a real inspiration.

Last October, we bought a house in a Bulgarian village. Most of the villages here don’t have a central sewage system, so every house has an underground reservoir for wastewater. Every time it’s full, you call a special truck that pumps it out and takes it to a water recycling station(hopefully 😈😈😈). I had never lived with such a system before. A few months after we started living in the house, we encountered a situation where the water rose up to the pipe. When it happened again a month ago, I knew I needed to get shit done.

The idea of my first DIY smart home device was simple but intriguing: take an ultrasonic sensor, a cheap controller board with WiFi (to make things easier), and connect it all to a battery powered by a solar panel. Connect the board to WiFi and start sending API requests to New Relic. New Relic is a top-notch monitoring system that takes care of drawing a dashboard and sending monitoring alerts to Telegram or any other messenger. As soon as everything was planned, I placed a few orders on AliExpress and started waiting.

The ultrasonic sensor AJ-SR04M came first. It’s a very interesting device that works in five different modes. You can choose one by soldering a particular resistor. Here is a very good blog explaining it all in detail. Fortunately, the mode I chose works if you simply short the resistor. While waiting for the controller board, I decided to play with the sensor and connected it to my Flipper Zero:

A week later, I received an ESP32 dev board. There is a lot of information on the internet about these boards, but for me, it was important that it comes with an embedded WiFi module and is easy to program using Arduino IDE or Visual Studio and PlatformIO:

I knew that the hardest part of the project would be related to the battery and solar panel, but here I cheated. Some months ago, I bought a solar-powered WiFi camera from AliExpress that was not working. Being too busy at the time, I missed the return window, so I disassembled it and took the entire power supply module.

What was left was to assemble everything and mount it on the reservoir. I decided not to go with a fancy 3D-printed solution and bought an electrical box the size of the solar panel:

The entire project took me 16-20 hours in total, spread equally over four weekends. Now I can see how close the water gets to the pipe. Also, once it’s getting close enough, I get a sarcastic notification in Telegram. But most importantly, I accomplished one of my annual goals and had lots of fun learning new things. It was also fun to joke with my Bulgarian friends that, while people are turning off lights by talking to Alexa, I probably found the idea for the most demanded smart home device in the country 😆😆😆.

Yet another ode to Vert.x, or how to write a performance-wise expiring map in less than 100 lines of code.

Oleg Agafonov — Wed, 22 Sep 2021 21:23:40 +0000

I've been working with the Vert.x framework for more than 4 years but I won't stop being excited how simple, lightweight and elegant it is (especially the event loop thread model). In this blog post I will tell you how we implemented PeriodicallyExpiringHashMap data structure in less than 100 lines of code. But first let me give you a bit of a context about why do we need it.

Problem

SIP3 is a very advanced VoIP monitoring and troubleshooting platform. To provide detailed information about calls quality we need to:

Aggregate RTP packets into RTP streams in a real-time
Periodically walk though all the RTP streams and terminate ones that haven't been updated for a certain period of time.

Let's stay away from telecom specific and take a look at a simplified code example:

class RtpStreamHandler : AbstractVerticle() {

    var expirationDelay: Long = 1000
    var aggregationTimeout: Long = 30000

    private val rtpStreams = mutableMapOf<String, RtpStream>()

    override fun start() {
        vertx.setPeriodic(expirationDelay) {
            val now = System.currentTimeMillis()

            rtpStreams.filterValues { rtpStream -> rtpStream.updatedAt + aggregationTimeout < now }
                .forEach { (rtpStreamId, rtpStream) ->
                    terminateRtpStream(rtpStream)
                    rtpStreams.remove(rtpStreamId)
                }
        }

        vertx.eventBus().localConsumer<RtpPacket>("on_rtp_packet") { event ->
            val rtpPacket = event.body()
            handleRtpPacket(rtpPacket)
        }
    }

    fun handleRtpPacket(rtpPacket: RtpPacket) {
        val rtpStream = rtpStreams.getOrPut(rtpPacket.rtpStreamId) { RtpStream() }
        rtpStream.addPacket(rtpPacket)
    }

    fun terminateRtpStream(rtpStream: RtpStream) {
        vertx.eventBus().localSend("on_rtp_stream", rtpStream)
    }
}

Now let's imagine that we constantly have 30K of active RTP streams. Also every second we terminate approximately a thousand of old steams but get a thousand of new ones instead. In these circumstances our code doesn't look very efficient and we certainly need a better solution.

Solution

As you can see from the first code snippet once an RTP stream was updated it won't be terminated at least for the next aggregationTimeout. This means that we can simply do not bother about it for some time.

And this is the key idea behind the SIP3 PeriodicallyExpiringHashMap implementation:

class PeriodicallyExpiringHashMap<K, V> private constructor(
    vertx: Vertx,
    private val delay: Long,
    private val period: Int,
    private val expireAt: (K, V) -> Long,
    private val onExpire: (K, V) -> Unit
) {

    private val objects = mutableMapOf<K, V>()
    private val expiringSlots = (0 until period).map { mutableMapOf<K, V>() }.toList()
    private var expiringSlotIdx = 0

    init {
        vertx.setPeriodic(delay) {
            terminateExpiringSlot()
            expiringSlotIdx += 1
            if (expiringSlotIdx >= period) {
                expiringSlotIdx = 0
            }
        }
    }

    fun getOrPut(key: K, defaultValue: () -> V): V {
        return objects.getOrPut(key) {
            defaultValue.invoke().also { expiringSlots[expiringSlotIdx][key] = it }
        }
    }

    private fun terminateExpiringSlot() {
        val now = System.currentTimeMillis()

        expiringSlots[expiringSlotIdx].apply {
            forEach { (k, v) ->
                val expireAt = expireAt(k, v)

                when {
                    expireAt <= now -> {
                        objects.remove(k)?.let { onExpire(k, it) }
                    }
                    else -> {
                        var shift = ((expireAt - now) / delay).toInt() + 1
                        if (shift >= period) {
                            shift = period - 1
                        }
                        val nextExpiringSlotIdx = (expiringSlotIdx + shift) % period

                        expiringSlots[nextExpiringSlotIdx][k] = v
                    }
                }
            }
            clear()
        }
    }

    data class Builder<K, V>(
        var delay: Long = 1000,
        var period: Int = 60,
        var expireAt: (K, V) -> Long = { _: K, _: V -> Long.MAX_VALUE },
        var onExpire: (K, V) -> Unit = { _: K, _: V -> }
    ) {
        fun delay(delay: Long) = apply { this.delay = delay }
        fun period(period: Int) = apply { this.period = period }
        fun expireAt(expireAt: (K, V) -> Long) = apply { this.expireAt = expireAt }
        fun onExpire(onExpire: (K, V) -> Unit) = apply { this.onExpire = onExpire }

        fun build(vertx: Vertx) = PeriodicallyExpiringHashMap(vertx, delay, period, expireAt, onExpire)
    }
}

Here are the benefits of this data structure:

Now we just have a bunch of time slots. So, instead of walking through all the objects in our map every expirationDelay we can walk trough a single slot. So, instead of checking on 30K objects every second we will check on 1K only.
We don't need to create a copy of original map every time we decide to walk though it. In the previous example it also was an issue, because rtpSteams.filtervalues creates a copy of the original map.
The last and the most important. Our implementation will stay consistent within a particular verticle context. That means you can simply extend it and implement the rest of the methods (including tricky ones, like size()).

Conclusions

Finally let's see how our verticle will look like with the new PeriodicallyExpiringHashMap data structure:

class RtpStreamHandler : AbstractVerticle() {

    var expirationDelay: Long = 1000
    var aggregationTimeout: Long = 30000

    private lateinit var rtpStreams: PeriodicallyExpiringHashMap<String, RtpStream>

    override fun start() {
        rtpStreams = PeriodicallyExpiringHashMap.Builder<String, RtpStream>()
            .delay(expirationDelay)
            .period((aggregationTimeout / expirationDelay).toInt())
            .expireAt { _, rtpStream -> rtpStream.updatedAt + aggregationTimeout }
            .onExpire { _, rtpStream -> terminateRtpStream(rtpStream) }
            .build(vertx)

        vertx.eventBus().localConsumer<RtpPacket>("on_rtp_packet") { event ->
            val rtpPacket = event.body()
            handleRtpPacket(rtpPacket)
        }
    }

    fun handleRtpPacket(rtpPacket: RtpPacket) {
        val rtpStream = rtpStreams.getOrPut(rtpPacket.rtpStreamId) { RtpStream() }
        rtpStream.addPacket(rtpPacket)
    }

    fun terminateRtpStream(rtpStream: RtpStream) {
        vertx.eventBus().localSend("on_rtp_stream", rtpStream)
    }
}

And here are the load tests results (purple - is our new implementation):

The tests look great and the code looks clean and simple thanks to the Vert.x event loop thread model.

👨‍💻 Happy coding,
Your SIP3 team.

Java iterators are dead. Long live Kotlin iterators!

Oleg Agafonov — Tue, 17 Nov 2020 19:11:05 +0000

Iterator is a great pattern. It's extremely simple and yet very powerful. Long time ago I saw iterators literally everywhere but, when Java 8 came with streams API, things have changed. It doesn't mean that iterators disappeared fully. No, they just hid deep into libraries and frameworks you use every day. But few of them are still the most important players in modern and widely used Java SDKs.

SIP3 uses MongoDB Java Driver to read insanely big amounts of data. And as you know from my previous blog post we prefer to work with MongoCursor<Document> explicitly. Guess what? MongoCursor<Document> is nothing else but iterator 😊.

Having MongoCursor<Document> as iterator totally makes sense though. Iterator's contract forces you to call hasNext() method every time when you want to fetch a new document from MongoDB collection. So, MongoDB Java Driver just makes you read not all documents but as many as you need. This approach saves a lot of resources on data transferring cause data now can be stored in MongoDB server and pulled from there on demand in multiple batches.

In the previous blog post I showed how we work with multiple MongoDB collections partitioned by time the same way you will work with just a single collection. However, it wasn't enough to aggregate the data from all the partitions we have. That's why, inspired by streams API, we decided to introduce a few extension functions.

⚠️ WARNING: Examples below are not perfect and didn't mean to be though. So, keep in mind that calling map() and merge() methods will change a calling iterator state too.

map()

No doubts map() is the most popular function of streams API. Of course it can be a great addition to our iterator:

fun <T, R> Iterator<T>.map(transform: (T) -> R): Iterator<R> {
    val i = this

    return object : Iterator<R> {

        override fun hasNext(): Boolean {
            return i.hasNext()
        }

        override fun next(): R {
            if (!hasNext()) throw NoSuchElementException()
            return transform.invoke(i.next())
        }
    }
}

Now let's see how we can call it:

// Int: 1, 2, 3
val numbers = listOf(1, 2, 3).iterator()

// String: "1", "2", "3"
val strings = numbers.map(Any::toString)

The cherry on the top 🍒 is that transform function will be called lazily.

merge()

Not like map() this method is not a part of streams API. However, it let's us gather data from logically partitioned in multiple collections (you can read more about it in the previous blog post). Implementation is a bit complicated.

First we need to write a helper method:

fun <T> Iterator<T>.nextOrNull(): T? {
    return if (hasNext()) next() else null
}

,and then our method extension itself:

fun <T> Iterator<T>.merge(o: Iterator<T>, comparator: Comparator<T>? = null): Iterator<T> {
    val i = this

    return object : Iterator<T> {

        var vi: T? = i.nextOrNull()
        var vo: T? = o.nextOrNull()

        override fun hasNext(): Boolean {
            return vi != null || vo != null
        }

        override fun next(): T {
            if (!hasNext()) throw NoSuchElementException()

            val v: T?
            if (vi != null && (vo == null || comparator == null || comparator.compare(vi, vo) <= 0)) {
                v = vi
                vi = i.nextOrNull()
            } else {
                v = vo
                vo = o.nextOrNull()
            }

            return v!!
        }
    }
}

Now let's see it in action:

val list1 = listOf(1, 3, 5)
val list2 = listOf(2, 4, 6)

var iterator1 = list1.iterator()
var iterator2 = list2.iterator()

// Int: 1, 3, 5, 2, 4, 6
val merged = iterator1.merge(iterator2)

iterator1 = list1.iterator()
iterator2 = list2.iterator()

// Int: 1, 2, 3, 4, 5, 6
val sorted = iterator1.merge(iterator2, Comparator.comparingInt<Int> { i -> i })

So, we just merged and even sorted two iterators 🤯. But... As Leonardo says...

Sure thing:

object IteratorUtil {

    fun <T> merge(vararg iterators: Iterator<T>): Iterator<T> {
        var i = Collections.emptyIterator<T>()
        iterators.forEach { i = i.merge(it) }
        return i
    }
}

Whaaaaat 🤯🤯🤯

Jokes aside in our case this merge exercises are very needed even though they look scary. You can always see this code in action in our Github projects.

In the post I just wanted to show all the power and beauty of Kotlin extension functions and maybe give you some motivation to use them.

👨‍💻 Happy coding,
Your SIP3 team.

MongoDB best practices from the SIP3 team.

Oleg Agafonov — Sun, 15 Nov 2020 14:49:11 +0000

In SIP3 we deal with hundreds of thousands SIP messages per second. Each message has to be aggregated into a SIP session and also saved to MongoDB.

In this blog post I won't explain a rational behind choosing MongoDB but will concentrate on how to work with this database (and any other databases) as efficiently as possible.

✍️ Writing to MongoDB

If you want to increase write throughput while working with your database follow these three rules:

Use a connection pool to the database.
Use bulk writes.
Introduce data partitioning.

You can read about the first two in official MongoDB documentation here and here.

However, I would love to say a few words about partitioning. As you can read here, partitioning is a set of rules/practices you can use to divide your data into distinct independent parts. Often case you may want to have a few levels of partitioning as we do it in SIP3:

Level 1: Logical - Splits SIP traffic into calls and registrations. It let's us to configure separately for how long to keep each of the data.

Level 2: Vertical - Splits SIP session into CDRs (Call Details Records) and SIP messages content. It helps us to save on data indexing.

Level 3: Horizontal - Splits our data by time. This split literally kills two birds with one stone: helps to manage historical depth (by easily dropping old collections) and to save on data indexing.

An implementation of data partitioning may vary and fully depends on the application's specific. However, on a code level writing data to partitions is never a problem. That's how we do it in the SIP3 Salto code:

open fun writeToDatabase(prefix: String, packet: Packet, message: SIPMessage) {
    val collection = prefix + "_raw_" + timeSuffix.format(packet.timestamp)

    val operation = JsonObject().apply {
        put("document", JsonObject().apply {
            val timestamp = packet.timestamp
            put("created_at", timestamp.time)
            put("nanos", timestamp.nanos)

            val src = packet.srcAddr
            put("src_addr", src.addr)
            put("src_port", src.port)
            src.host?.let { put("src_host", it) }

            val dst = packet.dstAddr
            put("dst_addr", dst.addr)
            put("dst_port", dst.port)
            dst.host?.let { put("dst_host", it) }

            put("call_id", message.callId())
            put("raw_data", String(packet.payload, Charsets.ISO_8859_1))
        })
    }

    vertx.eventBus().localRequest<Any>(RoutesCE.mongo_bulk_writer, Pair(collection, operation))
}

Now, when we know how to write data let's check how to read it.

📚 Reading from MongoDB

No doubts, reading data from partitions is a tough cookie 🍪.

In our projects we prefer to work with MongoDB Java Driver explicitly avoiding all existing ORM implementations.

MongoDB Java Driver introduces MongoCursor<Document> which is basically a very efficient iterator because it pulls data from the database in batches.

Iterator is a very old pattern but with modern Kotlin and method extensions it indeed got the second wind. In the next blog post I will show how we introduced Iterator.merge and Iterator.map Kotlin extension functions (do you like this teaser 😈).

But let's get back to reading data partitioned by time from multiple MongoDB collections. Here is our simple recipe:

1 - Retrieve and cache a list of collections by prefix:

@Cacheable(value = ["listCollectionNames"], key = "#prefix")
open fun listCollectionNames(prefix: String): List<String> {
    return client.getDatabase(db).listCollectionNames().asSequence()
            .filter { name -> name.startsWith(prefix) }
            .sorted()
            .toList()
}

2 - Introduce find() method (optionally make it an extension method of existing MongoClient API):

open fun find(prefix: String, timeRange: Pair<Long, Long>,
              filter: Bson, sort: Bson? = null, limit: Int? = null): Iterator<Document> {
    // Retrieve list of collections by prefix and time range
    val collectionNames = listCollectionNames(prefix).asSequence()
            .filter { name -> "${prefix}_${suffix.format(timeRange.first)}" <= name }
            .filter { name -> "${prefix}_${suffix.format(timeRange.second)}" >= name }
            .iterator()

    // Create `Iterator<Document>` which will imitate working with a simple `MongoCursor<Document>` 
    return object : Iterator<Document> {

        var cursor: MongoCursor<Document>? = null

        override fun hasNext(): Boolean {
            if (cursor?.hasNext() == true) return true

            if (collectionNames.hasNext()) {
                cursor = client.getDatabase(db)
                        .getCollection(collectionNames.next())
                        .run {
                            filter?.let { find(filter) } ?: find()
                        }
                        .apply {
                            maxTime(maxExecutionTime, TimeUnit.MILLISECONDS)
                            batchSize(limit ?: batchSize)
                            sort?.let { sort(it) }
                        }
                        .iterator()

                return hasNext()
            }

            return false
        }

        override fun next(): Document {
            if (!hasNext()) throw NoSuchElementException()
            return cursor!!.next()
        }
    }
}

3 - Work with your documents as you are used to do with MongoCursor<Document>:

fun findInRawBySessionRequest(req: SessionRequest): Iterator<Document> {
    val filters = mutableListOf<Bson>().apply {
        add(gte("created_at", req.createdAt))
        add(lte("created_at", req.terminatedAt))
        add(`in`("call_id", req.callId))
    }

    return mongoClient.find("sip_call_raw", Pair(req.createdAt, req.terminatedAt), and(filters))
}

4 - Profit 👌

As you can see now, we managed to work with partitioned data in the way we usually work with just a single collection.

Please, leave us a comment if you liked the idea and share with us how you implement data partitioning in your projects!

👨‍💻 Happy coding,
Your SIP3 team.

How to extend Vert.x EventBus API to save on serialization.

Oleg Agafonov — Tue, 28 Apr 2020 21:03:39 +0000

Alexey Soshin has a great blog post: Understanding Vert.x: Event Bus. In the blog post he explains how to extend MessageCodec interface and start sending objects via Event Bus without paying anything for serialization.

SIP3 is a monitoring and troubleshooting platform that has to handle hundreds of thousands SIP packets and millions of RTP packets per second. That's why we've started using this approach from the very beginning. However, as a Kotlin users, we couldn't do not add some sugar 🍬🍭🍰.

Here is our recipe:

🍬 Register `MessageCodec`

Let's use Kotlin method extensions to define our local codec:

fun Vertx.registerLocalCodec() {
    eventBus().unregisterCodec("local")
    eventBus().registerCodec(object : MessageCodec<Any, Any> {
        override fun decodeFromWire(pos: Int, buffer: Buffer?) = throw NotImplementedError()
        override fun encodeToWire(buffer: Buffer?, s: Any?) = throw NotImplementedError()
        override fun transform(s: Any?) = s
        override fun name() = "local"
        override fun systemCodecID(): Byte = -1
    })
}

All is left is just to register our codec explicitly while initializing Vertx object.

🍭 Extend `EventBus`

Kotlin has two great features - method extensions and named arguments. We decided to use a combination of those to introduce a custom method called localSend:

object EventBusUtil {

    val USE_LOCAL_CODEC = deliveryOptionsOf(codecName = "local", localOnly = true)
}

fun EventBus.localSend(address: String, message: Any, options: DeliveryOptions? = null) {
    options?.apply {
        codecName = "local"
        isLocalOnly = true
    }
    send(address, message, options ?: USE_LOCAL_CODEC)
}

🍰 Just enjoy

Now we can start sending messages within the application without paying for serialization:

fun main() {
    val vertx = Vertx.vertx()
    vertx.registerLocalCodec()

    val answer = BigDecimal(42)
    vertx.eventBus().localSend("question", answer)
}

I hope you loved the approach and will start using it in your projects. If you have any questions, just leave me a message 😄

Happy coding...

How to write beautiful unit tests in Vert.x

Oleg Agafonov — Fri, 24 Apr 2020 15:08:38 +0000

As you already know from my previous blog post the SIP3 team is in love with the incredible Vert.x framework.

Reactive nature makes Vert.x extremely fast and scalable, but (as everything else) it all comes with a price 👿... And here I'm talking about writing unit tests (😈😈😈 of course if you do write them 😈😈😈)...

Is it really that complicated to write tests with Vert.x?

Let's assume that we want to check a simple thing: that a message sent via event bus was received by its consumer.

In a perfect world we would expect to see something like this (please pay attention to the test structure):

class MyVertxTest {

    @Test
    fun `Send message to the address`() {
        // 1. Define our message
        val message = "Hello, world!"

        val vertx = Vertx.vertx()

        // 2. Send it to the address
        vertx.eventBus().send("address", message)
        // 3. Retrieve and check the message
        vertx.eventBus().consumer<String>("address") { event ->
            assertEquals(message, event.body())
        }
    }
}

Obviously this code won't be working because:

There is no consumer at the moment we've sent the message;
Even if there was a consumer assertEquals() will be executed not in the context of our test thread.

Of course, the problem is not new and Vert.x team has already resolved it by introducing a TestContext object (learn more about that from official documentation: here and here).

With the TestContext and JUnit5 our code will look like this:

@ExtendWith(VertxExtension::class)
class MyVertxTest {

    @Test
    fun `Send message to the address`() {
        // 1. Define our message
        val message = "Hello, world!"

        val context = VertxTestContext()
        val vertx = Vertx.vertx()

        // 3. Retrieve and check the message
        vertx.eventBus().consumer<String>("address") { event ->
            context.verify {
                assertEquals(message, event.body())
            }
            context.completeNow()
        }

        // 2. Send it to the address
        vertx.eventBus().send("address", message)

        assertTrue(context.awaitCompletion(5, TimeUnit.SECONDS))
        if (context.failed()) {
            throw context.causeOfFailure();
        }
    }
}

Not like in our previous example, this code will be working fine, but:

Assertion on the context object brings too much overhead;
We would love to keep the order - define test context, execute scenario and only after make all the assertions.

Trying to satisfy the last two requirements the SIP3 team decided to introduce its own test class wrapper:

@ExtendWith(VertxExtension::class)
open class VertxTest {

    lateinit var context: VertxTestContext
    lateinit var vertx: Vertx

    fun runTest(deploy: (suspend () -> Unit)? = null, execute: (suspend () -> Unit)? = null,
                assert: (suspend () -> Unit)? = null, cleanup: (() -> Unit)? = null, timeout: Long = 10) {
        context = VertxTestContext()
        vertx = Vertx.vertx()
        GlobalScope.launch(vertx.dispatcher()) {
            assert?.invoke()
            deploy?.invoke()
            execute?.invoke()
        }
        assertTrue(context.awaitCompletion(timeout, TimeUnit.SECONDS))
        cleanup?.invoke()
        if (context.failed()) {
            throw context.causeOfFailure()
        }
    }
}

We put a context object assertion within the runTest() method. Also we used Kotlin named arguments(which is a really cool) to define test stages.

Let's rewrite our test class using VertxTest:

class MyVertxTest : VertxTest() {

    @Test
    fun `Send message to the address`() {
        // 1. Define our message
        val message = "Hello, world!"

        runTest(
                execute = {
                    // 2. Send it to the address
                    vertx.eventBus().send("address", message)
                },
                assert = {
                    // 3. Retrieve and check the message
                    vertx.eventBus().consumer<String>("address") { event ->
                        context.verify {
                            assertEquals(message, event.body())
                        }
                        context.completeNow()
                    }
                }
        )
    }
}

Of course, it's a matter of taste but we think that the code above is clean and simple even though there is always room for improvement 😄.

The SIP3 team uses this approach in all it projects. Feel free to check out our github and let us know if you loved it.

Cheers...

Scaling Vert.x application for session dependent data processing.

Oleg Agafonov — Wed, 22 Apr 2020 20:01:03 +0000

I'm working on VoIP monitoring and troubleshooting platform called SIP3.

Under the hood SIP3 is an ETL engine powered by incredible Vert.x framework.

Simple ETL pipeline consists of input data, processor and output data as it's shown on the picture below:

Let's assume that our DataProcessingVerticle looks like this:

class DataProcessingVerticle : AbstractVerticle() {

    override fun start() {
        vertx.eventBus().consumer<JsonObject>("DataProcessingVerticle") {
            event -> handle(event.body())
        }
    }

    private fun handle(inputData: JsonObject) {
        val outputData = doTransformation(inputData)
        vertx.eventBus().send("ConsoleOutputVerticle", outputData)
    }
}

Given that doTransformation() is a very fast method we may be happy handling our data using only one server's core.

But what if we have tons of data and one core is not enough?

Our input data is a steam of events. Let's assume that every event is independent from each other. In this case doTransformation() is a function of event itself and scaling will look like this:

We just need to deploy multiple instances of DataProcessingVerticle accordingly to amount of our event-loop threads:

    val deploymentOptions = deploymentOptionsOf(
            instances = threads
    )
    vertx.deployVerticle(DataProcessingVerticle::class.java, deploymentOptions)

As you can see it's as easy as a simple ETL :)

But what if some of our events are dependent from each other?

In case of SIP3 we have SIP messages as an input data and SIP sessions - as an output. Each SIP session consists of a bunch of SIP messages with the same Call-ID. So, how to scale it?

The solution is a little more complicated. But as everything written in Vert.x is simple enough:

We will be balancing our SIP messages using LoadBalancerVerticle to a dedicated instance of DataProcessingVerticle:

class LoadBalancerVerticle : AbstractVerticle() {

    var threads: Int = 1

    override fun start() {
        config().getInteger("instances")?.let {
            threads = it
        }

        vertx.eventBus().consumer<JsonObject>("LoadBalancerVerticle") {
            event -> handle(event.body())
        }
    }

    private fun handle(inputData: JsonObject) {
        val id = inputData.getString("call_id")
        val index = abs(id.hashCode() % threads)
        vertx.eventBus().send("DataProcessingVerticle_$index", inputData)
    }
}

In this case each DataProcessingVerticle has to be started with the unique address. We can guarantee this by using local counter from Vert.x shared data:

class DataProcessingVerticle : AbstractVerticle() {

    override fun start() {
        GlobalScope.launch(vertx.dispatcher()) {
            val index = vertx.sharedData().getLocalCounterAwait(DataProcessingVerticle)
            val address = "DataProcessingVerticle_${index.getAndIncrementAwait()}"
            vertx.eventBus().consumer<JsonObject>(address) {
                event -> handle(event.body())
            }
        }
    }

    private fun handle(inputData: JsonObject) {
        val outputData = doTransformation(inputData)
        vertx.eventBus().send("ConsoleOutputVerticle", outputData)
    }
}

That's all I wanted to show you today. Check out our project to learn more of advanced techniques we use to work with Vert.x framework. If you have any questions, just leave me a message :)

Cheers...

How to monitor remote JVM applications with `logback-webhook-appender`.

Oleg Agafonov — Mon, 20 Apr 2020 09:50:20 +0000

Overview

This tutorial is for anyone who is actively using logback in their JVM applications. For those who are not familiar with the framework, I strongly recommend checking out its documentation.

Out-of-the-box logback can send WARNING and ERROR logging events via email. Sometimes receiving those events is the best way to monitor and debug remote applications. However, email as notifications channel isn't that attractive. That's why the SIP3 team wrote a handy logback extension that uses webhooks and makes it possible to send the same events to unlimited amount of channels including the most popular - Slack and Telegram.

In this tutorial we will build a simple Spring Boot application and configure it to start sending ERROR messages via Incoming Webhook for Slack.

1. Creating Spring Boot Application

Open Spring Initializr project to create a dummy Spring Boot application:

Let's choose Kotlin and press the GENERATE button.

For this showcase we will add some code in the DemoApplication class:

@SpringBootApplication
class DemoApplication {

    private val logger: Logger = LoggerFactory.getLogger(DemoApplication::class.java)

    @PostConstruct
    fun init() {
        try {
            throw RuntimeException("Houston, we have a problem!")
        } catch (e: Exception) {
            logger.error("Got exception...", e)
        }       
    }
}

fun main(args: Array<String>) {
    runApplication<DemoApplication>(*args)
}

As you can see the code above throws RuntimeException every time we start the application.

2. Adding logback-webhook-appender

Let's pull logback-webhook-appender and install it to a local Maven repository using mvn clean install.

Now we can add it as a dependency to the project's pom.xml(in some cases you will have to resolve a conflict with okhttp3 library by explicitly specifying its version):

<properties>
    ...
    <okhttp3.version>4.5.0</okhttp3.version>
</properties>

<dependencies>
    ...     
    <dependency>
        <groupId>io.sip3</groupId>
        <artifactId>logback-webhook-appender</artifactId>
        <version>1.0.0-SNAPSHOT</version>
    </dependency>   
    ...
</dependencies>

3. Configuring logback-webhook-appender

Prior to the next step we need to set up an Incoming Webhook for our Slack.

Once it's done we can create and configure src/main/resources/logback-spring.xml:

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <appender name="webhook" class="io.sip3.logback.WebhookAppender">
        <url>https://hooks.slack.com/services/%YOUR_WEBHOOK_URI%</url>     // (1)
        <json>                                                             // (2)
            {
                "text": ":fire: Houston, we have a problem! :fire:",
                "attachments": [
                    {
                        "text": "{message}"                                // (3)
                    }
                ]
            }
        </json>
        <encoder>
            <pattern>%d{dd-MM-yyyy HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>

    <root level="ERROR">
        <appender-ref ref="webhook"/>
    </root>
</configuration>

(1) - Use <url> to set up a webhook URL
(2) - Use <json> to set up a payload template
(3) - Use {message} as a placeholder for encoded logging event

4. Testing logback-webhook-appender

So far so good. Let's start our application using mvn spring-boot:run command. Then open the configured Slack channel to receive an ERROR notification:

And that is how you setup a very simple but useful monitoring in a matter of 2-5 minutes.

I hope you loved the approach and will start using it in your projects. If you have any questions, just leave me a message :)

Cheers...

DEV Community: Oleg Agafonov

Building a DIY Smart Home Device: Get S**t Done

Yet another ode to Vert.x, or how to write a performance-wise expiring map in less than 100 lines of code.

Problem

Solution

Conclusions

Java iterators are dead. Long live Kotlin iterators!

map()

merge()

MongoDB best practices from the SIP3 team.

✍️ Writing to MongoDB

📚 Reading from MongoDB

How to extend Vert.x EventBus API to save on serialization.

🍬 Register MessageCodec

🍭 Extend EventBus

🍰 Just enjoy

How to write beautiful unit tests in Vert.x

Scaling Vert.x application for session dependent data processing.

How to monitor remote JVM applications with `logback-webhook-appender`.

Overview

1. Creating Spring Boot Application

2. Adding logback-webhook-appender

3. Configuring logback-webhook-appender

4. Testing logback-webhook-appender

🍬 Register `MessageCodec`

🍭 Extend `EventBus`