Coverage Report

Coverage Report - org.kuali.rice.kew.batch.XmlPollerServiceImpl

Classes in this File

Line Coverage

Branch Coverage

Complexity

XmlPollerServiceImpl

0/127

0/56

3.105

 /*
  * Copyright 2005-2007 The Kuali Foundation
  * 
  * 
  * Licensed under the Educational Community License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  * 
  * http://www.opensource.org/licenses/ecl2.php
  * 
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.kuali.rice.kew.batch;
 
 import org.kuali.rice.kew.service.KEWServiceLocator;
 
 import java.io.*;
 import java.text.Format;
 import java.text.SimpleDateFormat;
 import java.util.*;
 
 
 /**
  * Utility class responsible for polling and ingesting XML data files
  * containing various forms of workflow engine data (e.g. document types
  * and rules).
  * Loaded files and problem files are placed into a subdirectory of a
  * configured 'loaded' and 'problem' directory, respectively.
  * "Problem-ness" is determined by inspecting a 'processed' flag on each <code>XmlDoc</code>
  * in each collection.  If not all <code>XmlDoc</code>s are marked 'processed' an
  * error is assumed, and the collection file (e.g. for a Zip, the Zip file) is moved
  * to the 'problem' directory.
  * As such, it is the <b><code>XmlIngesterService</code>'s responsibility</b> to mark
  * any unknown or otherwise innocuous non-failure non-processed files, as 'processed'.
  * A different mechanism can be developed if this proves to be a problem, but for now
  * it is simple enough for the <code>XmlIngesterService</code> to determine this.
  * @see org.kuali.rice.kew.batch.XmlPollerService
  * @see org.kuali.rice.kew.batch.XmlIngesterServiceImpl
  * @author Kuali Rice Team (rice.collab@kuali.org)
  */
 public class XmlPollerServiceImpl implements XmlPollerService {
 
     private static final org.apache.log4j.Logger LOG = org.apache.log4j.Logger
             .getLogger(XmlPollerServiceImpl.class);
     private static final Format DIR_FORMAT = new SimpleDateFormat("yyyy-MM-dd_HH-mm-ss-SSS");
 
     /**
      * Specifies the polling interval that should be used with this task.
      */
     private int pollIntervalSecs = 5 * 60; // default to 5 minutes
     /**
      * Specifies the initial delay the poller should wait before starting to poll
      */
     private int initialDelaySecs = 30; // default to 30 seconds
     /**
      * Location in which to find XML files to load.
      */
     private String xmlPendingLocation;
     /**
      * Location in which to place successfully loaded XML files.
      */
     private String xmlCompletedLocation;
     /**
      * Location in which to place XML files which have failed to load.
      */
     private String xmlProblemLocation;
     
     private String xmlParentDirectory;
     private static final String PENDING_MOVE_FAILED_ARCHIVE_FILE = "movesfailed";
     private static final String NEW_LINE = "\n";
 
     public void run() {
         // if(!directoriesWritable()){
         //     LOG.error("Error writing to xml data directories. Stopping xmlLoader ...");
         //     this.cancel();
         // }
         LOG.debug("checking for xml data files...");
         File[] files = getXmlPendingDir().listFiles();
         if (files == null || files.length == 0) {
                 return;
         }
         LOG.info("Found " + files.length + " files to ingest.");
         List<XmlDocCollection> collections = new ArrayList<XmlDocCollection>();
         for (File file : files)
         {
             if (file.isDirectory())
             {
                 collections.add(new DirectoryXmlDocCollection(file));
             } else if (file.getName().equals(PENDING_MOVE_FAILED_ARCHIVE_FILE))
             {
                 // the movesfailed file...ignore this
                 continue;
             } else if (file.getName().toLowerCase().endsWith(".zip"))
             {
                 try
                 {
                     collections.add(new ZipXmlDocCollection(file));
                 } catch (IOException ioe)
                 {
                     LOG.error("Unable to load file: " + file);
                 }
             } else if (file.getName().endsWith(".xml"))
             {
                 collections.add(new FileXmlDocCollection(file));
             } else
             {
                 LOG.warn("Ignoring extraneous file in xml pending directory: " + file);
             }
         }
 
         // Cull any resources which were already processed and whose moves failed
         Iterator collectionsIt = collections.iterator();
         Collection<XmlDocCollection> culled = new ArrayList<XmlDocCollection>();
         while (collectionsIt.hasNext()) {
             XmlDocCollection container = (XmlDocCollection) collectionsIt.next();
             // if a move has already failed for this archive, ignore it
             if (inPendingMoveFailedArchive(container.getFile())) {
                 LOG.info("Ignoring previously processed resource: " + container);
                 culled.add(container);
             }
         }
         collections.removeAll(culled);
 
         if (collections.size() == 0) {
             LOG.debug("No valid new resources found to ingest");
             return;
         }
 
         Date LOAD_TIME = Calendar.getInstance().getTime();
         // synchronization around date format should not be an issue as this code is single-threaded
         File completeDir = new File(getXmlCompleteDir(), DIR_FORMAT.format(LOAD_TIME));
         File failedDir = new File(getXmlProblemDir(), DIR_FORMAT.format(LOAD_TIME));
 
         // now ingest the containers
         Collection failed = null;
         try {
             failed = KEWServiceLocator.getXmlIngesterService().ingest(collections);
         } catch (Exception e) {
             LOG.error("Error ingesting data", e);
             //throw new RuntimeException(e);
         }
     
         // now iterate through all containers again, and move containers to approprate dir
         LOG.info("Moving files...");
         collectionsIt = collections.iterator();
         while (collectionsIt.hasNext()) {
             XmlDocCollection container = (XmlDocCollection) collectionsIt.next();
             LOG.debug("container: " + container);
             try {
                 // "close" the container
                 // this only matters for ZipFiles for now
                 container.close();
             } catch (IOException ioe) {
                 LOG.warn("Error closing " + container, ioe);
             }
             if (failed.contains(container)) {
                 // some docs must have failed, move the whole
                 // container to the failed dir
                 if (container.getFile() != null) {
                     LOG.error("Moving " + container.getFile() + " to problem dir.");
                     if ((!failedDir.isDirectory() && !failedDir.mkdirs())
                         || !moveFile(failedDir, container.getFile())) {
                         LOG.error("Could not move: " + container.getFile());
                         recordUnmovablePendingFile(container.getFile(), LOAD_TIME);         
                     }
                 }
             } else {
                 if (container.getFile() != null) {
                     LOG.info("Moving " + container.getFile() + " to loaded dir.");
                     if((!completeDir.isDirectory() && !completeDir.mkdirs())
                         || !moveFile(completeDir, container.getFile())){
                         LOG.error("Could not move: " + container.getFile());
                         recordUnmovablePendingFile(container.getFile(), LOAD_TIME);         
                     }
                 }
             }
         }
     }
 
     private boolean inPendingMoveFailedArchive(File xmlDataFile){
         if (xmlDataFile == null) return false;
         BufferedReader inFile = null;
         File movesFailedFile = new File(getXmlPendingDir(), PENDING_MOVE_FAILED_ARCHIVE_FILE);
         if (!movesFailedFile.isFile()) return false;
         try {
             inFile = new BufferedReader(new FileReader(movesFailedFile));
             String line;
             
             while((line = inFile.readLine()) != null){
                 String trimmedLine = line.trim();
                 if(trimmedLine.equals(xmlDataFile.getName()) ||
                    trimmedLine.startsWith(xmlDataFile.getName() + "=")) { 
                     return true;
                 }
             }
         } catch (IOException e){
             LOG.warn("Error reading file " + movesFailedFile);
             //TODO try reading the pending file or stop?
         } finally {
             if (inFile != null) try {
                 inFile.close();
             } catch (Exception e) {
                 LOG.warn("Error closing buffered reader for " + movesFailedFile);
             }
         }
       
         return false;
     }
 
     private boolean recordUnmovablePendingFile(File unMovablePendingFile, Date dateLoaded){
         boolean recorded = false;
         FileWriter archiveFile = null;
         try{
             archiveFile = new FileWriter(new File(getXmlPendingDir(), PENDING_MOVE_FAILED_ARCHIVE_FILE), true);  
             archiveFile.write(unMovablePendingFile.getName() + "=" + dateLoaded.getTime() + NEW_LINE);
             recorded = true;
         } catch (IOException e){
             LOG.error("Unable to record unmovable pending file " + unMovablePendingFile.getName() + "in the archive file " + PENDING_MOVE_FAILED_ARCHIVE_FILE);
         } finally {
             if (archiveFile != null) {
                 try {
                     archiveFile.close();
                 } catch (IOException ioe) {
                     LOG.error("Error closing unmovable pending file", ioe);
                 }
             }
         }
         return recorded;       
     }
 
     private boolean moveFile(File toDirectory, File fileToMove){
         boolean moved = true;
         if (!fileToMove.renameTo(new File(toDirectory.getPath(), fileToMove.getName()))){
             LOG.error("Unable to move file " + fileToMove.getName() + " to directory " + toDirectory.getPath());
             moved = false;
         }
         return moved;
     }
 
     private File getXmlPendingDir() {
         return new File(getXmlPendingLocation());
     }
 
     private File getXmlCompleteDir() {
         return new File(getXmlCompletedLocation());
     }
 
     private File getXmlProblemDir() {
         return new File(getXmlProblemLocation());
     }
 
     public String getXmlCompletedLocation() {
         return xmlCompletedLocation;
     }
 
     public void setXmlCompletedLocation(String xmlCompletedLocation) {
         this.xmlCompletedLocation = xmlCompletedLocation;
     }
 
     public String getXmlPendingLocation() {
         return xmlPendingLocation;
     }
 
     /*public boolean validate(File uploadedFile) {
         XmlDataLoaderFileFilter filter = new XmlDataLoaderFileFilter();
         return filter.accept(uploadedFile);
     }*/
 
     public void setXmlPendingLocation(String xmlPendingLocation) {
         this.xmlPendingLocation = xmlPendingLocation;
     }
 
     public String getXmlProblemLocation() {
         return xmlProblemLocation;
     }
 
     public void setXmlProblemLocation(String xmlProblemLocation) {
         this.xmlProblemLocation = xmlProblemLocation;
     }
     public String getXmlParentDirectory() {
         return xmlParentDirectory;
     }
     public void setXmlParentDirectory(String xmlDataParentDirectory) {
         this.xmlParentDirectory = xmlDataParentDirectory;
     }
 
     /**
      * Sets the polling interval time in seconds
      * @param seconds the polling interval time in seconds
      */
     public void setPollIntervalSecs(int seconds) {
         this.pollIntervalSecs = seconds;
     }
 
     /**
      * Gets the polling interval time in seconds
      * @return the polling interval time in seconds
      */
     public int getPollIntervalSecs() {
         return this.pollIntervalSecs;
     }
 
     /**
      * Sets the initial delay time in seconds
      * @param seconds the initial delay time in seconds
      */
     public void setInitialDelaySecs(int seconds) {
         this.initialDelaySecs = seconds;
     }
 
     /**
      * Gets the initial delay time in seconds
      * @return the initial delay time in seconds
      */
     public int getInitialDelaySecs() {
         return this.initialDelaySecs;
     }
 }

1		/*
2		* Copyright 2005-2007 The Kuali Foundation
3		*
4		*
5		* Licensed under the Educational Community License, Version 2.0 (the "License");
6		* you may not use this file except in compliance with the License.
7		* You may obtain a copy of the License at
8		*
9		* http://www.opensource.org/licenses/ecl2.php
10		*
11		* Unless required by applicable law or agreed to in writing, software
12		* distributed under the License is distributed on an "AS IS" BASIS,
13		* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14		* See the License for the specific language governing permissions and
15		* limitations under the License.
16		*/
17		package org.kuali.rice.kew.batch;
18
19		import org.kuali.rice.kew.service.KEWServiceLocator;
20
21		import java.io.*;
22		import java.text.Format;
23		import java.text.SimpleDateFormat;
24		import java.util.*;
25
26
27		/**
28		* Utility class responsible for polling and ingesting XML data files
29		* containing various forms of workflow engine data (e.g. document types
30		* and rules).
31		* Loaded files and problem files are placed into a subdirectory of a
32		* configured 'loaded' and 'problem' directory, respectively.
33		* "Problem-ness" is determined by inspecting a 'processed' flag on each <code>XmlDoc</code>
34		* in each collection. If not all <code>XmlDoc</code>s are marked 'processed' an
35		* error is assumed, and the collection file (e.g. for a Zip, the Zip file) is moved
36		* to the 'problem' directory.
37		* As such, it is the <b><code>XmlIngesterService</code>'s responsibility</b> to mark
38		* any unknown or otherwise innocuous non-failure non-processed files, as 'processed'.
39		* A different mechanism can be developed if this proves to be a problem, but for now
40		* it is simple enough for the <code>XmlIngesterService</code> to determine this.
41		* @see org.kuali.rice.kew.batch.XmlPollerService
42		* @see org.kuali.rice.kew.batch.XmlIngesterServiceImpl
43		* @author Kuali Rice Team (rice.collab@kuali.org)
44		*/
45	0	public class XmlPollerServiceImpl implements XmlPollerService {
46
47	0	private static final org.apache.log4j.Logger LOG = org.apache.log4j.Logger
48		.getLogger(XmlPollerServiceImpl.class);
49	0	private static final Format DIR_FORMAT = new SimpleDateFormat("yyyy-MM-dd_HH-mm-ss-SSS");
50
51		/**
52		* Specifies the polling interval that should be used with this task.
53		*/
54	0	private int pollIntervalSecs = 5 * 60; // default to 5 minutes
55		/**
56		* Specifies the initial delay the poller should wait before starting to poll
57		*/
58	0	private int initialDelaySecs = 30; // default to 30 seconds
59		/**
60		* Location in which to find XML files to load.
61		*/
62		private String xmlPendingLocation;
63		/**
64		* Location in which to place successfully loaded XML files.
65		*/
66		private String xmlCompletedLocation;
67		/**
68		* Location in which to place XML files which have failed to load.
69		*/
70		private String xmlProblemLocation;
71
72		private String xmlParentDirectory;
73		private static final String PENDING_MOVE_FAILED_ARCHIVE_FILE = "movesfailed";
74		private static final String NEW_LINE = "\n";
75
76		public void run() {
77		// if(!directoriesWritable()){
78		// LOG.error("Error writing to xml data directories. Stopping xmlLoader ...");
79		// this.cancel();
80		// }
81	0	LOG.debug("checking for xml data files...");
82	0	File[] files = getXmlPendingDir().listFiles();
83	0	if (files == null \|\| files.length == 0) {
84	0	return;
85		}
86	0	LOG.info("Found " + files.length + " files to ingest.");
87	0	List<XmlDocCollection> collections = new ArrayList<XmlDocCollection>();
88	0	for (File file : files)
89		{
90	0	if (file.isDirectory())
91		{
92	0	collections.add(new DirectoryXmlDocCollection(file));
93	0	} else if (file.getName().equals(PENDING_MOVE_FAILED_ARCHIVE_FILE))
94		{
95		// the movesfailed file...ignore this
96	0	continue;
97	0	} else if (file.getName().toLowerCase().endsWith(".zip"))
98		{
99		try
100		{
101	0	collections.add(new ZipXmlDocCollection(file));
102	0	} catch (IOException ioe)
103		{
104	0	LOG.error("Unable to load file: " + file);
105	0	}
106	0	} else if (file.getName().endsWith(".xml"))
107		{
108	0	collections.add(new FileXmlDocCollection(file));
109		} else
110		{
111	0	LOG.warn("Ignoring extraneous file in xml pending directory: " + file);
112		}
113		}
114
115		// Cull any resources which were already processed and whose moves failed
116	0	Iterator collectionsIt = collections.iterator();
117	0	Collection<XmlDocCollection> culled = new ArrayList<XmlDocCollection>();
118	0	while (collectionsIt.hasNext()) {
119	0	XmlDocCollection container = (XmlDocCollection) collectionsIt.next();
120		// if a move has already failed for this archive, ignore it
121	0	if (inPendingMoveFailedArchive(container.getFile())) {
122	0	LOG.info("Ignoring previously processed resource: " + container);
123	0	culled.add(container);
124		}
125	0	}
126	0	collections.removeAll(culled);
127
128	0	if (collections.size() == 0) {
129	0	LOG.debug("No valid new resources found to ingest");
130	0	return;
131		}
132
133	0	Date LOAD_TIME = Calendar.getInstance().getTime();
134		// synchronization around date format should not be an issue as this code is single-threaded
135	0	File completeDir = new File(getXmlCompleteDir(), DIR_FORMAT.format(LOAD_TIME));
136	0	File failedDir = new File(getXmlProblemDir(), DIR_FORMAT.format(LOAD_TIME));
137
138		// now ingest the containers
139	0	Collection failed = null;
140		try {
141	0	failed = KEWServiceLocator.getXmlIngesterService().ingest(collections);
142	0	} catch (Exception e) {
143	0	LOG.error("Error ingesting data", e);
144		//throw new RuntimeException(e);
145	0	}
146
147		// now iterate through all containers again, and move containers to approprate dir
148	0	LOG.info("Moving files...");
149	0	collectionsIt = collections.iterator();
150	0	while (collectionsIt.hasNext()) {
151	0	XmlDocCollection container = (XmlDocCollection) collectionsIt.next();
152	0	LOG.debug("container: " + container);
153		try {
154		// "close" the container
155		// this only matters for ZipFiles for now
156	0	container.close();
157	0	} catch (IOException ioe) {
158	0	LOG.warn("Error closing " + container, ioe);
159	0	}
160	0	if (failed.contains(container)) {
161		// some docs must have failed, move the whole
162		// container to the failed dir
163	0	if (container.getFile() != null) {
164	0	LOG.error("Moving " + container.getFile() + " to problem dir.");
165	0	if ((!failedDir.isDirectory() && !failedDir.mkdirs())
166		\|\| !moveFile(failedDir, container.getFile())) {
167	0	LOG.error("Could not move: " + container.getFile());
168	0	recordUnmovablePendingFile(container.getFile(), LOAD_TIME);
169		}
170		}
171		} else {
172	0	if (container.getFile() != null) {
173	0	LOG.info("Moving " + container.getFile() + " to loaded dir.");
174	0	if((!completeDir.isDirectory() && !completeDir.mkdirs())
175		\|\| !moveFile(completeDir, container.getFile())){
176	0	LOG.error("Could not move: " + container.getFile());
177	0	recordUnmovablePendingFile(container.getFile(), LOAD_TIME);
178		}
179		}
180		}
181	0	}
182	0	}
183
184		private boolean inPendingMoveFailedArchive(File xmlDataFile){
185	0	if (xmlDataFile == null) return false;
186	0	BufferedReader inFile = null;
187	0	File movesFailedFile = new File(getXmlPendingDir(), PENDING_MOVE_FAILED_ARCHIVE_FILE);
188	0	if (!movesFailedFile.isFile()) return false;
189		try {
190	0	inFile = new BufferedReader(new FileReader(movesFailedFile));
191		String line;
192
193	0	while((line = inFile.readLine()) != null){
194	0	String trimmedLine = line.trim();
195	0	if(trimmedLine.equals(xmlDataFile.getName()) \|\|
196		trimmedLine.startsWith(xmlDataFile.getName() + "=")) {
197	0	return true;
198		}
199	0	}
200	0	} catch (IOException e){
201	0	LOG.warn("Error reading file " + movesFailedFile);
202		//TODO try reading the pending file or stop?
203	0	} finally {
204	0	if (inFile != null) try {
205	0	inFile.close();
206	0	} catch (Exception e) {
207	0	LOG.warn("Error closing buffered reader for " + movesFailedFile);
208	0	}
209	0	}
210
211	0	return false;
212		}
213
214		private boolean recordUnmovablePendingFile(File unMovablePendingFile, Date dateLoaded){
215	0	boolean recorded = false;
216	0	FileWriter archiveFile = null;
217		try{
218	0	archiveFile = new FileWriter(new File(getXmlPendingDir(), PENDING_MOVE_FAILED_ARCHIVE_FILE), true);
219	0	archiveFile.write(unMovablePendingFile.getName() + "=" + dateLoaded.getTime() + NEW_LINE);
220	0	recorded = true;
221	0	} catch (IOException e){
222	0	LOG.error("Unable to record unmovable pending file " + unMovablePendingFile.getName() + "in the archive file " + PENDING_MOVE_FAILED_ARCHIVE_FILE);
223	0	} finally {
224	0	if (archiveFile != null) {
225		try {
226	0	archiveFile.close();
227	0	} catch (IOException ioe) {
228	0	LOG.error("Error closing unmovable pending file", ioe);
229	0	}
230		}
231	0	}
232	0	return recorded;
233		}
234
235		private boolean moveFile(File toDirectory, File fileToMove){
236	0	boolean moved = true;
237	0	if (!fileToMove.renameTo(new File(toDirectory.getPath(), fileToMove.getName()))){
238	0	LOG.error("Unable to move file " + fileToMove.getName() + " to directory " + toDirectory.getPath());
239	0	moved = false;
240		}
241	0	return moved;
242		}
243
244		private File getXmlPendingDir() {
245	0	return new File(getXmlPendingLocation());
246		}
247
248		private File getXmlCompleteDir() {
249	0	return new File(getXmlCompletedLocation());
250		}
251
252		private File getXmlProblemDir() {
253	0	return new File(getXmlProblemLocation());
254		}
255
256		public String getXmlCompletedLocation() {
257	0	return xmlCompletedLocation;
258		}
259
260		public void setXmlCompletedLocation(String xmlCompletedLocation) {
261	0	this.xmlCompletedLocation = xmlCompletedLocation;
262	0	}
263
264		public String getXmlPendingLocation() {
265	0	return xmlPendingLocation;
266		}
267
268		/*public boolean validate(File uploadedFile) {
269		XmlDataLoaderFileFilter filter = new XmlDataLoaderFileFilter();
270		return filter.accept(uploadedFile);
271		}*/
272
273		public void setXmlPendingLocation(String xmlPendingLocation) {
274	0	this.xmlPendingLocation = xmlPendingLocation;
275	0	}
276
277		public String getXmlProblemLocation() {
278	0	return xmlProblemLocation;
279		}
280
281		public void setXmlProblemLocation(String xmlProblemLocation) {
282	0	this.xmlProblemLocation = xmlProblemLocation;
283	0	}
284		public String getXmlParentDirectory() {
285	0	return xmlParentDirectory;
286		}
287		public void setXmlParentDirectory(String xmlDataParentDirectory) {
288	0	this.xmlParentDirectory = xmlDataParentDirectory;
289	0	}
290
291		/**
292		* Sets the polling interval time in seconds
293		* @param seconds the polling interval time in seconds
294		*/
295		public void setPollIntervalSecs(int seconds) {
296	0	this.pollIntervalSecs = seconds;
297	0	}
298
299		/**
300		* Gets the polling interval time in seconds
301		* @return the polling interval time in seconds
302		*/
303		public int getPollIntervalSecs() {
304	0	return this.pollIntervalSecs;
305		}
306
307		/**
308		* Sets the initial delay time in seconds
309		* @param seconds the initial delay time in seconds
310		*/
311		public void setInitialDelaySecs(int seconds) {
312	0	this.initialDelaySecs = seconds;
313	0	}
314
315		/**
316		* Gets the initial delay time in seconds
317		* @return the initial delay time in seconds
318		*/
319		public int getInitialDelaySecs() {
320	0	return this.initialDelaySecs;
321		}
322		}