runtime/org.argeo.slc.detached/src/main/java/org/argeo/slc/detached/drivers/FileDriver.java

   1 package org.argeo.slc.detached.drivers;
   2
   3 import java.io.File;
   4 import java.io.FileFilter;
   5 import java.io.FileInputStream;
   6 import java.io.FileOutputStream;
   7 import java.io.IOException;
   8 import java.io.ObjectInputStream;
   9 import java.io.ObjectOutputStream;
  10 import java.text.SimpleDateFormat;
  11 import java.util.Date;
  12
  13 import javax.xml.transform.stream.StreamResult;
  14 import javax.xml.transform.stream.StreamSource;
  15
  16 import org.apache.commons.io.FileUtils;
  17 import org.apache.commons.io.FilenameUtils;
  18 import org.apache.commons.io.IOUtils;
  19 import org.apache.commons.io.filefilter.NotFileFilter;
  20 import org.apache.commons.io.filefilter.SuffixFileFilter;
  21 import org.apache.commons.logging.Log;
  22 import org.apache.commons.logging.LogFactory;
  23 import org.argeo.slc.detached.DetachedAnswer;
  24 import org.argeo.slc.detached.DetachedClient;
  25 import org.argeo.slc.detached.DetachedCommunication;
  26 import org.argeo.slc.detached.DetachedException;
  27 import org.argeo.slc.detached.DetachedRequest;
  28 import org.springframework.beans.factory.InitializingBean;
  29
  30 public class FileDriver extends AbstractDriver implements DetachedClient,
  31                 InitializingBean {
  32         private final static Log log = LogFactory.getLog(FileDriver.class);
  33         private final static SimpleDateFormat sdf = new SimpleDateFormat(
  34                         "yyMMdd_HHmmss_SSS");
  35
  36         private File baseDir;
  37         private File requestsDir;
  38         private File answersDir;
  39         private File processedRequestsDir;
  40         private File processedAnswersDir;
  41         private File cleanedRequestsDir;
  42         private File cleanedAnswersDir;
  43
  44         private String lockFileExt = "lck";
  45         private FileFilter notLockFileFilter = new NotFileFilter(
  46                         new SuffixFileFilter("." + lockFileExt));
  47
  48         public synchronized DetachedRequest receiveRequest() throws Exception {
  49                 DetachedRequest request = (DetachedRequest) receiveFile(requestsDir,
  50                                 processedRequestsDir, 0);
  51                 if (request != null)
  52                         if (log.isTraceEnabled())
  53                                 log.trace("Received detached request #" + request.getUuid()
  54                                                 + " for ref '" + request.getRef() + "', path="
  55                                                 + request.getPath());
  56                 return request;
  57         }
  58
  59         public void sendAnswer(DetachedAnswer answer) throws Exception {
  60                 sendFile(answersDir, answer);
  61                 if (log.isTraceEnabled())
  62                         log.trace("Sent     detached answer  #" + answer.getUuid());
  63         }
  64
  65         public DetachedAnswer receiveAnswer() throws Exception {
  66                 DetachedAnswer answer = (DetachedAnswer) receiveFile(answersDir,
  67                                 processedAnswersDir, getReceiveAnswerTimeout());
  68                 if (answer != null)
  69                         if (log.isTraceEnabled())
  70                                 log.trace("Received detached answer  #" + answer.getUuid());
  71                 return answer;
  72         }
  73
  74         public void sendRequest(DetachedRequest request) throws Exception {
  75                 sendFile(requestsDir, request);
  76                 if (log.isTraceEnabled())
  77                         log.trace("Sent     detached request #" + request.getUuid()
  78                                         + " for ref '" + request.getRef() + "', path="
  79                                         + request.getPath());
  80         }
  81
  82         protected void sendFile(File dir, DetachedCommunication detCom)
  83                         throws Exception {
  84                 final File file;
  85                 if (getXmlConverter() != null)
  86                         file = new File(dir.getPath() + File.separator
  87                                         + sdf.format(new Date()) + '-' + detCom.getUuid() + ".xml");
  88                 else
  89                         file = new File(dir.getPath() + File.separator + detCom.getUuid());
  90
  91                 File lockFile = createLockFile(file);
  92                 if (getXmlConverter() != null) {
  93                         FileOutputStream outFile = new FileOutputStream(file);
  94                         try {
  95                                 StreamResult result = new StreamResult(outFile);
  96                                 getXmlConverter().marshallCommunication(detCom, result);
  97                         } finally {
  98                                 IOUtils.closeQuietly(outFile);
  99                         }
 100                 } else {
 101                         ObjectOutputStream out = new ObjectOutputStream(
 102                                         new FileOutputStream(file));
 103                         try {
 104                                 out.writeObject(detCom);
 105                         } finally {
 106                                 IOUtils.closeQuietly(out);
 107                         }
 108                 }
 109                 lockFile.delete();
 110         }
 111
 112         /**
 113          * @param timeout
 114          *            in ms, 0 is no timeout
 115          */
 116         protected synchronized DetachedCommunication receiveFile(File dir,
 117                         File processedDir, long timeout) throws Exception {
 118                 long begin = System.currentTimeMillis();
 119                 File file = null;
 120                 while (file == null && isActive()) {
 121                         if (!dir.exists())
 122                                 throw new DetachedException("Dir " + dir + " does not exist.");
 123
 124                         File[] files = dir.listFiles(notLockFileFilter);
 125                         if (files.length > 0)
 126                                 file = files[0];
 127                         else {
 128                                 try {
 129                                         wait(100);
 130                                 } catch (InterruptedException e) {
 131                                         // silent
 132                                 }
 133                         }
 134
 135                         long duration = System.currentTimeMillis() - begin;
 136                         if (timeout != 0 && duration > timeout) {
 137                                 throw new DetachedException("Receive file timed out after "
 138                                                 + duration + "ms.");
 139                         }
 140                 }
 141
 142                 if (!isActive())
 143                         return null;
 144
 145                 File lockFile = nameLockFile(file);
 146                 while (lockFile.exists())
 147                         // FIXME: implements time out
 148                         Thread.sleep(100);
 149
 150                 // Read the file
 151                 final DetachedCommunication detCom;
 152                 if (FilenameUtils.getExtension(file.getName()).equals("xml")) {
 153                         if (getXmlConverter() == null)
 154                                 throw new DetachedException("No XML converter defined.");
 155                         FileInputStream in = new FileInputStream(file);
 156                         try {
 157                                 StreamSource source = new StreamSource(in);
 158                                 detCom = getXmlConverter().unmarshallCommunication(source);
 159                         } finally {
 160                                 IOUtils.closeQuietly(in);
 161                         }
 162                 } else {
 163                         ObjectInputStream in = new ObjectInputStream(new FileInputStream(
 164                                         file));
 165                         try {
 166                                 detCom = (DetachedCommunication) in.readObject();
 167                         } finally {
 168                                 IOUtils.closeQuietly(in);
 169                         }
 170                 }
 171                 // Move to processed dir
 172                 FileUtils.moveFileToDirectory(file, processedDir, false);
 173                 return detCom;
 174         }
 175
 176         protected File createLockFile(File file) {
 177                 File lockFile = nameLockFile(file);
 178                 try {
 179                         lockFile.createNewFile();
 180                 } catch (IOException e) {
 181                         throw new DetachedException("Cannot create lock file " + lockFile);
 182                 }
 183                 return lockFile;
 184         }
 185
 186         protected File nameLockFile(File file) {
 187                 return new File(file.getAbsolutePath() + "." + lockFileExt);
 188         }
 189
 190         public void setBaseDir(File baseDir) {
 191                 this.baseDir = baseDir;
 192         }
 193
 194         private void createIfNotExist(File dir) {
 195                 if (!dir.exists()) {
 196                         log.warn("Dir " + dir.getAbsolutePath()
 197                                         + " does not exist. Creating it...");
 198                         dir.mkdirs();
 199                 }
 200         }
 201
 202         public void afterPropertiesSet() throws Exception {
 203                 this.requestsDir = new File(baseDir.getAbsolutePath() + File.separator
 204                                 + "requests");
 205                 this.answersDir = new File(baseDir.getAbsolutePath() + File.separator
 206                                 + "answers");
 207                 this.processedRequestsDir = new File(baseDir.getAbsolutePath()
 208                                 + File.separator + "processed" + File.separator + "requests");
 209                 this.processedAnswersDir = new File(baseDir.getAbsolutePath()
 210                                 + File.separator + "processed" + File.separator + "answers");
 211                 this.cleanedRequestsDir = new File(baseDir.getAbsolutePath()
 212                                 + File.separator + "cleaned" + File.separator + "requests");
 213                 this.cleanedAnswersDir = new File(baseDir.getAbsolutePath()
 214                                 + File.separator + "cleaned" + File.separator + "answers");
 215
 216                 createIfNotExist(requestsDir);
 217                 createIfNotExist(answersDir);
 218                 createIfNotExist(processedRequestsDir);
 219                 createIfNotExist(processedAnswersDir);
 220                 createIfNotExist(cleanedRequestsDir);
 221                 createIfNotExist(cleanedAnswersDir);
 222                 log.info("Detached File Driver initialized on " + baseDir);
 223         }
 224
 225         public void cleanPreviousRuns() throws Exception {
 226
 227                 // Clean requests and answers from previous builds
 228                 File[] remainingRequests = requestsDir.listFiles();
 229                 for (int i = 0; i < remainingRequests.length; i++) {
 230                         FileUtils.moveFileToDirectory(remainingRequests[i],
 231                                         cleanedRequestsDir, false);
 232                 }
 233
 234                 File[] remainingAnswers = answersDir.listFiles();
 235                 for (int i = 0; i < remainingAnswers.length; i++) {
 236                         FileUtils.moveFileToDirectory(remainingAnswers[i],
 237                                         cleanedAnswersDir, false);
 238                 }
 239                 log.info("Clean previous runs of File Driver on " + baseDir);
 240
 241         }
 242
 243 }