3.x: merged gar1t's fork

.gitignore

@@ -1,5 +1,6 @@
 ebin/*.app
 ebin/*.beam
+test/*.beam
 priv/*.so
 c_src/*.o
 deps
@@ -8,9 +9,9 @@ perf/*.beam
 .eunit
 doc/edoc-info
 doc/erlang.png
-doc/ezmq.html
-doc/index.html
-doc/overview-summary.html
-doc/packages-frame.html
+doc/*.html
 doc/stylesheet.css
-doc/modules-frame.html
+*~
+*.swa
+*.swp
+*.swo

include/erlzmq.hrl

@@ -125,3 +125,4 @@
 %% Possible option values for {@link erlzmq:setsockopt/3. setsockopt/3}.
 -type erlzmq_sockopt_value() :: integer() | iolist().
 
+-record(zmq, {socket, data, flags}).

src/erlzmq.erl

@@ -300,7 +300,11 @@ setsockopt({I, Socket}, Name, Value) when is_integer(I), is_atom(Name) ->
     {ok, erlzmq_sockopt_value()} |
     erlzmq_error().
 getsockopt({I, Socket}, Name) when is_integer(I), is_atom(Name) ->
-    erlzmq_nif:getsockopt(Socket, option_name(Name)).
+    coerce_sockopt(Name, erlzmq_nif:getsockopt(Socket, option_name(Name))).
+
+coerce_sockopt(rcvmore, {ok, 0}) -> {ok, false};
+coerce_sockopt(rcvmore, {ok, 1}) -> {ok, true};
+coerce_sockopt(_Other, Val) -> Val.
 
 %% @equiv close(Socket, infinity)
 -spec close(Socket :: erlzmq_socket()) ->
@@ -368,6 +372,9 @@ term(Context, Timeout) ->
     end.
 
 %% @doc Returns the 0MQ library version.
+%% <br />
+%% <i>For more information see
+%% <a href="http://api.zeromq.org/master:zmq_version">zmq_version</a>.</i>
 %% @end
 -spec version() -> {integer(), integer(), integer()}.
 

src/erlzmq_device.erl

@@ -0,0 +1,60 @@
+-module(erlzmq_device).
+
+-export([queue/2]).
+
+-import(proplists, [get_bool/2]).
+
+%%--------------------------------------------------------------------
+%% @doc A queue device implemented in Erlang.
+%%
+%% Frontend and Backend must be sockets in active mode.
+%%
+%% This function will not return.
+%%
+%% @spec queue(Frontend, Backend) -> any()
+%% Frontend = erlzmq_socket()
+%% Backend = erlzmq_socket()
+%% @end
+%%--------------------------------------------------------------------
+
+queue(Frontend, Backend) ->
+    receive
+        {zmq, Frontend, Msg, Flags} ->
+            Parts = lists:reverse(queue_recv_acc(Frontend, Flags, [Msg])),
+            queue_send(Backend, Parts),
+            queue(Frontend, Backend);
+        {zmq, Backend, Msg, Flags} ->
+            Parts = lists:reverse(queue_recv_acc(Backend, Flags, [Msg])),
+            queue_send(Frontend, Parts),
+            queue(Frontend, Backend);
+        {shutdown} ->
+            ok
+    end.
+
+%%--------------------------------------------------------------------
+%% @doc Accumulates messages from Socket.
+%% @spec queue_recv_acc(Socket, Flags, Acc0) -> Acc
+%% @end
+%%--------------------------------------------------------------------
+
+queue_recv_acc(Socket, Flags0, Acc) ->
+    case get_bool(rcvmore, Flags0) of
+        true ->
+            receive
+                {zmq, Socket, Msg, Flags} ->
+                    queue_recv_acc(Socket, Flags, [Msg|Acc])
+            end;
+        false -> Acc
+    end.
+
+%%--------------------------------------------------------------------
+%% @doc Sends a multipart message to Out.
+%% @spec queue_send(erlzmq_socket(), Parts) -> ok
+%% @end
+%%--------------------------------------------------------------------
+
+queue_send(Out, [LastPart]) ->
+    ok = erlzmq:send(Out, LastPart);
+queue_send(Out, [Part|Rest]) ->
+    ok = erlzmq:send(Out, Part, [sndmore]),
+    queue_send(Out, Rest).

src/erlzmq_parts.erl

@@ -0,0 +1,26 @@
+-module(erlzmq_parts).
+
+-export([new/0, part_in/2]).
+
+new() -> [].
+
+part_in({zmq, Socket, Part, [rcvmore]}, Parts) ->
+    {rcvmore, add_part(Socket, Part, Parts)};
+part_in({zmq, Socket, Part, []}, Parts) ->
+    {SocketParts, NewParts} =
+        find_socket_parts(Socket, add_part(Socket, Part, Parts)),
+    {msg, Socket, SocketParts, NewParts}.
+
+add_part(Socket, Part, Parts) ->
+    [{Socket, Part}|Parts].
+
+find_socket_parts(Socket, Parts) ->
+    find_socket_parts_acc(Socket, Parts, [], []).
+
+find_socket_parts_acc(_Socket, [], SocketParts, NewParts) ->
+    {SocketParts, lists:reverse(NewParts)};
+find_socket_parts_acc(Socket, [{Socket, Part}|Rest], SocketParts, NewParts) ->
+    find_socket_parts_acc(Socket, Rest, [Part|SocketParts], NewParts);
+find_socket_parts_acc(Socket, [OtherPart|Rest], SocketParts, NewParts) ->
+    find_socket_parts_acc(Socket, Rest, SocketParts, [OtherPart|NewParts]).
+

src/erlzmq_proxy.erl

@@ -0,0 +1,40 @@
+-module(erlzmq_proxy).
+-export([create/2]).
+
+
+%%---------------------------------------------------
+%% @doc A proxy implemented in Erlang.
+%% 
+%% Frontend and Backend must be sockets in default, 
+%% i.e. non-active, mode
+%%
+%% @spec create(Frontend, Backend) -> any()
+%% Frontend = erlzmq_socket()
+%% Backend = erlzmq_socket()
+%% @end
+%%---------------------------------------------------
+
+create(Frontend, Backend) ->
+    case erlzmq:recv(Backend, [dontwait]) of
+        {error, eagain} -> ok;
+        {ok, MsgB} ->
+            case erlzmq:getsockopt(Backend, rcvmore) of
+                {ok, true} -> erlzmq:send(Frontend, MsgB, [sndmore]);
+                {ok, false} -> erlzmq:send(Frontend, MsgB)
+            end
+    end,
+    case erlzmq:recv(Frontend, [dontwait]) of
+        {error, eagain} -> ok;
+        {ok, MsgF} ->
+            case erlzmq:getsockopt(Frontend, rcvmore) of
+                {ok, true} -> erlzmq:send(Backend, MsgF, [sndmore]);
+                {ok, false} -> erlzmq:send(Backend, MsgF)
+            end
+    end,
+    receive
+        shutdown -> 
+            ok
+    after 
+        0 ->
+            create(Frontend, Backend)
+    end.

src/erlzmq_util.erl

@@ -0,0 +1,99 @@
+-module(erlzmq_util).
+
+-export([dump/1, recv_parts/1, recv_parts/2]).
+
+%%--------------------------------------------------------------------
+%% @doc Reads available messages from Socket, printing them to stdout.
+%% @spec dump(Socket) -> any()
+%% Socket = erlzmq_socket()
+%% @end
+%%--------------------------------------------------------------------
+
+dump(Socket) ->
+    {ok, Msg} = erlzmq:recv(Socket),
+    io:format("----------------------------------------~n"),
+    dump_msg(Msg, Socket).
+
+%%--------------------------------------------------------------------
+%% @doc Receives message parts for Socket, returning them as a list.
+%% @equiv recv_parts(Socket, infinity)
+%% @end
+%%--------------------------------------------------------------------
+
+recv_parts(Socket) ->
+    recv_parts(Socket, infinity).
+
+%%--------------------------------------------------------------------
+%% @doc Receives message parts for Socket, returning them as a list.
+%% @spec recv_parts(Socket, Timeout) -> [binary()]
+%% Socket = erlzmq_socket()
+%% Timeout = integer() | infinity
+%% @end
+%%--------------------------------------------------------------------
+
+recv_parts(Socket, Timeout) ->
+    recv_parts(Socket, Timeout, []).
+
+%%--------------------------------------------------------------------
+%% @doc Accumulator for recv_parts/2
+%% @spec recv_parts(Socket, Timeout, Acc0) -> Acc
+%% @end
+%%--------------------------------------------------------------------
+recv_parts(Socket, Timeout, Acc) ->
+    receive
+        {zmq, Socket, Part, []} ->
+            lists:reverse([Part|Acc]);
+        {zmq, Socket, Part, [rcvmore]} ->
+            recv_parts(Socket, Timeout, [Part|Acc])
+    after
+        Timeout -> exit({zmq_recv_timeout, Socket})
+    end.
+
+%%--------------------------------------------------------------------
+%% @doc Print a socket message, including subsequent parts.
+%% @spec dump_msg(Msg, Socket) -> ok
+%% @end
+%%--------------------------------------------------------------------
+
+dump_msg(Msg, Socket) ->
+    io:format("[~3..0B] ", [size(Msg)]),
+    Str = binary_to_list(Msg),
+    case io_lib:printable_list(Str) of
+        true -> io:format(Str);
+        false -> io:format(bin_to_hex(Msg))
+    end,
+    io:format("~n"),
+    case erlzmq:getsockopt(Socket, rcvmore) of
+        {ok, true} ->
+            {ok, Next} = erlzmq:recv(Socket),
+            dump_msg(Next, Socket);
+        {ok, false} ->
+            ok
+    end.
+
+%%--------------------------------------------------------------------
+%% @doc Convert a binary to a hex string.
+%% @spec bin_to_hex(binary()) -> list()
+%% @end
+%%--------------------------------------------------------------------
+
+bin_to_hex(B) when is_binary(B) ->
+    lists:flatten(lists:map(fun int_to_hex/1, binary_to_list(B))).
+
+%%--------------------------------------------------------------------
+%% @doc Convert an int to a two char hex string.
+%% @spec int_to_hex(integer()) -> string()
+%% @end
+%%--------------------------------------------------------------------
+
+int_to_hex(N) when N < 256 ->
+    [hex(N div 16), hex(N rem 16)].
+
+%%--------------------------------------------------------------------
+%% @doc Converts an integer to a hex char.
+%% @spec hex(integer()) -> char()
+%% @end
+%%--------------------------------------------------------------------
+
+hex(N) when N < 10 -> $0 + N;
+hex(N) when N >= 10, N < 16 -> $a + (N - 10).