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