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

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