runtime/org.argeo.slc.server/src/main/java/org/argeo/slc/services/impl/AgentServiceImpl.java

   1 /*
   2  * Copyright (C) 2007-2012 Mathieu Baudier
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *         http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package org.argeo.slc.services.impl;
  17
  18 import java.util.ArrayList;
  19 import java.util.List;
  20 import java.util.concurrent.Executor;
  21
  22 import org.apache.commons.logging.Log;
  23 import org.apache.commons.logging.LogFactory;
  24 import org.argeo.slc.dao.runtime.SlcAgentDescriptorDao;
  25 import org.argeo.slc.runtime.SlcAgent;
  26 import org.argeo.slc.runtime.SlcAgentDescriptor;
  27 import org.argeo.slc.runtime.SlcAgentFactory;
  28 import org.argeo.slc.services.AgentService;
  29 import org.springframework.beans.factory.DisposableBean;
  30 import org.springframework.beans.factory.InitializingBean;
  31
  32 public class AgentServiceImpl implements AgentService, InitializingBean,
  33                 DisposableBean {
  34         private final static Log log = LogFactory.getLog(AgentServiceImpl.class);
  35
  36         private final SlcAgentDescriptorDao slcAgentDescriptorDao;
  37         private final SlcAgentFactory agentFactory;
  38
  39         private Executor systemExecutor;
  40
  41         private Long pingCycle = 20000l;
  42
  43         private Boolean pingThreadActive = true;
  44
  45         public AgentServiceImpl(SlcAgentDescriptorDao slcAgentDescriptorDao,
  46                         SlcAgentFactory agentFactory) {
  47                 this.slcAgentDescriptorDao = slcAgentDescriptorDao;
  48                 this.agentFactory = agentFactory;
  49         }
  50
  51         public void register(SlcAgentDescriptor slcAgentDescriptor) {
  52                 if (slcAgentDescriptorDao.getAgentDescriptor(slcAgentDescriptor
  53                                 .getUuid()) == null)
  54                         slcAgentDescriptorDao.create(slcAgentDescriptor);
  55                 log.info("Registered agent #" + slcAgentDescriptor.getUuid());
  56         }
  57
  58         public void unregister(SlcAgentDescriptor slcAgentDescriptor) {
  59                 slcAgentDescriptorDao.delete(slcAgentDescriptor);
  60                 log.info("Unregistered agent #" + slcAgentDescriptor.getUuid());
  61         }
  62
  63         public void afterPropertiesSet() throws Exception {
  64                 // if (pingCycle > 0)
  65                 // new PingThread().start();
  66                 if (pingCycle > 0) {
  67                         Thread authenticatedThread = new Thread("SLC Agents Ping") {
  68                                 public void run() {
  69                                         systemExecutor.execute(new AgentsPing());
  70                                 }
  71                         };
  72                         authenticatedThread.start();
  73
  74                 }
  75         }
  76
  77         public synchronized void destroy() throws Exception {
  78                 pingThreadActive = false;
  79                 notifyAll();
  80         }
  81
  82         public void setPingCycle(Long pingCycle) {
  83                 this.pingCycle = pingCycle;
  84         }
  85
  86         public void setSystemExecutor(Executor securityService) {
  87                 this.systemExecutor = securityService;
  88         }
  89
  90         protected class AgentsPing implements Runnable {
  91                 public void run() {
  92
  93                         // FIXME: temporary hack so that the ping starts after the server
  94                         // has been properly started.
  95                         try {
  96                                 Thread.sleep(10 * 1000);
  97                         } catch (InterruptedException e1) {
  98                                 // silent
  99                         }
 100
 101                         log.info("Start pinging agents.");
 102                         while (pingThreadActive) {
 103                                 List<SlcAgentDescriptor> lst = slcAgentDescriptorDao
 104                                                 .listSlcAgentDescriptors();
 105                                 List<String> agentIds = new ArrayList<String>();
 106                                 for (SlcAgentDescriptor ad : lst)
 107                                         agentIds.add(ad.getUuid());
 108
 109                                 if (log.isTraceEnabled())
 110                                         log.trace("Ping " + agentIds.size() + " agent(s).");
 111                                 for (String agentId : agentIds) {
 112                                         SlcAgent agent = agentFactory.getAgent(agentId);
 113                                         if (!agent.ping()) {
 114                                                 log.info("Agent " + agentId + " did not reply to ping,"
 115                                                                 + " removing its descriptor...");
 116                                                 slcAgentDescriptorDao.delete(agentId);
 117                                         }
 118                                 }
 119
 120                                 lst = slcAgentDescriptorDao.listSlcAgentDescriptors();
 121                                 agentIds = new ArrayList<String>();
 122                                 for (SlcAgentDescriptor ad : lst)
 123                                         agentIds.add(ad.getUuid());
 124                                 agentFactory.pingAll(agentIds);
 125
 126                                 synchronized (AgentServiceImpl.this) {
 127                                         try {
 128                                                 AgentServiceImpl.this.wait(pingCycle);
 129                                         } catch (InterruptedException e) {
 130                                                 // silent
 131                                         }
 132                                 }
 133                         }
 134                         log.info("Stopped pinging agents.");
 135                 }
 136
 137         }
 138
 139 }