update diff parser API

update/diff/parser.go

@@ -3,6 +3,7 @@ package diff
 import (
 	"compress/gzip"
 	"encoding/xml"
+	"io"
 	"os"
 	"strconv"
 	"time"
@@ -22,37 +23,77 @@ type DiffElem struct {
 	Rel  *element.Relation
 }
 
-func Parse(diff string) (chan DiffElem, chan error) {
-	elems := make(chan DiffElem)
-	errc := make(chan error)
-	go parse(diff, elems, errc, false)
-	return elems, errc
+type Parser struct {
+	reader   io.Reader
+	elems    chan DiffElem
+	errc     chan error
+	metadata bool
+	running  bool
+	onClose  func() error
 }
 
-func ParseFull(diff string) (chan DiffElem, chan error) {
-	elems := make(chan DiffElem)
-	errc := make(chan error)
-	go parse(diff, elems, errc, true)
-	return elems, errc
+func (p *Parser) SetWithMetadata(metadata bool) {
+	p.metadata = metadata
 }
 
-func parse(diff string, elems chan DiffElem, errc chan error, metadata bool) {
-	defer close(elems)
-	defer close(errc)
-
-	file, err := os.Open(diff)
-	if err != nil {
-		errc <- err
-		return
+func (p *Parser) Next() (DiffElem, error) {
+	if !p.running {
+		p.running = true
+		go parse(p.reader, p.elems, p.errc, p.metadata)
+	}
+	select {
+	case elem, ok := <-p.elems:
+		if !ok {
+			p.elems = nil
+		} else {
+			return elem, nil
+		}
+	case err, ok := <-p.errc:
+		if !ok {
+			p.errc = nil
+		} else {
+			if p.onClose != nil {
+				p.onClose()
+				p.onClose = nil
+			}
+			return DiffElem{}, err
+		}
+	}
+	if p.onClose != nil {
+		err := p.onClose()
+		p.onClose = nil
+		return DiffElem{}, err
+	}
+	return DiffElem{}, nil
+}
+
+func NewDecoder(r io.Reader) *Parser {
+	elems := make(chan DiffElem)
+	errc := make(chan error)
+	return &Parser{reader: r, elems: elems, errc: errc}
+}
+
+func NewOscGzDecoder(fname string) (*Parser, error) {
+	file, err := os.Open(fname)
+	if err != nil {
+		return nil, err
 	}
-	defer file.Close()
 
 	reader, err := gzip.NewReader(file)
 	if err != nil {
-		errc <- err
-		return
+		file.Close()
+		return nil, err
 	}
 
+	elems := make(chan DiffElem)
+	errc := make(chan error)
+	return &Parser{reader: reader, elems: elems, errc: errc, onClose: file.Close}, nil
+}
+
+func parse(reader io.Reader, elems chan DiffElem, errc chan error, metadata bool) {
+	defer close(elems)
+	defer close(errc)
+
 	decoder := xml.NewDecoder(reader)
 
 	add := false
@@ -190,6 +231,7 @@ NextToken:
 				rel = &element.Relation{}
 				newElem = true
 			case "osmChange":
+				errc <- io.EOF
 				return
 			}
 

update/process.go

@@ -3,6 +3,7 @@ package update
 import (
 	"errors"
 	"fmt"
+	"io"
 	"path/filepath"
 	"runtime"
 
@@ -100,7 +101,10 @@ func Update(oscFile string, geometryLimiter *limit.Limiter, expireor expire.Expi
 
 	defer log.StopStep(log.StartStep(fmt.Sprintf("Processing %s", oscFile)))
 
-	elems, errc := diff.Parse(oscFile)
+	parser, err := diff.NewOscGzDecoder(oscFile)
+	if err != nil {
+		return err
+	}
 
 	tagmapping, err := mapping.NewMapping(config.BaseOptions.MappingFile)
 	if err != nil {
@@ -197,11 +201,12 @@ func Update(oscFile string, geometryLimiter *limit.Limiter, expireor expire.Expi
 	g := geos.NewGeos()
 
 	for {
-		select {
-		case elem, ok := <-elems:
-			if !ok {
-				elems = nil
-				break
+		elem, err := parser.Next()
+		if err == io.EOF {
+			break // finished
+		}
+		if err != nil {
+			return diffError(err, "")
 		}
 		if elem.Rel != nil {
 			relTagFilter.Filter(&elem.Rel.Tags)
@@ -299,16 +304,6 @@ func Update(oscFile string, geometryLimiter *limit.Limiter, expireor expire.Expi
 				}
 			}
 		}
-		case err, ok := <-errc:
-			if !ok {
-				errc = nil
-				break
-			}
-			return diffError(err, "")
-		}
-		if errc == nil && elems == nil {
-			break
-		}
 	}
 
 	// mark member ways from deleted relations for re-insert