XmlSlurperStaticExtensions.groovy

package reactor.groovy.ext

import groovy.transform.CompileStatic
import groovy.util.slurpersupport.GPathResult
import org.reactivestreams.Publisher
import reactor.io.buffer.Buffer
import reactor.rx.Stream
import reactor.rx.Streams

@CompileStatic
class XmlSlurperStaticExtensions {

    static Stream<GPathResult> parse(XmlSlurper selfType, Publisher<Buffer> p) {
        def x = new XmlSlurper()
        def b = new Buffer()
        Streams.createWith { req, sub ->
            Streams.wrap(p).
                    observeComplete {
                        b.flip()
                        sub.onNext(x.parse(new ByteArrayInputStream(b.asBytes())))
                    }.
                    consume {
                        b.append(it)
                    }
        }
    }

}