JGroups Example With Addressing

Address.kt

/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

package org.libraryweasel.pack

import org.jgroups.*
import java.io.Serializable
import java.lang.Exception
import java.lang.RuntimeException
import java.util.concurrent.ConcurrentHashMap

data class IntroductionMessage(val identity: Any): Serializable

class LibraryWeaselPack(
        val identity: Any,
        val trackingFilter: (Any) -> Boolean,
        val messageListener: MessageListener
    ) {
    private lateinit var jChannel: JChannel
    private var connected: Boolean = false
    private val addressBook = ConcurrentHashMap<Address, Any>()
    private var currentView: View? = null

    fun join() {
        System.setProperty("java.net.preferIPv4Stack", "true")
        System.setProperty("jgroups.bind_addr", "127.0.0.1")
        jChannel = JChannel()
        jChannel.discardOwnMessages = true

        jChannel.receiver = object: ReceiverAdapter() {
            override fun viewAccepted(view: View?) {
                if (view == null) {
                    currentView = view
                    jChannel.send(null, IntroductionMessage(identity))
                } else {
                    val oldView = currentView
                    currentView = view
                    View.leftMembers(oldView, currentView)?.forEach {
                        addressBook.remove(it)
                    }
                    View.newMembers(oldView, currentView)?.forEach {
                        jChannel.send(it, IntroductionMessage(identity))
                    }
                }
            }

            override fun receive(msg: Message) {
                val body: Any? = msg.getObject()
                if (body != null && body is IntroductionMessage) {
                    val introduction = body.identity
                    if (trackingFilter.invoke(introduction)) {
                        addressBook[msg.src] = introduction
                    }
                } else {
                    messageListener.receive(msg)
                }
            }
        }

        jChannel.connect("lw")

        connected = true
    }

    fun sendFirst(filter : (Any) -> Boolean, body: Any): Boolean {
        if (!connected) {
            throw RuntimeException("Not connected.")
        }
        var sent = false;

        for(entry in addressBook.entries) {
            if (filter.invoke(entry.component2())) {
                try {
                    jChannel.send(entry.key, body)
                    sent = true
                    break
                } catch (ex: Exception) {
                    sent = false
                }
            }
        }
        return sent
    }

    fun sendAll(filter : (Any) -> Boolean, body: Any): Boolean {
        if (!connected) {
            throw RuntimeException("Not connected.")
        }
        var sent = false
        addressBook.forEach { address: Address, jsonObject: Any ->
            if (filter.invoke(jsonObject)) {
                try {
                    jChannel.send(address, body)
                    sent = true
                } catch (ex: Exception) {
                    //do nothing
                }
            }
        }
        return sent
    }
}