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

   1 package org.argeo.slc.services.impl;
   2
   3 import java.util.ArrayList;
   4 import java.util.List;
   5
   6 import org.apache.commons.logging.Log;
   7 import org.apache.commons.logging.LogFactory;
   8 import org.argeo.security.ArgeoSecurityService;
   9 import org.argeo.slc.dao.runtime.SlcAgentDescriptorDao;
  10 import org.argeo.slc.runtime.SlcAgent;
  11 import org.argeo.slc.runtime.SlcAgentDescriptor;
  12 import org.argeo.slc.runtime.SlcAgentFactory;
  13 import org.argeo.slc.services.AgentService;
  14 import org.springframework.beans.factory.DisposableBean;
  15 import org.springframework.beans.factory.InitializingBean;
  16
  17 public class AgentServiceImpl implements AgentService, InitializingBean,
  18                 DisposableBean {
  19         private final static Log log = LogFactory.getLog(AgentServiceImpl.class);
  20
  21         private final SlcAgentDescriptorDao slcAgentDescriptorDao;
  22         private final SlcAgentFactory agentFactory;
  23
  24         private ArgeoSecurityService securityService;
  25
  26         private Long pingCycle = 20000l;
  27
  28         private Boolean pingThreadActive = true;
  29
  30         public AgentServiceImpl(SlcAgentDescriptorDao slcAgentDescriptorDao,
  31                         SlcAgentFactory agentFactory) {
  32                 this.slcAgentDescriptorDao = slcAgentDescriptorDao;
  33                 this.agentFactory = agentFactory;
  34         }
  35
  36         public void register(SlcAgentDescriptor slcAgentDescriptor) {
  37                 if (slcAgentDescriptorDao.getAgentDescriptor(slcAgentDescriptor
  38                                 .getUuid()) == null)
  39                         slcAgentDescriptorDao.create(slcAgentDescriptor);
  40                 log.info("Registered agent #" + slcAgentDescriptor.getUuid());
  41         }
  42
  43         public void unregister(SlcAgentDescriptor slcAgentDescriptor) {
  44                 slcAgentDescriptorDao.delete(slcAgentDescriptor);
  45                 log.info("Unregistered agent #" + slcAgentDescriptor.getUuid());
  46         }
  47
  48         public void afterPropertiesSet() throws Exception {
  49                 // if (pingCycle > 0)
  50                 // new PingThread().start();
  51                 if (pingCycle > 0) {
  52                         Thread authenticatedThread = new Thread(securityService
  53                                         .wrapWithSystemAuthentication(new AgentsPing()),
  54                                         "SLC Agents Ping");
  55                         authenticatedThread.start();
  56
  57                 }
  58         }
  59
  60         public synchronized void destroy() throws Exception {
  61                 pingThreadActive = false;
  62                 notifyAll();
  63         }
  64
  65         public void setPingCycle(Long pingCycle) {
  66                 this.pingCycle = pingCycle;
  67         }
  68
  69         public void setSecurityService(ArgeoSecurityService securityService) {
  70                 this.securityService = securityService;
  71         }
  72
  73         protected class AgentsPing implements Runnable {
  74                 public void run() {
  75
  76                         // FIXME: temporary hack so that the ping starts after the server
  77                         // has been properly started.
  78                         try {
  79                                 Thread.sleep(10 * 1000);
  80                         } catch (InterruptedException e1) {
  81                                 // silent
  82                         }
  83
  84                         log.info("Start pinging agents.");
  85                         while (pingThreadActive) {
  86                                 List<SlcAgentDescriptor> lst = slcAgentDescriptorDao
  87                                                 .listSlcAgentDescriptors();
  88                                 List<String> agentIds = new ArrayList<String>();
  89                                 for (SlcAgentDescriptor ad : lst)
  90                                         agentIds.add(ad.getUuid());
  91
  92                                 if (log.isTraceEnabled())
  93                                         log.debug("Ping " + agentIds.size() + " agent.");
  94                                 for (String agentId : agentIds) {
  95                                         SlcAgent agent = agentFactory.getAgent(agentId);
  96                                         if (!agent.ping()) {
  97                                                 log.info("Agent " + agentId + " did not reply to ping,"
  98                                                                 + " removing its descriptor...");
  99                                                 slcAgentDescriptorDao.delete(agentId);
 100                                         }
 101                                 }
 102
 103                                 lst = slcAgentDescriptorDao.listSlcAgentDescriptors();
 104                                 agentIds = new ArrayList<String>();
 105                                 for (SlcAgentDescriptor ad : lst)
 106                                         agentIds.add(ad.getUuid());
 107                                 agentFactory.pingAll(agentIds);
 108
 109                                 synchronized (AgentServiceImpl.this) {
 110                                         try {
 111                                                 AgentServiceImpl.this.wait(pingCycle);
 112                                         } catch (InterruptedException e) {
 113                                                 // silent
 114                                         }
 115                                 }
 116                         }
 117                         log.info("Stopped pinging agents.");
 118                 }
 119
 120         }
 121
 122 }