src/main/java/org/xerial/silk/SilkStreamReader.java

   1 /*--------------------------------------------------------------------------\r
   2  *  Copyright 2009 Taro L. Saito\r
   3  *\r
   4  *  Licensed under the Apache License, Version 2.0 (the "License");\r
   5  *  you may not use this file except in compliance with the License.\r
   6  *  You may obtain a copy of the License at\r
   7  *\r
   8  *     http://www.apache.org/licenses/LICENSE-2.0\r
   9  *\r
  10  *  Unless required by applicable law or agreed to in writing, software\r
  11  *  distributed under the License is distributed on an "AS IS" BASIS,\r
  12  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  13  *  See the License for the specific language governing permissions and\r
  14  *  limitations under the License.\r
  15  *--------------------------------------------------------------------------*/\r
  16 //--------------------------------------\r
  17 // XerialJ\r
  18 //\r
  19 // SilkStreamReader.java\r
  20 // Since: 2009/03/31 19:53:33\r
  21 //\r
  22 // $URL$\r
  23 // $Author$\r
  24 //--------------------------------------\r
  25 package org.xerial.silk;\r
  26 \r
  27 import java.io.BufferedReader;\r
  28 import java.io.IOException;\r
  29 import java.io.InputStream;\r
  30 import java.io.InputStreamReader;\r
  31 import java.io.Reader;\r
  32 import java.lang.reflect.Field;\r
  33 import java.net.URL;\r
  34 import java.util.ArrayList;\r
  35 import java.util.Collections;\r
  36 import java.util.Comparator;\r
  37 import java.util.HashMap;\r
  38 import java.util.List;\r
  39 import java.util.Map;\r
  40 import java.util.TreeMap;\r
  41 \r
  42 import org.xerial.core.XerialError;\r
  43 import org.xerial.core.XerialErrorCode;\r
  44 import org.xerial.core.XerialException;\r
  45 import org.xerial.json.JSONArray;\r
  46 import org.xerial.json.JSONException;\r
  47 import org.xerial.json.JSONObject;\r
  48 import org.xerial.json.JSONUtil;\r
  49 import org.xerial.json.JSONValue;\r
  50 import org.xerial.json.JSONValueType;\r
  51 import org.xerial.silk.impl.SilkDataLine;\r
  52 import org.xerial.silk.impl.SilkFunction;\r
  53 import org.xerial.silk.impl.SilkFunctionArg;\r
  54 import org.xerial.silk.impl.SilkJSONValue;\r
  55 import org.xerial.silk.impl.SilkNode;\r
  56 import org.xerial.silk.impl.SilkNodeOccurrence;\r
  57 import org.xerial.silk.impl.SilkValue;\r
  58 import org.xerial.silk.plugin.SilkFunctionArgument;\r
  59 import org.xerial.silk.plugin.SilkFunctionPlugin;\r
  60 import org.xerial.util.ArrayDeque;\r
  61 import org.xerial.util.FileResource;\r
  62 import org.xerial.util.bean.TypeInformation;\r
  63 import org.xerial.util.log.Logger;\r
  64 import org.xerial.util.reflect.ReflectionUtil;\r
  65 import org.xerial.util.tree.TreeEvent;\r
  66 import org.xerial.util.tree.TreeStreamReader;\r
  67 import org.xerial.util.xml.impl.TreeEventQueue;\r
  68 \r
  69 /**\r
  70  * {@link TreeStreamReader} implementation for the Silk data format.\r
  71  * \r
  72  * @author leo\r
  73  * \r
  74  */\r
  75 public class SilkStreamReader implements TreeStreamReader\r
  76 {\r
  77     private static Logger _logger = Logger.getLogger(SilkStreamReader.class);\r
  78 \r
  79     private final SilkPullParser parser;\r
  80     private final SilkEnv parseContext;\r
  81     private TreeEventQueue eventQueue = new TreeEventQueue();\r
  82     private final ArrayDeque<TreeStreamReader> readerStack = new ArrayDeque<TreeStreamReader>();\r
  83 \r
  84     private int numReadLine = 0;\r
  85 \r
  86     /**\r
  87      * Creates a new reader with the specified input stream\r
  88      * \r
  89      * @param input\r
  90      *            `@throws IOException\r
  91      */\r
  92     protected SilkStreamReader(InputStream input) throws IOException\r
  93     {\r
  94         this(new InputStreamReader(input));\r
  95     }\r
  96 \r
  97     /**\r
  98      * Creates a new reader with the specified reader\r
  99      * \r
 100      * @param input\r
 101      * @throws IOException\r
 102      */\r
 103     protected SilkStreamReader(Reader input) throws IOException\r
 104     {\r
 105         this(input, SilkEnv.newEnv());\r
 106     }\r
 107 \r
 108     /**\r
 109      * Creates a new reader inherited the given environment\r
 110      * \r
 111      * @param input\r
 112      * @param env\r
 113      * @throws IOException\r
 114      */\r
 115     public SilkStreamReader(Reader input, SilkEnv env) throws IOException\r
 116     {\r
 117         this.parser = new SilkPullParser(input);\r
 118         this.parseContext = env;\r
 119     }\r
 120 \r
 121     /**\r
 122      * Concatenates the base path and the resource name\r
 123      * \r
 124      * @param resourceBasePath\r
 125      * @param resourceName\r
 126      * @return\r
 127      */\r
 128     private static String getResourcePath(String resourceBasePath, String resourceName)\r
 129     {\r
 130         String resourcePath = resourceBasePath;\r
 131         if (!resourcePath.endsWith("/"))\r
 132             resourcePath += "/";\r
 133         resourcePath += resourceName;\r
 134         return resourcePath;\r
 135     }\r
 136 \r
 137     /**\r
 138      * Create a new reader for reading local resources\r
 139      * \r
 140      * @param resourceBasePath\r
 141      * @param resourceName\r
 142      * @throws IOException\r
 143      */\r
 144     public SilkStreamReader(String resourceBasePath, String resourceName) throws IOException\r
 145     {\r
 146         this.parser = new SilkPullParser(new BufferedReader(new InputStreamReader(SilkWalker.class\r
 147                 .getResourceAsStream(getResourcePath(resourceBasePath, resourceName)))));\r
 148         this.parseContext = SilkEnv.newEnv(resourceBasePath);\r
 149     }\r
 150 \r
 151     /**\r
 152      * Create a new reader for reading the specified resource URL\r
 153      * \r
 154      * @param resourcePath\r
 155      * @throws IOException\r
 156      */\r
 157     public SilkStreamReader(URL resourcePath) throws IOException\r
 158     {\r
 159         this(resourcePath, SilkEnv.newEnv());\r
 160     }\r
 161 \r
 162     public SilkStreamReader(URL resource, SilkEnv env) throws IOException\r
 163     {\r
 164         String path = resource.toExternalForm();\r
 165         int fileNamePos = path.lastIndexOf("/");\r
 166         String resourceBasePath = fileNamePos > 0 ? path.substring(0, fileNamePos) : null;\r
 167 \r
 168         this.parser = new SilkPullParser(new BufferedReader(new InputStreamReader(resource.openStream())));\r
 169         this.parseContext = SilkEnv.newEnv(env, resourceBasePath);\r
 170     }\r
 171 \r
 172     public TreeEvent peekNext() throws XerialException\r
 173     {\r
 174         if (hasNext())\r
 175             return eventQueue.peekFirst();\r
 176         else\r
 177             return null;\r
 178     }\r
 179 \r
 180     public TreeEvent next() throws XerialException\r
 181     {\r
 182         if (hasNext())\r
 183             return getNextEvent();\r
 184         else\r
 185             return null;\r
 186     }\r
 187 \r
 188     /**\r
 189      * Enqueues a visit event.\r
 190      * \r
 191      * @param nodeName\r
 192      * @param immediateNodeValue\r
 193      * @throws XerialException\r
 194      */\r
 195     private void visit(String nodeName, String immediateNodeValue) throws XerialException\r
 196     {\r
 197         eventQueue.push(TreeEvent.newVisitEvent(nodeName, immediateNodeValue));\r
 198     }\r
 199 \r
 200     /**\r
 201      * Enqueues a leave event\r
 202      * \r
 203      * @param nodeName\r
 204      * @throws XerialException\r
 205      */\r
 206     private void leave(String nodeName) throws XerialException\r
 207     {\r
 208         eventQueue.push(TreeEvent.newLeaveEvent(nodeName));\r
 209     }\r
 210 \r
 211     /**\r
 212      * Enqueues a text event\r
 213      * \r
 214      * @param textFragment\r
 215      * @throws XerialException\r
 216      */\r
 217     private void text(String textFragment) throws XerialException\r
 218     {\r
 219         eventQueue.push(TreeEvent.newTextEvent(parseContext.getContextNode().getName(), textFragment));\r
 220     }\r
 221 \r
 222     /**\r
 223      * Closed pre-opened contexts up to the specified indent level\r
 224      * \r
 225      * @param newIndentLevel\r
 226      * @throws XerialException\r
 227      */\r
 228     private void closeContextUpTo(int newIndentLevel) throws XerialException\r
 229     {\r
 230         while (!parseContext.isContextNodeStackEmpty())\r
 231         {\r
 232             SilkContext context = parseContext.peekLastContext();\r
 233             SilkNode node = context.contextNode;\r
 234             if (node.getIndentLevel() >= newIndentLevel)\r
 235             {\r
 236                 parseContext.popLastContext();\r
 237 \r
 238                 if (parseContext.isAttributeOpen)\r
 239                 {\r
 240                     // close attribute \r
 241                     SilkNode attribute = node.getChildNodes().get(parseContext.contextNodeAttributeCursor);\r
 242                     leave(attribute.getName());\r
 243                     leave(node.getName());\r
 244                     parseContext.isAttributeOpen = false;\r
 245                 }\r
 246 \r
 247                 if (context.isOpen)\r
 248                 {\r
 249                     // close context\r
 250                     String nodeName = node.getName();\r
 251                     leave(nodeName);\r
 252                 }\r
 253             }\r
 254             else\r
 255                 return;\r
 256         }\r
 257     }\r
 258 \r
 259     /**\r
 260      * Opens a new context for the given node\r
 261      * \r
 262      * @param node\r
 263      *            new context node\r
 264      * @param visitor\r
 265      * @throws XerialException\r
 266      */\r
 267     private void openContext(SilkNode node) throws XerialException\r
 268     {\r
 269         int indentLevel = node.getIndentLevel();\r
 270         if (indentLevel != SilkNode.NO_INDENT)\r
 271             indentLevel += parseContext.getIndentationOffset();\r
 272 \r
 273         closeContextUpTo(indentLevel);\r
 274         openContext_internal(node);\r
 275     }\r
 276 \r
 277     private void openContext_internal(SilkNode node) throws XerialException\r
 278     {\r
 279         if (node.getName() == null)\r
 280         {\r
 281             // no name nodes must hierarchically organize attribute nodes\r
 282             for (SilkNode eachChild : node.getChildNodes())\r
 283             {\r
 284                 eachChild.setNodeIndent(node.getNodeIndent());\r
 285                 openContext_internal(eachChild);\r
 286             }\r
 287             return;\r
 288         }\r
 289 \r
 290         SilkContext currentContext = new SilkContext(node, true);\r
 291         parseContext.pushContext(currentContext);\r
 292 \r
 293         SilkNodeOccurrence occurrence = node.getOccurrence();\r
 294         if (occurrence.isSchemaOnlyNode())\r
 295         {\r
 296             currentContext.isOpen = false;\r
 297             // reset the attribute cursor\r
 298             parseContext.contextNodeAttributeCursor = 0;\r
 299             parseContext.isAttributeOpen = false;\r
 300             return; // do not invoke visit events\r
 301         }\r
 302 \r
 303         String nodeName = node.getName();\r
 304         SilkValue textValue = node.getValue();\r
 305 \r
 306         // process text values attached to the node\r
 307         if (textValue != null)\r
 308         {\r
 309             // When the text data is JSON, traverses the JSON data \r
 310             if (textValue.isJSON())\r
 311             {\r
 312 \r
 313                 SilkJSONValue jsonValue = SilkJSONValue.class.cast(textValue);\r
 314                 if (jsonValue.isObject())\r
 315                 {\r
 316                     visit(nodeName, null);\r
 317                     JSONObject jsonObj = new JSONObject(jsonValue.getValue());\r
 318                     walkJSONObject(jsonObj);\r
 319                 }\r
 320                 else\r
 321                 {\r
 322                     currentContext.isOpen = false;\r
 323                     JSONArray jsonArray = new JSONArray(jsonValue.getValue());\r
 324                     walkJSONAray(jsonArray, nodeName);\r
 325                 }\r
 326             }\r
 327             else if (textValue.isFunction())\r
 328             {\r
 329                 // evaluate the function \r
 330                 visit(nodeName, null);\r
 331                 SilkFunction function = SilkFunction.class.cast(textValue);\r
 332                 evalFunction(function);\r
 333 \r
 334                 return;\r
 335             }\r
 336             else\r
 337             {\r
 338                 // Simple text value will be reported as it is.\r
 339                 visit(nodeName, textValue.toString());\r
 340             }\r
 341         }\r
 342         else\r
 343         {\r
 344             if (occurrence == SilkNodeOccurrence.ZERO_OR_MORE)\r
 345             {\r
 346                 // CSV data\r
 347                 return; // do not invoke visit events\r
 348             }\r
 349 \r
 350             // Report a visit event without text value\r
 351             visit(nodeName, null);\r
 352         }\r
 353 \r
 354         // Traverse attribute nodes having text values. If no text value is specified for these attributes, \r
 355         // they are schema elements for the following DATA_LINE. \r
 356         for (SilkNode eachChild : node.getChildNodes())\r
 357         {\r
 358             if (eachChild.hasValue())\r
 359             {\r
 360                 openContext(eachChild);\r
 361             }\r
 362         }\r
 363 \r
 364     }\r
 365 \r
 366     private static class FunctionReader implements TreeStreamReader\r
 367     {\r
 368         SilkFunctionPlugin plugin;\r
 369 \r
 370         public FunctionReader(SilkFunctionPlugin plugin)\r
 371         {\r
 372             this.plugin = plugin;\r
 373         }\r
 374 \r
 375         public TreeEvent peekNext() throws XerialException\r
 376         {\r
 377             return plugin.peekNext();\r
 378         }\r
 379 \r
 380         public TreeEvent next() throws XerialException\r
 381         {\r
 382             return plugin.next();\r
 383         }\r
 384     }\r
 385 \r
 386     /**\r
 387      * Evaluate the function\r
 388      * \r
 389      * @param function\r
 390      * @throws XerialException\r
 391      */\r
 392     private void evalFunction(SilkFunction function) throws XerialException\r
 393     {\r
 394         SilkFunctionPlugin plugin = getPlugin(function.getName());\r
 395         if (plugin == null)\r
 396         {\r
 397             _logger.error(String.format("plugin %s not found", function.getName()));\r
 398             return;\r
 399         }\r
 400         // fill the function argument to the plugin instance\r
 401         populate(plugin, function.getArgumentList());\r
 402 \r
 403         // evaluate the function\r
 404         SilkEnv env = parseContext.newEnvFor(function);\r
 405         plugin.init(env);\r
 406 \r
 407         readerStack.addLast(new FunctionReader(plugin));\r
 408     }\r
 409 \r
 410     /**\r
 411      * Has finished reading the stream?\r
 412      */\r
 413     private boolean hasFinished = false;\r
 414 \r
 415     /**\r
 416      * Is next event available?\r
 417      * \r
 418      * @return true if there are remaining events, otherwise fales\r
 419      * @throws XerialException\r
 420      */\r
 421     private boolean hasNext() throws XerialException\r
 422     {\r
 423         if (!eventQueue.isEmpty())\r
 424             return true;\r
 425 \r
 426         if (hasFinished)\r
 427             return false;\r
 428 \r
 429         while (!hasFinished && eventQueue.isEmpty())\r
 430             fillQueue();\r
 431 \r
 432         return hasNext();\r
 433     }\r
 434 \r
 435     /**\r
 436      * Retrieves the next event from the queue. If the event queue is empty,\r
 437      * fill the queue with the next event\r
 438      * \r
 439      * @return the next event.\r
 440      * @throws XerialException\r
 441      */\r
 442     private TreeEvent getNextEvent() throws XerialException\r
 443     {\r
 444         if (!eventQueue.isEmpty())\r
 445             return eventQueue.pop();\r
 446 \r
 447         if (hasFinished)\r
 448             throw new XerialError(XerialErrorCode.INVALID_STATE,\r
 449                     "hasNext() value must be checked before calling getNextEvent()");\r
 450 \r
 451         fillQueue();\r
 452         return getNextEvent();\r
 453     }\r
 454 \r
 455     private void walkMicroFormatRoot(SilkNode schemaNode, JSONArray value) throws XerialException\r
 456     {\r
 457         // e.g., exon(start, name)\r
 458 \r
 459         if (schemaNode.hasManyOccurrences())\r
 460         {\r
 461             if (schemaNode.hasChildren())\r
 462             {\r
 463                 // e.g., exon(start, name)*\r
 464                 // multiple occurrences: [[start, end], [start, end], ... ] \r
 465                 for (int i = 0; i < value.size(); i++)\r
 466                 {\r
 467                     JSONArray eachElement = value.getJSONArray(i);\r
 468                     if (eachElement == null)\r
 469                         continue;\r
 470 \r
 471                     visit(schemaNode.getName(), null);\r
 472                     int index = 0;\r
 473                     for (SilkNode eachSubSchema : schemaNode.getChildNodes())\r
 474                     {\r
 475                         walkMicroFormatElement(eachSubSchema, eachElement.get(index++));\r
 476                     }\r
 477                     leave(schemaNode.getName());\r
 478                 }\r
 479             }\r
 480             else\r
 481             {\r
 482                 // e.g. QV*: [20, 50, 50]\r
 483                 for (int i = 0; i < value.size(); i++)\r
 484                 {\r
 485                     visit(schemaNode.getName(), value.get(i).toString());\r
 486                     leave(schemaNode.getName());\r
 487                 }\r
 488             }\r
 489         }\r
 490         else\r
 491         {\r
 492             // [e1, e2, ...]\r
 493             visit(schemaNode.getName(), null);\r
 494             int index = 0;\r
 495             if (schemaNode.getChildNodes().size() != value.size())\r
 496             {\r
 497                 throw new XerialException(XerialErrorCode.INVALID_INPUT, String.format(\r
 498                         "data format doesn't match: schema=%s, value=%s", schemaNode, value));\r
 499             }\r
 500             for (SilkNode each : schemaNode.getChildNodes())\r
 501             {\r
 502                 walkMicroFormatElement(each, value.get(index++));\r
 503             }\r
 504             leave(schemaNode.getName());\r
 505         }\r
 506     }\r
 507 \r
 508     private void walkMicroFormatElement(SilkNode schemaNode, JSONValue value) throws XerialException\r
 509     {\r
 510         if (schemaNode.hasChildren())\r
 511         {\r
 512             JSONArray array = value.getJSONArray();\r
 513             if (array != null)\r
 514                 walkMicroFormatRoot(schemaNode, array);\r
 515             else\r
 516                 throw new XerialException(XerialErrorCode.INVALID_INPUT, String.format(\r
 517                         "data format doesn't match: schema=%s, value=%s", schemaNode, value));\r
 518         }\r
 519         else\r
 520         {\r
 521             visit(schemaNode.getName(), value.toString());\r
 522             leave(schemaNode.getName());\r
 523         }\r
 524     }\r
 525 \r
 526     private void evalDatalineColumn(SilkNode node, String columnData) throws XerialException\r
 527     {\r
 528         // 7600 lines/sec\r
 529 \r
 530         if (node.hasChildren())\r
 531         {\r
 532             JSONArray array = new JSONArray(columnData);\r
 533             walkMicroFormatRoot(node, array);\r
 534             return;\r
 535         }\r
 536 \r
 537         switch (node.getOccurrence())\r
 538         {\r
 539         case ZERO_OR_MORE:\r
 540         case ONE_OR_MORE:\r
 541             if (columnData.startsWith("["))\r
 542             {\r
 543                 // micro-data format\r
 544 \r
 545                 // 7900 lines/sec \r
 546 \r
 547                 JSONArray array = new JSONArray(columnData);\r
 548                 // 1400 lines/sec (ANTLR)   4200 lines/sec (JSONTokener)\r
 549 \r
 550                 // 5233 lines/sec w/o traversing JSONArray\r
 551                 //                // dummy code \r
 552                 //                if (true)\r
 553                 //                {\r
 554                 //                    visit(node.getName(), null);\r
 555                 //                    leave(node.getName());\r
 556                 //                    return;\r
 557                 //                }\r
 558                 walkMicroFormatRoot(node, array);\r
 559                 return;\r
 560             }\r
 561             else\r
 562             {\r
 563                 String[] csv = columnData.split(",");\r
 564                 for (String each : csv)\r
 565                 {\r
 566                     String value = each.trim();\r
 567                     evalColumnData(node, value);\r
 568                 }\r
 569                 return;\r
 570             }\r
 571         default:\r
 572             evalColumnData(node, columnData);\r
 573             return;\r
 574         }\r
 575 \r
 576     }\r
 577 \r
 578     private void evalColumnData(SilkNode node, String columnData) throws XerialException\r
 579     {\r
 580         try\r
 581         {\r
 582             if (node.hasChildren())\r
 583             {\r
 584                 // micro-data format\r
 585                 JSONArray array = new JSONArray(columnData);\r
 586                 walkMicroFormatRoot(node, array);\r
 587                 return;\r
 588             }\r
 589 \r
 590             String dataType = node.getDataType();\r
 591             if (dataType != null && dataType.equalsIgnoreCase("json"))\r
 592             {\r
 593                 JSONValue json = JSONUtil.parseJSON(columnData);\r
 594                 if (json.getJSONObject() != null)\r
 595                 {\r
 596                     if (node.getName().equals("_")) // no name object\r
 597                     {\r
 598                         walkJSONValue(json, node.getName());\r
 599                     }\r
 600                     else\r
 601                     {\r
 602                         visit(node.getName(), null);\r
 603                         walkJSONValue(json, node.getName());\r
 604                         leave(node.getName());\r
 605                     }\r
 606                 }\r
 607                 else\r
 608                     walkJSONValue(json, node.getName());\r
 609             }\r
 610             else\r
 611             {\r
 612                 visit(node.getName(), columnData);\r
 613                 leave(node.getName());\r
 614             }\r
 615         }\r
 616         catch (JSONException e)\r
 617         {\r
 618             throw new XerialException(e.getErrorCode(), String.format("line=%d: %s", parser.getNumReadLine(), e\r
 619                     .getMessage()));\r
 620         }\r
 621 \r
 622     }\r
 623 \r
 624     /**\r
 625      * Fill the queue by retrieving the next event from the pull parser.\r
 626      * \r
 627      * @throws XerialException\r
 628      */\r
 629     private void fillQueue() throws XerialException\r
 630     {\r
 631         if (!readerStack.isEmpty())\r
 632         {\r
 633             TreeEvent e = readerStack.peekLast().next();\r
 634             if (e == null)\r
 635             {\r
 636                 readerStack.removeLast();\r
 637                 fillQueue();\r
 638             }\r
 639             else\r
 640             {\r
 641                 eventQueue.push(e);\r
 642                 return;\r
 643             }\r
 644         }\r
 645 \r
 646         if (!parser.hasNext())\r
 647         {\r
 648             // no more input data\r
 649             closeContextUpTo(parseContext.getIndentationOffset());\r
 650             hasFinished = true;\r
 651             return;\r
 652         }\r
 653 \r
 654         SilkEvent currentEvent = parser.next();\r
 655 \r
 656         // update the line count\r
 657         numReadLine = parser.getNumReadLine();\r
 658 \r
 659         if (_logger.isTraceEnabled())\r
 660         {\r
 661             _logger.trace("stack: " + parseContext.getContextNodeStack());\r
 662             _logger.trace(currentEvent);\r
 663         }\r
 664 \r
 665         switch (currentEvent.getType())\r
 666         {\r
 667         case NODE:\r
 668             // push context node\r
 669             SilkNode newContextNode = SilkNode.class.cast(currentEvent.getElement());\r
 670             openContext(newContextNode);\r
 671             break;\r
 672         case FUNCTION:\r
 673             SilkFunction function = SilkFunction.class.cast(currentEvent.getElement());\r
 674             evalFunction(function);\r
 675             break;\r
 676         case DATA_LINE:\r
 677             // pop the context stack until finding a node for stream data node occurrence\r
 678             while (!parseContext.isContextNodeStackEmpty())\r
 679             {\r
 680                 SilkContext context = parseContext.peekLastContext();\r
 681                 SilkNode node = context.contextNode;\r
 682                 if (!node.getOccurrence().isFollowedByStreamData())\r
 683                 {\r
 684                     parseContext.popLastContext();\r
 685                     if (context.isOpen)\r
 686                         leave(node.getName());\r
 687                 }\r
 688                 else\r
 689                     break;\r
 690             }\r
 691 \r
 692             if (parseContext.isContextNodeStackEmpty())\r
 693             {\r
 694                 // use default column names(c1, c2, ...) \r
 695                 SilkDataLine line = SilkDataLine.class.cast(currentEvent.getElement());\r
 696                 String[] columns = line.getDataLine().trim().split("\t");\r
 697                 int index = 1;\r
 698                 visit("row", null);\r
 699                 for (String each : columns)\r
 700                 {\r
 701                     String columnName = String.format("c%d", index++);\r
 702 \r
 703                     // TODO use evalColumnData\r
 704                     visit(columnName, each);\r
 705                     leave(columnName);\r
 706                 }\r
 707                 leave("row");\r
 708             }\r
 709             else\r
 710             {\r
 711                 SilkContext context = parseContext.peekLastContext();\r
 712                 SilkNode schema = context.contextNode;\r
 713                 SilkDataLine line = SilkDataLine.class.cast(currentEvent.getElement());\r
 714                 switch (schema.getOccurrence())\r
 715                 {\r
 716                 case SEQUENCE:\r
 717                     text(line.getDataLine());\r
 718                     break;\r
 719                 case ZERO_OR_MORE:\r
 720                     // CSV data\r
 721                 {\r
 722                     evalDatalineColumn(schema, line.getDataLine());\r
 723                 }\r
 724                     break;\r
 725                 case TABBED_SEQUENCE:\r
 726                 {\r
 727                     String[] columns = line.getDataLine().trim().split("\t");\r
 728                     int columnIndex = 0;\r
 729                     visit(schema.getName(), schema.hasValue() ? schema.getValue().toString() : null);\r
 730                     for (int i = 0; i < schema.getChildNodes().size(); i++)\r
 731                     {\r
 732                         SilkNode child = schema.getChildNodes().get(i);\r
 733                         if (child.hasValue())\r
 734                         {\r
 735                             // output the default value for the column \r
 736                             evalDatalineColumn(child, child.getValue().toString());\r
 737                         }\r
 738                         else\r
 739                         {\r
 740                             if (columnIndex < columns.length)\r
 741                             {\r
 742                                 String columnData = columns[columnIndex++];\r
 743                                 evalDatalineColumn(child, columnData);\r
 744                             }\r
 745                         }\r
 746                     }\r
 747                     leave(schema.getName());\r
 748                     break;\r
 749                 }\r
 750                 case MULTILINE_SEQUENCE:\r
 751                 {\r
 752                     int cursor = parseContext.contextNodeAttributeCursor;\r
 753 \r
 754                     if (cursor >= schema.getChildNodes().size())\r
 755                         break;\r
 756 \r
 757                     SilkNode attributeNode = schema.getChildNodes().get(cursor);\r
 758                     if (cursor == 0 && !parseContext.isAttributeOpen)\r
 759                     {\r
 760                         visit(schema.getName(), schema.hasValue() ? schema.getValue().toString() : null);\r
 761                     }\r
 762                     if (!parseContext.isAttributeOpen)\r
 763                     {\r
 764                         if (attributeNode.hasValue())\r
 765                             visit(attributeNode.getName(), attributeNode.getValue().toString());\r
 766                         else\r
 767                             visit(attributeNode.getName(), null);\r
 768 \r
 769                         parseContext.isAttributeOpen = true;\r
 770                     }\r
 771                     text(line.getDataLine().trim());\r
 772                     break;\r
 773                 }\r
 774                 }\r
 775             }\r
 776             break;\r
 777         case MULTILINE_ENTRY_SEPARATOR: // >>\r
 778         {\r
 779             SilkContext context = parseContext.peekLastContext();\r
 780             SilkNode schema = context.contextNode;\r
 781             if (parseContext.isAttributeOpen)\r
 782             {\r
 783                 SilkNode attributeNode = schema.getChildNodes().get(parseContext.contextNodeAttributeCursor);\r
 784                 leave(attributeNode.getName());\r
 785             }\r
 786             leave(schema.getName());\r
 787             // reset\r
 788             parseContext.contextNodeAttributeCursor = 0;\r
 789             parseContext.isAttributeOpen = false;\r
 790             break;\r
 791         }\r
 792         case MULTILINE_SEPARATOR: // --\r
 793         {\r
 794             SilkContext context = parseContext.peekLastContext();\r
 795             SilkNode schema = context.contextNode;\r
 796             if (parseContext.isAttributeOpen)\r
 797             {\r
 798                 SilkNode attributeNode = schema.getChildNodes().get(parseContext.contextNodeAttributeCursor);\r
 799                 leave(attributeNode.getName());\r
 800             }\r
 801             parseContext.contextNodeAttributeCursor++;\r
 802             parseContext.isAttributeOpen = false;\r
 803             break;\r
 804         }\r
 805         case BLANK_LINE:\r
 806             break;\r
 807 \r
 808         case PREAMBLE:\r
 809             break;\r
 810 \r
 811         }\r
 812 \r
 813     }\r
 814 \r
 815     private void walkJSONAray(JSONArray jsonArray, String parentNodeName) throws XerialException\r
 816     {\r
 817         for (JSONValue each : jsonArray)\r
 818         {\r
 819             walkJSONValue(each, parentNodeName);\r
 820         }\r
 821     }\r
 822 \r
 823     private void walkJSONObject(JSONObject jsonObj) throws XerialException\r
 824     {\r
 825         for (String key : jsonObj.keys())\r
 826         {\r
 827             JSONValue val = jsonObj.get(key);\r
 828             walkJSONValue(val, key);\r
 829         }\r
 830     }\r
 831 \r
 832     private void walkJSONValue(JSONValue value, String parentNodeName) throws XerialException\r
 833     {\r
 834         JSONValueType type = value.getValueType();\r
 835         switch (type)\r
 836         {\r
 837         case Array:\r
 838             walkJSONAray(value.getJSONArray(), parentNodeName);\r
 839             break;\r
 840         case Object:\r
 841             walkJSONObject(value.getJSONObject());\r
 842             break;\r
 843         case Boolean:\r
 844             visit(parentNodeName, value.toString());\r
 845             leave(parentNodeName);\r
 846             break;\r
 847         case Double:\r
 848             visit(parentNodeName, value.toString());\r
 849             leave(parentNodeName);\r
 850             break;\r
 851         case Integer:\r
 852             visit(parentNodeName, value.toString());\r
 853             leave(parentNodeName);\r
 854             break;\r
 855         case Null:\r
 856             visit(parentNodeName, value.toString());\r
 857             leave(parentNodeName);\r
 858             break;\r
 859         case String:\r
 860             visit(parentNodeName, value.toString());\r
 861             leave(parentNodeName);\r
 862             break;\r
 863         }\r
 864 \r
 865     }\r
 866 \r
 867     /**\r
 868      * Plugin holder\r
 869      */\r
 870     private static Map<String, Class<SilkFunctionPlugin>> pluginTable = null;\r
 871 \r
 872     /**\r
 873      * Get the plugin of the specified name\r
 874      * \r
 875      * @param name\r
 876      *            plugin name\r
 877      * @return plugin instance or null if no corresponding plugin is found.\r
 878      */\r
 879     private SilkFunctionPlugin getPlugin(String name)\r
 880     {\r
 881         Class<SilkFunctionPlugin> pluginClass = getPluginTable().get(name);\r
 882         if (pluginClass == null)\r
 883             return null;\r
 884 \r
 885         try\r
 886         {\r
 887             SilkFunctionPlugin pluginInstance = pluginClass.newInstance();\r
 888             return pluginInstance;\r
 889         }\r
 890         catch (InstantiationException e)\r
 891         {\r
 892             _logger.warn(e);\r
 893             return null;\r
 894         }\r
 895         catch (IllegalAccessException e)\r
 896         {\r
 897             _logger.warn(e);\r
 898             return null;\r
 899         }\r
 900     }\r
 901 \r
 902     private Map<String, Class<SilkFunctionPlugin>> getPluginTable()\r
 903     {\r
 904         if (pluginTable == null)\r
 905         {\r
 906             pluginTable = new TreeMap<String, Class<SilkFunctionPlugin>>();\r
 907             // load plugins \r
 908             for (Class<SilkFunctionPlugin> each : FileResource.findClasses(SilkFunctionPlugin.class.getPackage(),\r
 909                     SilkFunctionPlugin.class, SilkWalker.class.getClassLoader()))\r
 910             {\r
 911                 String functionName = each.getSimpleName().toLowerCase();\r
 912                 _logger.trace("loaded " + functionName);\r
 913                 pluginTable.put(functionName, each);\r
 914             }\r
 915         }\r
 916 \r
 917         return pluginTable;\r
 918     }\r
 919 \r
 920     /**\r
 921      * Information of the function (plugin) arguments (\r
 922      * {@link SilkFunctionArgument}) described in the Class definition, which\r
 923      * implements {@link SilkFunctionPlugin}.\r
 924      * \r
 925      * @author leo\r
 926      * \r
 927      */\r
 928     private static class PluginField\r
 929     {\r
 930         Field field;\r
 931         SilkFunctionArgument argInfo;\r
 932 \r
 933         private PluginField(SilkFunctionArgument argInfo, Field field)\r
 934         {\r
 935             this.argInfo = argInfo;\r
 936             this.field = field;\r
 937         }\r
 938     }\r
 939 \r
 940     private static class PluginHolder\r
 941     {\r
 942         Class< ? extends SilkFunctionPlugin> pluginClass;\r
 943         ArrayList<PluginField> argumentFieldList = new ArrayList<PluginField>();\r
 944         Map<String, PluginField> keyValueFieldTable = new HashMap<String, PluginField>();\r
 945 \r
 946         public PluginHolder(Class< ? extends SilkFunctionPlugin> pluginClass)\r
 947         {\r
 948             this.pluginClass = pluginClass;\r
 949 \r
 950             //ArrayList<SilkFunctionArgument> argDefs = new ArrayList<SilkFunctionArgument>();\r
 951             for (Field eachField : pluginClass.getDeclaredFields())\r
 952             {\r
 953                 SilkFunctionArgument argInfo = eachField.getAnnotation(SilkFunctionArgument.class);\r
 954                 if (argInfo != null)\r
 955                 {\r
 956                     PluginField pf = new PluginField(argInfo, eachField);\r
 957                     if (argInfo.name().equals(SilkFunctionArgument.NO_VALUE))\r
 958                         argumentFieldList.add(pf);\r
 959                     else\r
 960                         keyValueFieldTable.put(argInfo.name(), pf);\r
 961                 }\r
 962             }\r
 963 \r
 964             // sort arguments in the order of their ordinal\r
 965             Collections.sort(argumentFieldList, new Comparator<PluginField>() {\r
 966                 public int compare(PluginField o1, PluginField o2)\r
 967                 {\r
 968                     return o1.argInfo.ordinal() - o2.argInfo.ordinal();\r
 969                 }\r
 970             });\r
 971 \r
 972         }\r
 973 \r
 974         /**\r
 975          * Bind function arguments to the plug-in instance\r
 976          * \r
 977          * @param plugin\r
 978          *            the instance of the plug-in\r
 979          * @param args\r
 980          *            the function arguments\r
 981          */\r
 982         public void populate(SilkFunctionPlugin plugin, List<SilkFunctionArg> args)\r
 983         {\r
 984             int noNameArgCount = 0;\r
 985             for (SilkFunctionArg eachArgument : args)\r
 986             {\r
 987                 String argValue = eachArgument.getValue().toString();\r
 988                 try\r
 989                 {\r
 990                     if (eachArgument.hasName())\r
 991                     {\r
 992                         // key value arg\r
 993                         PluginField f = keyValueFieldTable.get(eachArgument.getName());\r
 994                         if (f == null)\r
 995                         {\r
 996                             _logger.warn("unknown argument: " + eachArgument);\r
 997                             continue;\r
 998                         }\r
 999                         ReflectionUtil.setFieldValue(plugin, f.field, argValue);\r
1000                     }\r
1001                     else\r
1002                     {\r
1003                         // unnamed argument\r
1004                         // matching argument order\r
1005                         if (noNameArgCount >= argumentFieldList.size())\r
1006                         {\r
1007                             _logger.warn(String.format("no corresponding field for the argument %s is found",\r
1008                                     eachArgument));\r
1009                             continue;\r
1010                         }\r
1011                         PluginField f = argumentFieldList.get(noNameArgCount);\r
1012                         ReflectionUtil.setFieldValue(plugin, f.field, argValue);\r
1013 \r
1014                         if (!TypeInformation.isCollection(f.field.getType()))\r
1015                             noNameArgCount++;\r
1016                     }\r
1017                 }\r
1018                 catch (XerialException e)\r
1019                 {\r
1020                     _logger.error(e);\r
1021                 }\r
1022 \r
1023             }\r
1024         }\r
1025 \r
1026     }\r
1027 \r
1028     /**\r
1029      * Fill the plug-in argument fields with the given arguments\r
1030      * \r
1031      * @param plugin\r
1032      *            plug-in instance.\r
1033      * @param args\r
1034      *            function arguments.\r
1035      */\r
1036     private static void populate(SilkFunctionPlugin plugin, List<SilkFunctionArg> args)\r
1037     {\r
1038         PluginHolder holder = new PluginHolder(plugin.getClass());\r
1039         holder.populate(plugin, args);\r
1040     }\r
1041 \r
1042     public int getNumReadLine()\r
1043     {\r
1044         return numReadLine;\r
1045     }\r
1046 \r
1047 }\r