code.grnet.gr Git - aquarium/blob - src/main/scala/gr/grnet/aquarium/service/StoreWatcherService.scala

   1 /*
   2  * Copyright 2011-2012 GRNET S.A. All rights reserved.
   3  *
   4  * Redistribution and use in source and binary forms, with or
   5  * without modification, are permitted provided that the following
   6  * conditions are met:
   7  *
   8  *   1. Redistributions of source code must retain the above
   9  *      copyright notice, this list of conditions and the following
  10  *      disclaimer.
  11  *
  12  *   2. Redistributions in binary form must reproduce the above
  13  *      copyright notice, this list of conditions and the following
  14  *      disclaimer in the documentation and/or other materials
  15  *      provided with the distribution.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
  18  * OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  19  * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
  20  * PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
  21  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  22  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  23  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  24  * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  25  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  26  * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
  27  * ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
  28  * POSSIBILITY OF SUCH DAMAGE.
  29  *
  30  * The views and conclusions contained in the software and
  31  * documentation are those of the authors and should not be
  32  * interpreted as representing official policies, either expressed
  33  * or implied, of GRNET S.A.
  34  */
  35
  36 package gr.grnet.aquarium.service
  37
  38 import gr.grnet.aquarium.util.{Tags, Loggable, Lifecycle, Tag}
  39 import gr.grnet.aquarium.util.safeUnit
  40 import java.util.concurrent.atomic.AtomicBoolean
  41 import gr.grnet.aquarium.service.event.{StoreIsAliveBusEvent, StoreIsDeadBusEvent}
  42 import gr.grnet.aquarium.connector.rabbitmq.service.RabbitMQService.RabbitMQConKeys
  43 import gr.grnet.aquarium.{Configurable, Aquarium}
  44 import com.ckkloverdos.props.Props
  45 import gr.grnet.aquarium.store.StoreProvider
  46
  47 /**
  48  * Watches for liveliness of stores.
  49  *
  50  * @author Christos KK Loverdos <loverdos@gmail.com>
  51  */
  52
  53 final class StoreWatcherService extends Lifecycle with Configurable with Loggable {
  54   private[this] var _reconnectPeriodMillis = 1000L
  55
  56   private[this] val _pingIsScheduled = new AtomicBoolean(false)
  57   private[this] val _stopped = new AtomicBoolean(false)
  58   private[this] val _rcIsAlive = new AtomicBoolean(true)
  59   private[this] val _imIsAlive = new AtomicBoolean(true)
  60
  61   def aquarium = Aquarium.Instance
  62
  63
  64   def propertyPrefix = Some(StoreProvider.Prefix)
  65
  66   /**
  67    * Configure this instance with the provided properties.
  68    *
  69    * If `propertyPrefix` is defined, then `props` contains only keys that start with the given prefix.
  70    */
  71   def configure(props: Props) = {
  72     this._reconnectPeriodMillis = props.getLongEx(StoreProvider.Keys.reconnect_period_millis)
  73   }
  74
  75   private[this] def safePingStore(tag: Tag,
  76                                   pinger: () ⇒ Any,
  77                                   getStatus: () ⇒ Boolean,
  78                                   setStatus: (Boolean) ⇒ Any): Unit = {
  79     try {
  80       val wasAlive = getStatus()
  81       pinger()
  82       // No exception happened, so we are alive
  83       setStatus(true)
  84       if(!wasAlive) {
  85         logger.info("Reconnected %s store".format(tag))
  86         aquarium.eventBus ! StoreIsAliveBusEvent(tag)
  87       }
  88     }
  89     catch {
  90       case e: Throwable ⇒
  91         setStatus(false)
  92         logger.info("Store %s detected down".format(tag))
  93         aquarium.eventBus ! StoreIsDeadBusEvent(tag)
  94     }
  95   }
  96
  97   private[this] def doSchedulePing(tag: Tag,
  98                                    info: String,
  99                                    pinger: () ⇒ Any,
 100                                    getStatus: () ⇒ Boolean,
 101                                    setStatus: (Boolean) ⇒ Any): Unit = {
 102     aquarium.timerService.scheduleOnce(
 103       info,
 104       {
 105         if(!aquarium.isStopping() && !_stopped.get()) {
 106 //          logger.debug("Pinging %s store".format(tag))
 107           safePingStore(tag, pinger, getStatus, setStatus)
 108
 109           doSchedulePing(tag, info, pinger, getStatus, setStatus)
 110         }
 111       },
 112       this._reconnectPeriodMillis,
 113       true
 114     )
 115   }
 116
 117   def pingResourceEventStore(): Unit = {
 118     val tag = Tags.ResourceEventTag
 119
 120     logger.info("Scheduling ping for %s store".format(tag))
 121
 122     doSchedulePing(
 123       tag,
 124       aquarium.resourceEventStore.toString,
 125       () ⇒ aquarium.resourceEventStore.pingResourceEventStore(),
 126       () ⇒ _rcIsAlive.get(),
 127       alive ⇒ _rcIsAlive.set(alive)
 128     )
 129   }
 130
 131   def pingIMEventStore(): Unit = {
 132     val tag = Tags.IMEventTag
 133
 134     logger.info("Scheduling ping for %s store".format(tag))
 135
 136     doSchedulePing(
 137       tag,
 138       aquarium.imEventStore.toString,
 139       () ⇒ aquarium.imEventStore.pingIMEventStore(),
 140       () ⇒ _imIsAlive.get(),
 141       alive ⇒ _imIsAlive.set(alive)
 142     )
 143   }
 144
 145   def start(): Unit = {
 146     if(!_pingIsScheduled.get()) {
 147       // First time pings (once)
 148       safeUnit {
 149         safePingStore(
 150           Tags.ResourceEventTag,
 151           () ⇒ aquarium.resourceEventStore.pingResourceEventStore(),
 152           () ⇒ _rcIsAlive.get(),
 153           alive ⇒ _rcIsAlive.set(alive)
 154         )
 155       }
 156
 157       safeUnit {
 158         safePingStore(
 159           Tags.IMEventTag,
 160           () ⇒ aquarium.resourceEventStore.pingResourceEventStore(),
 161           () ⇒ _imIsAlive.get(),
 162           alive ⇒ _imIsAlive.set(alive)
 163         )
 164       }
 165
 166       pingResourceEventStore()
 167       pingIMEventStore()
 168
 169       _stopped.set(false)
 170     }
 171   }
 172
 173   def stop(): Unit = {
 174     _stopped.set(true)
 175   }
 176 }