| 96 | } |
| 97 | |
| 98 | func (s *state) parseStream(done chan error) { |
| 99 | for line := range s.input { |
| 100 | line = strings.Trim(line, " \t") |
| 101 | if len(line) == 0 { |
| 102 | glog.Info("Empty line.") |
| 103 | continue |
| 104 | } |
| 105 | |
| 106 | glog.Debugf("Got line: %q", line) |
| 107 | nq, err := rdf.Parse(line) |
| 108 | if err != nil { |
| 109 | glog.WithError(err).Errorf("While parsing: %q", line) |
| 110 | done <- err |
| 111 | return |
| 112 | } |
| 113 | s.cnq <- nq |
| 114 | atomic.AddUint64(&s.ctr.parsed, 1) |
| 115 | } |
| 116 | done <- nil |
| 117 | } |
| 118 | |
| 119 | func (s *state) getEntity(db *neoism.Database, xid string) *neoism.Node { |
| 120 | p := neoism.Props{"_xid_": xid} |