it works

Author: rozap

lib/exkml.ex

@@ -20,6 +20,10 @@ defmodule Exkml do
     defstruct geometries: []
   end
 
+  defmodule KMLParseError do
+    defexception [:message, :event]
+  end
+
   defp do_str_to_point([x, y]) do
     with {x, _} <- Float.parse(x),
       {y, _} <- Float.parse(y) do
@@ -84,12 +88,14 @@ defmodule Exkml do
     defstruct [
       :receiver,
       :receiver_ref,
+      status: :out_kml,
       geom_stack: [],
       placemark: nil,
       stack: [],
       path: [],
       emit: [],
-      point_count: 0
+      point_count: 0,
+      batch_size: 64
     ]
   end
 
@@ -101,8 +107,8 @@ defmodule Exkml do
     %State{s | placemark: %Placemark{pm | attrs: Map.put(attrs, name, value)}}
   end
 
-  def put_error(state, reason) do
-    IO.inspect {:error, reason}
+  def put_error(state, _reason) do
+    # TODO: partial error handling?
     state
   end
 
@@ -147,7 +153,7 @@ defmodule Exkml do
     %State{state | geom_stack: [merge_up(child, parent, kind) | rest]}
   end
 
-  def pop_geom(state, kind) do
+  def pop_geom(_state, kind) do
     throw "Cannot pop #{kind}"
   end
 
@@ -178,7 +184,7 @@ defmodule Exkml do
   end
 
   defp merge_up(child, parent, _) do
-    throw "No merge_up impl #{inspect parent}"
+    throw "No merge_up impl #{inspect child} #{inspect parent}"
   end
 
   def put_point(%State{} = state, text) do
@@ -268,63 +274,85 @@ defmodule Exkml do
     %{emit(state) | stack: [], path: [], placemark: nil}
   end
 
-  def on_event(:endDocument, _, %State{receiver: r, receiver_ref: ref} = state) do
+  on_enter 'kml', _, state, do: %State{state | status: :kml}
+  on_exit  'kml', _, state, do: %State{state | status: :out_kml}
+
+  def on_event(:endDocument, _, %State{status: :out_kml, receiver: r, receiver_ref: ref} = state) do
     flush(state)
     send(r, {:done, ref})
     state
   end
 
+  def on_event(:endDocument, event, %State{status: :kml, receiver: r, receiver_ref: ref} = state) do
+    send(r, {:error, ref, self(), event})
+    state
+  end
+
   def on_event(_event, _, state), do: state
 
   defp flush(%State{receiver: r, receiver_ref: ref, emit: emit} = state) do
-    send(r, {:placemarks, ref, Enum.reverse(emit)})
+    send(r, {:placemarks, ref, self(), Enum.reverse(emit)})
+    receive do
+      {:ack, ^ref} -> :ok
+    end
     %State{state | emit: []}
   end
 
-
-  def emit(state) do
+  def emit(%State{batch_size: batch_size} = state) do
     case %State{state | emit: [state.placemark | state.emit]} do
-      %State{emit: emit} when length(emit) > 10 ->
-        flush(state)
+      %State{emit: emit} = new_state when length(emit) > batch_size -> flush(new_state)
       new_state -> new_state
     end
   end
 
+  def setup(binstream, chunk_size, ref) do
+    receiver = self()
 
+    spawn_link(fn ->
+      continuation = &Enumerable.reduce(binstream, &1, fn
+        x, {acc, counter} when counter <= 0 -> {:suspend, {[x | acc], 0}}
+        x, {acc, counter} -> {:cont, {[x | acc], counter - :erlang.byte_size(x)}}
+      end)
 
-  def stream!(binstream, chunk_size \\ 2048) do
-    continuation = &Enumerable.reduce(binstream, &1, fn
-      x, {acc, counter} when counter <= 0 -> {:suspend, {[x | acc], 0}}
-      x, {acc, counter} -> {:cont, {[x | acc], counter - :erlang.byte_size(x)}}
+      take = fn cont ->
+        case cont.({:cont, {[], chunk_size}}) do
+          {:suspended, {list, 0}, new_cont} ->
+            {:lists.reverse(list) |> Enum.join(), new_cont}
+          {status, {list, _}} ->
+            {:lists.reverse(list) |> Enum.join(), status}
+        end
+      end
+
+      :xmerl_sax_parser.stream("", [
+        continuation_fun: take,
+        continuation_state: continuation,
+        event_fun: &Exkml.on_event/3,
+        event_state: %State{receiver: receiver, receiver_ref: ref},
+        encoding: :utf8
+      ])
     end)
+  end
 
-    take = fn cont ->
-      case cont.({:cont, {[], chunk_size}}) do
-        {:suspended, {list, 0}, new_cont} ->
-          {:lists.reverse(list) |> Enum.join(), new_cont}
-        {status, {list, _}} ->
-          {:lists.reverse(list) |> Enum.join(), status}
-      end
-    end
+  def stage(binstream, chunk_size \\ 2048) do
+    Exkml.Stage.start_link(binstream, chunk_size)
+  end
 
-    ref = make_ref()
 
-    :xmerl_sax_parser.stream("", [
-      continuation_fun: take,
-      continuation_state: continuation,
-      event_fun: &Exkml.on_event/3,
-      event_state: %State{receiver: self(), receiver_ref: ref},
-      encoding: :utf8
-    ])
+  def stream!(binstream, chunk_size \\ 2048) do
+    ref = make_ref()
+    pid = setup(binstream, chunk_size, ref)
 
     Stream.resource(
       fn -> :ok end,
       fn state ->
         receive do
-          {:placemarks, ^ref, pms} ->
+          {:placemarks, ^ref, from, pms} ->
+            send from, {:ack, ref}
             {pms, state}
           {:done, ^ref} ->
             {:halt, state}
+          {:error, ^ref, ^pid, event} ->
+            raise KMLParseError, message: "Document ended prematurely", event: event
         end
       end,
       fn _ -> :ok end

lib/exkml/stage.ex

@@ -0,0 +1,57 @@
+defmodule Exkml.Stage do
+  use GenStage
+
+  def start_link(binstream, chunk_size) do
+    GenStage.start_link(__MODULE__, [binstream, chunk_size])
+  end
+
+  def init([binstream, chunk_size]) do
+    ref = make_ref()
+    Exkml.setup(binstream, chunk_size, ref)
+    buf = []
+    demand = 0
+    {:producer, {:started, ref, buf, demand, nil}}
+  end
+
+  def handle_demand(more_demand, {status, ref, buf, demand, from}) do
+    new_demand = demand + more_demand
+    {emit, keep} = Enum.split(buf, new_demand)
+    less_demand = new_demand - length(emit)
+
+    new_state = {status, ref, keep, less_demand, from}
+
+    ack(ref, from)
+
+    maybe_end(new_state)
+
+    {:noreply, emit, new_state}
+  end
+
+  def handle_info({:placemarks, ref, from, pms}, {status, ref, buf, 0, _}) do
+    {:noreply, [], {status, ref, buf ++ pms, 0, from}}
+  end
+
+  def handle_info({:placemarks, ref, from, pms}, {status, ref, buf, demand, _}) do
+    {emit, keep} = Enum.split(buf ++ pms, demand)
+    new_demand = demand - length(emit)
+    ack(ref, from)
+
+    {:noreply, emit, {status, ref, keep, new_demand, from}}
+  end
+
+  def handle_info({:done, ref}, {_, ref, buf, demand, from}) do
+    new_state = {:done, ref, buf, demand, from}
+    maybe_end(new_state)
+    {:noreply, [], new_state}
+  end
+
+  def handle_info(:stop, state) do
+    {:stop, :normal, state}
+  end
+
+  defp ack(_, nil), do: :ok
+  defp ack(ref, from), do: send from, {:ack, ref}
+
+  defp maybe_end({:done, _, [], _, _}), do: GenStage.async_info(self(), :stop)
+  defp maybe_end(_), do: :nope
+end

mix.exs

@@ -4,7 +4,7 @@ defmodule Exkml.Mixfile do
   def project do
     [
       app: :exkml,
-      version: "0.1.2",
+      version: "0.2.0",
       elixir: "~> 1.5",
       start_permanent: Mix.env == :prod,
       deps: deps(),
@@ -37,11 +37,8 @@ defmodule Exkml.Mixfile do
   # Run "mix help deps" to learn about dependencies.
   defp deps do
     [
-      {:sweet_xml, "~> 0.6.5"},
       {:ex_doc, ">= 0.0.0", only: :dev},
-      {:fast_xml, "~> 1.1"},
-      {:gen_stage, "~> 0.12.2"},
-      {:erlsom, "~> 1.4.1"}
+      {:gen_stage, "~> 0.12.2"}
     ]
   end
 end

test.sh

@@ -1,8 +1,4 @@
 #!/bin/bash
-
-# Simple thing to watch your project and run tests on change
-# Takes one optional argument, which will grep tests and run the ones that match
-
 while inotifywait -r -e modify ./test ./lib; do
   mix test test/exkml_test.exs
 done

test/exkml_test.exs

@@ -257,56 +257,77 @@ defmodule ExkmlTest do
     end)
   end
 
-  # test "malformed" do
-  #   Process.flag(:trap_exit, true)
-
-  #   assert "malformed_kml"
-  #   |> kml_fixture
-  #   |> Exkml.placemarks!()
-  #   |> Enum.into([])
-
-  #   assert_receive {:EXIT, _, {:fatal, _}}
-  #   Process.flag(:trap_exit, false)
-  # end
+  test "malformed" do
+    assert_raise Exkml.KMLParseError, ~r"ended prematurely", fn ->
+      "malformed_kml"
+      |> kml_fixture
+      |> Exkml.stream!()
+      |> Enum.into([])
+    end
+  end
 
 
   Enum.each([
-    {"boundaries", [Multigeometry]},
-    {"cgis-en-6393", [Point]},
-    {"la_bikelanes", [Multigeometry, Line]},
-    {"noaa", [Point]},
-    {"terrassa", [Multigeometry, Point]},
-    {"wards", [Polygon, Multigeometry]}
-  ], fn {name, kinds} ->
+    {"boundaries", [Multigeometry], 163},
+    {"cgis-en-6393", [Point], 233},
+    {"la_bikelanes", [Multigeometry, Line], 12844},
+    {"noaa", [Point], 227},
+    {"terrassa", [Multigeometry, Point], 73},
+    {"wards", [Polygon, Multigeometry], 53}
+  ], fn {name, kinds, expected_length} ->
     test "smoke #{name}" do
+
       expected_set = MapSet.new(unquote(kinds))
 
-      out = "smoke/#{unquote(name)}"
+      "smoke/#{unquote(name)}"
       |> kml_fixture
       |> Exkml.stream!()
-      |> Enum.into([])
+      |> compare_stream(unquote(expected_length), expected_set)
+
 
-      assert length(out) > 0
+      {:ok, stage} = "smoke/#{unquote(name)}"
+      |> kml_fixture
+      |> Exkml.stage()
+
+
+      GenStage.stream([{stage, max_demand: 2, cancel: :transient}])
+      |> compare_stream(unquote(expected_length), expected_set)
 
-      out
-      |> Enum.each(fn %Placemark{geoms: shapes} ->
-        Enum.each(shapes, fn actual ->
-          assert actual.__struct__ in expected_set
-        end)
-      end)
     end
   end)
 
-  # test "large" do
-  #   File.stream!("/home/chris/Downloads/large.kml", [], 2048)
-  #   |> Exkml.placemarks!
-  #   |> Enum.take(20_000)
+  def compare_stream(stream, expected_length, expected_set) do
+    out = Enum.into(stream, [])
+
+    assert length(out) == expected_length
+
+    actual_set = Enum.flat_map(out, fn %Placemark{geoms: shapes} ->
+      Enum.map(shapes, fn actual -> actual.__struct__ end)
+    end)
+    |> MapSet.new
+
+    assert actual_set == expected_set
+  end
+
+
+  # test "prof" do
+  #   proc = self()
+  #   spawn(fn ->
+  #     :fprof.trace([:start, {:procs, :all}])
+  #     :timer.sleep(5_000)
+  #     :fprof.trace([:stop])
+  #     :fprof.profile
+  #     :fprof.analyse({:dest, 'outfile.analysis'})
+  #     send proc, :done
+  #   end)
+
+  #   out = "smoke/usbr"
+  #   |> kml_fixture
+  #   |> Exkml.stream!()
+  #   |> Enum.take(1)
 
   #   receive do
-  #     any -> IO.inspect {:wat, any}
-  #   after
-  #     1000 -> IO.inspect :nope
+  #     :done -> :ok
   #   end
   # end
-
 end