HazelcastのEntryListenerをちょっと試してみる

ホントは、別の機能を試していたのですが、ちょっと気になったことがあったので先に確認してみました。

Hazelcastの、EntryListenerについてです。

Distributed Events
http://www.hazelcast.com/docs/3.1/manual/single_html/#Events

正確には、分散オブジェクトで発生したイベントに対するListenerで、Mapに紐付けるものがEntryListener、それ以外のSetなどに紐付けるのがItemListenerみたいですが。

それぞれ、分散オブジェクト（IMap、ISetなどなど…）に対する、追加、更新、削除などのイベントに対して、通知を受けるListenerを仕込むことができます。

で、今回はその中でもEntryListenerを扱います。

なお、これの発展形であるHazelcastのContinuous Queryは、以前試しています。

HazelcastのContinuous Queryを試す
http://d.hatena.ne.jp/Kazuhira/20131027/1382867045

準備

build.sbtの依存関係の定義。

libraryDependencies ++= Seq(
  "com.hazelcast" % "hazelcast-wm" % "3.1.3",
  "org.scalatest" %% "scalatest" % "2.0" % "test"
)

作成した、EntryListenerはこんな感じです。
src/test/scala/org/littlewings/hazelcast/entrylistener/CounteredEntryListener.scala

package org.littlewings.hazelcast.entrylistener

import com.hazelcast.core.{EntryEvent, EntryListener}

class CounteredEntryListener extends EntryListener[String, String] {
  var added: Int = _
  var updated: Int = _
  var removed: Int = _
  var evicted: Int = _

  override def entryAdded(event: EntryEvent[String, String]): Unit =
    added += 1

  override def entryUpdated(event: EntryEvent[String, String]): Unit =
    updated += 1

  override def entryRemoved(event: EntryEvent[String, String]): Unit =
    removed += 1

  override def entryEvicted(event: EntryEvent[String, String]): Unit =
    evicted += 1
}

発生したイベントの種類に応じて、カウントアップしていこうという目論見です。

そして、それを確認するテストコードの雛形。
src/test/scala/org/littlewings/hazelcast/entrylistener/EntryListenerSpec.scala

package org.littlewings.hazelcast.entrylistener

import com.hazelcast.config.Config
import com.hazelcast.core.{Hazelcast, HazelcastInstance}

import org.scalatest.FunSpec
import org.scalatest.Matchers._

class EntryListenerSpec extends FunSpec {
  describe("entry listener spec") {
    // ここに、テストコードを書く
  }

  def withHazelcast(fun: HazelcastInstance => Unit): Unit = {
    val hazelcast = Hazelcast.newHazelcastInstance(new Config)

    try {
      fun(hazelcast)
    } finally {
      hazelcast.getLifecycleService.shutdown
    }
  }
}

テストケースごとに、HazelcastInstanceを作成して、シャットダウンします。

確認。

それでは、テストケースと合わせて確認しましょう。

追加。

    it("add event") {
      withHazelcast { hazelcast =>
        val listener = new CounteredEntryListener
        val map = hazelcast.getMap[String, String]("default")
        map.addEntryListener(listener, false)

        map.put("key1", "value1")

        Thread.sleep(1 * 1000L)

        listener.added should be (1)
        listener.updated should be (0)
        listener.removed should be (0)
        listener.evicted should be (0)
      }
    }

更新その1。

    it("update event #1") {
      withHazelcast { hazelcast =>
        val listener = new CounteredEntryListener
        val map = hazelcast.getMap[String, String]("default")
        map.addEntryListener(listener, false)

        map.put("key1", "value1")
        map.put("key1", "value1-1")

        Thread.sleep(1 * 1000L)

        listener.added should be (1)
        listener.updated should be (1)
        listener.removed should be (0)
        listener.evicted should be (0)
      }
    }

更新その2。

    it("update event #2") {
      withHazelcast { hazelcast =>
        val listener = new CounteredEntryListener
        val map = hazelcast.getMap[String, String]("default")
        map.addEntryListener(listener, false)

        map.put("key1", "value1")
        map.replace("key1", "value1-1")

        Thread.sleep(1 * 1000L)

        listener.added should be (1)
        listener.updated should be (1)
        listener.removed should be (0)
        listener.evicted should be (0)
      }
    }

削除。

    it("remove event") {
      withHazelcast { hazelcast =>
        val listener = new CounteredEntryListener
        val map = hazelcast.getMap[String, String]("default")
        map.addEntryListener(listener, false)

        map.put("key1", "value1")
        map.remove("key1")

        Thread.sleep(1 * 1000L)

        listener.added should be (1)
        listener.updated should be (0)
        listener.removed should be (1)
        listener.evicted should be (0)
      }
    }

エビクト。

    it("evict event") {
      withHazelcast { hazelcast =>
        val listener = new CounteredEntryListener
        val map = hazelcast.getMap[String, String]("default")
        map.addEntryListener(listener, false)

        map.put("key1", "value1")
        map.evict("key1")

        Thread.sleep(1 * 1000L)

        listener.added should be (1)
        listener.updated should be (0)
        listener.removed should be (0)
        listener.evicted should be (1)
      }
    }

とまあ、あまり面白みのないごく当たり前の結果が得られました…。

なお、結果の確認の前にThread.sleepが入っているのは

        map.put("key1", "value1")

        Thread.sleep(1 * 1000L)

        listener.added should be (1)

Continuous Queryの時もそうでしたが、Listenerは別スレッドで動作するからです。

というわけで、EntryListenerの確認はできたのですが、自分が確認したかったこととはあまり関係なかったみたいで…。

さて、次のネタを仕込みますか。