X-Git-Url: http://git.argeo.org/?a=blobdiff_plain;ds=sidebyside;f=org.argeo.server.jcr%2Fsrc%2Fmain%2Fjava%2Forg%2Fargeo%2Fjcr%2Ftabular%2FJcrTabularRowIterator.java;fp=org.argeo.server.jcr%2Fsrc%2Fmain%2Fjava%2Forg%2Fargeo%2Fjcr%2Ftabular%2FJcrTabularRowIterator.java;h=ca0635c95887f65f4af2464aeb83aa39fd4fbd82;hb=1df1bf64759d35d3d72b9d96b26b71118fdbe031;hp=0000000000000000000000000000000000000000;hpb=3a3d316af102ba410d1d9e6de349d0c8f7ac044f;p=lgpl%2Fargeo-commons.git

diff --git a/org.argeo.server.jcr/src/main/java/org/argeo/jcr/tabular/JcrTabularRowIterator.java b/org.argeo.server.jcr/src/main/java/org/argeo/jcr/tabular/JcrTabularRowIterator.java
new file mode 100644
index 000000000..ca0635c95
--- /dev/null
+++ b/org.argeo.server.jcr/src/main/java/org/argeo/jcr/tabular/JcrTabularRowIterator.java
@@ -0,0 +1,185 @@
+/*
+ * Copyright (C) 2007-2012 Argeo GmbH
+ *
+ * Licensed under the Apache 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.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.argeo.jcr.tabular;
+
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ArrayBlockingQueue;
+
+import javax.jcr.Binary;
+import javax.jcr.Node;
+import javax.jcr.NodeIterator;
+import javax.jcr.Property;
+import javax.jcr.PropertyType;
+import javax.jcr.RepositoryException;
+
+import org.apache.commons.io.IOUtils;
+import org.argeo.ArgeoException;
+import org.argeo.jcr.ArgeoTypes;
+import org.argeo.util.CsvParser;
+import org.argeo.util.tabular.ArrayTabularRow;
+import org.argeo.util.tabular.TabularColumn;
+import org.argeo.util.tabular.TabularRow;
+import org.argeo.util.tabular.TabularRowIterator;
+
+/** Iterates over the rows of a {@link ArgeoTypes#ARGEO_TABLE} node. */
+public class JcrTabularRowIterator implements TabularRowIterator {
+	private Boolean hasNext = null;
+	private Boolean parsingCompleted = false;
+
+	private Long currentRowNumber = 0l;
+
+	private List<TabularColumn> header = new ArrayList<TabularColumn>();
+
+	/** referenced so that we can close it */
+	private Binary binary;
+	private InputStream in;
+
+	private CsvParser csvParser;
+	private ArrayBlockingQueue<List<String>> textLines;
+
+	public JcrTabularRowIterator(Node tableNode) {
+		try {
+			for (NodeIterator it = tableNode.getNodes(); it.hasNext();) {
+				Node node = it.nextNode();
+				if (node.isNodeType(ArgeoTypes.ARGEO_COLUMN)) {
+					Integer type = PropertyType.valueFromName(node.getProperty(
+							Property.JCR_REQUIRED_TYPE).getString());
+					TabularColumn tc = new TabularColumn(node.getProperty(
+							Property.JCR_TITLE).getString(), type);
+					header.add(tc);
+				}
+			}
+			Node contentNode = tableNode.getNode(Property.JCR_CONTENT);
+			if (contentNode.isNodeType(ArgeoTypes.ARGEO_CSV)) {
+				textLines = new ArrayBlockingQueue<List<String>>(1000);
+				csvParser = new CsvParser() {
+					protected void processLine(Integer lineNumber,
+							List<String> header, List<String> tokens) {
+						try {
+							textLines.put(tokens);
+						} catch (InterruptedException e) {
+							// TODO Auto-generated catch block
+							e.printStackTrace();
+						}
+						// textLines.add(tokens);
+						if (hasNext == null) {
+							hasNext = true;
+							synchronized (JcrTabularRowIterator.this) {
+								JcrTabularRowIterator.this.notifyAll();
+							}
+						}
+					}
+				};
+				csvParser.setNoHeader(true);
+				binary = contentNode.getProperty(Property.JCR_DATA).getBinary();
+				in = binary.getStream();
+				Thread thread = new Thread(contentNode.getPath() + " reader") {
+					public void run() {
+						try {
+							csvParser.parse(in);
+						} finally {
+							parsingCompleted = true;
+							IOUtils.closeQuietly(in);
+						}
+					}
+				};
+				thread.start();
+			}
+		} catch (RepositoryException e) {
+			throw new ArgeoException("Cannot read table " + tableNode, e);
+		}
+	}
+
+	public synchronized boolean hasNext() {
+		// we don't know if there is anything available
+		// while (hasNext == null)
+		// try {
+		// wait();
+		// } catch (InterruptedException e) {
+		// // silent
+		// // FIXME better deal with interruption
+		// Thread.currentThread().interrupt();
+		// break;
+		// }
+
+		// buffer not empty
+		if (!textLines.isEmpty())
+			return true;
+
+		// maybe the parsing is finished but the flag has not been set
+		while (!parsingCompleted && textLines.isEmpty())
+			try {
+				wait(100);
+			} catch (InterruptedException e) {
+				// silent
+				// FIXME better deal with interruption
+				Thread.currentThread().interrupt();
+				break;
+			}
+
+		// buffer not empty
+		if (!textLines.isEmpty())
+			return true;
+
+		// (parsingCompleted && textLines.isEmpty())
+		return false;
+
+		// if (!hasNext && textLines.isEmpty()) {
+		// if (in != null) {
+		// IOUtils.closeQuietly(in);
+		// in = null;
+		// }
+		// if (binary != null) {
+		// JcrUtils.closeQuietly(binary);
+		// binary = null;
+		// }
+		// return false;
+		// } else
+		// return true;
+	}
+
+	public synchronized TabularRow next() {
+		try {
+			List<String> tokens = textLines.take();
+			List<Object> objs = new ArrayList<Object>(tokens.size());
+			for (String token : tokens) {
+				// TODO convert to other formats using header
+				objs.add(token);
+			}
+			currentRowNumber++;
+			return new ArrayTabularRow(objs);
+		} catch (InterruptedException e) {
+			// silent
+			// FIXME better deal with interruption
+		}
+		return null;
+	}
+
+	public void remove() {
+		throw new UnsupportedOperationException();
+	}
+
+	public Long getCurrentRowNumber() {
+		return currentRowNumber;
+	}
+
+	public List<TabularColumn> getHeader() {
+		return header;
+	}
+
+}