server/runtime/org.argeo.server.jcr/src/main/java/org/argeo/jcr/tabular/JcrTabularRowIterator.java

   1 package org.argeo.jcr.tabular;
   2
   3 import java.io.InputStream;
   4 import java.util.ArrayList;
   5 import java.util.List;
   6 import java.util.concurrent.ArrayBlockingQueue;
   7
   8 import javax.jcr.Binary;
   9 import javax.jcr.Node;
  10 import javax.jcr.NodeIterator;
  11 import javax.jcr.Property;
  12 import javax.jcr.PropertyType;
  13 import javax.jcr.RepositoryException;
  14
  15 import org.apache.commons.io.IOUtils;
  16 import org.argeo.ArgeoException;
  17 import org.argeo.jcr.ArgeoTypes;
  18 import org.argeo.jcr.JcrUtils;
  19 import org.argeo.util.CsvParser;
  20 import org.argeo.util.tabular.ArrayTabularRow;
  21 import org.argeo.util.tabular.TabularColumn;
  22 import org.argeo.util.tabular.TabularRow;
  23 import org.argeo.util.tabular.TabularRowIterator;
  24
  25 /** Iterates over the rows of a {@link ArgeoTypes#ARGEO_TABLE} node. */
  26 public class JcrTabularRowIterator implements TabularRowIterator {
  27         private Boolean hasNext = null;
  28         private Boolean parsingCompleted = false;
  29
  30         private Long currentRowNumber = 0l;
  31
  32         private List<TabularColumn> header = new ArrayList<TabularColumn>();
  33
  34         /** referenced so that we can close it */
  35         private Binary binary;
  36         private InputStream in;
  37
  38         private CsvParser csvParser;
  39         private ArrayBlockingQueue<List<String>> textLines;
  40
  41         public JcrTabularRowIterator(Node tableNode) {
  42                 try {
  43                         Node contentNode = tableNode.getNode(Property.JCR_CONTENT);
  44                         for (NodeIterator it = tableNode.getNodes(); it.hasNext();) {
  45                                 Node node = it.nextNode();
  46                                 if (node.isNodeType(ArgeoTypes.ARGEO_COLUMN)) {
  47                                         Integer type = PropertyType.valueFromName(node.getProperty(
  48                                                         Property.JCR_REQUIRED_TYPE).getString());
  49                                         TabularColumn tc = new TabularColumn(node.getName(), type);
  50                                         header.add(tc);
  51                                         // } else if (node.getName().equals(Property.JCR_CONTENT)) {
  52                                         // contentNode = node;
  53                                 }
  54                         }
  55                         // should not happen since content is mandatory
  56                         assert contentNode != null;
  57
  58                         if (contentNode.isNodeType(ArgeoTypes.ARGEO_CSV)) {
  59                                 textLines = new ArrayBlockingQueue<List<String>>(1000);
  60                                 csvParser = new CsvParser() {
  61                                         protected void processLine(Integer lineNumber,
  62                                                         List<String> header, List<String> tokens) {
  63                                                 try {
  64                                                         textLines.put(tokens);
  65                                                 } catch (InterruptedException e) {
  66                                                         // TODO Auto-generated catch block
  67                                                         e.printStackTrace();
  68                                                 }
  69                                                 //textLines.add(tokens);
  70                                                 if (hasNext == null) {
  71                                                         hasNext = true;
  72                                                         synchronized (JcrTabularRowIterator.this) {
  73                                                                 JcrTabularRowIterator.this.notifyAll();
  74                                                         }
  75                                                 }
  76                                         }
  77                                 };
  78                                 csvParser.setNoHeader(true);
  79                                 binary = contentNode.getProperty(Property.JCR_DATA).getBinary();
  80                                 in = binary.getStream();
  81                                 Thread thread = new Thread(contentNode.getPath() + " reader") {
  82                                         public void run() {
  83                                                 try {
  84                                                         csvParser.parse(in);
  85                                                 } finally {
  86                                                         parsingCompleted = true;
  87                                                         IOUtils.closeQuietly(in);
  88                                                 }
  89                                         }
  90                                 };
  91                                 thread.start();
  92                         }
  93                 } catch (RepositoryException e) {
  94                         throw new ArgeoException("Cannot read table " + tableNode, e);
  95                 }
  96         }
  97
  98         public synchronized boolean hasNext() {
  99                 // we don't know if there is anything available
 100                 // while (hasNext == null)
 101                 // try {
 102                 // wait();
 103                 // } catch (InterruptedException e) {
 104                 // // silent
 105                 // // FIXME better deal with interruption
 106                 // Thread.currentThread().interrupt();
 107                 // break;
 108                 // }
 109
 110                 // buffer not empty
 111                 if (!textLines.isEmpty())
 112                         return true;
 113
 114                 // maybe the parsing is finished but the flag has not been set
 115                 while (!parsingCompleted && textLines.isEmpty())
 116                         try {
 117                                 wait(100);
 118                         } catch (InterruptedException e) {
 119                                 // silent
 120                                 // FIXME better deal with interruption
 121                                 Thread.currentThread().interrupt();
 122                                 break;
 123                         }
 124
 125                 // buffer not empty
 126                 if (!textLines.isEmpty())
 127                         return true;
 128
 129                 // (parsingCompleted && textLines.isEmpty())
 130                         return false;
 131
 132 //              if (!hasNext && textLines.isEmpty()) {
 133 //                      if (in != null) {
 134 //                              IOUtils.closeQuietly(in);
 135 //                              in = null;
 136 //                      }
 137 //                      if (binary != null) {
 138 //                              JcrUtils.closeQuietly(binary);
 139 //                              binary = null;
 140 //                      }
 141 //                      return false;
 142 //              } else
 143 //                      return true;
 144         }
 145
 146         public synchronized TabularRow next() {
 147                 try {
 148                         List<String> tokens = textLines.take();
 149                         List<Object> objs = new ArrayList<Object>(tokens.size());
 150                         for (String token : tokens) {
 151                                 // TODO convert to other formats using header
 152                                 objs.add(token);
 153                         }
 154                         currentRowNumber++;
 155                         return new ArrayTabularRow(objs);
 156                 } catch (InterruptedException e) {
 157                         // silent
 158                         // FIXME better deal with interruption
 159                 }
 160                 return null;
 161         }
 162
 163         public void remove() {
 164                 throw new UnsupportedOperationException();
 165         }
 166
 167         public Long getCurrentRowNumber() {
 168                 return currentRowNumber;
 169         }
 170
 171 }