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 java.util.concurrent.atomic.AtomicBoolean
  40 import gr.grnet.aquarium.service.event.{StoreIsAliveBusEvent, StoreIsDeadBusEvent}
  41 import gr.grnet.aquarium.{AquariumAwareSkeleton, Configurable, Aquarium}
  42 import com.ckkloverdos.props.Props
  43 import gr.grnet.aquarium.store.StoreProvider
  44
  45 /**
  46  * Watches for liveliness of stores.
  47  *
  48  * @author Christos KK Loverdos <loverdos@gmail.com>
  49  */
  50
  51 final class StoreWatcherService extends Lifecycle with Configurable with AquariumAwareSkeleton with Loggable {
  52   private[this] var _reconnectPeriodMillis = 1000L
  53
  54   private[this] val _pingIsScheduled = new AtomicBoolean(false)
  55   private[this] val _stopped = new AtomicBoolean(false)
  56   private[this] val _rcIsAlive = new AtomicBoolean(true)
  57   private[this] val _imIsAlive = new AtomicBoolean(true)
  58
  59   def propertyPrefix = Some(StoreProvider.Prefix)
  60
  61   /**
  62    * Configure this instance with the provided properties.
  63    *
  64    * If `propertyPrefix` is defined, then `props` contains only keys that start with the given prefix.
  65    */
  66   def configure(props: Props) = {
  67     this._reconnectPeriodMillis = props.getLongEx(StoreProvider.Keys.reconnect_period_millis)
  68   }
  69
  70   private[this] def safePingStore(
  71       tag: Tag,
  72       pinger: () ⇒ Any,
  73       getStatus: () ⇒ Boolean,
  74       setStatus: (Boolean) ⇒ Any
  75   ): Unit = {
  76
  77     try {
  78       val wasAlive = getStatus()
  79       pinger()
  80       // No exception happened, so we are alive
  81       setStatus(true)
  82       if(!wasAlive) {
  83         logger.info("Reconnected store for %s".format(tag))
  84         aquarium.eventBus ! StoreIsAliveBusEvent(tag)
  85       }
  86     }
  87     catch {
  88       case e: Throwable ⇒
  89         setStatus(false)
  90         logger.info("Store for %s detected down".format(tag))
  91         aquarium.eventBus ! StoreIsDeadBusEvent(tag)
  92     }
  93   }
  94
  95   private[this] def doSchedulePing(
  96       tag: Tag,
  97       info: String,
  98       pinger: () ⇒ Any,
  99       getStatus: () ⇒ Boolean,
 100       setStatus: (Boolean) ⇒ Any
 101   ): Unit = {
 102
 103     aquarium.timerService.scheduleOnce(
 104       info,
 105       {
 106         if(!aquarium.isStopping() && !_stopped.get()) {
 107 //          logger.debug("Pinging %s store".format(tag))
 108           safePingStore(tag, pinger, getStatus, setStatus)
 109
 110           doSchedulePing(tag, info, pinger, getStatus, setStatus)
 111         }
 112       },
 113       this._reconnectPeriodMillis,
 114       true
 115     )
 116   }
 117
 118   def pingResourceEventStore(): Unit = {
 119     val tag = Tags.ResourceEventTag
 120
 121     logger.info("Scheduling ping for %s store".format(tag))
 122
 123     doSchedulePing(
 124       tag,
 125       aquarium.resourceEventStore.toString,
 126       () ⇒ aquarium.resourceEventStore.pingResourceEventStore(),
 127       () ⇒ _rcIsAlive.get(),
 128       alive ⇒ _rcIsAlive.set(alive)
 129     )
 130   }
 131
 132   def pingIMEventStore(): Unit = {
 133     val tag = Tags.IMEventTag
 134
 135     logger.info("Scheduling ping for %s store".format(tag))
 136
 137     doSchedulePing(
 138       tag,
 139       aquarium.imEventStore.toString,
 140       () ⇒ aquarium.imEventStore.pingIMEventStore(),
 141       () ⇒ _imIsAlive.get(),
 142       alive ⇒ _imIsAlive.set(alive)
 143     )
 144   }
 145
 146   def start(): Unit = {
 147     if(!_pingIsScheduled.get()) {
 148       // First time pings (once)
 149       safePingStore(
 150         Tags.ResourceEventTag,
 151         () ⇒ aquarium.resourceEventStore.pingResourceEventStore(),
 152         () ⇒ _rcIsAlive.get(),
 153         alive ⇒ _rcIsAlive.set(alive)
 154       )
 155
 156       safePingStore(
 157         Tags.IMEventTag,
 158         () ⇒ aquarium.resourceEventStore.pingResourceEventStore(),
 159         () ⇒ _imIsAlive.get(),
 160         alive ⇒ _imIsAlive.set(alive)
 161       )
 162
 163       pingResourceEventStore()
 164       pingIMEventStore()
 165
 166       _stopped.set(false)
 167     }
 168   }
 169
 170   def stop(): Unit = {
 171     _stopped.set(true)
 172   }
 173 }