AgentServiceImpl.java

   1 package org.argeo.slc.services.impl.runtime;
   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.slc.dao.runtime.SlcAgentDescriptorDao;
   9 import org.argeo.slc.runtime.SlcAgent;
  10 import org.argeo.slc.runtime.SlcAgentDescriptor;
  11 import org.argeo.slc.runtime.SlcAgentFactory;
  12 import org.argeo.slc.services.runtime.AgentService;
  13 import org.springframework.beans.factory.DisposableBean;
  14 import org.springframework.beans.factory.InitializingBean;
  15
  16 public class AgentServiceImpl implements AgentService, InitializingBean,
  17                 DisposableBean {
  18         private final static Log log = LogFactory.getLog(AgentServiceImpl.class);
  19
  20         private final SlcAgentDescriptorDao slcAgentDescriptorDao;
  21         private final SlcAgentFactory agentFactory;
  22
  23         private Long pingCycle = 60000l;
  24
  25         private Boolean pingThreadActive = true;
  26
  27         public AgentServiceImpl(SlcAgentDescriptorDao slcAgentDescriptorDao,
  28                         SlcAgentFactory agentFactory) {
  29                 this.slcAgentDescriptorDao = slcAgentDescriptorDao;
  30                 this.agentFactory = agentFactory;
  31         }
  32
  33         public void register(SlcAgentDescriptor slcAgentDescriptor) {
  34                 slcAgentDescriptorDao.create(slcAgentDescriptor);
  35                 log.info("Registered agent #" + slcAgentDescriptor.getUuid());
  36         }
  37
  38         public void unregister(SlcAgentDescriptor slcAgentDescriptor) {
  39                 slcAgentDescriptorDao.delete(slcAgentDescriptor);
  40                 log.info("Unregistered agent #" + slcAgentDescriptor.getUuid());
  41         }
  42
  43         public void afterPropertiesSet() throws Exception {
  44                 if (pingCycle > 0)
  45                         new PingThread().start();
  46         }
  47
  48         public synchronized void destroy() throws Exception {
  49                 pingThreadActive = false;
  50                 notifyAll();
  51         }
  52
  53         public void setPingCycle(Long pingCycle) {
  54                 this.pingCycle = pingCycle;
  55         }
  56
  57         protected class PingThread extends Thread {
  58                 public void run() {
  59                         log.info("Start pinging agents.");
  60                         while (pingThreadActive) {
  61                                 List<SlcAgentDescriptor> lst = slcAgentDescriptorDao
  62                                                 .listSlcAgentDescriptors();
  63                                 List<String> agentIds = new ArrayList<String>();
  64                                 for (SlcAgentDescriptor ad : lst)
  65                                         agentIds.add(ad.getUuid());
  66
  67                                 if (log.isTraceEnabled())
  68                                         log.debug("Ping " + agentIds.size() + " agent.");
  69                                 for (String agentId : agentIds) {
  70                                         SlcAgent agent = agentFactory.getAgent(agentId);
  71                                         if (!agent.ping()) {
  72                                                 log.info("Agent " + agentId + " did not reply to ping,"
  73                                                                 + " removing its descriptor...");
  74                                                 slcAgentDescriptorDao.delete(agentId);
  75                                         }
  76                                 }
  77
  78                                 lst = slcAgentDescriptorDao.listSlcAgentDescriptors();
  79                                 agentIds = new ArrayList<String>();
  80                                 for (SlcAgentDescriptor ad : lst)
  81                                         agentIds.add(ad.getUuid());
  82                                 agentFactory.pingAll(agentIds);
  83
  84                                 synchronized (AgentServiceImpl.this) {
  85                                         try {
  86                                                 AgentServiceImpl.this.wait(pingCycle);
  87                                         } catch (InterruptedException e) {
  88                                                 // silent
  89                                         }
  90                                 }
  91                         }
  92                         log.info("Stopped pinging agents.");
  93                 }
  94
  95         }
  96 }