Document streaming. Test error handling for stream.

README.md

@@ -33,8 +33,6 @@ This is meant to run commands which you don't care about the return code. `cmd!/
     {:ok, res, 0} = SSHEx.run conn, 'ls /some/path'
 ```
 
-If `:ssh` returns any error (i.e. `{:error, reason}`, `:failure`, etc.), `SSHEx` will raise a `RuntimeError` with the message containing an `inspect` of whichever return value it got from `:ssh` (i.e. `"{:error, reason}"`, and so on). No attempt to ease the pain.
-
 You can pass the option `:separate_streams` to get separated stdout and stderr. Like this:
 
 ```elixir
@@ -45,6 +43,29 @@ You can pass the option `:separate_streams` to get separated stdout and stderr.
 You will be reusing the same SSH connection all over.
 
 
+## Streaming
+
+You can use `SSHEx` to run some command and create a [`Stream`](http://elixir-lang.org/docs/stable/elixir/Stream.html), so you can lazily process an arbitrarily long output as it arrives. Internally `Stream.resource/3` is used to create the `Stream`, and every response from `:ssh` is emitted so it can be easily matched with a simple `case`.
+
+You just have to use `stream/3` like this:
+
+```elixir
+  str = SSHEx.stream conn, 'somecommand'
+
+  Stream.each(str, fn(x)->
+    case x do
+      {:stdout,row}    -> process_output(row)
+      {:stderr,row}    -> process_error(row)
+      {:status,status} -> process_exit_status(status)
+    end
+  end)
+```
+
+## Error handling
+
+If `:ssh` returns any error (i.e. `{:error, reason}`, `:failure`, etc.), `SSHEx` will raise a `RuntimeError` with the message containing an `inspect` of whichever return value it got from `:ssh` (i.e. `"{:error, reason}"`, and so on). No attempt to ease the pain by now (will see in 2.0).
+
+
 ## Alternative keys
 
 To use alternative keys you should save them somewhere on disk and then set the `:user_dir` option for `:ssh.connect/4`. See [ssh library docs](http://www.erlang.org/doc/man/ssh.html) for more options.
@@ -60,6 +81,7 @@ To use alternative keys you should save them somewhere on disk and then set the
 
 ### master
 
+* Support streaming
 * Stop using global mocks (i.e. `:meck`)
 
 ### 1.2

lib/sshex.ex

@@ -54,7 +54,47 @@ defmodule SSHEx do
   end
 
   @doc """
-    TODO: docs
+    Gets an open SSH connection reference (as returned by `:ssh.connect/4`),
+    and a command to execute.
+
+    Optionally it gets a timeout for the underlying SSH channel opening,
+    and for the execution itself.
+
+    Supported options are:
+
+    * `:channel_timeout`
+    * `:exec_timeout`
+    * `:connection_module`
+
+    Any failure related with the SSH connection itself is raised without mercy (by now).
+
+    Returns a `Stream` that you can use to lazily retrieve each line of output
+    for the given command.
+
+    Each iteration of the stream will read from the underlying connection and
+    return one of these:
+
+    * `{:stdout,row}`
+    * `{:stderr,row}`
+    * `{:status,status}`
+
+    Keep in mind that rows may not be received in order.
+
+    Ex:
+    ```
+      {:ok, conn} = :ssh.connect('123.123.123.123', 22,
+                    [ {:user,'myuser'}, {:silently_accept_hosts, true} ], 5000)
+
+      str = SSHEx.stream conn, 'somecommand'
+
+      Stream.each(str, fn(x)->
+        case x do
+          {:stdout,row}    -> process_output(row)
+          {:stderr,row}    -> process_error(row)
+          {:status,status} -> process_exit_status(status)
+        end
+      end)
+    ```
   """
   def stream(conn, cmd, opts \\ []) do
     opts = opts |> H.defaults(connection_module: :ssh_connection,
@@ -65,7 +105,7 @@ defmodule SSHEx do
 
     next_fun = fn(channel)->
       if channel == :halt_next do # halt if asked
-        {:halt, :bogus}
+        {:halt, 'Halt requested on previous iteration'}
       else
         res = receive_and_parse_response(channel, opts[:exec_timeout])
         case res do
@@ -74,7 +114,8 @@ defmodule SSHEx do
           {:loop, {_, _, "",  x, nil, false}} -> {[ {:stderr,x} ], channel}
           {:loop, {_, _, "", "",   x, false}} -> {[ {:status,x} ], channel}
           {:loop, {_, _, "", "", nil, true }} -> {:halt, channel}
-          {:error, reason} = x -> {[x], :halt_next} # emit error, then halt
+          # TODO: wait until 2.0 to really handle errors
+          # {:error, reason} = x -> {[x], :halt_next} # emit error, then halt
           any -> raise inspect(any)
         end
       end

test/sshex_test.exs

@@ -76,10 +76,23 @@ defmodule SSHExTest do
     assert Enum.to_list(stream) == response
   end
 
-  defp send_long_sequence(lines) do
+  test "`:ssh` error message when `stream`" do
+    lines = ["some", "long", "output", "sequence"]
+    send_long_sequence(lines, error: true)
+
+    stream = SSHEx.stream :mocked, 'somecommand', connection_module: AllOKMock
+    assert_raise RuntimeError, "{:error, :reason}", fn ->
+      Enum.to_list(stream)
+    end
+  end
+
+  defp send_long_sequence(lines, opts \\ []) do
     for l <- lines do
       send self(), {:ssh_cm, :mocked, {:data, :mocked, 0, l}}
     end
+
+    if opts[:error], do: send(self(), {:ssh_cm, :mocked, {:error, :reason}})
+
     send self(), {:ssh_cm, :mocked, {:data, :mocked, 1, "mockederror"}}
     send self(), {:ssh_cm, :mocked, {:eof, :mocked}}
     send self(), {:ssh_cm, :mocked, {:exit_status, :mocked, 0}}